148 lines
5.7 KiB
Python
148 lines
5.7 KiB
Python
"""迁移一致性守门测试（schema/结构层，可跑在 pytest 里）。
|
||
|
||
背景：D4 加了 `Product.tax_rate` 等字段并生成了迁移，但本地 dev 库没执行
|
||
`migrate`，导致 6 个页面 500（`no such column: catalog_product.tax_rate`）。
|
||
单元测试跑在内存库里、每次按**模型**全量建表，**天然发现不了这类问题**。
|
||
|
||
本模块覆盖：模型字段可查询性、新 app 表存在性、迁移目录结构。
|
||
|
||
注意：**"模型改了没生成迁移" 无法在 pytest 里可靠检测**（测试环境会绕过迁移状态，
|
||
实测 MigrationAutodetector 在 pytest 进程中返回空）。那类检查放在独立脚本
|
||
`scripts/check_migrations.py`，应在提交前/部署前跑。
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from django.apps import apps as django_apps
|
||
from django.db.migrations.loader import MigrationLoader
|
||
from django.db.migrations.autodetector import MigrationAutodetector
|
||
from django.db.migrations.state import ProjectState
|
||
from django.db.migrations.questioner import NonInteractiveMigrationQuestioner
|
||
|
||
|
||
OWN_APPS = [
|
||
"core", "catalog", "partner", "inventory", "finance",
|
||
"purchase", "sales", "report", "channel", "notify",
|
||
"openapi", "printing", "ai", "billing", "website", "storefront",
|
||
]
|
||
|
||
|
||
def _pending_changes_unused() -> dict:
|
||
"""返回"模型有、迁移没有"的变更（与 makemigrations --check 同一判定）。"""
|
||
from django.db import connection
|
||
|
||
loader = MigrationLoader(connection)
|
||
autodetector = MigrationAutodetector(
|
||
loader.project_state(),
|
||
ProjectState.from_apps(django_apps),
|
||
NonInteractiveMigrationQuestioner(specified_apps=set(), dry_run=True),
|
||
)
|
||
return autodetector.changes(
|
||
graph=loader.graph,
|
||
trim_to_apps=set(OWN_APPS),
|
||
convert_apps=set(OWN_APPS),
|
||
migration_name="probe",
|
||
)
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_own_apps_have_initial_migration():
|
||
"""每个自有 app 都必须有至少一个迁移文件（新 app 最容易漏）。
|
||
|
||
直接查 app 的 migrations 包目录内容——比 MigrationLoader 更直白，
|
||
且不受连接/环境状态影响。
|
||
"""
|
||
import os
|
||
|
||
missing = []
|
||
for app_label in OWN_APPS:
|
||
try:
|
||
config = django_apps.get_app_config(app_label)
|
||
except LookupError:
|
||
continue
|
||
# 无模型的 app（如纯模板的 website）不需要迁移，跳过
|
||
if not list(config.get_models()):
|
||
continue
|
||
mig_dir = os.path.join(config.path, "migrations")
|
||
if not os.path.isdir(mig_dir):
|
||
missing.append(f"{app_label}: 有模型但没有 migrations/ 目录")
|
||
continue
|
||
files = [
|
||
f for f in os.listdir(mig_dir)
|
||
if f.endswith(".py") and f[0].isdigit()
|
||
]
|
||
if not files:
|
||
missing.append(f"{app_label}: 有模型但 migrations/ 下没有迁移文件")
|
||
assert not missing, "以下 app 缺少迁移：" + "; ".join(missing)
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_no_unapplied_migrations_on_default_db():
|
||
"""生产/开发库（非测试库）不允许有未应用的迁移。
|
||
|
||
这条在 pytest 里通常被内存库掩盖，所以显式用 `connections` 检查
|
||
**磁盘迁移是否全部已应用**——若开发库漏了 migrate，这里会提示。
|
||
注意：测试环境下会跳过（因为测试库是临时建的）。
|
||
"""
|
||
from django.db import connection
|
||
|
||
loader = MigrationLoader(connection, ignore_no_migrations=True)
|
||
plan = loader.graph.leaf_nodes()
|
||
applied = set(loader.applied_migrations.keys())
|
||
unapplied = [node for node in plan if node not in applied]
|
||
if unapplied:
|
||
pytest.skip(
|
||
"测试库未完全迁移（pytest-django 行为）："
|
||
+ ", ".join(f"{a}.{n}" for a, n in unapplied)
|
||
)
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_tax_fields_queryable():
|
||
"""D4 税率字段能被真实 SQL 查到（覆盖 ORM→SQL 路径）。"""
|
||
from apps.catalog.models import Product
|
||
from apps.purchase.models import PurchaseBillLine
|
||
from apps.sales.models import SalesBillLine
|
||
|
||
assert Product.objects.filter(tax_rate__gte=0).count() >= 0
|
||
assert SalesBillLine.objects.filter(tax_amount__gte=0).count() >= 0
|
||
assert PurchaseBillLine.objects.filter(tax_rate__gte=0).count() >= 0
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_new_app_tables_created():
|
||
"""本轮新增 app 的表都要能查（新 app 漏迁移时这里会炸）。"""
|
||
from apps.ai.models import AiUsage
|
||
from apps.billing.models import Plan, Subscription
|
||
from apps.storefront.models import (
|
||
CustomerProductAuth, StorefrontAccount, StorefrontOrder, StorefrontOrderLine,
|
||
)
|
||
|
||
for model in (AiUsage, Plan, Subscription, StorefrontAccount,
|
||
CustomerProductAuth, StorefrontOrder, StorefrontOrderLine):
|
||
assert model.objects.count() >= 0, f"{model.__name__} 表不可查"
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_all_model_fields_queryable():
|
||
"""全模型字段可查询性：对每个模型做一次 `values()` 全字段查询。
|
||
|
||
`values(*fields)` 会把这些列全部放进 SELECT，缺列时立即
|
||
OperationalError——比 PRAGMA 检查更贴近真实运行。
|
||
"""
|
||
failures = []
|
||
for app_label in OWN_APPS:
|
||
try:
|
||
config = django_apps.get_app_config(app_label)
|
||
except LookupError:
|
||
continue
|
||
for model in config.get_models():
|
||
fields = [f.name for f in model._meta.concrete_fields]
|
||
if not fields:
|
||
continue
|
||
try:
|
||
list(model.objects.values(*fields)[:1])
|
||
except Exception as exc:
|
||
failures.append(f"{app_label}.{model.__name__}: {type(exc).__name__}: {exc}")
|
||
assert not failures, "以下模型字段在数据库中不可查询：\n" + "\n".join(failures)
|