feat(#66): 版本发布+回滚点（基于 config_store 快照，semver 单调递增，回滚可追溯）
This commit is contained in:
@@ -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()
|
||||
]
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user