linkeyu
96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
"""Regression coverage for comparison terminal-status normalization."""
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import sqlalchemy as sa
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
|
|
|
|
def _migration_module():
|
|
path = (
|
|
Path(__file__).parents[1]
|
|
/ "alembic"
|
|
/ "versions"
|
|
/ "comparison_below_minimum_as_success.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location(path.stem, path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_business_status_migration_upgrade_and_downgrade() -> None:
|
|
engine = sa.create_engine("sqlite:///:memory:")
|
|
metadata = sa.MetaData()
|
|
records = sa.Table(
|
|
"comparison_record",
|
|
metadata,
|
|
sa.Column("id", sa.Integer, primary_key=True),
|
|
sa.Column("status", sa.String(16), nullable=False),
|
|
sa.Column("fail_reason", sa.String(256)),
|
|
sa.Column("raw_payload", sa.JSON, nullable=False),
|
|
)
|
|
metadata.create_all(engine)
|
|
module = _migration_module()
|
|
|
|
with engine.begin() as connection:
|
|
connection.execute(
|
|
records.insert(),
|
|
[
|
|
{
|
|
"status": "below_minimum",
|
|
"fail_reason": "旧失败原因",
|
|
"raw_payload": {"record_status": "below_minimum"},
|
|
},
|
|
{
|
|
"status": "success",
|
|
"fail_reason": None,
|
|
"raw_payload": {"record_status": "success"},
|
|
},
|
|
{
|
|
"status": "failed",
|
|
"fail_reason": "技术异常",
|
|
"raw_payload": {"status": "failed"},
|
|
},
|
|
{
|
|
"status": "store_closed",
|
|
"fail_reason": "店铺打烊",
|
|
"raw_payload": {"record_status": "store_closed"},
|
|
},
|
|
{
|
|
"status": "items_not_found",
|
|
"fail_reason": "商品未找到",
|
|
"raw_payload": {"status": "items_not_found"},
|
|
},
|
|
],
|
|
)
|
|
module.op = Operations(MigrationContext.configure(connection))
|
|
|
|
module.upgrade()
|
|
upgraded = connection.execute(
|
|
sa.select(records.c.status, records.c.fail_reason).order_by(records.c.id)
|
|
).all()
|
|
assert upgraded == [
|
|
("success", None),
|
|
("success", None),
|
|
("failed", "技术异常"),
|
|
("failed", "店铺打烊"),
|
|
("failed", "商品未找到"),
|
|
]
|
|
|
|
module.downgrade()
|
|
downgraded = connection.execute(
|
|
sa.select(records.c.status, records.c.fail_reason).order_by(records.c.id)
|
|
).all()
|
|
assert downgraded == [
|
|
("below_minimum", None),
|
|
("success", None),
|
|
("failed", "技术异常"),
|
|
("store_closed", "店铺打烊"),
|
|
("items_not_found", "商品未找到"),
|
|
]
|