baseline: 批次A-D 成果 + membership 半成品(测试红)

This commit is contained in:
agent
2026-09-11 23:11:35 +08:00
commit b3f3095d53
311 changed files with 40540 additions and 0 deletions
+147
View File
@@ -0,0 +1,147 @@
"""迁移一致性守门测试(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)