# -*- 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()