Clawby-Q/test_activate_v3_product.py at main · ZarksLi/Clawby-Q · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import json
import tempfile
import unittest
from pathlib import Path

from activate_v3_product import run_once


class ActivateV3ProductTests(unittest.TestCase):
    def _fixture(self, root: Path, now: str) -> str:
        run_id = "new-product-run"
        for path in (
            root / "config",
            root / "results",
            root / "data/v3_paper_next",
            root / "data/v3_robot_subaccounts",
            root / "data/v3_shadow",
        ):
            path.mkdir(parents=True, exist_ok=True)
        payloads = {
            root / "config/v3_product_candidate.json": {
                "status": "PREPARED_NOT_ACTIVE", "run_id": run_id,
                "parent_run_id": "old-run", "paper_trading_only": True,
            },
            root / "config/v3_product_runtime_manifest.json": {
                "run_id": run_id, "parent_run_id": "old-run", "paper_trading_only": True,
            },
            root / "config/v3_validation_registry.json": {
                "run_id": run_id, "parent_run_id": "old-run", "records": {},
                "paper_trading_only": True, "automatic_promotion_allowed": False,
            },
            root / "config/v3_decision_policy.json": {
                "run_id": run_id, "paper_trading_only": True, "real_order_allowed": False,
                "formal_forward_evidence": False,
            },
            root / "config/product_policy_v1.json": {"paper_trading_only": True},
            root / "config/v3_institutional_group.json": {"paper_trading_only": True},
            root / "results/v3_product_acceptance.json": {
                "status": "PASS", "run_id": run_id, "order_sent": False,
                "paper_trading_only": True,
            },
            root / "results/v3_shadow_runtime_status.json": {
                "status": "DEGRADED", "run_id": run_id, "generated_at": now,
                "paper_trading_only": True, "automatic_order_allowed": False,
                "new_formal_positions_allowed": False,
                "formal_forward_ledgers_modified": False,
                "shadow_pipeline_errors": [],
            },
            root / "data/v3_paper_next/paper_account.json": {
                "run_id": "old-run", "paper_trading_only": True,
                "real_order_allowed": False, "open_positions": [{"ticker": "IBM"}],
                "closed_positions": [],
            },
            root / "data/v3_robot_subaccounts/accounts.json": {
                "run_id": "old-run", "paper_trading_only": True,
                "real_order_allowed": False, "accounts": {"G1-US01": {}},
            },
        }
        for path, payload in payloads.items():
            path.write_text(json.dumps(payload), encoding="utf-8")
        (root / "data/v3_shadow/robot_event_ledger.jsonl").write_text(
            json.dumps({"event_id": "legacy", "run_id": "old-run"}) + "\n",
            encoding="utf-8",
        )
        (root / "results/v3_independent_forward_events.jsonl").write_text(
            json.dumps({"event_id": "formal-old", "run_id": "old-run"}) + "\n",
            encoding="utf-8",
        )
        return run_id

    def test_activation_archives_legacy_and_starts_clean_without_rewriting_ledgers(self):
        now = "2026-07-22T01:02:03+00:00"
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            run_id = self._fixture(root, now)
            result = run_once(
                run_id, base_dir=root, now=now,
                regenerate_acceptance=False, max_status_age_seconds=60,
            )
            candidate = json.loads((root / "config/v3_product_candidate.json").read_text())
            account = json.loads((root / "data/v3_paper_next/paper_account.json").read_text())
            ledger = (root / "results/v3_independent_forward_events.jsonl").read_text()
            manifest = json.loads(Path(result["manifest"]).read_text())
        self.assertEqual(candidate["status"], "FROZEN_ACTIVE")
        self.assertEqual(account["run_id"], run_id)
        self.assertEqual(account["open_positions"], [])
        self.assertIn("formal-old", ledger)
        self.assertTrue(manifest["shadow_and_independent_ledgers_preserved"])
        self.assertFalse(manifest["order_sent"])

    def test_activation_requires_exact_run_id(self):
        now = "2026-07-22T01:02:03+00:00"
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            self._fixture(root, now)
            with self.assertRaisesRegex(RuntimeError, "exact_run_id"):
                run_once(
                    "wrong-run", base_dir=root, now=now,
                    regenerate_acceptance=False, max_status_age_seconds=60,
                )


if __name__ == "__main__":
    unittest.main()