from __future__ import annotations import hashlib import json import os import shutil import subprocess import sys import tempfile import threading import time import unittest import urllib.error from concurrent.futures import ThreadPoolExecutor from unittest.mock import patch from datetime import datetime, timedelta from pathlib import Path from zoneinfo import ZoneInfo REPO = Path(__file__).resolve().parents[4] PROJECT_DEV = REPO / "dev" / "project-dev" ANA_DEV = REPO / "dev" / "ana-dev" if str(PROJECT_DEV) not in sys.path: sys.path.insert(0, str(PROJECT_DEV)) from stock_valuation_pipeline_v2.cache import ( # noqa: E402 ContentCache, canonical_bytes, request_fingerprint, sha256_bytes, ) from stock_valuation_pipeline_v2.acquisition import acquire_all # noqa: E402 from stock_valuation_pipeline_v2.full_report import HEADINGS # noqa: E402 from stock_valuation_pipeline_v2.http_client import ( # noqa: E402 HttpClient, HttpRequest, NetworkBudgetExceeded, SourceContractError, ) from stock_valuation_pipeline_v2.judgment import ( # noqa: E402 JudgmentConflict, apply_overlay, load_overlay, ) from stock_valuation_pipeline_v2.providers import ( # noqa: E402 DEBT_KEYS, FINANCIAL_SINGLE_KEYS, LIQUID_FV_ALIAS_KEYS, acquire_announcements, acquire_forecast, acquire_market, parse_balance_record, ) from stock_valuation_pipeline_v2.report_qa import run_qa # noqa: E402 from stock_valuation_pipeline_v2.snapshot_builder import ( # noqa: E402 CORE_VALUE_PATHS, CoreInputError, build_data_snapshot, ) from stock_valuation_pipeline_v2.v1_bridge import V1Bridge # noqa: E402 from stock_valuation_pipeline_v2.workflow import run_ticker_pipeline # noqa: E402 FIXTURE = Path(__file__).with_name("fixtures") / "chengchang_20260801" SOURCE_SNAPSHOT = REPO / "outputs" / "20260801_chengchang_technology_valuation" / "valuation_snapshot.json" SOURCE_RESULTS = REPO / "outputs" / "20260801_chengchang_technology_valuation" / "generated" / "valuation_results.json" GREAT_WALL = ANA_DEV / "test" / "stock_valuation_pipeline" / "fixtures" / "great_wall_military_20260731.json" TASK_START = "2026-08-01T21:20:48+08:00" EXPECTED_KEYS = { "schema_version", "status", "run_id", "exit_code", "output_dir", "failed_dir", "manifest_path", "manifest_sha256", "receipt_path", "receipt_sha256", "receipt_bytes", "task_start", "task_start_source", "process_start", "terminal_at", "task_wall_seconds", "process_wall_seconds", "gap_count", "error_code", "error", } def tree_hash(root: Path) -> str: digest = hashlib.sha256() for path in sorted(root.rglob("*")): if path.is_file(): digest.update(path.relative_to(root).as_posix().encode()) digest.update(path.read_bytes()) return digest.hexdigest() class PipelineV2Tests(unittest.TestCase): def run_fixture(self, root: Path, name: str, judgment: bool = False, force: bool = False, fault=None): return run_ticker_pipeline( ticker="001270.SZ", as_of="2026-08-01", output_dir=root / name, cache_dir=root / "cache", judgment_path=FIXTURE / "judgment_overlay.json" if judgment else None, fixture_dir=FIXTURE, task_start=TASK_START, force=force, network_budget_seconds=5.0, fault_injector=fault, ) def clone_fixture(self, root: Path) -> Path: target = root / "fixture" shutil.copytree(FIXTURE, target) return target def run_fixture_dir( self, root: Path, name: str, fixture: Path, judgment: bool = False, force: bool = False, fault=None, ): return run_ticker_pipeline( ticker="001270.SZ", as_of="2026-08-01", output_dir=root / name, cache_dir=root / "cache", judgment_path=fixture / "judgment_overlay.json" if judgment else None, fixture_dir=fixture, task_start=TASK_START, force=force, network_budget_seconds=5.0, fault_injector=fault, ) def install_baseline_state(self, root: Path, kind: str, state: str) -> bytes: """Replace the selected temp baseline with one deterministic kind state.""" cache = ContentCache(root / "cache") baseline = cache.select_baseline("001270.SZ", "2026-08-01") self.assertIsNotNone(baseline) assert baseline is not None replacement = json.loads(json.dumps(baseline)) replacement.pop("data_hash", None) if state == "STALE": replacement["data_kinds"][kind]["expires_at"] = "2000-01-01T00:00:00+08:00" elif state == "MISSING": replacement["data_kinds"].pop(kind) else: raise AssertionError(f"unsupported state {state}") baselines = root / "cache" / "companies" / "001270.SZ" / "baselines" for path in baselines.glob("*.json"): path.unlink() current = root / "cache" / "companies" / "001270.SZ" / "current.json" current.unlink(missing_ok=True) cache.write_baseline("001270.SZ", replacement) reusable = root / "cache" / "reusable" if reusable.exists(): shutil.rmtree(reusable) return current.read_bytes() def test_01_v1_contract_and_frozen_fixture(self): bridge = V1Bridge.load() self.assertEqual(bridge.module.__version__, "1.0.0") self.assertEqual(GREAT_WALL.stat().st_size, 7142) self.assertEqual( hashlib.sha256(GREAT_WALL.read_bytes()).hexdigest().upper(), "06AB6A7C45C2E43F71654C616F15A826B21E468C4BA72AFB0B253C9DD89FACD8", ) def test_02_fixture_data_ready_artifacts(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) code, summary = self.run_fixture(root, "out") self.assertEqual(code, 0) self.assertEqual(summary["status"], "DATA_READY_NEEDS_JUDGMENT") self.assertEqual(set(summary), EXPECTED_KEYS) out = root / "out" required = { "data_snapshot.json", "snapshot_build_report.json", "report.md", "qa_report.json", "gaps.json", "source_evidence.json", "provider_results.json", "runtime_metrics.json", "runtime.log", "manifest.json", } self.assertEqual({p.name for p in out.iterdir()}, required) self.assertFalse((out / "valuation_snapshot.json").exists()) self.assertFalse((out / "valuation_results.json").exists()) report = (out / "report.md").read_text(encoding="utf-8") self.assertEqual(sum(1 for line in report.splitlines() if line.startswith("## ")), 16) self.assertIn("GAP：需要人工判断覆盖层", report) gaps = json.loads((out / "gaps.json").read_text(encoding="utf-8")) self.assertEqual( [item["gap_id"] for item in gaps], ["W_FORECAST_COVERAGE", "W_JUDGMENT_REQUIRED"], ) def test_03_judgment_matches_independent_v1_baseline(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) code, summary = self.run_fixture(root, "out", judgment=True) self.assertEqual(code, 0) self.assertEqual(summary["status"], "COMPLETE_WITH_GAPS") actual = json.loads((root / "out" / "valuation_results.json").read_text(encoding="utf-8")) expected = json.loads(SOURCE_RESULTS.read_text(encoding="utf-8")) for key in ( "market_cap", "ttm_revenue", "ttm_attributable_profit", "ttm_deduct_profit", "normalized_profit", "pb", "ps", ): self.assertEqual(actual["metrics"][key], expected["metrics"][key]) actual_base = next(item for item in actual["scenarios"] if item["role"] == "base") expected_base = next(item for item in expected["scenarios"] if item["role"] == "base") self.assertEqual(actual_base, expected_base) self.assertEqual(len(json.loads((root / "out" / "gaps.json").read_text(encoding="utf-8"))), 1) def test_04_fixture_cold_and_warm_performance(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) start = time.perf_counter() self.run_fixture(root, "cold") cold = time.perf_counter() - start start = time.perf_counter() _, warm_summary = self.run_fixture(root, "warm") warm = time.perf_counter() - start self.assertLess(cold, 5.0) self.assertLess(warm, 2.0) providers = json.loads((root / "warm" / "provider_results.json").read_text(encoding="utf-8")) self.assertTrue(all(result.get("baseline_reused") for result in providers.values())) telemetry = [item for result in providers.values() for item in result["request_telemetry"]] self.assertEqual(telemetry, []) self.assertGreaterEqual(warm_summary["task_wall_seconds"], warm_summary["process_wall_seconds"]) def test_05_manifest_receipt_hashes_and_stdout(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) _, summary = self.run_fixture(root, "out") manifest = Path(summary["manifest_path"]) receipt = Path(summary["receipt_path"]) self.assertEqual(hashlib.sha256(manifest.read_bytes()).hexdigest(), summary["manifest_sha256"]) self.assertEqual(hashlib.sha256(receipt.read_bytes()).hexdigest(), summary["receipt_sha256"]) self.assertEqual(receipt.stat().st_size, summary["receipt_bytes"]) payload = json.loads(receipt.read_text(encoding="utf-8")) self.assertEqual(payload["schema_version"], 2) self.assertEqual(payload["wall_scope"], "through_receipt_write_started") self.assertEqual(payload["run_id"], summary["run_id"]) self.assertNotIn("task_wall_seconds", payload) self.assertNotIn("process_wall_seconds", payload) self.assertNotIn("terminal_at", payload) self.assertGreaterEqual( datetime.fromisoformat(summary["terminal_at"]), datetime.fromisoformat(payload["receipt_write_started_at"]), ) runtime = json.loads((manifest.parent / "runtime_metrics.json").read_text(encoding="utf-8")) self.assertGreaterEqual(summary["task_wall_seconds"], runtime["task_wall_seconds"]) self.assertGreaterEqual(summary["process_wall_seconds"], runtime["process_wall_seconds"]) manifest_payload = json.loads(manifest.read_text(encoding="utf-8")) for rel, expected in manifest_payload["artifacts"].items(): path = manifest.parent / rel self.assertEqual(path.stat().st_size, expected["bytes"]) self.assertEqual(hashlib.sha256(path.read_bytes()).hexdigest(), expected["sha256"]) def test_06_forecast_four_three_is_one_gap(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") gaps = json.loads((root / "out" / "gaps.json").read_text(encoding="utf-8")) self.assertEqual( [gap["gap_id"] for gap in gaps], ["W_FORECAST_COVERAGE", "W_JUDGMENT_REQUIRED"], ) providers = json.loads((root / "out" / "provider_results.json").read_text(encoding="utf-8")) self.assertEqual(len(providers["forecast"]["institutions"]["forecasts"]), 3) def test_07_cache_fingerprint_is_canonical(self): left = request_fingerprint({"b": 2, "a": 1}) right = request_fingerprint({"a": 1, "b": 2}) self.assertEqual(left, right) self.assertNotEqual(left, request_fingerprint({"a": 1, "b": 3})) def test_08_cache_tamper_is_not_reused(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) cache = ContentCache(root) now = datetime.now().astimezone() meta = cache.store( "p", "f", b"good", now, 3600, True, {"run_id":"r","transport_complete":True,"parse_ok":True,"schema_ok":True,"as_of_ok":True,"semantic_status":"OK","http_status":200,"content_type":"application/json"}, ) (root / meta["blob_path"]).write_bytes(b"tampered") self.assertIsNone(cache.load_reusable("p", "f", now)) def test_09_negative_response_not_reusable(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) cache = ContentCache(root) now = datetime.now().astimezone() cache.store( "p", "negative", b"404", now, 60, False, {"run_id":"r","transport_complete":True,"parse_ok":False,"schema_ok":False,"as_of_ok":True,"semantic_status":"GAP","http_status":404,"content_type":"application/json"}, ) self.assertIsNone(cache.load_reusable("p", "negative", now)) def test_10_unknown_domain_rejected_before_transport(self): with tempfile.TemporaryDirectory() as raw: registry = json.loads((PROJECT_DEV / "stock_valuation_pipeline_v2" / "provider_registry.json").read_text(encoding="utf-8")) client = HttpClient(registry, ContentCache(Path(raw)), time.monotonic() + 1, "r", datetime.now().astimezone(), FIXTURE) request = HttpRequest("eastmoney.market", "1.0.0", "market_close", "GET", "https://example.com/x", "001270.SZ", "2026-08-01", "x") with self.assertRaises(SourceContractError): client.fetch(request) def test_11_judgment_protected_field_rejected(self): with tempfile.TemporaryDirectory() as raw: path = Path(raw) / "bad.json" path.write_text(json.dumps({"normalization":{},"valuation":{"scenarios":[{"role":"base","price":1}]},"analysis":{}}), encoding="utf-8") with self.assertRaises(JudgmentConflict): load_overlay(path) def test_12_future_task_start_is_input_error_without_artifacts(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) code, summary = run_ticker_pipeline( ticker="001270.SZ", as_of="2026-08-01", output_dir=root/"out", cache_dir=root/"cache", fixture_dir=FIXTURE, task_start=(datetime.now().astimezone()+timedelta(days=1)).isoformat(), ) self.assertEqual(code, 2) self.assertEqual(summary["status"], "INPUT_ERROR") self.assertEqual(set(summary), EXPECTED_KEYS) self.assertFalse((root / "out").exists()) def test_13_existing_output_rejected_before_cache_or_network(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) (root / "out").mkdir() (root / "out" / "sentinel").write_text("old", encoding="utf-8") code, _ = self.run_fixture(root, "out") self.assertEqual(code, 2) self.assertFalse((root / "cache").exists()) self.assertEqual((root / "out" / "sentinel").read_text(encoding="utf-8"), "old") def test_14_force_replaces_valid_output(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") (root / "out" / "user-extra.txt").write_text("old", encoding="utf-8") code, summary = self.run_fixture(root, "out", judgment=True, force=True) self.assertEqual(code, 0) self.assertEqual(summary["status"], "COMPLETE_WITH_GAPS") self.assertFalse((root / "out" / "user-extra.txt").exists()) self.assertTrue((root / "out" / "valuation_results.json").exists()) def test_15_keyboard_interrupt_restores_old_output_and_receipt(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") old_hash = tree_hash(root / "out") old_receipt = (root / "out.commit-receipt.json").read_bytes() def fault(step): if step == "rename_stage_to_output": raise KeyboardInterrupt("injected") with self.assertRaises(KeyboardInterrupt): self.run_fixture(root, "out", judgment=True, force=True, fault=fault) self.assertEqual(tree_hash(root / "out"), old_hash) self.assertEqual((root / "out.commit-receipt.json").read_bytes(), old_receipt) self.assertTrue(any(root.glob("out.failed-*"))) def test_16_build_rejects_future_source(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") providers = json.loads((root / "out" / "provider_results.json").read_text(encoding="utf-8")) providers["announcements"]["sources"][0]["publish_date"] = "2026-08-02" with self.assertRaisesRegex(CoreInputError, "E_ASOF_VIOLATION"): build_data_snapshot("001270.SZ", "2026-08-01", providers) def test_17_v1_input_bridge_matches_direct_cli(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) direct = root / "direct" bridge = root / "bridge" env_direct = os.environ.copy() env_direct["PYTHONPATH"] = str(ANA_DEV) env_bridge = os.environ.copy() env_bridge["PYTHONPATH"] = str(PROJECT_DEV) a = subprocess.run( [sys.executable, "-m", "stock_valuation_pipeline", "--input", str(GREAT_WALL), "--output-dir", str(direct)], cwd=REPO, env=env_direct, text=True, capture_output=True, check=False, ) b = subprocess.run( [sys.executable, "-m", "stock_valuation_pipeline_v2", "--input", str(GREAT_WALL), "--output-dir", str(bridge)], cwd=REPO, env=env_bridge, text=True, capture_output=True, check=False, ) self.assertEqual((a.returncode, b.returncode), (0, 0), (a.stderr, b.stderr)) self.assertEqual(json.loads(a.stdout)["status"], json.loads(b.stdout)["status"]) for name in ("valuation_results.json", "valuation_report.md", "run_manifest.json"): self.assertEqual((direct / name).read_bytes(), (bridge / name).read_bytes(), name) a2 = subprocess.run( [sys.executable, "-m", "stock_valuation_pipeline", "--input", str(GREAT_WALL), "--output-dir", str(direct)], cwd=REPO, env=env_direct, text=True, capture_output=True, check=False, ) b2 = subprocess.run( [sys.executable, "-m", "stock_valuation_pipeline_v2", "--input", str(GREAT_WALL), "--output-dir", str(bridge)], cwd=REPO, env=env_bridge, text=True, capture_output=True, check=False, ) self.assertEqual(json.loads(a2.stdout)["status"], "REUSED") self.assertEqual(json.loads(b2.stdout)["status"], "REUSED") def test_18_same_asof_baseline_selection_is_deterministic_max(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) cache = ContentCache(root) base = { "schema_version":1,"adapter_bundle_version":"2.0.0","ticker":"001270.SZ", "baseline_as_of":"2026-08-01","created_at":"2026-08-01T00:00:00+08:00", "watermark":{"market_date":"2026-07-31"},"provider_results":{},"field_lineage":{},"source_hashes":[], } first = dict(base, marker="a") second = dict(base, marker="z") cache.write_baseline("001270.SZ", first) cache.write_baseline("001270.SZ", second) current = json.loads((root/"companies"/"001270.SZ"/"current.json").read_text(encoding="utf-8")) hashes = [] for item in (first, second): material = {k:v for k,v in item.items() if k not in {"created_at","data_hash"}} hashes.append(sha256_bytes(canonical_bytes(material))) self.assertEqual(current["data_hash"], max(hashes)) def test_19_balance_complete_schema_and_exact_totals(self): row = json.loads((FIXTURE / "eastmoney_finance_balance_success.json").read_text(encoding="utf-8"))["result"]["data"][0] parsed = parse_balance_record(row) self.assertEqual(parsed["non_operating_financial_assets"], 95338761.64) self.assertEqual(parsed["interest_bearing_debt"], 825466.20) self.assertEqual(parsed["minority_interest"], 0) self.assertNotIn("OTHER_EQUITY_INVEST", LIQUID_FV_ALIAS_KEYS + FINANCIAL_SINGLE_KEYS) def test_20_balance_each_required_key_deletion_blocks(self): row = json.loads((FIXTURE / "eastmoney_finance_balance_success.json").read_text(encoding="utf-8"))["result"]["data"][0] for key in LIQUID_FV_ALIAS_KEYS + FINANCIAL_SINGLE_KEYS + DEBT_KEYS + ("MINORITY_EQUITY",): with self.subTest(key=key): damaged = dict(row) damaged.pop(key) with self.assertRaisesRegex(ValueError, "E_BALANCE_SCHEMA_DRIFT"): parse_balance_record(damaged) def test_21_balance_alias_equal_single_count_and_conflict(self): row = json.loads((FIXTURE / "eastmoney_finance_balance_success.json").read_text(encoding="utf-8"))["result"]["data"][0] equal = dict(row, TRADE_FINASSET=95338761.64) self.assertEqual(parse_balance_record(equal)["non_operating_financial_assets"], 95338761.64) conflict = dict(row, TRADE_FINASSET=1) with self.assertRaisesRegex(ValueError, "E_BALANCE_ALIAS_CONFLICT"): parse_balance_record(conflict) def test_22_balance_complete_all_null_is_zero(self): row = json.loads((FIXTURE / "eastmoney_finance_balance_success.json").read_text(encoding="utf-8"))["result"]["data"][0] for key in LIQUID_FV_ALIAS_KEYS + FINANCIAL_SINGLE_KEYS + DEBT_KEYS + ("MINORITY_EQUITY",): row[key] = None parsed = parse_balance_record(row) self.assertEqual(parsed, {"non_operating_financial_assets":0.0,"interest_bearing_debt":0.0,"minority_interest":0.0}) def test_23_http_success_is_not_reusable_before_adapter_confirmation(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) cache = ContentCache(root) now = datetime.now().astimezone() meta = cache.store( "p", "pending", b"{}", now, 3600, False, {"run_id":"r","transport_complete":True,"parse_ok":False,"schema_ok":False,"as_of_ok":False,"semantic_status":"PENDING_PARSE","http_status":200,"content_type":"application/json"}, ) self.assertIsNone(cache.load_reusable("p", "pending", now)) cache.promote_reusable("p", "pending", meta) self.assertIsNotNone(cache.load_reusable("p", "pending", now)) def test_24_four_worker_deadline_is_cooperative_and_bounded(self): class Client: def __init__(self): self.deadline = time.monotonic() + 0.08 def remaining(self): return max(0.0, self.deadline - time.monotonic()) client = Client() def slow(_client, _ticker, as_of): while _client.remaining() > 0: time.sleep(min(0.005, _client.remaining())) return {"provider_id":"stub","adapter_version":"1","status":"OK","as_of_date":as_of,"records":[],"sources":[],"gaps":[],"warnings":[],"raw_artifact_hashes":[],"request_telemetry":[]} started = time.perf_counter() with patch("stock_valuation_pipeline_v2.acquisition.acquire_announcements", slow), patch( "stock_valuation_pipeline_v2.acquisition.acquire_market", slow ), patch("stock_valuation_pipeline_v2.acquisition.acquire_finance", slow), patch( "stock_valuation_pipeline_v2.acquisition.acquire_forecast", slow ): results = acquire_all(client, "001270.SZ", "2026-08-01") elapsed = time.perf_counter() - started self.assertEqual(set(results), {"announcements","market","finance","forecast"}) self.assertLess(elapsed, 0.5) def test_25_market_uses_max_eligible_kline_and_real_quote_time(self): quote_body = canonical_bytes({"data":{"f84":100,"f116":2000,"f124":1785481200}}) kline_body = canonical_bytes({"data":{"klines":[ "2026-07-30,1,19,1,1,1", "2026-08-02,1,99,1,1,1", "2026-07-31,1,20,1,1,1" ]}}) class Client: process_start = datetime(2026, 8, 1, 12, tzinfo=ZoneInfo("Asia/Shanghai")) fixture_dir = Path("fixture") def fetch(self, request): body = quote_body if request.data_kind == "shares_market_cap" else kline_body return {"body":body,"from_cache":False,"fingerprint":request.data_kind,"meta":{ "blob_hash":sha256_bytes(body),"bytes":len(body),"http_status":200, "fetched_at":"2026-08-01T12:00:00+08:00","expires_at":"2026-08-02T12:00:00+08:00", "data_kind":request.data_kind, }} def confirm_reusable(self, *_args): pass result = acquire_market(Client(), "001270.SZ", "2026-08-01") self.assertEqual(result["market"]["price"], 20) self.assertEqual(result["market"]["shares_date"], "2026-07-31") self.assertEqual(result["field_lineage"]["market.price"]["data_date"], "2026-07-31") def test_26_market_rejects_unproven_historical_shares(self): quote_body = canonical_bytes({"data":{"f84":100,"f116":2000,"f124":0}}) kline_body = canonical_bytes({"data":{"klines":["2026-07-31,1,20,1,1,1"]}}) class Client: process_start = datetime(2026, 8, 2, 12, tzinfo=ZoneInfo("Asia/Shanghai")) fixture_dir = None def fetch(self, request): body = quote_body if request.data_kind == "shares_market_cap" else kline_body return {"body":body,"from_cache":False,"fingerprint":request.data_kind,"meta":{ "blob_hash":sha256_bytes(body),"bytes":len(body),"http_status":200, "fetched_at":"2026-08-02T12:00:00+08:00","expires_at":"2026-08-03T12:00:00+08:00", "data_kind":request.data_kind, }} def confirm_reusable(self, *_args): pass with self.assertRaisesRegex(ValueError, "E_HISTORICAL_SHARES_UNPROVEN"): acquire_market(Client(), "001270.SZ", "2026-08-01") def test_27_cninfo_cutoff_is_shanghai_end_of_day(self): published = datetime(2026, 8, 1, 23, 30, tzinfo=ZoneInfo("Asia/Shanghai")) stock_body = canonical_bytes({"stockList":[{"code":"001270","orgId":"x","zwjc":"铖昌科技"}]}) ann_body = canonical_bytes({"announcements":[{ "announcementTime":int(published.timestamp()*1000),"announcementId":"1", "announcementTitle":"2025年年度报告","adjunctUrl":"finalpage/a.pdf" }]}) class Client: process_start = datetime(2026, 8, 2, tzinfo=ZoneInfo("Asia/Shanghai")) def fetch(self, request): body = stock_body if request.data_kind == "stock_identity" else ann_body return {"body":body,"from_cache":False,"fingerprint":request.data_kind,"meta":{ "blob_hash":sha256_bytes(body),"bytes":len(body),"http_status":200, "fetched_at":"2026-08-02T00:00:00+08:00","expires_at":"2026-08-03T00:00:00+08:00", "data_kind":request.data_kind, }} def confirm_reusable(self, *_args): pass result = acquire_announcements(Client(), "001270.SZ", "2026-08-01") self.assertEqual(result["records"][0]["publish_date"], "2026-08-01") def test_28_forecast_never_fabricates_requested_asof_dates(self): summary_body = canonical_bytes({"result":{"data":[{ "REPORT_DATE":"2026-08-02","RATING_ORG_NUM":4 }]}}) detail_body = b'' class Client: def fetch(self, request): body = summary_body if request.data_kind == "forecast_summary" else detail_body return {"body":body,"from_cache":False,"fingerprint":request.data_kind,"meta":{ "blob_hash":sha256_bytes(body),"bytes":len(body),"http_status":200, "fetched_at":"2026-08-02T00:00:00+08:00","expires_at":"2026-08-02T06:00:00+08:00", "data_kind":request.data_kind, }} def confirm_reusable(self, *_args): pass result = acquire_forecast(Client(), "001270.SZ", "2026-08-01") self.assertEqual(result["sources"], []) self.assertEqual(result["institutions"]["forecasts"], []) self.assertEqual([g["gap_id"] for g in result["gaps"]], ["W_FORECAST_COVERAGE"]) def test_29_all_core_fields_have_exact_lineage_a1_and_raw_hash(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") build = json.loads((root/"out"/"snapshot_build_report.json").read_text(encoding="utf-8")) sources = json.loads((root/"out"/"source_evidence.json").read_text(encoding="utf-8")) source_by_id = {item["id"]: item for item in sources["sources"]} source_ids = set(source_by_id) raw_hashes = set(sources["raw_hashes"]) for field in CORE_VALUE_PATHS: item = build["field_lineage"][field] self.assertIn(item["source_id"], source_ids, field) self.assertIn(item["raw_hash"], raw_hashes, field) self.assertLessEqual(item["publish_date"], "2026-08-01", field) self.assertLessEqual(item["data_date"], "2026-08-01", field) if field.startswith(("financials.", "balance_sheet.")): self.assertIn(item["a1_source_id"], source_ids, field) self.assertIn(item["a1_raw_hash"], raw_hashes, field) a1 = source_by_id[item["a1_source_id"]] self.assertEqual(a1["period_end"], item["a1_period_end"], field) self.assertEqual(a1["publish_date"], item["a1_publish_date"], field) self.assertIn(item["a1_support"], a1["supports"], field) if field.startswith("financials.prior_year_same_period."): relation = item["comparison_relation"] self.assertEqual(relation["type"], "same_response_comparative_row") self.assertEqual(relation["comparison_period_end"], item["data_date"]) self.assertEqual(relation["current_period_end"], item["a1_period_end"]) self.assertEqual(relation["shared_publish_date"], item["publish_date"]) else: self.assertEqual(item["a1_period_end"], item["data_date"], field) def test_30_qa_rejects_tampered_field_lineage_and_protected_value(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") report = (root/"out"/"report.md").read_text(encoding="utf-8") snapshot = json.loads((root/"out"/"data_snapshot.json").read_text(encoding="utf-8")) build = json.loads((root/"out"/"snapshot_build_report.json").read_text(encoding="utf-8")) build["field_lineage"]["market.price"]["raw_hash"] = "0"*64 snapshot["market"]["price"] = 1 qa = run_qa(report, snapshot, build, None, root/"out") self.assertEqual(qa["status"], "FAIL") self.assertTrue(any("raw hash" in item or "保护字段" in item for item in qa["errors"])) def test_31_every_data_kind_fresh_stale_missing_through_entrypoint(self): kinds = sorted({ "stock_identity", "announcement_index", "market_close", "shares_market_cap", "finance_main", "finance_income", "finance_balance", "finance_cashflow", "forecast_summary", "forecast_detail", }) for kind in kinds: for requested_state in ("STALE", "MISSING"): with self.subTest(kind=kind, state=requested_state), tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "cold") self.install_baseline_state(root, kind, requested_state) cache = ContentCache(root / "cache") _, before = cache.baseline_states( "001270.SZ", "2026-08-01", datetime.now().astimezone() ) self.assertEqual(before[kind], requested_state) self.assertTrue(all( state == "FRESH" for name, state in before.items() if name != kind ), before) code, summary = self.run_fixture(root, "refresh") self.assertEqual((code, summary["status"]), (0, "DATA_READY_NEEDS_JUDGMENT")) providers = json.loads( (root / "refresh" / "provider_results.json").read_text(encoding="utf-8") ) transported = [ item["data_kind"] for result in providers.values() for item in result["request_telemetry"] ] self.assertEqual(transported, [kind]) _, after = cache.baseline_states( "001270.SZ", "2026-08-01", datetime.now().astimezone() ) self.assertTrue(all(state == "FRESH" for state in after.values()), after) def test_32_baseline_current_update_is_concurrent_and_deterministic(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) cache = ContentCache(root) bases = [{ "schema_version":1,"adapter_bundle_version":"2.1.0","ticker":"001270.SZ", "baseline_as_of":"2026-08-01","created_at":f"2026-08-01T00:00:{i:02d}+08:00", "watermark":{"market_date":"2026-07-31"},"provider_results":{},"data_kinds":{}, "field_lineage":{},"source_hashes":[],"marker":str(i), } for i in range(12)] with ThreadPoolExecutor(max_workers=6) as pool: list(pool.map(lambda item: cache.write_baseline("001270.SZ", item), reversed(bases))) current = json.loads((root/"companies"/"001270.SZ"/"current.json").read_text(encoding="utf-8")) hashes = [sha256_bytes(canonical_bytes({k:v for k,v in item.items() if k not in {"created_at","data_hash"}})) for item in bases] self.assertEqual(current["data_hash"], max(hashes)) def test_33_fingerprint_covers_headers_form_body_and_repeated_query_order(self): with tempfile.TemporaryDirectory() as raw: registry = json.loads((PROJECT_DEV/"stock_valuation_pipeline_v2"/"provider_registry.json").read_text(encoding="utf-8")) client = HttpClient(registry, ContentCache(Path(raw)), time.monotonic()+1, "r", datetime.now().astimezone(), FIXTURE) base = dict(provider_id="cninfo.announcement_index",adapter_version="1.0.0",data_kind="announcement_index",method="POST",url="https://www.cninfo.com.cn/new/hisAnnouncement/query",ticker="001270.SZ",as_of="2026-08-01",fixture_id="x",content_type="application/x-www-form-urlencoded") a = HttpRequest(**base, query=[], body=b"b=2&a=1", headers={"X-Test":"v"}) b = HttpRequest(**base, query=[], body=b"a=1&b=2", headers={"x-test":"v"}) self.assertEqual(client._fingerprint(a), client._fingerprint(b)) q1 = HttpRequest(**{**base,"data_kind":"announcement_index"}, query=[("x","1"),("x","2")], body=b"") q2 = HttpRequest(**{**base,"data_kind":"announcement_index"}, query=[("x","2"),("x","1")], body=b"") self.assertNotEqual(client._fingerprint(q1), client._fingerprint(q2)) def test_34_registry_rejects_path_query_reportname_and_content_type_variants(self): with tempfile.TemporaryDirectory() as raw: registry = json.loads((PROJECT_DEV/"stock_valuation_pipeline_v2"/"provider_registry.json").read_text(encoding="utf-8")) client = HttpClient(registry, ContentCache(Path(raw)), time.monotonic()+1, "r", datetime.now().astimezone(), FIXTURE) valid_query = {"reportName":"RPT_DMSK_FN_INCOME","columns":"ALL","filter":'(SECUCODE="001270.SZ")',"pageNumber":"1","pageSize":"20","sortTypes":"-1","sortColumns":"REPORT_DATE"} variants = [ HttpRequest("eastmoney.finance","1.0.0","finance_income","GET","https://datacenter-web.eastmoney.com/api/data/v1/other","001270.SZ","2026-08-01","x",query=valid_query), HttpRequest("eastmoney.finance","1.0.0","finance_income","GET","https://datacenter-web.eastmoney.com/api/data/v1/get","001270.SZ","2026-08-01","x",query={**valid_query,"extra":"1"}), HttpRequest("eastmoney.finance","1.0.0","finance_income","GET","https://datacenter-web.eastmoney.com/api/data/v1/get","001270.SZ","2026-08-01","x",query={**valid_query,"reportName":"WRONG"}), ] for request in variants: with self.subTest(request=request.url, query=request.query): with self.assertRaises(SourceContractError): client.fetch(request) def test_35_deadline_clips_backoff_and_forbids_cache_mutation(self): with tempfile.TemporaryDirectory() as raw: registry = json.loads((PROJECT_DEV/"stock_valuation_pipeline_v2"/"provider_registry.json").read_text(encoding="utf-8")) client = HttpClient(registry, ContentCache(Path(raw)), time.monotonic()+0.08, "r", datetime.now().astimezone()) class Opener: def open(self, *_args, **_kwargs): raise urllib.error.URLError("reset") client.opener = Opener() request = HttpRequest("eastmoney.market","1.0.0","market_close","GET","https://push2his.eastmoney.com/api/qt/stock/kline/get","001270.SZ","2026-08-01","x",query={"secid":"0.001270","klt":"101","fqt":"1","beg":"20260718","end":"20260801","fields1":"f1,f2,f3,f4,f5,f6","fields2":"f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61"}) started = time.perf_counter() with self.assertRaises((NetworkBudgetExceeded, SourceContractError)): client.fetch(request) self.assertLess(time.perf_counter()-started, 1.08) self.assertEqual(client.cache_mutations, 0) attempts = client.network_attempts time.sleep(0.12) self.assertEqual(client.network_attempts, attempts) self.assertEqual(list(Path(raw).rglob("*.json")), []) def test_36_cleanup_backup_failure_keeps_new_success_and_backup(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") old_hash = tree_hash(root/"out") def fault(step): if step == "cleanup_backup": raise OSError("cleanup injected") code, summary = self.run_fixture(root, "out", judgment=True, force=True, fault=fault) self.assertEqual((code, summary["status"]), (0, "COMPLETE_WITH_GAPS")) self.assertNotEqual(tree_hash(root/"out"), old_hash) self.assertTrue((root/"out"/"valuation_results.json").is_file()) self.assertTrue(any(root.glob(".out.backup-*"))) self.assertTrue((root/"out.commit-receipt.json").is_file()) def test_37_receipt_failure_restores_old_output_and_receipt(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") old_hash = tree_hash(root/"out") old_receipt = (root/"out.commit-receipt.json").read_bytes() def fault(step): if step == "receipt": raise SystemExit("receipt injected") with self.assertRaises(SystemExit): self.run_fixture(root, "out", judgment=True, force=True, fault=fault) self.assertEqual(tree_hash(root/"out"), old_hash) self.assertEqual((root/"out.commit-receipt.json").read_bytes(), old_receipt) failure = json.loads(next(root.glob("out.failed-*"), None).joinpath("failure.json").read_text(encoding="utf-8")) self.assertEqual(failure["error_type"], "SystemExit") def test_38_rollback_secondary_error_is_preserved(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") def fault(step): if step == "rename_stage_to_output": raise KeyboardInterrupt("primary") if step == "restore": raise RuntimeError("secondary") with self.assertRaises(KeyboardInterrupt): self.run_fixture(root, "out", judgment=True, force=True, fault=fault) failed = next(root.glob("out.failed-*")) payload = json.loads((failed/"failure.json").read_text(encoding="utf-8")) self.assertEqual(payload["error_type"], "KeyboardInterrupt") self.assertTrue(any("secondary" in item for item in payload["rollback_errors"])) self.assertTrue(any(root.glob(".out.backup-*"))) def test_39_lexical_reparse_output_is_rejected_before_cache(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) (root/"out").mkdir() with patch("stock_valuation_pipeline_v2.workflow._is_reparse", side_effect=lambda p: p == root/"out"): code, summary = self.run_fixture(root, "out", force=True) self.assertEqual((code, summary["status"]), (2, "INPUT_ERROR")) self.assertFalse((root/"cache").exists()) def test_40_six_terminal_contracts_and_exact_artifact_sets(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) complete_fixture = self.clone_fixture(root) summary_path = complete_fixture/"eastmoney_forecast_summary_success.json" summary_data = json.loads(summary_path.read_text(encoding="utf-8")) summary_data["result"]["data"][0]["RATING_ORG_NUM"] = 3 summary_path.write_text(json.dumps(summary_data), encoding="utf-8") def run(name, fixture, judgment=False, fault=None): return run_ticker_pipeline(ticker="001270.SZ",as_of="2026-08-01",output_dir=root/name,cache_dir=root/f"cache-{name}",judgment_path=(fixture/"judgment_overlay.json") if judgment else None,fixture_dir=fixture,task_start=TASK_START,network_budget_seconds=5,fault_injector=fault) code, complete = run("complete", complete_fixture, True) self.assertEqual((code, complete["status"], complete["gap_count"]), (0,"COMPLETE",0)) code, with_gaps = run("with-gaps", FIXTURE, True) self.assertEqual((code, with_gaps["status"]), (0,"COMPLETE_WITH_GAPS")) code, data_ready = run("data-ready", FIXTURE, False) self.assertEqual((code, data_ready["status"]), (0,"DATA_READY_NEEDS_JUDGMENT")) blocked_fixture = root/"blocked-fixture" shutil.copytree(FIXTURE, blocked_fixture) quote = json.loads((blocked_fixture/"eastmoney_market_quote_success.json").read_text(encoding="utf-8")) quote["data"]["f124"] = 0 (blocked_fixture/"eastmoney_market_quote_success.json").write_text(json.dumps(quote), encoding="utf-8") code, blocked = run("blocked", blocked_fixture) self.assertEqual((code, blocked["status"]), (4,"BLOCKED")) code, failed = run("failed", FIXTURE, fault=lambda step: (_ for _ in ()).throw(RuntimeError("x")) if step == "write:data_snapshot.json" else None) self.assertEqual((code, failed["status"]), (5,"FAILED")) code, input_error = run_ticker_pipeline(ticker="bad",as_of="2026-08-01",output_dir=root/"input",cache_dir=root/"input-cache") self.assertEqual((code, input_error["status"]), (2,"INPUT_ERROR")) for summary in (complete, with_gaps, data_ready, blocked, failed, input_error): self.assertEqual(set(summary), EXPECTED_KEYS) success_base = {"data_snapshot.json","snapshot_build_report.json","report.md","qa_report.json","gaps.json","source_evidence.json","provider_results.json","runtime_metrics.json","runtime.log","manifest.json"} self.assertEqual({p.name for p in (root/"data-ready").iterdir()}, success_base) self.assertEqual({p.name for p in (root/"complete").iterdir()}, success_base|{"valuation_snapshot.json","valuation_results.json"}) self.assertEqual({p.name for p in (root/"with-gaps").iterdir()}, success_base|{"valuation_snapshot.json","valuation_results.json"}) for summary in (complete, with_gaps, data_ready): for key in ( "run_id", "output_dir", "manifest_path", "manifest_sha256", "receipt_path", "receipt_sha256", "receipt_bytes", "task_start", "task_start_source", "process_start", "terminal_at", "task_wall_seconds", "process_wall_seconds", "gap_count", ): self.assertIsNotNone(summary[key], (summary["status"], key)) self.assertIsNone(summary["failed_dir"]) self.assertIsNone(summary["error_code"]) self.assertIsNone(summary["error"]) self.assertGreaterEqual(summary["task_wall_seconds"], summary["process_wall_seconds"]) for summary in (blocked, failed): names = {p.name for p in Path(summary["failed_dir"]).iterdir()} required_failure = {"failure.json","runtime_metrics.json","runtime.log"} optional_failure = {"provider_results.json","source_evidence.json"} self.assertTrue(required_failure.issubset(names)) self.assertTrue(names.issubset(required_failure | optional_failure), names) for key in ( "run_id", "output_dir", "failed_dir", "task_start", "task_start_source", "process_start", "terminal_at", "task_wall_seconds", "process_wall_seconds", "gap_count", "error_code", "error", ): self.assertIsNotNone(summary[key], (summary["status"], key)) for key in ( "manifest_path", "manifest_sha256", "receipt_path", "receipt_sha256", "receipt_bytes", ): self.assertIsNone(summary[key], (summary["status"], key)) self.assertEqual(input_error["exit_code"], 2) for key in ( "run_id", "output_dir", "failed_dir", "manifest_path", "manifest_sha256", "receipt_path", "receipt_sha256", "receipt_bytes", "gap_count", ): self.assertIsNone(input_error[key], key) self.assertIsNotNone(input_error["error_code"]) self.assertIsNotNone(input_error["error"]) self.assertFalse(any(root.glob("input.failed-*"))) def test_41_cli_parser_error_is_one_canonical_json_stdout(self): env = os.environ.copy() env["PYTHONPATH"] = str(PROJECT_DEV) proc = subprocess.run( [sys.executable,"-m","stock_valuation_pipeline_v2","--ticker","001270.SZ","--output-dir","x"], cwd=REPO,env=env,text=True,capture_output=True,check=False, ) self.assertEqual(proc.returncode, 2) self.assertEqual(proc.stderr, "") self.assertEqual(len(proc.stdout.splitlines()), 1) payload = json.loads(proc.stdout) self.assertEqual((payload["status"], set(payload)), ("INPUT_ERROR", EXPECTED_KEYS)) def test_42_entrypoint_blocks_http_schema_and_future_variants(self): cases = ("http_404","http_500","schema","future") for case in cases: with self.subTest(case=case), tempfile.TemporaryDirectory() as raw: root = Path(raw) fixture = self.clone_fixture(root) manifest_path = fixture/"fixture_manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if case.startswith("http_"): manifest["requests"]["eastmoney_finance_income"] = { "file":"http_error.json","status":int(case.split("_")[1]), "headers":{"content-type":"application/json"}, } elif case == "schema": manifest["requests"]["eastmoney_finance_income"]["file"] = "schema_drift.json" else: data = json.loads((fixture/"eastmoney_finance_income_success.json").read_text(encoding="utf-8")) for row in data["result"]["data"]: row["NOTICE_DATE"] = "2026-08-02" (fixture/"future_income.json").write_text(json.dumps(data), encoding="utf-8") manifest["requests"]["eastmoney_finance_income"]["file"] = "future_income.json" manifest_path.write_text(json.dumps(manifest), encoding="utf-8") code, summary = run_ticker_pipeline(ticker="001270.SZ",as_of="2026-08-01",output_dir=root/"out",cache_dir=root/"cache",fixture_dir=fixture,task_start=TASK_START,network_budget_seconds=5) self.assertEqual((code, summary["status"]), (4,"BLOCKED")) self.assertFalse((root/"out").exists()) self.assertIsNone(summary["manifest_path"]) def test_43_all_staging_write_manifest_and_rename1_faults_are_atomic(self): points = [ "mkdir_staging", "write:data_snapshot.json", "write:snapshot_build_report.json", "write:provider_results.json", "write:gaps.json", "write:source_evidence.json", "write:report.md", "write:qa_report.json", "write:runtime.log", "write:runtime_metrics.json", "manifest", ] for point in points: with self.subTest(point=point), tempfile.TemporaryDirectory() as raw: root = Path(raw) def fault(step, target=point): if step == target: raise SystemExit(target) with self.assertRaises(SystemExit): self.run_fixture(root, "out", fault=fault) self.assertFalse((root/"out").exists()) self.assertFalse((root/"out.commit-receipt.json").exists()) failed = next(root.glob("out.failed-*")) self.assertFalse((failed/"manifest.json").exists()) with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") old_hash = tree_hash(root/"out") old_receipt = (root/"out.commit-receipt.json").read_bytes() def rename1_fault(step): if step == "rename_old_to_backup": raise KeyboardInterrupt("rename1") with self.assertRaises(KeyboardInterrupt): self.run_fixture(root, "out", judgment=True, force=True, fault=rename1_fault) self.assertEqual(tree_hash(root/"out"), old_hash) self.assertEqual((root/"out.commit-receipt.json").read_bytes(), old_receipt) def test_44_v1_bridge_force_and_invalid_input_match_direct_v1(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) output = root / "same-output" env_direct, env_bridge = os.environ.copy(), os.environ.copy() env_direct["PYTHONPATH"], env_bridge["PYTHONPATH"] = str(ANA_DEV), str(PROJECT_DEV) base_direct = [sys.executable,"-m","stock_valuation_pipeline","--input",str(GREAT_WALL),"--output-dir",str(output)] base_bridge = [sys.executable,"-m","stock_valuation_pipeline_v2","--input",str(GREAT_WALL),"--output-dir",str(output)] # No-force generation: use the exact same path so stdout/stderr are byte-comparable. direct_generated = subprocess.run(base_direct,cwd=REPO,env=env_direct,capture_output=True,check=False) self.assertEqual(direct_generated.returncode, 0, direct_generated.stderr) direct_files = { name: (output/name).read_bytes() for name in ("valuation_results.json","valuation_report.md","run_manifest.json") } shutil.rmtree(output) bridge_generated = subprocess.run(base_bridge,cwd=REPO,env=env_bridge,capture_output=True,check=False) self.assertEqual( (bridge_generated.returncode, bridge_generated.stdout, bridge_generated.stderr), (direct_generated.returncode, direct_generated.stdout, direct_generated.stderr), ) for name, expected in direct_files.items(): self.assertEqual((output/name).read_bytes(), expected, name) # Reuse and force paths retain the exact V1 CLI contract. direct_reused = subprocess.run(base_direct,cwd=REPO,env=env_direct,capture_output=True,check=False) bridge_reused = subprocess.run(base_bridge,cwd=REPO,env=env_bridge,capture_output=True,check=False) self.assertEqual( (bridge_reused.returncode, bridge_reused.stdout, bridge_reused.stderr), (direct_reused.returncode, direct_reused.stdout, direct_reused.stderr), ) direct_forced = subprocess.run(base_direct+["--force"],cwd=REPO,env=env_direct,capture_output=True,check=False) forced_files = { name: (output/name).read_bytes() for name in ("valuation_results.json","valuation_report.md","run_manifest.json") } bridge_forced = subprocess.run(base_bridge+["--force"],cwd=REPO,env=env_bridge,capture_output=True,check=False) self.assertEqual( (bridge_forced.returncode, bridge_forced.stdout, bridge_forced.stderr), (direct_forced.returncode, direct_forced.stdout, direct_forced.stderr), ) for name, expected in forced_files.items(): self.assertEqual((output/name).read_bytes(), expected, name) # Exit 2 (input) and exit 3 (runtime) are compared on identical paths too. bad = root/"bad.json" bad.write_text("{}",encoding="utf-8") bad_output = root / "bad-output" a = subprocess.run([sys.executable,"-m","stock_valuation_pipeline","--input",str(bad),"--output-dir",str(bad_output)],cwd=REPO,env=env_direct,capture_output=True,check=False) b = subprocess.run([sys.executable,"-m","stock_valuation_pipeline_v2","--input",str(bad),"--output-dir",str(bad_output)],cwd=REPO,env=env_bridge,capture_output=True,check=False) self.assertEqual((b.returncode,b.stdout,b.stderr),(a.returncode,a.stdout,a.stderr)) self.assertEqual(a.returncode, 2) blocked_output = root / "blocked-output" blocked_output.write_text("not a directory", encoding="utf-8") a = subprocess.run([sys.executable,"-m","stock_valuation_pipeline","--input",str(GREAT_WALL),"--output-dir",str(blocked_output)],cwd=REPO,env=env_direct,capture_output=True,check=False) b = subprocess.run([sys.executable,"-m","stock_valuation_pipeline_v2","--input",str(GREAT_WALL),"--output-dir",str(blocked_output)],cwd=REPO,env=env_bridge,capture_output=True,check=False) self.assertEqual((b.returncode,b.stdout,b.stderr),(a.returncode,a.stdout,a.stderr)) self.assertEqual(a.returncode, 3) def test_45_market_empty_weekend_and_future_only_are_blocking(self): quote_body = canonical_bytes({"data":{"f84":100,"f116":2000,"f124":1785481200}}) variants = ([], ["2026-08-02,1,20,1,1,1"]) for klines in variants: with self.subTest(klines=klines): kline_body = canonical_bytes({"data":{"klines":klines}}) class Client: process_start = datetime(2026,8,1,12,tzinfo=ZoneInfo("Asia/Shanghai")) fixture_dir = Path("fixture") def fetch(self, request): body = quote_body if request.data_kind == "shares_market_cap" else kline_body return {"body":body,"from_cache":False,"fingerprint":request.data_kind,"meta":{ "blob_hash":sha256_bytes(body),"bytes":len(body),"http_status":200, "fetched_at":"2026-08-01T12:00:00+08:00","expires_at":"2026-08-02T12:00:00+08:00","data_kind":request.data_kind, }} def confirm_reusable(self, *_args): pass with self.assertRaisesRegex(ValueError, "无 as-of|无不晚于"): acquire_market(Client(), "001270.SZ", "2026-08-01") def test_46_four_blocking_transports_stop_without_background_cache_effects(self): with tempfile.TemporaryDirectory() as raw: registry = json.loads((PROJECT_DEV/"stock_valuation_pipeline_v2"/"provider_registry.json").read_text(encoding="utf-8")) # Leave enough time for all four workers to enter the registered # transport, then keep that transport blocked past the shared # deadline. This proves cancellation after a real request start, # not merely the pre-request deadline guard. client = HttpClient(registry, ContentCache(Path(raw)), time.monotonic()+2.0, "r", datetime.now().astimezone()) started_four = threading.Barrier(4) class Response: status = 200 headers = {"content-type":"application/json"} def __enter__(self): return self def __exit__(self, *_args): return False def read(self, _size): return b"{}" class Opener: def open(self, *_args, **_kwargs): started_four.wait(timeout=1.5) time.sleep(2.50) return Response() client.opener = Opener() def blocking(c, _ticker, _as_of): req = HttpRequest("eastmoney.market","1.0.0","market_close","GET","https://push2his.eastmoney.com/api/qt/stock/kline/get","001270.SZ","2026-08-01","x",query={"secid":"0.001270","klt":"101","fqt":"1","beg":"20260718","end":"20260801","fields1":"f1,f2,f3,f4,f5,f6","fields2":"f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61"}) c.fetch(req) return {} started = time.perf_counter() with patch("stock_valuation_pipeline_v2.acquisition.acquire_announcements", blocking), patch( "stock_valuation_pipeline_v2.acquisition.acquire_market", blocking ), patch("stock_valuation_pipeline_v2.acquisition.acquire_finance", blocking), patch( "stock_valuation_pipeline_v2.acquisition.acquire_forecast", blocking ): acquire_all(client, "001270.SZ", "2026-08-01") self.assertLess(time.perf_counter()-started, 3.00) self.assertEqual(client.network_attempts, 4) self.assertEqual(client.cache_mutations, 0) time.sleep(0.15) self.assertEqual(client.cache_mutations, 0) self.assertEqual(list(Path(raw).rglob("*.json")), []) def test_47_comparative_a1_support_and_qa_mismatch_are_blocking(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) fixture = self.clone_fixture(root) announcements = fixture / "cninfo_announcements_success.json" payload = json.loads(announcements.read_text(encoding="utf-8")) q1 = next(item for item in payload["v2_sources"] if item["id"] == "SRC-Q1-2026") q1["supports"].remove("financials.prior_year_same_period_comparative") announcements.write_text(json.dumps(payload), encoding="utf-8") code, summary = self.run_fixture_dir(root, "blocked", fixture) self.assertEqual((code, summary["status"]), (4, "BLOCKED")) self.assertIn("A1", summary["error"]) with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") report = (root / "out" / "report.md").read_text(encoding="utf-8") snapshot = json.loads((root / "out" / "data_snapshot.json").read_text(encoding="utf-8")) build = json.loads((root / "out" / "snapshot_build_report.json").read_text(encoding="utf-8")) item = build["field_lineage"]["financials.prior_year_same_period.revenue"] item["comparison_relation"]["current_period_end"] = "2025-03-31" qa = run_qa(report, snapshot, build, None, root / "out") self.assertEqual(qa["status"], "FAIL") self.assertTrue(any("比较关系" in error for error in qa["errors"]), qa["errors"]) def test_48_partial_finance_refresh_failure_does_not_advance_current(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "cold") current_before = self.install_baseline_state(root, "finance_income", "STALE") baseline_files_before = { path.name for path in (root / "cache" / "companies" / "001270.SZ" / "baselines").glob("*.json") } fixture = self.clone_fixture(root) manifest_path = fixture / "fixture_manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["requests"]["eastmoney_finance_income"] = { "file": "http_error.json", "status": 500, "headers": {"content-type": "application/json"}, } manifest_path.write_text(json.dumps(manifest), encoding="utf-8") code, summary = self.run_fixture_dir(root, "failed-refresh", fixture) self.assertEqual((code, summary["status"]), (4, "BLOCKED")) current = root / "cache" / "companies" / "001270.SZ" / "current.json" self.assertEqual(current.read_bytes(), current_before) self.assertEqual( { path.name for path in (root / "cache" / "companies" / "001270.SZ" / "baselines").glob("*.json") }, baseline_files_before, ) self.assertFalse((root / "failed-refresh").exists()) def test_49_slow_http_error_bodies_share_deadline_and_never_archive(self): registry = json.loads( (PROJECT_DEV / "stock_valuation_pipeline_v2" / "provider_registry.json").read_text(encoding="utf-8") ) query = { "secid": "0.001270", "klt": "101", "fqt": "1", "beg": "20260718", "end": "20260801", "fields1": "f1,f2,f3,f4,f5,f6", "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61", } request = HttpRequest( "eastmoney.market", "1.0.0", "market_close", "GET", "https://push2his.eastmoney.com/api/qt/stock/kline/get", "001270.SZ", "2026-08-01", "x", query=query, ) for status in (404, 500): with self.subTest(status=status), tempfile.TemporaryDirectory() as raw: class SlowBody: def __init__(self): self.reads = 0 self.closed = False def read(self, _size): self.reads += 1 time.sleep(1.20) return b'{"error":"slow"}' if self.reads == 1 else b"" def close(self): self.closed = True body = SlowBody() class Opener: calls = 0 def open(self, req, **_kwargs): self.calls += 1 raise urllib.error.HTTPError( req.full_url, status, "slow", {"content-type": "application/json"}, body ) opener = Opener() cache_root = Path(raw) client = HttpClient( registry, ContentCache(cache_root), time.monotonic() + 0.50, "r", datetime.now().astimezone(), ) client.opener = opener started = time.perf_counter() with self.assertRaises(NetworkBudgetExceeded): client.fetch(request) elapsed = time.perf_counter() - started self.assertLess(elapsed, 1.50) self.assertEqual((opener.calls, client.network_attempts, client.cache_mutations), (1, 1, 0)) time.sleep(1.25) self.assertEqual((opener.calls, client.cache_mutations), (1, 0)) self.assertEqual(list(cache_root.rglob("*.json")), []) def test_50_receipt_schema2_final_wall_covers_write_validation_and_cleanup(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) def delayed_receipt(step): if step == "file:receipt:write": time.sleep(0.25) started = time.perf_counter() code, summary = self.run_fixture(root, "out", fault=delayed_receipt) observed = time.perf_counter() - started self.assertEqual(code, 0) self.assertGreaterEqual(summary["process_wall_seconds"], 0.24) self.assertLessEqual(summary["process_wall_seconds"], observed + 0.05) self.assertLess(observed - summary["process_wall_seconds"], 0.10) receipt = json.loads((root / "out.commit-receipt.json").read_text(encoding="utf-8")) self.assertEqual(receipt["schema_version"], 2) self.assertNotIn("terminal_at", receipt) self.assertGreaterEqual( datetime.fromisoformat(summary["terminal_at"]), datetime.fromisoformat(receipt["receipt_write_started_at"]), ) with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") def delayed_cleanup(step): if step == "cleanup_backup": time.sleep(0.25) code, summary = self.run_fixture(root, "out", judgment=True, force=True, fault=delayed_cleanup) self.assertEqual(code, 0) self.assertGreaterEqual(summary["process_wall_seconds"], 0.24) def test_51_receipt_all_boundaries_fail_atomically_for_exception_and_baseexception(self): points = [ "file:receipt:temp", "file:receipt:write", "file:receipt:flush", "file:receipt:fsync", "file:receipt:replace", "receipt:read", "receipt:hash", "receipt:stat", ] for point in points: with self.subTest(point=point, error="RuntimeError"), tempfile.TemporaryDirectory() as raw: root = Path(raw) def fault(step, target=point): if step == target: raise RuntimeError(target) code, summary = self.run_fixture(root, "out", fault=fault) self.assertEqual((code, summary["status"]), (5, "FAILED")) self.assertFalse((root / "out").exists()) self.assertFalse((root / "out.commit-receipt.json").exists()) for error_type in (KeyboardInterrupt, SystemExit): with self.subTest(point=point, error=error_type.__name__), tempfile.TemporaryDirectory() as raw: root = Path(raw) def fault(step, target=point, exc_type=error_type): if step == target: raise exc_type(target) with self.assertRaises(error_type): self.run_fixture(root, "out", fault=fault) self.assertFalse((root / "out").exists()) self.assertFalse((root / "out.commit-receipt.json").exists()) failure = json.loads( (next(root.glob("out.failed-*")) / "failure.json").read_text(encoding="utf-8") ) self.assertEqual(failure["error_type"], error_type.__name__) def test_52_force_receipt_boundaries_restore_exact_old_truth(self): points = [ "file:receipt:temp", "file:receipt:write", "file:receipt:flush", "file:receipt:fsync", "file:receipt:replace", "receipt:read", "receipt:hash", "receipt:stat", ] for point in points: with self.subTest(point=point), tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "out") old_hash = tree_hash(root / "out") old_receipt = (root / "out.commit-receipt.json").read_bytes() def fault(step, target=point): if step == target: raise SystemExit(target) with self.assertRaises(SystemExit): self.run_fixture(root, "out", judgment=True, force=True, fault=fault) self.assertEqual(tree_hash(root / "out"), old_hash) self.assertEqual((root / "out.commit-receipt.json").read_bytes(), old_receipt) def test_53_every_provider_success_empty_http_schema_and_future_entrypoint_matrix(self): providers = { "announcements": "cninfo_announcements", "market": "eastmoney_market_kline", "finance": "eastmoney_finance_income", "forecast": "eastmoney_forecast_detail", } cases = ("success", "empty", "http_404", "http_500", "schema", "future") for provider, fixture_id in providers.items(): for case in cases: with self.subTest(provider=provider, case=case), tempfile.TemporaryDirectory() as raw: root = Path(raw) fixture = self.clone_fixture(root) manifest_path = fixture / "fixture_manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) entry = manifest["requests"][fixture_id] if case == "empty": (fixture / "matrix-empty.bin").write_bytes(b"") entry["file"] = "matrix-empty.bin" elif case.startswith("http_"): entry.update(file="http_error.json", status=int(case.split("_")[1])) elif case == "schema": entry["file"] = "schema_drift.json" elif case == "future": if provider == "announcements": data = json.loads((fixture / entry["file"]).read_text(encoding="utf-8")) data["v2_sources"][0]["publish_date"] = "2026-08-02" (fixture / "matrix-future.json").write_text(json.dumps(data), encoding="utf-8") elif provider == "market": data = json.loads((fixture / entry["file"]).read_text(encoding="utf-8")) data["data"]["klines"] = ["2026-08-02,1,92.56,1,1,1"] (fixture / "matrix-future.json").write_text(json.dumps(data), encoding="utf-8") elif provider == "finance": data = json.loads((fixture / entry["file"]).read_text(encoding="utf-8")) for row in data["result"]["data"]: row["NOTICE_DATE"] = "2026-08-02" (fixture / "matrix-future.json").write_text(json.dumps(data), encoding="utf-8") else: text = (fixture / entry["file"]).read_text(encoding="utf-8") text = __import__("re").sub( r'"report_date":"[^"]+"', '"report_date":"2026-08-02"', text ) (fixture / "matrix-future.html").write_text(text, encoding="utf-8") entry["file"] = "matrix-future.html" if provider != "forecast": entry["file"] = "matrix-future.json" manifest_path.write_text(json.dumps(manifest), encoding="utf-8") code, summary = self.run_fixture_dir(root, "out", fixture) if case == "success" or provider == "forecast": self.assertEqual(code, 0, summary) self.assertIn( summary["status"], {"DATA_READY_NEEDS_JUDGMENT", "COMPLETE", "COMPLETE_WITH_GAPS"}, ) if provider == "forecast" and case != "success": gaps = json.loads((root / "out" / "gaps.json").read_text(encoding="utf-8")) self.assertEqual( sum(item["gap_id"] == "W_FORECAST_COVERAGE" for item in gaps), 1 ) else: self.assertEqual((code, summary["status"]), (4, "BLOCKED"), summary) self.assertFalse((root / "out").exists()) def test_54_every_staged_atomic_replace_and_nonordinary_exception_has_no_false_success(self): files = [ "valuation_snapshot.json", "valuation_results.json", "data_snapshot.json", "snapshot_build_report.json", "provider_results.json", "gaps.json", "source_evidence.json", "report.md", "qa_report.json", "runtime.log", "runtime_metrics.json", "manifest.json", ] boundaries = ("temp", "write", "flush", "fsync", "replace") with tempfile.TemporaryDirectory() as seed_raw: seed = Path(seed_raw) self.run_fixture(seed, "seed") for name in files: for boundary in boundaries: point = f"file:{name}:{boundary}" for error_type in (KeyboardInterrupt, SystemExit): with self.subTest(point=point, error=error_type.__name__), tempfile.TemporaryDirectory() as raw: root = Path(raw) shutil.copytree(seed / "cache", root / "cache") def fault(step, target=point, exc_type=error_type): if step == target: raise exc_type(target) with self.assertRaises(error_type): self.run_fixture(root, "out", judgment=True, fault=fault) self.assertFalse((root / "out").exists()) self.assertFalse((root / "out.commit-receipt.json").exists()) failure = json.loads( (next(root.glob("out.failed-*")) / "failure.json").read_text(encoding="utf-8") ) self.assertEqual(failure["error_type"], error_type.__name__) def test_55_concurrent_incremental_supersede_keeps_deterministic_max_current(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) cache = ContentCache(root) seed = { "schema_version": 1, "adapter_bundle_version": "2.1.0", "ticker": "001270.SZ", "baseline_as_of": "2026-08-01", "created_at": "2026-08-01T00:00:00+08:00", "watermark": {"market_date": "2026-07-31"}, "provider_results": {}, "data_kinds": {}, "field_lineage": {}, "source_hashes": [], "marker": "seed", } seed_path = cache.write_baseline("001270.SZ", seed) seed_hash = json.loads(seed_path.read_text(encoding="utf-8"))["data_hash"] candidates = [dict(seed, marker=marker) for marker in ("refresh-a", "refresh-z")] for item in candidates: item["created_at"] = "2026-08-02T00:00:00+08:00" with ThreadPoolExecutor(max_workers=2) as pool: list(pool.map( lambda item: cache.write_baseline( "001270.SZ", item, supersede_data_hash=seed_hash ), reversed(candidates), )) expected_hashes = [ sha256_bytes(canonical_bytes({ key: value for key, value in item.items() if key not in {"created_at", "data_hash"} })) for item in candidates ] current = json.loads( (root / "companies" / "001270.SZ" / "current.json").read_text(encoding="utf-8") ) self.assertEqual(current["data_hash"], max(expected_hashes)) active = list((root / "companies" / "001270.SZ" / "baselines").glob("*.json")) self.assertEqual(len(active), 2) self.assertTrue( (root / "companies" / "001270.SZ" / "history" / seed_path.name).is_file() ) def test_56_each_data_kind_deleted_or_tampered_blob_isolated_and_repaired(self): kinds = ( "stock_identity", "announcement_index", "market_close", "shares_market_cap", "finance_main", "finance_income", "finance_balance", "finance_cashflow", "forecast_summary", "forecast_detail", ) for kind in kinds: for damage in ("deleted", "tampered"): with self.subTest(kind=kind, damage=damage), tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "cold") cache = ContentCache(root / "cache") baseline = cache.select_baseline("001270.SZ", "2026-08-01") self.assertIsNotNone(baseline) assert baseline is not None digest = baseline["data_kinds"][kind]["raw_hash"] blob = root / "cache" / "blobs" / "sha256" / digest[:2] / f"{digest}.bin" if damage == "deleted": blob.unlink() else: blob.write_bytes(b"corrupted-data-kind") selected, before = cache.baseline_states( "001270.SZ", "2026-08-01", datetime.now().astimezone() ) self.assertIsNotNone(selected) self.assertEqual(before[kind], "STALE") self.assertTrue(all( state == "FRESH" for name, state in before.items() if name != kind ), before) code, summary = self.run_fixture(root, "refresh") self.assertEqual((code, summary["status"]), (0, "DATA_READY_NEEDS_JUDGMENT")) providers = json.loads( (root / "refresh" / "provider_results.json").read_text(encoding="utf-8") ) transported = [ item["data_kind"] for result in providers.values() for item in result["request_telemetry"] ] self.assertEqual(transported, [kind]) repaired = blob.read_bytes() self.assertEqual(sha256_bytes(repaired), digest) merged, after = cache.baseline_states( "001270.SZ", "2026-08-01", datetime.now().astimezone() ) self.assertIsNotNone(merged) self.assertTrue(all(state == "FRESH" for state in after.values()), after) assert merged is not None for source_hash in merged["source_hashes"]: source_blob = ( root / "cache" / "blobs" / "sha256" / source_hash[:2] / f"{source_hash}.bin" ) self.assertEqual(sha256_bytes(source_blob.read_bytes()), source_hash) def test_57_blob_repair_failure_does_not_promote_or_advance_baseline(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "cold") cache = ContentCache(root / "cache") baseline = cache.select_baseline("001270.SZ", "2026-08-01") self.assertIsNotNone(baseline) assert baseline is not None entry = baseline["data_kinds"]["market_close"] digest = entry["raw_hash"] blob = root / "cache" / "blobs" / "sha256" / digest[:2] / f"{digest}.bin" blob.write_bytes(b"corrupted-market-close") reusable = ( root / "cache" / "reusable" / entry["provider_id"] / f"{entry['request_fingerprint']}.json" ) reusable.unlink(missing_ok=True) current = root / "cache" / "companies" / "001270.SZ" / "current.json" current_before = current.read_bytes() active_root = root / "cache" / "companies" / "001270.SZ" / "baselines" active_before = {path.name: path.read_bytes() for path in active_root.glob("*.json")} from stock_valuation_pipeline_v2 import cache as cache_module real_atomic_write = cache_module.atomic_write def fail_only_blob(path, data): if Path(path) == blob: raise OSError("injected blob repair failure") return real_atomic_write(path, data) with patch("stock_valuation_pipeline_v2.cache.atomic_write", side_effect=fail_only_blob): code, summary = self.run_fixture(root, "repair-failed") self.assertEqual((code, summary["status"]), (4, "BLOCKED")) self.assertEqual(current.read_bytes(), current_before) self.assertEqual( {path.name: path.read_bytes() for path in active_root.glob("*.json")}, active_before, ) self.assertFalse(reusable.exists()) self.assertEqual(blob.read_bytes(), b"corrupted-market-close") self.assertFalse((root / "repair-failed").exists()) def test_58_noncore_blob_repair_failure_keeps_company_baseline(self): with tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "cold") cache = ContentCache(root / "cache") baseline = cache.select_baseline("001270.SZ", "2026-08-01") self.assertIsNotNone(baseline) assert baseline is not None entry = baseline["data_kinds"]["forecast_detail"] digest = entry["raw_hash"] blob = root / "cache" / "blobs" / "sha256" / digest[:2] / f"{digest}.bin" blob.write_bytes(b"corrupted-forecast-detail") reusable = ( root / "cache" / "reusable" / entry["provider_id"] / f"{entry['request_fingerprint']}.json" ) reusable.unlink(missing_ok=True) current = root / "cache" / "companies" / "001270.SZ" / "current.json" current_before = current.read_bytes() active_root = root / "cache" / "companies" / "001270.SZ" / "baselines" active_before = {path.name: path.read_bytes() for path in active_root.glob("*.json")} from stock_valuation_pipeline_v2 import cache as cache_module real_atomic_write = cache_module.atomic_write def fail_only_blob(path, data): if Path(path) == blob: raise OSError("injected noncore blob repair failure") return real_atomic_write(path, data) with patch("stock_valuation_pipeline_v2.cache.atomic_write", side_effect=fail_only_blob): code, summary = self.run_fixture(root, "noncore-repair-failed") self.assertEqual((code, summary["status"]), (0, "DATA_READY_NEEDS_JUDGMENT")) self.assertEqual(current.read_bytes(), current_before) self.assertEqual( {path.name: path.read_bytes() for path in active_root.glob("*.json")}, active_before, ) self.assertFalse(reusable.exists()) selected, states = cache.baseline_states( "001270.SZ", "2026-08-01", datetime.now().astimezone() ) self.assertIsNotNone(selected) self.assertEqual(states["forecast_detail"], "STALE") self.assertTrue(all( state == "FRESH" for kind, state in states.items() if kind != "forecast_detail" ), states) def test_59_postwrite_hash_failure_never_advances_company_baseline(self): for kind in ("market_close", "forecast_detail"): with self.subTest(kind=kind), tempfile.TemporaryDirectory() as raw: root = Path(raw) self.run_fixture(root, "cold") cache = ContentCache(root / "cache") baseline = cache.select_baseline("001270.SZ", "2026-08-01") self.assertIsNotNone(baseline) assert baseline is not None entry = baseline["data_kinds"][kind] digest = entry["raw_hash"] blob = root / "cache" / "blobs" / "sha256" / digest[:2] / f"{digest}.bin" blob.write_bytes(f"corrupted-{kind}".encode()) reusable = ( root / "cache" / "reusable" / entry["provider_id"] / f"{entry['request_fingerprint']}.json" ) reusable.unlink(missing_ok=True) current = root / "cache" / "companies" / "001270.SZ" / "current.json" current_before = current.read_bytes() active_root = root / "cache" / "companies" / "001270.SZ" / "baselines" active_before = { path.name: path.read_bytes() for path in active_root.glob("*.json") } from stock_valuation_pipeline_v2 import cache as cache_module real_atomic_write = cache_module.atomic_write def write_wrong_blob(path, data): if Path(path) == blob: return real_atomic_write(path, b"post-write-hash-mismatch") return real_atomic_write(path, data) output_name = f"verify-{kind}" with patch( "stock_valuation_pipeline_v2.cache.atomic_write", side_effect=write_wrong_blob, ): code, summary = self.run_fixture(root, output_name) expected = (4, "BLOCKED") if kind == "market_close" else ( 0, "DATA_READY_NEEDS_JUDGMENT" ) self.assertEqual((code, summary["status"]), expected) self.assertEqual(current.read_bytes(), current_before) self.assertEqual( {path.name: path.read_bytes() for path in active_root.glob("*.json")}, active_before, ) self.assertFalse(reusable.exists()) self.assertEqual(blob.read_bytes(), b"post-write-hash-mismatch") if kind == "forecast_detail": providers = json.loads( (root / output_name / "provider_results.json").read_text(encoding="utf-8") ) self.assertTrue(providers["forecast"]["cache_integrity_failure"]) else: self.assertFalse((root / output_name).exists()) if __name__ == "__main__": unittest.main()