import html import re import sys import unittest from pathlib import Path from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from app import product_status REPO_ROOT = Path(__file__).resolve().parent.parent def _sample_alert(filename): source = (REPO_ROOT / "docs" / "html" / filename).read_text(encoding="utf-8") title_match = re.search(r'class="eds-alert-title">\s*([^<]+)', source) desc_match = re.search(r'class="eds-alert-desc"[^>]*>(.*?)

', source, re.DOTALL) if title_match is None or desc_match is None: raise AssertionError("商品状态样本缺少标题或说明") description = re.sub(r"<[^>]+>", "", desc_match.group(1)) return { "title": html.unescape(title_match.group(1)), "description": html.unescape(description), } class ProductStatusTests(unittest.TestCase): def test_real_warning_samples_are_classified(self): reviewing = product_status.classify_alerts([_sample_alert("审核中商品提示.html")]) unlisted = product_status.classify_alerts([_sample_alert("未上架商品提示.html")]) self.assertEqual(product_status.STATUS_REVIEWING, reviewing["product_status"]) self.assertEqual(product_status.STATUS_UNLISTED, unlisted["product_status"]) def test_empty_unknown_and_multiple_alerts_follow_contract(self): self.assertEqual( product_status.STATUS_NORMAL, product_status.classify_alerts([])["product_status"], ) self.assertEqual( product_status.STATUS_UNKNOWN, product_status.classify_alerts([{"title": "其他警告", "description": "说明"}])["product_status"], ) result = product_status.classify_alerts( [ {"title": "其他警告", "description": "忽略"}, {"title": "您的商品未上架", "description": "已下架"}, {"title": "审核中", "description": "后续提示"}, ] ) self.assertEqual(product_status.STATUS_UNLISTED, result["product_status"]) self.assertIn("您的商品未上架", result["product_status_note"]) def test_promotion_edit_restriction_is_not_a_product_status_alert(self): traditional = { "title": "由於正在進行促銷，以下一些欄位無法進行編輯，數值將顯示為灰色。", "description": "", } simplified = { "title": "由于正在进行促销，以下一些字段无法进行编辑，数值将显示为灰色。", "description": "", } for alert in (traditional, simplified): result = product_status.classify_alerts([alert]) self.assertEqual(product_status.STATUS_NORMAL, result["product_status"]) self.assertIsNone(result["product_status_note"]) self.assertTrue( product_status.is_promotion_edit_restriction_note( f"标题：{alert['title']}" ) ) reviewing = product_status.classify_alerts( [traditional, {"title": "审核中", "description": "等待审核"}] ) unlisted = product_status.classify_alerts( [simplified, {"title": "您的商品未上架", "description": "已下架"}] ) self.assertEqual(product_status.STATUS_REVIEWING, reviewing["product_status"]) self.assertEqual(product_status.STATUS_UNLISTED, unlisted["product_status"]) def test_invalid_values_are_unknown_and_partitioning_is_consistent(self): tasks = [ SimpleNamespace(product_status="normal"), SimpleNamespace(product_status="reviewing"), SimpleNamespace(product_status=None), SimpleNamespace(product_status="invalid"), ] grouped = product_status.partition_tasks(tasks) self.assertEqual("状态未知", product_status.status_label(None)) self.assertTrue(product_status.is_normal("normal")) self.assertFalse(product_status.is_normal(None)) self.assertTrue(product_status.is_known_abnormal("unlisted")) self.assertEqual(1, len(grouped["normal"])) self.assertEqual(2, len(grouped["unknown"])) def test_note_is_normalized_and_limited(self): result = product_status.classify_alerts( [{"title": " 审核中\n", "description": " " + "x" * 2200}] ) self.assertEqual(product_status.STATUS_REVIEWING, result["product_status"]) self.assertLessEqual(len(result["product_status_note"]), 2000) self.assertTrue(result["product_status_note"].startswith("标题：审核中；说明：")) def test_generation_plan_filters_unknown_by_default_and_freezes_snapshot(self): normal = SimpleNamespace( id=1, updated_at="2026-07-18T10:00:00", product_status="normal", new_title=None, new_cover_path=None, ) unlisted = { "id": 2, "updated_at": "2026-07-18T10:00:01", "product_status": "unlisted", "new_title": "已有标题", "new_cover_path": None, } unknown = SimpleNamespace( id=3, updated_at="2026-07-18T10:00:02", product_status=None, new_title=None, new_cover_path=None, ) normal_only = product_status.build_generation_plan( [normal, unlisted, unknown, normal], "title_cover", ) all_statuses = product_status.build_generation_plan( [normal, unlisted, unknown], "title_cover", product_status.SCOPE_ALL, ) changed = product_status.build_generation_plan( [ normal, { **unlisted, "updated_at": "2026-07-18T10:01:00", }, unknown, ], "title_cover", ) self.assertEqual([1, 2, 3], [task.id if hasattr(task, "id") else task["id"] for task in normal_only["base_candidates"]]) self.assertEqual([1], [task.id for task in normal_only["execution_tasks"]]) self.assertEqual(2, normal_only["scope_excluded"]) self.assertEqual( {"normal": 1, "unlisted": 1, "reviewing": 0, "unknown": 1}, normal_only["status_counts"], ) self.assertEqual([1, 2, 3], [task.id if hasattr(task, "id") else task["id"] for task in all_statuses["execution_tasks"]]) self.assertNotEqual(normal_only["fingerprint"], changed["fingerprint"]) def test_apply_plan_excludes_abnormal_before_content_and_deduplicates_skips(self): tasks = [ SimpleNamespace( id=1, updated_at="2026-07-18T10:00:00", product_status="normal", new_title="新标题", new_cover_path="new.jpg", ), SimpleNamespace( id=2, updated_at="2026-07-18T10:00:01", product_status="unlisted", new_title=None, new_cover_path=None, ), SimpleNamespace( id=3, updated_at="2026-07-18T10:00:02", product_status="reviewing", new_title=None, new_cover_path="new.jpg", ), SimpleNamespace( id=4, updated_at="2026-07-18T10:00:03", product_status=None, new_title="新标题", new_cover_path=None, ), SimpleNamespace( id=5, updated_at="2026-07-18T10:00:04", product_status="normal", new_title=None, new_cover_path=None, ), ] plan = product_status.build_apply_plan(tasks + [tasks[0]], "title_cover") changed = product_status.build_apply_plan( tasks[:4] + [ SimpleNamespace( id=5, updated_at="2026-07-18T10:01:00", product_status="normal", new_title=None, new_cover_path=None, ) ], "title_cover", ) self.assertEqual([1], [task.id for task in plan["executable"]]) self.assertEqual([2], [task.id for task in plan["unlisted"]]) self.assertEqual([3], [task.id for task in plan["reviewing"]]) self.assertEqual([4], [task.id for task in plan["unknown"]]) self.assertEqual([5], [task.id for task in plan["missing_title"]]) self.assertEqual([5], [task.id for task in plan["missing_cover"]]) self.assertEqual(3, plan["status_scope_excluded"]) self.assertEqual(1, plan["content_scope_excluded"]) self.assertEqual(4, plan["scope_excluded"]) self.assertNotEqual(plan["fingerprint"], changed["fingerprint"])