from __future__ import annotations import json import tempfile import unittest from pathlib import Path from deployments.online.runtime.cleanup_prod_retention import ( CURRENT_SCHEMA, IN_PROGRESS_SCHEMA, cleanup_plan, execute, run_cleanup, ) REPO_ROOT = Path(__file__).resolve().parents[1] class ProdD3RetentionTests(unittest.TestCase): def make_release(self, releases: Path, tag: str) -> Path: root = releases / tag root.mkdir(parents=True) (root / "payload.txt").write_text(tag, encoding="utf-8") return root def write_authority( self, path: Path, *, schema: str, release_tag: str, release_root: Path ) -> None: path.write_text( json.dumps( { "schema": schema, "releaseTag": release_tag, "releaseRoot": str(release_root.resolve()), } ) + "\n", encoding="utf-8", ) def test_retention_uses_current_v2_authority_schemas(self) -> None: self.assertEqual( "nexuscrm.serviceification-d3-prod-current/v2", CURRENT_SCHEMA ) self.assertEqual( "nexuscrm.serviceification-d3-prod-in-progress/v2", IN_PROGRESS_SCHEMA, ) def test_cleanup_protects_current_and_in_progress_releases(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" releases = d3 / "releases" old = self.make_release(releases, "staging-d3-release-20260820") kept = self.make_release(releases, "staging-d3-release-20260821") current = self.make_release(releases, "staging-d3-release-20260822") pending = self.make_release(releases, "staging-d3-release-20260823") (d3 / "current").symlink_to(current) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=current.name, release_root=current, ) self.write_authority( d3 / "in-progress.json", schema=IN_PROGRESS_SCHEMA, release_tag=pending.name, release_root=pending, ) payload = run_cleanup( multi_prod_root=multi_prod, keep_inactive_releases=1, dry_run=False, ) self.assertFalse(old.exists()) self.assertTrue(kept.is_dir()) self.assertTrue(current.is_dir()) self.assertTrue(pending.is_dir()) self.assertEqual(1, payload["deleted_count"]) self.assertEqual(current.name, payload["active_release_tag"]) self.assertEqual(pending.name, payload["in_progress_release_tag"]) def test_dry_run_does_not_delete_planned_release(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" releases = d3 / "releases" release = self.make_release(releases, "staging-d3-release-old") current = self.make_release( releases, "staging-d3-release-current" ) (d3 / "current").symlink_to(current) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=current.name, release_root=current, ) payload = run_cleanup( multi_prod_root=multi_prod, keep_inactive_releases=0, dry_run=True, ) self.assertTrue(release.is_dir()) self.assertEqual(1, payload["planned_delete_count"]) self.assertEqual(0, payload["deleted_count"]) def test_release_directories_without_current_authority_fail_closed(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" release = self.make_release( multi_prod / "d3" / "releases", "staging-d3-release-unowned" ) with self.assertRaisesRegex( RuntimeError, "without durable current authority" ): cleanup_plan(multi_prod, 0) self.assertTrue(release.is_dir()) def test_execute_revalidates_authority_before_each_deletion(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" releases = d3 / "releases" planned = self.make_release(releases, "staging-d3-release-old") current = self.make_release(releases, "staging-d3-release-current") current_link = d3 / "current" current_link.symlink_to(current) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=current.name, release_root=current, ) summary, actions = cleanup_plan(multi_prod, 0) self.assertEqual([planned.resolve()], [action.path for action in actions]) current_link.unlink() current_link.symlink_to(planned) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=planned.name, release_root=planned, ) with self.assertRaisesRegex(RuntimeError, "became protected"): execute( actions, d3_root=Path(str(summary["d3_root"])), releases_root=Path(str(summary["releases_root"])), dry_run=False, ) self.assertTrue(planned.is_dir()) def test_execute_rejects_a_replaced_release_directory(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" releases = d3 / "releases" planned = self.make_release(releases, "staging-d3-release-old") current = self.make_release(releases, "staging-d3-release-current") (d3 / "current").symlink_to(current) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=current.name, release_root=current, ) summary, actions = cleanup_plan(multi_prod, 0) original = releases / "original-old" planned.rename(original) replacement = self.make_release(releases, planned.name) with self.assertRaisesRegex(RuntimeError, "identity changed"): execute( actions, d3_root=Path(str(summary["d3_root"])), releases_root=Path(str(summary["releases_root"])), dry_run=False, ) self.assertTrue(original.is_dir()) self.assertTrue(replacement.is_dir()) def test_current_pointer_must_stay_inside_d3_releases(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" d3.mkdir(parents=True) outside = Path(temp) / "outside" outside.mkdir() (d3 / "current").symlink_to(outside) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=outside.name, release_root=outside, ) with self.assertRaisesRegex(RuntimeError, "escaped Prod D3 releases"): cleanup_plan(multi_prod, 3) def test_current_pointer_and_marker_must_agree(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" releases = d3 / "releases" current = self.make_release(releases, "staging-d3-release-current") other = self.make_release(releases, "staging-d3-release-other") (d3 / "current").symlink_to(current) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=other.name, release_root=other, ) with self.assertRaisesRegex(RuntimeError, "disagrees with current.json"): cleanup_plan(multi_prod, 3) def test_current_marker_without_pointer_fails_closed(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" current = self.make_release( d3 / "releases", "staging-d3-release-current" ) self.write_authority( d3 / "current.json", schema=CURRENT_SCHEMA, release_tag=current.name, release_root=current, ) with self.assertRaisesRegex(RuntimeError, "without current pointer"): cleanup_plan(multi_prod, 3) def test_release_root_symlink_fails_closed(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" releases = multi_prod / "d3" / "releases" releases.mkdir(parents=True) outside = Path(temp) / "outside" outside.mkdir() (releases / "staging-d3-release-symlink").symlink_to(outside) with self.assertRaisesRegex(RuntimeError, "must not be a symlink"): cleanup_plan(multi_prod, 3) def test_releases_directory_symlink_fails_before_traversal(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" d3 = multi_prod / "d3" d3.mkdir(parents=True) outside = Path(temp) / "outside-releases" protected = self.make_release(outside, "must-not-delete") (d3 / "releases").symlink_to(outside) with self.assertRaisesRegex( RuntimeError, "Prod D3 releases root must not be a symlink" ): cleanup_plan(multi_prod, 0) self.assertTrue(protected.is_dir()) def test_d3_root_symlink_fails_before_traversal(self) -> None: with tempfile.TemporaryDirectory() as temp: multi_prod = Path(temp) / "multi_prod" multi_prod.mkdir() outside = Path(temp) / "outside-d3" protected = self.make_release( outside / "releases", "must-not-delete" ) (multi_prod / "d3").symlink_to(outside) with self.assertRaisesRegex( RuntimeError, "Prod D3 root must not be a symlink" ): cleanup_plan(multi_prod, 0) self.assertTrue(protected.is_dir()) def test_multi_prod_root_symlink_fails_before_traversal(self) -> None: with tempfile.TemporaryDirectory() as temp: outside = Path(temp) / "outside-multi-prod" protected = self.make_release( outside / "d3" / "releases", "must-not-delete" ) multi_prod = Path(temp) / "multi_prod" multi_prod.symlink_to(outside) with self.assertRaisesRegex( RuntimeError, "multi_prod root must not be a symlink" ): cleanup_plan(multi_prod, 0) self.assertTrue(protected.is_dir()) def test_negative_retention_window_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as temp: with self.assertRaisesRegex(RuntimeError, "must be non-negative"): cleanup_plan(Path(temp) / "multi_prod", -1) def test_retention_workflow_uses_the_d3_runtime_module(self) -> None: workflow = ( REPO_ROOT / ".github" / "workflows" / "cleanup-runtime-retention.yml" ).read_text(encoding="utf-8") self.assertIn( "python3 deployments/online/runtime/cleanup_prod_retention.py", workflow, ) self.assertNotIn("deploy/online/runtime/cleanup_prod_retention.py", workflow) prod_job = workflow.split(" cleanup-prod-retention:", 1)[1].split( " cleanup-staging-retention:", 1 )[0] self.assertIn("group: prod-release-tidb-boundary", prod_job) self.assertIn("cancel-in-progress: false", prod_job) if __name__ == "__main__": unittest.main()