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 / "ana-data" / "result" / "股票估值" / "20260801_chengchang_technology_valuation" / "valuation_snapshot.json"
|
SOURCE_RESULTS = REPO / "ana-data" / "result" / "股票估值" / "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'<script id="v2-fixture" type="application/json">{"forecasts":[{"institution":"x","estimates":{}}]}</script>'
|
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()
|
