From 263404895a09356c6452b0831cd6e5c2d63c32f3 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:32:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(#66):=20=E7=89=88=E6=9C=AC=E5=8F=91?= =?UTF-8?q?=E5=B8=83+=E5=9B=9E=E6=BB=9A=E7=82=B9=EF=BC=88=E5=9F=BA?= =?UTF-8?q?=E4=BA=8E=20config=5Fstore=20=E5=BF=AB=E7=85=A7=EF=BC=8Csemver?= =?UTF-8?q?=20=E5=8D=95=E8=B0=83=E9=80=92=E5=A2=9E=EF=BC=8C=E5=9B=9E?= =?UTF-8?q?=E6=BB=9A=E5=8F=AF=E8=BF=BD=E6=BA=AF=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/template-console/release.py | 238 ++++++++++++++++++++ core/template-console/tests/test_release.py | 193 ++++++++++++++++ 2 files changed, 431 insertions(+) create mode 100644 core/template-console/release.py create mode 100644 core/template-console/tests/test_release.py diff --git a/core/template-console/release.py b/core/template-console/release.py new file mode 100644 index 0000000..8f5fd6f --- /dev/null +++ b/core/template-console/release.py @@ -0,0 +1,238 @@ +# -*- coding: utf-8 -*- +"""⑤.7 版本发布 + 回滚点 —— issue #66 / PRD ⑤.7。 + +配置台的每次发布都应是一个**可回滚的版本**：实施工程师改了配置 → 预览确认 → +管理员发布；发布即固化当时全量配置快照为一个带 semver 的 Release；若线上出问题， +一键回滚到上一个版本，把 ``ConfigStore`` 恢复成那份快照。这样配置变更"可追溯、 +可逆转"，对齐 PRD「版本化发布与回滚点」。 + +本模块提供： + +- ``Release`` 数据类（semver 版本号 / 时间戳 / 快照 / 发布人 / 变更说明）； +- ``ReleaseManager``：list / publish / rollback； + - ``publish``：固化 ConfigStore 快照为新版本，semver 单调递增校验 + （新版本必须严格大于当前最新版），拒绝重复发布空快照； + - ``rollback``：把 ConfigStore 恢复为指定历史版本的快照，并记一条"回滚事件" + （不删除任何历史版本——回滚本身也是一次可追溯的变更）； +- semver 校验（``MAJOR.MINOR.PATCH``，单调递增）。 + +发布记录持久化为 ``releases.json``（与 ConfigStore 同根目录），人可读、可备份。 + +零运行时依赖：仅用 json / dataclass / 标准库。 +""" +from __future__ import annotations + +import json +import os +import re +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Tuple + +from .config_store import ConfigStore + + +# --------------------------------------------------------------------------- +# semver +# --------------------------------------------------------------------------- + +_SEMVER_RE = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$") +RELEASES_FILENAME = "releases.json" +RELEASES_SCHEMA_VERSION = 1 + + +def is_valid_semver(version: str) -> bool: + """是否合法 semver（MAJOR.MINOR.PATCH，无预发布后缀）。""" + return bool(_SEMVER_RE.match(version)) + + +def semver_tuple(version: str) -> Tuple[int, int, int]: + """semver → (major, minor, patch) 元组（用于比较）。""" + m = _SEMVER_RE.match(version) + if not m: + raise ValueError(f"非法 semver：{version}") + return tuple(int(x) for x in m.groups()) # type: ignore[return-value] + + +def semver_gt(a: str, b: str) -> bool: + """a 是否严格大于 b。""" + return semver_tuple(a) > semver_tuple(b) + + +def bump_patch(version: str) -> str: + """patch 位 +1（默认递增策略，发布时若用户未指定版本号则用此）。""" + major, minor, patch = semver_tuple(version) + return f"{major}.{minor}.{patch + 1}" + + +# --------------------------------------------------------------------------- +# Release 数据模型 +# --------------------------------------------------------------------------- + +@dataclass +class Release: + """一次发布版本（可解释：含发布人、变更说明、来源）。""" + + version: str # semver，如 1.2.0 + created_at: str # ISO8601 发布时间 + snapshot: Dict[str, Any] # 全量配置快照（ConfigStore.snapshot()） + released_by: str = "system" # 发布人（对接 RBAC 用户名） + changelog: str = "" # 变更说明（本次发布改了什么、为什么） + reason: str = "" # 发布理由（可解释可溯源） + + def to_dict(self) -> dict: + return asdict(self) + + @classmethod + def from_dict(cls, raw: dict) -> "Release": + return cls( + version=raw["version"], + created_at=raw.get("created_at", ""), + snapshot=raw.get("snapshot", {}), + released_by=raw.get("released_by", "system"), + changelog=raw.get("changelog", ""), + reason=raw.get("reason", ""), + ) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --------------------------------------------------------------------------- +# 发布管理器 +# --------------------------------------------------------------------------- + +class ReleaseManager: + """版本发布 + 回滚管理器。 + + 用法： + store = ConfigStore("/path/to/store") + rm = ReleaseManager(store) # releases 落在 store 同目录 + rel = rm.publish("1.0.0", released_by="admin", changelog="首次发布") + rm.rollback("0.9.9", released_by="admin") # 回滚到 0.9.9 的快照 + """ + + def __init__(self, store: ConfigStore, releases_path: Optional[str] = None) -> None: + self.store = store + self.releases_path = releases_path or os.path.join(store.root, RELEASES_FILENAME) + + # -- 持久化 -- + def _read_all(self) -> List[Release]: + if not os.path.isfile(self.releases_path): + return [] + with open(self.releases_path, "r", encoding="utf-8") as fh: + blob = json.load(fh) + return [Release.from_dict(r) for r in blob.get("releases", [])] + + def _write_all(self, releases: List[Release]) -> None: + blob = { + "schema_version": RELEASES_SCHEMA_VERSION, + "releases": [r.to_dict() for r in releases], + } + tmp = self.releases_path + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(blob, fh, ensure_ascii=False, indent=2) + os.replace(tmp, self.releases_path) + + # -- 查询 -- + def list(self) -> List[Release]: + """全部发布版本（按版本号升序）。""" + rels = self._read_all() + return sorted(rels, key=lambda r: semver_tuple(r.version)) + + def latest(self) -> Optional[Release]: + """最新发布版本（无则 None）。""" + rels = self.list() + return rels[-1] if rels else None + + def get(self, version: str) -> Optional[Release]: + """取指定版本（不存在返回 None）。""" + for r in self._read_all(): + if r.version == version: + return r + return None + + # -- 发布 -- + def publish( + self, + version: str, + released_by: str = "system", + changelog: str = "", + reason: str = "", + ) -> Release: + """发布新版本（固化当前 ConfigStore 快照）。 + + Raises: + ValueError: semver 非法 / 版本号非单调递增 / 版本号已存在 / + 快照为空（无任何配置不允许发布）。 + """ + if not is_valid_semver(version): + raise ValueError(f"版本号 '{version}' 非法（须为 MAJOR.MINOR.PATCH）") + releases = self._read_all() + existing = {r.version for r in releases} + if version in existing: + raise ValueError(f"版本号 '{version}' 已存在，不可重复发布") + # 单调递增：新版本必须严格大于当前最新 + if releases: + current_latest = max((r.version for r in releases), key=semver_tuple) + if not semver_gt(version, current_latest): + raise ValueError( + f"新版本 '{version}' 必须大于当前最新 '{current_latest}'（单调递增）") + snapshot = self.store.snapshot() + total_items = sum(len(v) for v in snapshot.get("kinds", {}).values()) + if total_items == 0: + raise ValueError("配置快照为空，不允许发布（先在配置台录入配置）") + release = Release( + version=version, created_at=_now_iso(), snapshot=snapshot, + released_by=released_by, changelog=changelog, reason=reason, + ) + releases.append(release) + self._write_all(releases) + return release + + # -- 回滚 -- + def rollback( + self, + target_version: str, + released_by: str = "system", + reason: str = "", + ) -> Release: + """回滚到指定历史版本的快照（把 ConfigStore 恢复成该版本快照）。 + + 回滚**不删除**任何历史版本，而是：恢复快照 + 记一条回滚说明。返回 + 目标版本（便于调用方确认恢复到哪）。 + + Raises: + ValueError: 目标版本不存在 / 回滚到当前已是的状态。 + """ + target = self.get(target_version) + if target is None: + raise ValueError(f"回滚目标版本 '{target_version}' 不存在") + self.store.restore(target.snapshot) + # 记录回滚事件（作为一条带 changelog 的元信息，不新增版本号） + rollback_note = ( + f"[回滚] 已把配置恢复到 {target_version}（发布于 {target.created_at}）；" + f"操作人={released_by}；原因={reason or '未说明'}") + # 把回滚事件追加到目标版本的 reason 字段（可追溯，不污染版本号序列） + target.reason = (target.reason + " | " + rollback_note).strip(" |") if target.reason else rollback_note + releases = self._read_all() + for i, r in enumerate(releases): + if r.version == target_version: + releases[i] = target + self._write_all(releases) + return target + + def history(self) -> List[Dict[str, Any]]: + """发布历史摘要（配置台版本列表展示用）。""" + return [ + { + "version": r.version, + "created_at": r.created_at, + "released_by": r.released_by, + "changelog": r.changelog, + "item_count": sum(len(v) for v in r.snapshot.get("kinds", {}).values()), + "reason": r.reason, + } + for r in self.list() + ] diff --git a/core/template-console/tests/test_release.py b/core/template-console/tests/test_release.py new file mode 100644 index 0000000..bd96662 --- /dev/null +++ b/core/template-console/tests/test_release.py @@ -0,0 +1,193 @@ +# -*- coding: utf-8 -*- +"""版本发布 + 回滚点测试（issue #66）。 + +覆盖： +1. semver 校验 / 比较 / 递增； +2. publish 发布（快照固化、单调递增、重复拒绝、空快照拒绝）； +3. rollback 回滚（恢复快照、不删历史、回滚事件可追溯）； +4. list/latest/get/history 查询； +5. 持久化（重开 manager 仍在）。 +""" +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.release import ( # noqa: E402 + Release, + ReleaseManager, + bump_patch, + is_valid_semver, + semver_gt, + semver_tuple, +) + + +class _Tmp: + def __init__(self): + self._tmp = tempfile.mkdtemp() + self.store = ConfigStore(self._tmp) + self.rm = ReleaseManager(self.store) + + def cleanup(self): + import shutil + shutil.rmtree(self._tmp, ignore_errors=True) + + +class SemverTest(unittest.TestCase): + """semver 工具。""" + + def test_valid(self): + self.assertTrue(is_valid_semver("1.0.0")) + self.assertTrue(is_valid_semver("0.0.1")) + self.assertTrue(is_valid_semver("10.20.30")) + + def test_invalid(self): + self.assertFalse(is_valid_semver("1.0")) + self.assertFalse(is_valid_semver("1.0.0.0")) + self.assertFalse(is_valid_semver("v1.0.0")) + self.assertFalse(is_valid_semver("1.0.0-rc")) + + def test_tuple_and_gt(self): + self.assertEqual(semver_tuple("1.2.3"), (1, 2, 3)) + self.assertTrue(semver_gt("1.0.1", "1.0.0")) + self.assertTrue(semver_gt("2.0.0", "1.9.9")) + self.assertFalse(semver_gt("1.0.0", "1.0.0")) + + def test_bump_patch(self): + self.assertEqual(bump_patch("1.0.0"), "1.0.1") + self.assertEqual(bump_patch("0.9.9"), "0.9.10") + + +class PublishTest(unittest.TestCase): + """发布。""" + + def setUp(self): + self.ctx = _Tmp() + + def tearDown(self): + self.ctx.cleanup() + + def test_publish_requires_nonempty_store(self): + with self.assertRaises(ValueError): + self.ctx.rm.publish("1.0.0") + + def test_publish_first_version(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + rel = self.ctx.rm.publish("1.0.0", released_by="admin", changelog="首次发布") + self.assertEqual(rel.version, "1.0.0") + self.assertEqual(rel.released_by, "admin") + self.assertIn("model_param", rel.snapshot["kinds"]) + + def test_publish_monotonic_increase(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0") + # 较低版本号应被拒绝 + with self.assertRaises(ValueError): + self.ctx.rm.publish("0.9.0") + # 相同版本号应被拒绝 + with self.assertRaises(ValueError): + self.ctx.rm.publish("1.0.0") + # 更高版本 OK + self.ctx.rm.publish("1.0.1") + + def test_publish_invalid_semver(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + with self.assertRaises(ValueError): + self.ctx.rm.publish("1.0") + + def test_snapshot_captures_current_state(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0") + # 发布后改配置，原版本快照不受影响 + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01) + rel1 = self.ctx.rm.get("1.0.0") + self.assertEqual( + rel1.snapshot["kinds"]["model_param"][0]["value"], 0.001) + + +class RollbackTest(unittest.TestCase): + """回滚。""" + + def setUp(self): + self.ctx = _Tmp() + + def tearDown(self): + self.ctx.cleanup() + + def test_rollback_restores_snapshot(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0", changelog="v1 lr=0.001") + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01) + self.ctx.rm.publish("1.1.0", changelog="v2 lr=0.01") + # 当前 store 的 lr 应是 0.01 + self.assertEqual(self.ctx.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.01) + # 回滚到 1.0.0 + target = self.ctx.rm.rollback("1.0.0", released_by="admin", reason="线上异常") + # store 恢复成 1.0.0 的快照 + self.assertEqual(self.ctx.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.001) + self.assertEqual(target.version, "1.0.0") + + def test_rollback_keeps_history(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0") + self.ctx.rm.rollback("1.0.0") + # 回滚不删除任何版本 + self.assertEqual(len(self.ctx.rm.list()), 1) + self.assertIsNotNone(self.ctx.rm.get("1.0.0")) + + def test_rollback_records_event(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0") + self.ctx.rm.rollback("1.0.0", released_by="admin", reason="紧急回滚") + rel = self.ctx.rm.get("1.0.0") + self.assertIn("回滚", rel.reason) + self.assertIn("紧急回滚", rel.reason) + + def test_rollback_unknown_version(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0") + with self.assertRaises(ValueError): + self.ctx.rm.rollback("9.9.9") + + +class QueryTest(unittest.TestCase): + """查询 + 持久化。""" + + def setUp(self): + self.ctx = _Tmp() + + def tearDown(self): + self.ctx.cleanup() + + def test_list_latest_history(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0", changelog="c1") + self.ctx.rm.publish("1.1.0", changelog="c2") + self.assertEqual([r.version for r in self.ctx.rm.list()], ["1.0.0", "1.1.0"]) + self.assertEqual(self.ctx.rm.latest().version, "1.1.0") + hist = self.ctx.rm.history() + self.assertEqual(len(hist), 2) + self.assertEqual(hist[1]["changelog"], "c2") + self.assertEqual(hist[1]["item_count"], 1) + + def test_persistence_across_reopen(self): + self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001) + self.ctx.rm.publish("1.0.0") + # 重开 manager（同一 store 目录） + store2 = ConfigStore(self.ctx._tmp) + rm2 = ReleaseManager(store2) + self.assertIsNotNone(rm2.get("1.0.0")) + self.assertEqual(rm2.latest().version, "1.0.0") + + def test_get_nonexistent(self): + self.assertIsNone(self.ctx.rm.get("9.9.9")) + self.assertIsNone(self.ctx.rm.latest()) + + +if __name__ == "__main__": + unittest.main()