"""Declaration-level contract for the Credits approval seed. Every check here is one that was made by eye while adding a notification leg to the Credits chains, and got made wrong at least once. The seed's own ``--validate-only`` does not cover any of them: it validates the seed's internal shape, not its agreement with the things that read it. The count check exists because the counts are declared in THREE independent places and one of them fails closed before applying: ``apply_credits_approval_seed_adapter.PLANNED_COUNTS`` rejects a seed whose counts do not match, so raising the seed and the catalog while leaving the adapter behind aborts every exp and staging deploy. That is exactly what happened on a change that added two rows, and only review caught it. """ from __future__ import annotations import json import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] SEED_PATH = REPO_ROOT / "deployments/offline/runtime/seeds/credits-approvals.seed.json" CATALOG_PATH = REPO_ROOT / "deployments/offline/runtime/seeds/deploy-seed-catalog.json" ADAPTER_PATH = REPO_ROOT / "scripts/apply_credits_approval_seed_adapter.py" PATCH_BUNDLES = REPO_ROOT / "docs/migration/patch-bundles" L2_PACK = REPO_ROOT / "packs/Salesforce-PingCAP-L2" FINAL_PACK = REPO_ROOT / "packs/NOT-EDIT-final-merged-packs" STACK_LOCK = REPO_ROOT / "stack.lock.json" def _json(path: Path): return json.loads(path.read_text(encoding="utf-8")) def _adapter_planned_counts() -> dict: """Read PLANNED_COUNTS without importing the adapter (it imports deploy modules).""" namespace: dict = {} source = ADAPTER_PATH.read_text(encoding="utf-8") start = source.index("PLANNED_COUNTS = {") end = source.index("}", start) + 1 exec(source[start:end], namespace) # noqa: S102 - a literal dict from our own repo return namespace["PLANNED_COUNTS"] class CreditsApprovalSeedContractTests(unittest.TestCase): def setUp(self) -> None: self.seed = _json(SEED_PATH) def _actual_counts(self) -> dict: return { "email_templates": len(self.seed["emailTemplates"]), "process_actions": len(self.seed["processActions"]), "process_definitions": len(self.seed["processDefinitions"]), "process_nodes": len(self.seed["processNodes"]), } def test_seed_catalog_and_adapter_counts_are_exactly_aligned(self) -> None: catalog = _json(CATALOG_PATH)["modules"]["credits_approval_seed"]["declared_stats"] self.assertEqual(self._actual_counts(), catalog, "seed content vs catalog declared_stats") self.assertEqual(catalog, _adapter_planned_counts(), "catalog declared_stats vs adapter PLANNED_COUNTS") def test_every_bound_flow_exists_and_is_whitelisted(self) -> None: """A Flow-type binding whose Flow is not built into the pack is a notification that fails at approval time, silently, in a 200 response.""" whitelist = set(_json(STACK_LOCK)["pack_build"]["flow_whitelist"]) l2_components = { component["businessKey"] for component in _json(L2_PACK / "pack.json")["components"] if component.get("kind") == "standard_flow" } final_components = { component["businessKey"] for component in _json(FINAL_PACK / "pack.json")["components"] if component.get("kind") == "standard_flow" } for action in self.seed["processActions"]: if action["actionType"] != "Flow": continue name = action["apiName"] with self.subTest(flow=name): self.assertTrue((L2_PACK / "flows" / f"{name}.flow.json").is_file(), "missing L2 source") self.assertTrue((FINAL_PACK / "flows" / f"{name}.flow.json").is_file(), "missing from final pack") self.assertIn(name, whitelist, "absent from stack.lock flow_whitelist (built pack drops it silently)") self.assertIn(name, l2_components, "unregistered in the L2 pack manifest") self.assertIn(name, final_components, "unregistered in the final pack manifest") def test_every_referenced_email_template_is_seeded(self) -> None: """sendEmail resolves templateId at run time; an unseeded id fails the send.""" seeded = {template["id"] for template in self.seed["emailTemplates"]} for flow_path in sorted((L2_PACK / "flows").glob("Credits*.flow.json")): for call in _json(flow_path).get("actionCalls", []): if call.get("actionName") != "sendEmail": continue template_id = next( parameter["value"]["stringValue"] for parameter in call["inputParameters"] if parameter["name"] == "templateId" ) with self.subTest(flow=flow_path.name): self.assertIn(template_id, seeded) def test_action_order_is_unique_per_process_and_event(self) -> None: """Actions are fetched ORDER BY `Order` ASC, so a collision reshuffles delivery.""" seen: dict[tuple[str, str], set] = {} for action in self.seed["processActions"]: key = (action["processDeveloperName"], action["actionEvent"]) orders = seen.setdefault(key, set()) with self.subTest(process=key[0], event=key[1], order=action["order"]): self.assertNotIn(action["order"], orders, "duplicate Order on the same event") orders.add(action["order"]) def test_ids_and_developer_names_are_unique(self) -> None: for collection, field in (("processActions", "id"), ("emailTemplates", "id"), ("emailTemplates", "developerName")): values = [row[field] for row in self.seed[collection]] with self.subTest(collection=collection, field=field): self.assertEqual(len(values), len(set(values))) def test_no_template_hardcodes_a_credit_program_name(self) -> None: """Both chains bind the same templates, so a program name baked into one is wrong on the other. 'PoC Credits' was wrong on every top-up notification from 2026-07-22 until it was corrected.""" for template in self.seed["emailTemplates"]: body = template["subject"] + template["body"] + template["htmlValue"] with self.subTest(template=template["developerName"]): self.assertNotIn("PoC Credits", body) def test_patch_bundles_carry_the_seed_text_verbatim(self) -> None: """More than one patch bundle upserts the same _System_EmailTemplate rows by primary key. Replay is operator-selected -- one patch_dir per run, no index -- so a bundle left holding stale copy silently reverts a newer one the moment somebody picks it. That is how the 2026-07-22 bundle would have undone the program-name correction. The seed is the single source; every bundle that writes these rows must agree with it verbatim. """ seeded = {template["id"]: template for template in self.seed["emailTemplates"]} checked = 0 for sql_path in sorted(PATCH_BUNDLES.glob("*/*.sql")): sql = sql_path.read_text(encoding="utf-8") for template_id, template in seeded.items(): # Only rows this file actually writes; a mention in a comment or # a verification SELECT is not an upsert of the content. if f"VALUES ('{template_id}'" not in sql: continue checked += 1 for field in ("name", "developerName", "subject", "body", "htmlValue"): escaped = template[field].replace("\\", "\\\\").replace("'", "''") with self.subTest(bundle=sql_path.parent.name, template=template_id, field=field): self.assertIn( f"'{escaped}'", sql, "patch bundle text has drifted from credits-approvals.seed.json", ) self.assertGreater(checked, 0, "found no patch-bundle template upserts; the glob or the marker is wrong") if __name__ == "__main__": unittest.main()