533 lines
27 KiB
Python
533 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
"""内部狗粮运行保障工具的行为测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import subprocess
|
||
import tempfile
|
||
import unittest
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
TOOL = Path(__file__).resolve().parents[1] / "dogfood_ops.py"
|
||
OPS_DIR = TOOL.parent
|
||
|
||
|
||
class DogfoodOpsTest(unittest.TestCase):
|
||
"""通过命令行入口验证运维人员实际使用的行为。"""
|
||
|
||
def setUp(self) -> None:
|
||
"""每个用例使用独立临时证据根，避免测试相互污染。"""
|
||
self.temp_dir = tempfile.TemporaryDirectory()
|
||
self.root = Path(self.temp_dir.name) / "evidence"
|
||
|
||
def tearDown(self) -> None:
|
||
"""清理测试创建的本地证据。"""
|
||
self.temp_dir.cleanup()
|
||
|
||
def run_tool(self, *args: str) -> subprocess.CompletedProcess[str]:
|
||
"""调用真实 CLI，并保留标准输出供失败断言。"""
|
||
return subprocess.run(
|
||
["python3", str(TOOL), *args],
|
||
check=False,
|
||
text=True,
|
||
capture_output=True,
|
||
)
|
||
|
||
def init_wave(self, wave: str) -> Path:
|
||
"""初始化指定波次并返回证据目录。"""
|
||
result = self.run_tool("init", "--root", str(self.root), "--wave", wave)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
return self.root / wave
|
||
|
||
def write_json(self, path: Path, value: Any) -> None:
|
||
"""写入稳定 JSON fixture。"""
|
||
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
def add_attachment(self, wave_dir: Path, name: str, content: str = "evidence") -> tuple[str, str]:
|
||
"""创建目录内证据附件并返回相对路径与哈希。"""
|
||
attachment = wave_dir / "attachments" / name
|
||
attachment.parent.mkdir(exist_ok=True)
|
||
attachment.write_text(content, encoding="utf-8")
|
||
return attachment.relative_to(wave_dir).as_posix(), hashlib.sha256(attachment.read_bytes()).hexdigest()
|
||
|
||
def populate_valid_evidence(self, wave_dir: Path) -> None:
|
||
"""构造不依赖 staging 的完整本地通过 fixture。"""
|
||
manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8"))
|
||
wave = manifest["wave"]
|
||
commit = "1" * 40
|
||
artifact_ref, artifact_hash = self.add_attachment(wave_dir, "rc-artifact.bin")
|
||
config_ref, config_hash = self.add_attachment(wave_dir, "config-bundle.txt")
|
||
dump_ref, dump_hash = self.add_attachment(wave_dir, "mysql.sql.gz")
|
||
restore_ref, _ = self.add_attachment(wave_dir, "restore-report.json")
|
||
storage_ref, storage_hash = self.add_attachment(wave_dir, "storage-audit.json")
|
||
common_ref, _ = self.add_attachment(wave_dir, "browser-proof.png")
|
||
network_ref, _ = self.add_attachment(wave_dir, "main-chain.har")
|
||
log_ref, _ = self.add_attachment(wave_dir, "application.log")
|
||
database_ref, _ = self.add_attachment(wave_dir, "database.json")
|
||
|
||
self.write_json(wave_dir / "rc.json", {
|
||
"schema": "dogfood-rc/1", "commit": commit, "repoClean": True,
|
||
"artifactRef": artifact_ref, "artifactSha256": artifact_hash,
|
||
})
|
||
self.write_json(wave_dir / "config-identity.json", {
|
||
"schema": "dogfood-config-identity/1", "capturedAt": "2026-07-22T00:00:00Z",
|
||
"rcCommit": commit, "configBundleRef": config_ref, "configSha256": config_hash,
|
||
})
|
||
self.write_json(wave_dir / "snapshot.json", {
|
||
"schema": "dogfood-snapshot/1",
|
||
"mysql": {"snapshotRef": dump_ref, "sha256": dump_hash, "capturedAt": "2026-07-22T00:00:00Z"},
|
||
"isolatedRestore": {"result": "PASS", "target": "isolated-dogfood-restore-001", "completedAt": "2026-07-22T01:00:00Z", "evidenceRef": restore_ref},
|
||
})
|
||
self.write_json(wave_dir / "storage-audit.json", {
|
||
"schema": "dogfood-storage-audit/1", "result": "PASS", "capturedAt": "2026-07-22T00:30:00Z",
|
||
"reportRef": storage_ref, "reportSha256": storage_hash,
|
||
})
|
||
self.write_json(wave_dir / "accounts.json", {
|
||
"schema": "dogfood-accounts/1", "batchId": f"dogfood-{wave}",
|
||
"creatorCount": manifest["creatorCount"], "playerCount": manifest["playerCount"],
|
||
"freeQuotaCount": 4 if wave == "wave0" else 12,
|
||
})
|
||
|
||
permission_cases = {
|
||
"creator-own-project": "ALLOW",
|
||
"creator-cross-account-denied": "DENY",
|
||
"reviewer-review-only": "ALLOW",
|
||
"reviewer-config-write-denied": "DENY",
|
||
"readonly-write-denied": "DENY",
|
||
}
|
||
self.write_json(wave_dir / "role-permissions.json", {
|
||
"schema": "dogfood-role-permissions/1",
|
||
"checks": [{"case": case, "result": result, "screenshotRef": common_ref} for case, result in permission_cases.items()],
|
||
})
|
||
|
||
evidence_bundle = {"screenshotRef": common_ref, "networkRef": network_ref, "logRef": log_ref, "databaseRef": database_ref}
|
||
stages = ["create", "generate", "preview", "submit-review", "approve", "feed", "play", "telemetry"]
|
||
self.write_json(wave_dir / "main-chain.json", {
|
||
"schema": "dogfood-main-chain/1",
|
||
"identity": {"projectId": "p-1", "versionId": "v-1", "artifactSha256": artifact_hash, "gameId": "g-1"},
|
||
"stages": [{"stage": stage, "result": "PASS", **evidence_bundle} for stage in stages],
|
||
})
|
||
|
||
negative_cases = [
|
||
"draft-publish", "reviewing-publish", "rejected-publish", "direct-publish-bypass",
|
||
"artifact-drift", "delisted-feed", "delisted-detail", "delisted-deep-link",
|
||
]
|
||
self.write_json(wave_dir / "release-negatives.json", {
|
||
"schema": "dogfood-release-negatives/1",
|
||
"checks": [{"case": case, "blocked": True, "evidenceRef": common_ref} for case in negative_cases],
|
||
})
|
||
self.write_json(wave_dir / "manual-play.json", {
|
||
"schema": "dogfood-manual-play/1", "releasedGameCount": 1,
|
||
"games": [{"gameId": "g-1", "tester": "tester-1", "playedAt": "2026-07-22T02:00:00Z", "feedEnteredAt": "2026-07-22T02:10:00Z", "screenshotRef": common_ref, "source": "/gstack@mini-desktop"}],
|
||
})
|
||
|
||
samples = [
|
||
{"navigationId": f"nav-{index:02d}", "cache": "cold", "durationMs": 2000, "source": "/gstack@mini-desktop"}
|
||
for index in range(30)
|
||
]
|
||
self.write_json(wave_dir / "performance.json", {
|
||
"schema": "dogfood-first-frame-performance/1",
|
||
"context": {"host": "mini-desktop", "browser": "Chrome", "browserVersion": "146", "viewport": "390x844@1", "network": "Tailscale staging", "timing": "navigation-commit-to-first-interactive-game-frame"},
|
||
"samples": samples,
|
||
})
|
||
self.write_json(wave_dir / "telemetry.json", {
|
||
"schema": "dogfood-telemetry-completeness/1", "expectedCount": 20, "completeCount": 19,
|
||
"evidenceRef": common_ref,
|
||
})
|
||
self.write_json(wave_dir / "alerts.json", {
|
||
"schema": "dogfood-alert-delivery/1",
|
||
"deliveries": [{"alertId": "alert-1", "receivedAt": "2026-07-22T03:00:00Z", "recipient": "oncall-1", "externalMessageId": "msg-1", "evidenceRef": common_ref}],
|
||
})
|
||
|
||
start = datetime(2026, 7, 22, tzinfo=timezone.utc)
|
||
end = start + timedelta(hours=manifest["durationHours"])
|
||
(wave_dir / "oncall.tsv").write_text(
|
||
"start_at\tend_at\tprimary\tbackup\tcontact\n"
|
||
f"{start.isoformat().replace('+00:00', 'Z')}\t{end.isoformat().replace('+00:00', 'Z')}\toncall-1\toncall-2\tinternal-roster\n",
|
||
encoding="utf-8",
|
||
)
|
||
(wave_dir / "feedback.tsv").write_text(
|
||
"id\tcreated_at\treporter_id\trole\tcategory\tseverity\tsummary\tstatus\towner\tevidence_ref\n"
|
||
f"F-1\t2026-07-22T04:00:00Z\tu-1\tplayer\tux\tP2\tfixture feedback\tCLOSED\tops\t{common_ref}\n",
|
||
encoding="utf-8",
|
||
)
|
||
checklist_items = ["role-permissions", "main-chain", "release-negatives", "manual-play-before-feed"]
|
||
(wave_dir / "manual-acceptance.tsv").write_text(
|
||
"item\tresult\toperator\tchecked_at\tevidence_ref\n" + "".join(
|
||
f"{item}\tPASS\toperator-1\t2026-07-22T05:00:00Z\t{common_ref}\n" for item in checklist_items
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
self.write_json(wave_dir / "drills.json", {
|
||
"schema": "dogfood-drills/1",
|
||
"drills": [{"action": action, "result": "PASS", "ack": f"ACK-DOGFOOD-{wave}-{action}", "performedAt": "2026-07-22T06:00:00Z", "operator": "ops-1", "evidenceRef": common_ref} for action in ["stop", "delete", "rollback"]],
|
||
})
|
||
stop_conditions = [
|
||
"security-incident", "cross-account-access", "data-loss", "real-charge", "real-ad-impression",
|
||
"unreviewed-content-in-feed", "generation-success-below-80", "confirmed-false-positive",
|
||
"same-root-cause-blocked-two-users", "telemetry-missing-above-5", "review-backlog-over-4h",
|
||
"acceptance-cost-exceeded",
|
||
]
|
||
self.write_json(wave_dir / "safety-status.json", {
|
||
"schema": "dogfood-safety-status/1", "assessedAt": "2026-07-22T07:00:00Z", "assessor": "ops-1",
|
||
"conditions": [{"condition": condition, "triggered": False} for condition in stop_conditions],
|
||
})
|
||
|
||
def test_init_wave0_creates_fixed_ledgers_and_scale(self) -> None:
|
||
"""wave0 必须固定为 3 名创作者和 5 名玩家。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
|
||
manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8"))
|
||
self.assertEqual(manifest["wave"], "wave0")
|
||
self.assertEqual(manifest["creatorCount"], 3)
|
||
self.assertEqual(manifest["playerCount"], 5)
|
||
self.assertTrue((wave_dir / "oncall.tsv").is_file())
|
||
self.assertTrue((wave_dir / "feedback.tsv").is_file())
|
||
self.assertTrue((wave_dir / "p0.tsv").is_file())
|
||
self.assertTrue((wave_dir / "config-identity.json").is_file())
|
||
self.assertTrue((wave_dir / "manual-acceptance.tsv").is_file())
|
||
for name in (
|
||
"rc.json", "snapshot.json", "storage-audit.json", "accounts.json", "role-permissions.json",
|
||
"main-chain.json", "release-negatives.json", "manual-play.json", "performance.json",
|
||
"telemetry.json", "alerts.json", "drills.json", "safety-status.json",
|
||
):
|
||
self.assertTrue((wave_dir / name).is_file(), name)
|
||
self.assertTrue((wave_dir / "attachments").is_dir())
|
||
self.assertTrue((wave_dir / "freezes").is_dir())
|
||
|
||
def test_init_wave1_uses_approved_scale(self) -> None:
|
||
"""wave1 必须固定为累计 10 名创作者和 30 名玩家。"""
|
||
wave_dir = self.init_wave("wave1")
|
||
|
||
manifest = json.loads((wave_dir / "manifest.json").read_text(encoding="utf-8"))
|
||
self.assertEqual((manifest["creatorCount"], manifest["playerCount"]), (10, 30))
|
||
|
||
def test_repeated_init_is_idempotent_and_does_not_overwrite(self) -> None:
|
||
"""重复初始化应成功返回，但不得覆盖已经填写的证据。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
feedback = wave_dir / "feedback.tsv"
|
||
feedback.write_text(feedback.read_text(encoding="utf-8") + "F-1\t2026-07-22T00:00:00Z\tu1\tplayer\tbug\tP1\t卡住\tOPEN\tops\tattachments/f.txt\n", encoding="utf-8")
|
||
before = hashlib.sha256(feedback.read_bytes()).hexdigest()
|
||
|
||
result = self.run_tool("init", "--root", str(self.root), "--wave", "wave0")
|
||
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertEqual(hashlib.sha256(feedback.read_bytes()).hexdigest(), before)
|
||
self.assertIn("INIT_ALREADY_EXISTS", result.stdout)
|
||
|
||
def test_repeated_init_repairs_missing_templates_without_overwrite(self) -> None:
|
||
"""初始化中断后重试应补回缺失模板，同时保留已经填写的账本。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
feedback = wave_dir / "feedback.tsv"
|
||
feedback.write_text(feedback.read_text(encoding="utf-8") + "kept\n", encoding="utf-8")
|
||
(wave_dir / "rc.json").unlink()
|
||
(wave_dir / "attachments").rmdir()
|
||
|
||
result = self.run_tool("init", "--root", str(self.root), "--wave", "wave0")
|
||
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertIn("INIT_REPAIRED", result.stdout)
|
||
self.assertTrue((wave_dir / "rc.json").is_file())
|
||
self.assertTrue((wave_dir / "attachments").is_dir())
|
||
self.assertTrue(feedback.read_text(encoding="utf-8").endswith("kept\n"))
|
||
|
||
def test_init_rejects_unknown_wave_instead_of_creating_a_path(self) -> None:
|
||
"""波次名不得被当成任意相对路径使用。"""
|
||
result = self.run_tool("init", "--root", str(self.root), "--wave", "../wave0")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertFalse((Path(self.temp_dir.name) / "wave0").exists())
|
||
self.assertIn("invalid choice", result.stderr)
|
||
|
||
def test_readiness_fails_closed_when_evidence_is_missing(self) -> None:
|
||
"""仅初始化模板不得被误判为可开波。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("rc.missing", result.stdout)
|
||
self.assertNotIn("GATE_PASS", result.stdout)
|
||
|
||
def test_readiness_accepts_complete_structured_evidence(self) -> None:
|
||
"""完整且互相对账的结构化证据应通过开波门。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||
self.assertIn("GATE_PASS phase=readiness", result.stdout)
|
||
|
||
def test_performance_requires_at_least_30_cold_independent_samples(self) -> None:
|
||
"""冷样本少于 30 个时必须直接失败。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
path = wave_dir / "performance.json"
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
payload["samples"] = payload["samples"][:29]
|
||
self.write_json(path, payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("performance.sample-count", result.stdout)
|
||
|
||
def test_performance_p75_uses_nearest_rank_and_rejects_3000ms(self) -> None:
|
||
"""30 样本的第 23 个值是 P75，恰好 3000ms 不满足严格小于。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
path = wave_dir / "performance.json"
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
for index, sample in enumerate(payload["samples"]):
|
||
sample["durationMs"] = 1000 if index < 22 else 3000
|
||
self.write_json(path, payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("performance.p75", result.stdout)
|
||
|
||
def test_performance_accepts_p75_of_2999ms(self) -> None:
|
||
"""严格阈值内的第 23 个值应通过。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
path = wave_dir / "performance.json"
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
for index, sample in enumerate(payload["samples"]):
|
||
sample["durationMs"] = 1000 if index < 22 else 2999
|
||
self.write_json(path, payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||
|
||
def test_performance_rejects_hot_cache_mixed_into_cold_samples(self) -> None:
|
||
"""冷样本文件中混入热缓存记录时不得计算 P75。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
path = wave_dir / "performance.json"
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
payload["samples"][0]["cache"] = "hot"
|
||
self.write_json(path, payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("performance.cold-only", result.stdout)
|
||
|
||
def test_performance_rejects_non_finite_duration(self) -> None:
|
||
"""JSON NaN 不得绕过非负数和 P75 阈值。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
path = wave_dir / "performance.json"
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
for sample in payload["samples"]:
|
||
sample["durationMs"] = float("nan")
|
||
self.write_json(path, payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("performance.duration", result.stdout)
|
||
|
||
def test_main_chain_requires_distinct_evidence_modalities(self) -> None:
|
||
"""同一个占位附件不得同时冒充截图、网络、日志和数据库证据。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
path = wave_dir / "main-chain.json"
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
for field in ("networkRef", "logRef", "databaseRef"):
|
||
payload["stages"][0][field] = payload["stages"][0]["screenshotRef"]
|
||
self.write_json(path, payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("main-chain.evidence-modalities", result.stdout)
|
||
|
||
def test_telemetry_accepts_exactly_95_percent_and_rejects_below(self) -> None:
|
||
"""完整率 95% 可过，低于 95% 必须失败。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
telemetry = wave_dir / "telemetry.json"
|
||
|
||
pass_result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
self.assertEqual(pass_result.returncode, 0, pass_result.stdout + pass_result.stderr)
|
||
|
||
self.write_json(telemetry, {"schema": "dogfood-telemetry-completeness/1", "expectedCount": 100, "completeCount": 94, "evidenceRef": "attachments/browser-proof.png"})
|
||
fail_result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
self.assertNotEqual(fail_result.returncode, 0)
|
||
self.assertIn("telemetry.completeness", fail_result.stdout)
|
||
|
||
def test_top_level_pass_cannot_forge_restore_evidence(self) -> None:
|
||
"""写一个 PASS 字符串不能替代隔离恢复目标和证据。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
self.write_json(wave_dir / "snapshot.json", {"schema": "dogfood-snapshot/1", "result": "PASS"})
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("snapshot.mysql", result.stdout)
|
||
self.assertIn("snapshot.isolated-restore", result.stdout)
|
||
|
||
def test_evidence_reference_cannot_escape_wave_directory(self) -> None:
|
||
"""即使目录外文件存在，证据引用也不得越过波次根。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
outside = wave_dir.parent / "outside.png"
|
||
outside.write_text("outside", encoding="utf-8")
|
||
payload = json.loads((wave_dir / "release-negatives.json").read_text(encoding="utf-8"))
|
||
payload["checks"][0]["evidenceRef"] = "../outside.png"
|
||
self.write_json(wave_dir / "release-negatives.json", payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("reference.outside-wave", result.stdout)
|
||
|
||
def test_main_chain_artifact_must_match_the_rc_artifact(self) -> None:
|
||
"""主链不能用另一份产物哈希冒充当前 RC 的验收结果。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
path = wave_dir / "main-chain.json"
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
payload["identity"]["artifactSha256"] = "f" * 64
|
||
self.write_json(path, payload)
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "readiness")
|
||
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("main-chain.rc-artifact", result.stdout)
|
||
|
||
def test_close_requires_freeze_and_detects_post_freeze_tampering(self) -> None:
|
||
"""收口门必须验证最近一次 SHA-256 冻结，冻结后改动立即失效。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
|
||
missing = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
self.assertNotEqual(missing.returncode, 0)
|
||
self.assertIn("freeze.missing", missing.stdout)
|
||
|
||
freeze = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22")
|
||
self.assertEqual(freeze.returncode, 0, freeze.stdout + freeze.stderr)
|
||
passed = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
self.assertEqual(passed.returncode, 0, passed.stdout + passed.stderr)
|
||
|
||
with (wave_dir / "feedback.tsv").open("a", encoding="utf-8") as stream:
|
||
stream.write("F-2\t2026-07-22T08:00:00Z\tu-2\tplayer\tbug\tP1\ttampered\tOPEN\tops\tattachments/browser-proof.png\n")
|
||
tampered = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
self.assertNotEqual(tampered.returncode, 0)
|
||
self.assertIn("freeze.hash-mismatch", tampered.stdout)
|
||
|
||
def test_close_rejects_malformed_freeze_without_traceback(self) -> None:
|
||
"""畸形冻结清单应形成可追踪失败，而不是让闸门异常退出。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
freeze_dir = wave_dir / "freezes"
|
||
(freeze_dir / "2026-07-22.sha256.json").write_text("[]\n", encoding="utf-8")
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
|
||
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
|
||
self.assertIn("freeze.schema", result.stdout)
|
||
self.assertNotIn("Traceback", result.stderr)
|
||
|
||
def test_close_rejects_symlink_added_after_freeze_without_traceback(self) -> None:
|
||
"""冻结后加入符号链接时应 fail-closed，并保持稳定退出码。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
self.populate_valid_evidence(wave_dir)
|
||
freeze = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22")
|
||
self.assertEqual(freeze.returncode, 0, freeze.stderr)
|
||
(wave_dir / "attachments" / "late-link").symlink_to(wave_dir / "attachments" / "browser-proof.png")
|
||
|
||
result = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
|
||
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
|
||
self.assertIn("freeze.collection", result.stdout)
|
||
self.assertNotIn("Traceback", result.stderr)
|
||
|
||
def test_wave1_close_requires_three_chained_daily_freezes(self) -> None:
|
||
"""三天 wave1 不得用单日快照代替每日冻结，且历史清单必须受哈希链保护。"""
|
||
wave_dir = self.init_wave("wave1")
|
||
self.populate_valid_evidence(wave_dir)
|
||
first = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22")
|
||
self.assertEqual(first.returncode, 0, first.stderr)
|
||
|
||
too_few = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
self.assertNotEqual(too_few.returncode, 0)
|
||
self.assertIn("freeze.daily-count", too_few.stdout)
|
||
|
||
for day in ("2026-07-23", "2026-07-24"):
|
||
frozen = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", day)
|
||
self.assertEqual(frozen.returncode, 0, frozen.stderr)
|
||
passed = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
self.assertEqual(passed.returncode, 0, passed.stdout + passed.stderr)
|
||
|
||
first_path = wave_dir / "freezes" / "2026-07-22.sha256.json"
|
||
first_path.write_text(first_path.read_text(encoding="utf-8") + " ", encoding="utf-8")
|
||
tampered = self.run_tool("check", "--wave-dir", str(wave_dir), "--phase", "close")
|
||
self.assertNotEqual(tampered.returncode, 0)
|
||
self.assertIn("freeze.chain", tampered.stdout)
|
||
|
||
def test_record_drill_requires_exact_ack_and_is_idempotent(self) -> None:
|
||
"""演练记录必须使用精确 ACK，重复签收同一证据不得制造重复行。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
evidence_ref, _ = self.add_attachment(wave_dir, "stop-drill.txt")
|
||
common_args = (
|
||
"record-drill", "--wave-dir", str(wave_dir), "--action", "stop",
|
||
"--operator", "ops-1", "--performed-at", "2026-07-22T06:00:00Z",
|
||
"--evidence-ref", evidence_ref,
|
||
)
|
||
|
||
refused = self.run_tool(*common_args, "--ack", "yes")
|
||
self.assertNotEqual(refused.returncode, 0)
|
||
self.assertIn("ACK_REFUSED", refused.stderr)
|
||
|
||
accepted = self.run_tool(*common_args, "--ack", "ACK-DOGFOOD-wave0-stop")
|
||
self.assertEqual(accepted.returncode, 0, accepted.stdout + accepted.stderr)
|
||
repeated = self.run_tool(*common_args, "--ack", "ACK-DOGFOOD-wave0-stop")
|
||
self.assertEqual(repeated.returncode, 0, repeated.stdout + repeated.stderr)
|
||
self.assertIn("DRILL_ALREADY_RECORDED", repeated.stdout)
|
||
payload = json.loads((wave_dir / "drills.json").read_text(encoding="utf-8"))
|
||
self.assertEqual(len(payload["drills"]), 1)
|
||
|
||
def test_freeze_is_idempotent_but_refuses_same_day_rewrite(self) -> None:
|
||
"""同一证据可重复冻结，内容变化后不得覆盖同日清单。"""
|
||
wave_dir = self.init_wave("wave0")
|
||
first = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22")
|
||
self.assertEqual(first.returncode, 0, first.stderr)
|
||
repeated = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22")
|
||
self.assertEqual(repeated.returncode, 0, repeated.stderr)
|
||
self.assertIn("FREEZE_ALREADY_EXISTS", repeated.stdout)
|
||
|
||
(wave_dir / "feedback.tsv").write_text("changed\n", encoding="utf-8")
|
||
refused = self.run_tool("freeze", "--wave-dir", str(wave_dir), "--date", "2026-07-22")
|
||
self.assertNotEqual(refused.returncode, 0)
|
||
self.assertIn("FREEZE_REFUSED", refused.stderr)
|
||
|
||
def test_shell_entrypoints_delegate_to_the_single_cli(self) -> None:
|
||
"""值班入口应直接调用同一套实现，避免脚本间口径漂移。"""
|
||
scripts = [
|
||
"init-wave.sh", "check-readiness.sh", "check-close.sh",
|
||
"freeze-evidence.sh", "record-drill.sh",
|
||
]
|
||
for name in scripts:
|
||
result = subprocess.run(
|
||
["bash", str(OPS_DIR / name), "--help"],
|
||
check=False,
|
||
text=True,
|
||
capture_output=True,
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"{name}: {result.stdout}{result.stderr}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|