#!/usr/bin/env -S uv run -q # /// script # requires-python = ">=3.14" # dependencies = [ # "pyyaml>=6.0", # ] # /// import importlib.util import re import sys import unittest from pathlib import Path from types import SimpleNamespace import yaml SCRIPT_PATH = Path(__file__).with_name("run-weekly-test.py") SPEC = importlib.util.spec_from_file_location("run_weekly_test", SCRIPT_PATH) if SPEC is None or SPEC.loader is None: raise RuntimeError(f"Unable to load {SCRIPT_PATH}") weekly_test = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = weekly_test SPEC.loader.exec_module(weekly_test) class RegionNameGenerationTest(unittest.TestCase): def test_suffix_is_stable_six_character_base36(self) -> None: suffix = weekly_test._region_name_suffix(32324763735) self.assertEqual("om7blv", suffix) self.assertRegex(suffix, re.compile(r"^[0-9a-z]{6}$")) self.assertEqual(suffix, weekly_test._region_name_suffix(32324763735)) def test_materialize_region_name_updates_all_lifecycle_fields(self) -> None: config = weekly_test.Config( workflow="deploy-aws-byoc-region.yml", env="staging", logical_region_name="usw2-byoc-wkly", generated_region_prefix="usw2-byoc", params={"aws-region": "us-west-2"}, app_destroy={"cloud": "aws"}, ) weekly_test._materialize_region_name(config, 32324763735) self.assertEqual("usw2-byoc-om7blv", config.logical_region_name) self.assertEqual("usw2-byoc-om7blv", config.label) self.assertEqual("usw2-byoc-om7blv", config.params["region-name"]) self.assertNotIn("logical-region-name", config.params) self.assertEqual("us-west-2", config.params["aws-region"]) self.assertEqual( "aws/byoc/staging-usw2-byoc-om7blv.json", config.state_key ) self.assertEqual("usw2-byoc-om7blv", config.app_destroy["region"]) self.assertEqual( "staging-seed-usw2-byoc-om7blv", config.app_destroy["cluster"], ) def test_all_weekly_configs_use_generated_name(self) -> None: config_path = SCRIPT_PATH.parent.parent / "test-configs.yaml" raw = yaml.safe_load(config_path.read_text()) for item in raw["configs"]: with self.subTest(workflow=item["workflow"]): config = weekly_test.Config( workflow=item["workflow"], env=item["env"], logical_region_name=item["logical-region-name"], params=item["params"], generated_region_prefix=item.get("generated-region-prefix"), app_destroy=item["app_destroy"], ) generated_prefix = ( config.generated_region_prefix or config.logical_region_name ) weekly_test._materialize_region_name(config, 32324763735) weekly_test._validate_derived_region_names(config) expected_name = f"{generated_prefix}-om7blv" self.assertEqual(expected_name, config.label) name_input = weekly_test.WORKFLOW_REGION_NAME_INPUTS[item["workflow"]] self.assertEqual(expected_name, config.params[name_input]) self.assertEqual( [name_input], [ key for key in ("region-name", "logical-region-name") if key in config.params ], ) self.assertEqual(expected_name, config.app_destroy["region"]) self.assertEqual( f"{config.env}-seed-{expected_name}", config.app_destroy["cluster"], ) if item["workflow"] == "deploy-azure-byoc-region.yml": self.assertLessEqual(len(f"{config.env}-{expected_name}"), 29) def test_unknown_workflow_region_name_input_is_rejected(self) -> None: config = weekly_test.Config( workflow="unknown.yml", env="dev", logical_region_name="unknown-wkly", params={}, ) with self.assertRaisesRegex(ValueError, "No region name input mapping"): weekly_test._materialize_region_name(config, 32324763735) def test_invalid_run_id_is_rejected(self) -> None: with self.assertRaisesRegex(ValueError, "run ID must be positive"): weekly_test._region_name_suffix(0) class ActiveRegionLifecycleTest(unittest.TestCase): def test_overlong_aws_name_is_rejected_before_recording_or_execution(self) -> None: class MemoryStore: def load(self, slot): return None def create(self, slot, record): raise AssertionError("must not record an invalid generated name") def delete(self, slot): raise AssertionError("must not delete without executing") config = weekly_test.Config( workflow="deploy-aws-premium-region.yml", env="staging", logical_region_name="us-east-1-wkly", params={"aws-region": "us-east-1"}, ) def execute(materialized_config, *, skip_deploy, ref): self.fail("must not execute an invalid generated name") with self.assertRaisesRegex( ValueError, r"derived EKS IAM role name prefix .* is 43 characters; limit is 38; " r"generated-region-prefix must be at most 9 characters for " r"environment 'staging'", ): weekly_test.run_tracked_config( config, run_id=32324763735, store=MemoryStore(), execute=execute, ) def test_mismatched_active_record_blocks_cleanup_and_new_deployment(self) -> None: active_record = weekly_test.ActiveRegionRecord( slot="staging/usw2-byoc-wkly", run_id=32324763735, config=weekly_test.Config( workflow="deploy-azure-premium-region.yml", env="staging", logical_region_name="usw2-byoc-wkly-om7blv", params={"logical-region-name": "usw2-byoc-wkly-om7blv"}, ), ) class MemoryStore: def load(self, slot): return active_record def create(self, slot, record): self.fail("must not replace an active record") def delete(self, slot): self.fail("must not delete a mismatched record") config = weekly_test.Config( workflow="deploy-aws-byoc-region.yml", env="staging", logical_region_name="usw2-byoc-wkly", params={"aws-region": "us-west-2"}, ) def execute(materialized_config, *, skip_deploy, ref): self.fail("must not execute a mismatched active record") with self.assertRaisesRegex(ValueError, "does not match weekly config"): weekly_test.run_tracked_config( config, run_id=99999999999, store=MemoryStore(), execute=execute, ) def test_s3_store_round_trip_does_not_overwrite_active_record(self) -> None: objects = {} def run_command(command, *, check=True, env=None): if command[1:3] == ["s3", "cp"]: key = command[3].removeprefix("s3://state-bucket/") if key not in objects: return SimpleNamespace( returncode=1, stdout="", stderr="NoSuchKey", ) return SimpleNamespace( returncode=0, stdout=objects[key], stderr="", ) operation = command[2] key = command[command.index("--key") + 1] if operation == "head-object": if key not in objects: return SimpleNamespace( returncode=1, stdout="", stderr="404 Not Found", ) elif operation == "put-object": body_path = Path(command[command.index("--body") + 1]) if ( key in objects and command[command.index("--if-none-match") + 1] == "*" ): raise RuntimeError("precondition failed") objects[key] = body_path.read_text() elif operation == "delete-object": objects.pop(key, None) return SimpleNamespace(returncode=0, stdout="", stderr="") config = weekly_test.Config( workflow="deploy-aws-premium-region.yml", env="staging", logical_region_name="us-east-1-wkly-om7blv", params={"region-name": "us-east-1-wkly-om7blv"}, ) record = weekly_test.ActiveRegionRecord( slot="staging/us-east-1-wkly", run_id=32324763735, config=config, ) store = weekly_test.S3ActiveRegionStore( bucket="state-bucket", aws_env={}, run_command=run_command, ) self.assertFalse(store.state_exists(config)) objects[config.state_key] = "{}" self.assertTrue(store.state_exists(config)) objects.pop(config.state_key) store.create(record.slot, record) with self.assertRaisesRegex(RuntimeError, "precondition failed"): store.create(record.slot, record) self.assertEqual(record, store.load(record.slot)) self.assertEqual( ["weekly-test/active/staging/us-east-1-wkly.json"], list(objects), ) store.delete(record.slot) self.assertIsNone(store.load(record.slot)) def test_active_record_round_trip_preserves_cleanup_inputs(self) -> None: config = weekly_test.Config( workflow="deploy-azure-byoc-region.yml", env="dev", logical_region_name="jpeast-byoc-wkly-om7blv", params={ "logical-region-name": "jpeast-byoc-wkly-om7blv", "azure-region": "japaneast", "customer-id": "test0603", }, app_destroy={ "cloud": "azure", "region": "jpeast-byoc-wkly-om7blv", "cluster": "dev-seed-jpeast-byoc-wkly-om7blv", }, ) record = weekly_test.ActiveRegionRecord( slot="dev/jpeast-byoc-wkly", run_id=32324763735, config=config, ) restored = weekly_test.ActiveRegionRecord.from_json(record.to_json()) self.assertEqual(record, restored) def test_successful_region_is_recorded_before_execution_and_cleared_afterward( self, ) -> None: events = [] class MemoryStore: record = None def load(self, slot): return self.record def create(self, slot, record): self.record = record events.append(("create", record.config.logical_region_name)) def delete(self, slot): self.record = None events.append(("delete", slot)) store = MemoryStore() config = weekly_test.Config( workflow="deploy-aws-byoc-region.yml", env="staging", logical_region_name="usw2-byoc-wkly", generated_region_prefix="usw2-byoc", params={"aws-region": "us-west-2"}, app_destroy={"cloud": "aws"}, ) def execute(materialized_config, *, skip_deploy, ref): self.assertIsNotNone(store.record) self.assertFalse(skip_deploy) events.append(("execute", materialized_config.logical_region_name)) return weekly_test.ConfigResult( config=materialized_config, deploy=weekly_test.StepResult(conclusion="success"), destroy=weekly_test.StepResult(conclusion="success"), ) result = weekly_test.run_tracked_config( config, run_id=32324763735, store=store, execute=execute, ) self.assertTrue(result.passed) self.assertIsNone(store.record) self.assertEqual( [ ("create", "usw2-byoc-om7blv"), ("execute", "usw2-byoc-om7blv"), ("delete", "staging/usw2-byoc-wkly"), ], events, ) def test_absent_terraform_state_clears_record_without_cleanup(self) -> None: active_record = weekly_test.ActiveRegionRecord( slot="staging/usw2-byoc-wkly", run_id=33482042607, config=weekly_test.Config( workflow="deploy-aws-byoc-region.yml", env="staging", logical_region_name="usw2-byoc-wkly-dv0hba", params={"region-name": "usw2-byoc-wkly-dv0hba"}, ), ) class MemoryStore: record = active_record def load(self, slot): return self.record def state_exists(self, config): return False def create(self, slot, record): raise AssertionError("must not replace an active record") def delete(self, slot): self.record = None store = MemoryStore() config = weekly_test.Config( workflow="deploy-aws-byoc-region.yml", env="staging", logical_region_name="usw2-byoc-wkly", generated_region_prefix="usw2-byoc", params={"aws-region": "us-west-2"}, ) def execute(materialized_config, *, skip_deploy, ref): self.fail("must not run cleanup when terraform state is absent") result = weekly_test.run_tracked_config( config, run_id=99999999999, store=store, execute=execute, ) self.assertTrue(result.passed) self.assertEqual("skipped", result.deploy.conclusion) self.assertEqual("success", result.destroy.conclusion) self.assertIsNone(store.record) def test_next_run_cleans_failed_region_without_deploying_a_new_one(self) -> None: class MemoryStore: record = None create_count = 0 def load(self, slot): return self.record def state_exists(self, config): return True def create(self, slot, record): self.record = record self.create_count += 1 def delete(self, slot): self.record = None store = MemoryStore() def config(): return weekly_test.Config( workflow="deploy-aws-byoc-region.yml", env="staging", logical_region_name="usw2-byoc-wkly", generated_region_prefix="usw2-byoc", params={"aws-region": "us-west-2"}, app_destroy={"cloud": "aws"}, ) executions = [] def fail_destroy(materialized_config, *, skip_deploy, ref): executions.append((materialized_config.logical_region_name, skip_deploy)) return weekly_test.ConfigResult( config=materialized_config, deploy=weekly_test.StepResult(conclusion="success"), destroy=weekly_test.StepResult(conclusion="failure"), ) weekly_test.run_tracked_config( config(), run_id=32324763735, store=store, execute=fail_destroy, ) def clean_previous(materialized_config, *, skip_deploy, ref): executions.append((materialized_config.logical_region_name, skip_deploy)) return weekly_test.ConfigResult( config=materialized_config, deploy=weekly_test.StepResult(conclusion="skipped"), destroy=weekly_test.StepResult(conclusion="success"), ) result = weekly_test.run_tracked_config( config(), run_id=99999999999, store=store, execute=clean_previous, ) self.assertTrue(result.passed) self.assertIsNone(store.record) self.assertEqual(1, store.create_count) self.assertEqual( [ ("usw2-byoc-om7blv", False), ("usw2-byoc-om7blv", True), ], executions, ) class WeeklyWorkflowTest(unittest.TestCase): def test_each_region_slot_is_serialized_without_stopping_later_jobs(self) -> None: workflow_path = ( SCRIPT_PATH.parent.parent.parent / ".github/workflows/weekly-test.yml" ) workflow = yaml.safe_load(workflow_path.read_text()) expected_groups = { "test-aws-byoc": "weekly-test-usw2-byoc-wkly", "test-aws-premium": "weekly-test-us-east-1-wkly", "test-alicloud-premium": "weekly-test-ali-apse1-wkly", "test-azure-premium": "weekly-test-japaneast-wkly", "test-azure-byoc": "weekly-test-jpeast-byoc-wkly", } for job_name, group in expected_groups.items(): with self.subTest(job=job_name): job = workflow["jobs"][job_name] self.assertEqual(group, job["concurrency"]["group"]) self.assertFalse(job["concurrency"]["cancel-in-progress"]) if "needs" in job: self.assertIn("always()", job["if"]) if __name__ == "__main__": unittest.main()