feat(#67): 配置台↔内核配置推送契约（JSON manifest+SHA256 校验和+幂等推送/撤回）

2026-08-05 06:05:27 +08:00
parent 263404895a
commit 23b374f836
2 changed files with 478 additions and 0 deletions
@@ -0,0 +1,215 @@
+# -*- coding: utf-8 -*-
+"""配置推送契约测试（issue #67）。
+
+覆盖：
+1. manifest 构建（版本/快照/校验和/来源）；
+2. checksum 稳定性 + 完整性校验（verify）；
+3. 推送幂等（同版本不重复写文件、返回原 manifest）；
+4. force 强制重推；
+5. retract 撤回 + 重新推送；
+6. 模拟传输损坏（篡改 snapshot → verify 失败）；
+7. 推送日志与查询。
+"""
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401
+
+from template_console.config_store import ConfigKind, ConfigStore  # noqa: E402
+from template_console.push_channel import (  # noqa: E402
+    PushChannel,
+    PushManifest,
+    PushRecord,
+    checksum,
+)
+from template_console.release import Release, ReleaseManager  # noqa: E402
+
+
+def _make_release(version: str = "1.0.0") -> Release:
+    """构造一个带快照的 Release（不走文件系统，直接内存构造）。"""
+    snap = {
+        "schema_version": 1,
+        "captured_at": "2026-01-01T00:00:00Z",
+        "kinds": {"model_param": [
+            {"key": "lr", "value": 0.001, "kind": "model_param",
+             "meaning": "学习率", "updated_by": "li", "reason": "init",
+             "updated_at": "2026-01-01T00:00:00Z"}]},
+    }
+    return Release(version=version, created_at="2026-01-01T00:00:00Z",
+                   snapshot=snap, released_by="admin", changelog="t", reason="r")
+
+
+class _Tmp:
+    def __init__(self):
+        self._tmp = tempfile.mkdtemp()
+        self.inbox = os.path.join(self._tmp, "inbox")
+
+    def cleanup(self):
+        import shutil
+        shutil.rmtree(self._tmp, ignore_errors=True)
+
+
+class ChecksumTest(unittest.TestCase):
+    """校验和稳定性 + 完整性。"""
+
+    def test_checksum_stable(self):
+        snap = {"kinds": {"a": [1, 2, 3]}}
+        self.assertEqual(checksum(snap), checksum(snap))
+
+    def test_checksum_key_order_independent(self):
+        # 键序不影响校验和（规范序列化）
+        a = checksum({"x": 1, "y": 2})
+        b = checksum({"y": 2, "x": 1})
+        self.assertEqual(a, b)
+
+    def test_checksum_changes_on_value_change(self):
+        self.assertNotEqual(checksum({"v": 1}), checksum({"v": 2}))
+
+    def test_checksum_is_sha256_hex(self):
+        cs = checksum({"v": 1})
+        self.assertEqual(len(cs), 64)
+        self.assertTrue(all(c in "0123456789abcdef" for c in cs))
+
+
+class ManifestTest(unittest.TestCase):
+    """manifest 构建。"""
+
+    def test_build_manifest_has_checksum(self):
+        ch = PushChannel(inbox=tempfile.mkdtemp())
+        rel = _make_release()
+        m = ch.build_manifest(rel)
+        self.assertEqual(m.version, "1.0.0")
+        self.assertTrue(m.checksum)
+        self.assertEqual(m.source, "template-console")
+        self.assertTrue(m.generated_at)
+
+    def test_manifest_roundtrip(self):
+        m = PushManifest(version="1.0.0", snapshot={"a": 1},
+                         checksum=checksum({"a": 1}), generated_at="t")
+        text = m.to_json()
+        m2 = PushManifest.from_dict(json.loads(text))
+        self.assertEqual(m2.version, "1.0.0")
+        self.assertEqual(m2.checksum, m.checksum)
+
+
+class PushIdempotencyTest(unittest.TestCase):
+    """推送幂等。"""
+
+    def setUp(self):
+        self.ctx = _Tmp()
+        self.ch = PushChannel(inbox=self.ctx.inbox)
+        self.rel = _make_release()
+
+    def tearDown(self):
+        self.ctx.cleanup()
+
+    def test_push_writes_manifest_file(self):
+        self.ch.push(self.rel, pushed_by="admin")
+        path = os.path.join(self.ctx.inbox, "manifest-1.0.0.json")
+        self.assertTrue(os.path.isfile(path))
+
+    def test_push_is_idempotent(self):
+        m1 = self.ch.push(self.rel, pushed_by="a")
+        m2 = self.ch.push(self.rel, pushed_by="b")  # 重复推送
+        # 同版本返回同一 manifest（校验和一致）
+        self.assertEqual(m1.checksum, m2.checksum)
+        # 推送日志只有一条记录
+        self.assertEqual(len(self.ch.pushed_versions()), 1)
+
+    def test_force_overrides_idempotency(self):
+        self.ch.push(self.rel, pushed_by="a")
+        before = self.ch.pushed_versions()[0]["pushed_at"]
+        # force 重推（时间戳可能更新）
+        self.ch.push(self.rel, pushed_by="b", force=True)
+        records = self.ch.pushed_versions()
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["pushed_by"], "b")
+
+    def test_push_log_records_pushed_by(self):
+        self.ch.push(self.rel, pushed_by="admin_zhang")
+        rec = self.ch.pushed_versions()[0]
+        self.assertEqual(rec["pushed_by"], "admin_zhang")
+        self.assertEqual(rec["status"], "pushed")
+
+
+class RetractTest(unittest.TestCase):
+    """撤回 + 重新推送。"""
+
+    def setUp(self):
+        self.ctx = _Tmp()
+        self.ch = PushChannel(inbox=self.ctx.inbox)
+        self.rel = _make_release()
+
+    def tearDown(self):
+        self.ctx.cleanup()
+
+    def test_retract_marks_status(self):
+        self.ch.push(self.rel)
+        self.assertTrue(self.ch.retract("1.0.0", by="admin", reason="有问题"))
+        rec = self.ch.pushed_versions()[0]
+        self.assertEqual(rec["status"], "retracted")
+
+    def test_retract_unknown_returns_false(self):
+        self.assertFalse(self.ch.retract("9.9.9"))
+
+    def test_retract_allows_repush(self):
+        self.ch.push(self.rel)
+        self.ch.retract("1.0.0")
+        # 撤回后可重新推送（幂等解除）
+        m = self.ch.push(self.rel, pushed_by="admin2")
+        rec = self.ch.pushed_versions()[0]
+        self.assertEqual(rec["status"], "pushed")
+        self.assertEqual(rec["pushed_by"], "admin2")
+
+
+class VerifyTest(unittest.TestCase):
+    """完整性校验。"""
+
+    def test_verify_valid_manifest(self):
+        ch = PushChannel(inbox=tempfile.mkdtemp())
+        m = ch.build_manifest(_make_release())
+        self.assertTrue(PushChannel.verify(m))
+
+    def test_verify_tampered_snapshot_fails(self):
+        ch = PushChannel(inbox=tempfile.mkdtemp())
+        m = ch.build_manifest(_make_release())
+        # 篡改 snapshot 但不改 checksum → 校验失败
+        m.snapshot["kinds"]["model_param"][0]["value"] = 0.999
+        self.assertFalse(PushChannel.verify(m))
+
+    def test_verify_payload_dict(self):
+        ch = PushChannel(inbox=tempfile.mkdtemp())
+        m = ch.build_manifest(_make_release())
+        self.assertTrue(PushChannel.verify_payload(m.to_dict()))
+
+    def test_verify_payload_bad_dict(self):
+        self.assertFalse(PushChannel.verify_payload({"nope": 1}))
+
+
+class IntegrationTest(unittest.TestCase):
+    """端到端：store → publish → push → verify。"""
+
+    def test_store_publish_push_flow(self):
+        tmp = tempfile.mkdtemp()
+        try:
+            store = ConfigStore(tmp)
+            store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
+            rm = ReleaseManager(store)
+            rel = rm.publish("1.0.0", released_by="admin", changelog="首发")
+            inbox = os.path.join(tmp, "inbox")
+            ch = PushChannel(inbox=inbox)
+            m = ch.push(rel, pushed_by="admin")
+            # 内核侧校验通过
+            self.assertTrue(PushChannel.verify(m))
+            self.assertEqual(len(ch.pushed_versions()), 1)
+        finally:
+            import shutil
+            shutil.rmtree(tmp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    unittest.main()