import base64 import hashlib import io import os import stat import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock import yaml import generate_packed_backup_cronjob as generator ROOT = Path(__file__).resolve().parents[1] GPG_OWNER = "Fixture Owner " def crc24(payload): checksum = 0xB704CE for byte in payload: checksum ^= byte << 16 for _ in range(8): checksum <<= 1 if checksum & 0x1000000: checksum ^= 0x1864CFB return checksum & 0xFFFFFF GPG_PUBLIC_KEY_PACKET = ( b"\x04" + (1_700_000_000).to_bytes(4, "big") + b"\x01" + b"\x00\x09\x01\x01" + b"\x00\x11\x01\x00\x01" ) GPG_FINGERPRINT = hashlib.sha1( b"\x99" + len(GPG_PUBLIC_KEY_PACKET).to_bytes(2, "big") + GPG_PUBLIC_KEY_PACKET ).hexdigest().upper() GPG_PACKETS = ( bytes((0xC6, len(GPG_PUBLIC_KEY_PACKET))) + GPG_PUBLIC_KEY_PACKET + bytes((0xCD, len(GPG_OWNER.encode("utf-8")))) + GPG_OWNER.encode("utf-8") ) GPG_PUBLIC_KEY = ( "-----BEGIN PGP PUBLIC KEY BLOCK-----\n" "\n" + base64.b64encode(GPG_PACKETS).decode("ascii") + "\n=" + base64.b64encode(crc24(GPG_PACKETS).to_bytes(3, "big")).decode("ascii") + "\n-----END PGP PUBLIC KEY BLOCK-----\n" ).encode("ascii") MYSQL_DEFAULTS = ( b"[client]\n" b"host=mysql.example.test\n" b"port=4000\n" b"user=fixture-user\n" b"password=fixture-password\n" b"database=fixture-database\n" ) LARK_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/fixture-token" CRON_JOB_NAME = "on-demand-packed-export" EXECUTION_ID_BUDGET = generator.execution_id_length_budget(CRON_JOB_NAME, 1) def valid_config_payload(): return { "schemaVersion": 1, "cronJob": { "name": "on-demand-packed-export", "namespace": "tidb-serverless", "schedule": "0 1 * * *", "timeZone": "Etc/UTC", "startingDeadlineSeconds": 3600, "ttlSecondsAfterFinished": 123, "successfulJobsHistoryLimit": 3, "failedJobsHistoryLimit": 5, "suspend": False, "kubernetesContextName": "fixture-context", "serviceAccountName": "backup-restore-manus", "controllerImage": "controller-image", "imagePullPolicy": "Always", "nodeGroup": "general", "resources": { "requests": {"cpu": "100m", "memory": "256Mi"}, "limits": {"cpu": "1", "memory": "1Gi"}, }, }, "workflow": { "schemaVersion": 1, "pack": { "shards": 1, "cseCtlImage": "pack-image", "resources": { "cpuRequest": "4", "memoryRequest": "8Gi", }, "generatorArgs": [], "commandArgs": [], }, "export": { "outPrefix": "s3://bucket/shared/export", "cseCtlImage": "export-image", "dumplingImage": "dumpling-image", "runnerImage": "runner-image", "resources": { "cpuRequest": "4", "memoryRequest": "8Gi", }, "generatorArgs": [], "workerArgs": ["--parallel", "4"], "dumplingArgs": ["--threads", "4"], }, "watermark": { "image": "watermark-image", "awsRegion": "us-east-1", "resources": { "cpuRequest": "500m", "memoryRequest": "1Gi", }, "generatorArgs": [], "workerArgs": [], }, }, } class RunIdTests(unittest.TestCase): def test_scheduled_execution_id_and_hash_are_both_preserved(self): job_name = "on-demand-packed-export-1785728400" expected_hash = hashlib.sha256(job_name.encode("utf-8")).hexdigest()[:8] run_id = generator.derive_run_id( job_name, CRON_JOB_NAME, EXECUTION_ID_BUDGET, ) self.assertEqual(run_id, f"cron-1785728400-{expected_hash}") def test_same_controller_job_name_is_a_stable_retry(self): job_name = "on-demand-packed-export-reimport-0731" first = generator.derive_run_id( job_name, CRON_JOB_NAME, EXECUTION_ID_BUDGET, ) second = generator.derive_run_id( job_name, CRON_JOB_NAME, EXECUTION_ID_BUDGET, ) self.assertEqual(first, second) def test_manual_job_strips_the_exact_cron_job_name(self): job_name = "on-demand-packed-export-reimport-0731" run_id = generator.derive_run_id( job_name, CRON_JOB_NAME, EXECUTION_ID_BUDGET, ) self.assertEqual(run_id, "cron-reimport-0731-985f5bad") def test_hash_disambiguates_the_same_execution_id(self): first_cron_job = "first-cron" second_cron_job = "second-cron" first = generator.derive_run_id( "first-cron-1785728400", first_cron_job, generator.execution_id_length_budget(first_cron_job, 1), ) second = generator.derive_run_id( "second-cron-1785728400", second_cron_job, generator.execution_id_length_budget(second_cron_job, 1), ) self.assertNotEqual(first, second) self.assertTrue(first.startswith("cron-1785728400-")) self.assertTrue(second.startswith("cron-1785728400-")) def test_job_name_without_the_exact_cron_job_prefix_is_an_error(self): with self.assertRaisesRegex(generator.GeneratorError, "configured CronJob"): generator.derive_run_id( "other-packed-export-1785728400", CRON_JOB_NAME, EXECUTION_ID_BUDGET, ) def test_missing_or_invalid_execution_id_is_an_error_without_fallback(self): with self.assertRaisesRegex(generator.GeneratorError, "configured CronJob"): generator.derive_run_id( CRON_JOB_NAME, CRON_JOB_NAME, EXECUTION_ID_BUDGET, ) with self.assertRaisesRegex(generator.GeneratorError, "1-14 characters"): generator.derive_run_id( f"{CRON_JOB_NAME}-execution-id-is-too-long", CRON_JOB_NAME, EXECUTION_ID_BUDGET, ) def test_length_budget_covers_scheduled_and_hyphenated_manual_ids(self): self.assertEqual(EXECUTION_ID_BUDGET, 14) self.assertEqual( generator.execution_id_length_budget(CRON_JOB_NAME, 4), 14, ) self.assertEqual( generator.execution_id_length_budget(CRON_JOB_NAME, 10), 13, ) def test_embedded_runtime_program_matches_generator_logic(self): job_name = "on-demand-packed-export-reimport-0731" result = subprocess.run( [ sys.executable, "-c", generator.RUN_ID_PROGRAM, job_name, CRON_JOB_NAME, str(EXECUTION_ID_BUDGET), ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual( result.stdout.strip(), generator.derive_run_id( job_name, CRON_JOB_NAME, EXECUTION_ID_BUDGET, ), ) def test_embedded_runtime_program_errors_instead_of_falling_back(self): result = subprocess.run( [ sys.executable, "-c", generator.RUN_ID_PROGRAM, "other-packed-export-1785728400", CRON_JOB_NAME, str(EXECUTION_ID_BUDGET), ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertIn("configured CronJob", result.stderr) def test_controller_shell_stops_before_workflow_on_invalid_execution_id(self): environment = dict(os.environ) environment["CONTROLLER_JOB_NAME"] = ( "on-demand-packed-export-execution-id-is-too-long" ) environment["CRONJOB_NAME"] = CRON_JOB_NAME environment["MAX_EXECUTION_ID_LENGTH"] = str(EXECUTION_ID_BUDGET) result = subprocess.run( ["bash", "-ec", generator.controller_script(), "controller"], env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("1-14 characters", result.stderr) self.assertNotIn("run-packed-backup-export.py", result.stderr) class CronJobGeneratorTests(unittest.TestCase): def test_generation_summary_emphasizes_values_in_interactive_terminals(self): class InteractiveStderr(io.StringIO): def isatty(self): return True stderr = InteractiveStderr() gpg_public_key = generator.inspect_gpg_public_key(GPG_PUBLIC_KEY) mysql = generator.inspect_mysql_defaults(MYSQL_DEFAULTS) with ( mock.patch.object(generator.sys, "stderr", stderr), mock.patch.dict(os.environ, {"TERM": "xterm-256color"}, clear=True), ): generator.emit_generation_summary(None, gpg_public_key, mysql) summary = stderr.getvalue() self.assertIn("\x1b[2mGPG public key: version=\x1b[0m", summary) self.assertIn(f"\x1b[1m{GPG_FINGERPRINT}\x1b[0m", summary) self.assertIn("\x1b[2mMySQL target: host=\x1b[0m", summary) self.assertIn('\x1b[1m"mysql.example.test"\x1b[0m', summary) self.assertIn("\x1b[1mstdout\x1b[0m", summary) def test_example_config_requires_environment_placeholders_to_be_replaced(self): path = ROOT / "example" / "cronjob-config.example.yaml" payload = yaml.safe_load(path.read_text(encoding="utf-8")) self.assertEqual( set(payload["cronJob"]), {"name", "schedule", "serviceAccountName", "controllerImage"}, ) with self.assertRaisesRegex(generator.GeneratorError, "example placeholder"): generator.load_generator_config(str(path)) def test_full_example_covers_every_config_field_and_parses(self): path = ROOT / "example" / "full-example.yaml" payload = yaml.safe_load(path.read_text(encoding="utf-8")) self.assertEqual( set(payload), {"schemaVersion", "cronJob", "workflow"}, ) self.assertEqual( set(payload["cronJob"]), { "name", "namespace", "schedule", "timeZone", "startingDeadlineSeconds", "ttlSecondsAfterFinished", "successfulJobsHistoryLimit", "failedJobsHistoryLimit", "suspend", "kubernetesContextName", "serviceAccountName", "controllerImage", "imagePullPolicy", "nodeGroup", "resources", }, ) self.assertEqual( set(payload["workflow"]["pack"]), { "shards", "cseCtlImage", "resources", "ebs", "generatorArgs", "commandArgs", }, ) self.assertEqual( set(payload["workflow"]["pack"]["ebs"]), {"request", "storageClass"}, ) self.assertEqual( set(payload["workflow"]["export"]), { "outPrefix", "cseCtlImage", "dumplingImage", "runnerImage", "resources", "excludeMetaNameRegexp", "generatorArgs", "workerArgs", "dumplingArgs", }, ) self.assertEqual( set(payload["workflow"]["watermark"]), { "image", "awsRegion", "resources", "generatorArgs", "workerArgs", }, ) self.assertEqual( { field: payload["cronJob"][field] for field in ( "namespace", "timeZone", "startingDeadlineSeconds", "ttlSecondsAfterFinished", "successfulJobsHistoryLimit", "failedJobsHistoryLimit", "suspend", "kubernetesContextName", "imagePullPolicy", "nodeGroup", ) }, { "namespace": "tidb-serverless", "timeZone": "Etc/UTC", "startingDeadlineSeconds": 3600, "ttlSecondsAfterFinished": 2592000, "successfulJobsHistoryLimit": 3, "failedJobsHistoryLimit": 5, "suspend": False, "kubernetesContextName": "unknown", "imagePullPolicy": "Always", "nodeGroup": "general", }, ) payload["cronJob"]["controllerImage"] = "controller-image" payload["workflow"]["pack"]["cseCtlImage"] = "pack-image" payload["workflow"]["export"].update( { "outPrefix": "s3://bucket/export", "cseCtlImage": "export-cse-image", "dumplingImage": "dumpling-image", } ) payload["workflow"]["watermark"]["image"] = "watermark-image" config = generator.parse_generator_config(payload) self.assertIsNone(config.workflow.pack.ebs_request) self.assertIsNone(config.workflow.pack.ebs_storage_class) self.assertEqual(config.workflow.pack.generator_args, ()) self.assertEqual(config.workflow.pack.command_args, ()) self.assertEqual( config.workflow.export.runner_image, "public.ecr.aws/lambda/python:3.13", ) self.assertIsNone(config.workflow.export.exclude_meta_name_regexp) self.assertEqual(config.workflow.export.generator_args, ()) self.assertEqual(config.workflow.export.worker_args, ()) self.assertEqual(config.workflow.export.dumpling_args, ()) self.assertEqual(config.workflow.watermark.generator_args, ()) self.assertEqual(config.workflow.watermark.worker_args, ()) def test_optional_cron_job_fields_use_defaults(self): payload = valid_config_payload() for field in ( "namespace", "timeZone", "startingDeadlineSeconds", "ttlSecondsAfterFinished", "successfulJobsHistoryLimit", "failedJobsHistoryLimit", "suspend", "kubernetesContextName", "imagePullPolicy", "nodeGroup", "resources", ): del payload["cronJob"][field] config = generator.parse_generator_config(payload) self.assertEqual(config.cron_job.namespace, "tidb-serverless") self.assertEqual(config.cron_job.time_zone, "Etc/UTC") self.assertEqual(config.cron_job.starting_deadline_seconds, 3600) self.assertEqual(config.cron_job.ttl_seconds_after_finished, 2592000) self.assertEqual(config.workflow.ttl_seconds_after_finished, 2592000) self.assertEqual(config.cron_job.successful_jobs_history_limit, 3) self.assertEqual(config.cron_job.failed_jobs_history_limit, 5) self.assertFalse(config.cron_job.suspend) self.assertEqual(config.cron_job.kubernetes_context_name, "unknown") self.assertEqual(config.cron_job.image_pull_policy, "Always") self.assertEqual(config.cron_job.node_group, "general") self.assertIsNone(config.cron_job.resources) documents = generator.build_manifest( config, generator.inspect_gpg_public_key(GPG_PUBLIC_KEY).armored, MYSQL_DEFAULTS, LARK_WEBHOOK, ) cron_job = documents[-1] cron_job_spec = cron_job["spec"] self.assertEqual(cron_job["metadata"]["namespace"], "tidb-serverless") self.assertEqual(cron_job_spec["timeZone"], "Etc/UTC") self.assertEqual(cron_job_spec["startingDeadlineSeconds"], 3600) self.assertEqual(cron_job_spec["successfulJobsHistoryLimit"], 3) self.assertEqual(cron_job_spec["failedJobsHistoryLimit"], 5) self.assertFalse(cron_job_spec["suspend"]) self.assertEqual( cron_job_spec["jobTemplate"]["spec"][ "ttlSecondsAfterFinished" ], 2592000, ) pod_spec = cron_job_spec["jobTemplate"]["spec"]["template"]["spec"] self.assertEqual( pod_spec["nodeSelector"]["serverless.tidbcloud.com/node"], "general", ) container = pod_spec["containers"][0] self.assertEqual(container["imagePullPolicy"], "Always") self.assertNotIn("resources", container) plain_environment = { item["name"]: item["value"] for item in container["env"] if "value" in item } self.assertEqual( plain_environment["KUBERNETES_CONTEXT_NAME"], "unknown", ) def test_cron_job_identity_and_runtime_fields_remain_required(self): for field in ( "name", "schedule", "serviceAccountName", "controllerImage", ): with self.subTest(field=field): payload = valid_config_payload() del payload["cronJob"][field] with self.assertRaisesRegex( generator.GeneratorError, "missing required field", ): generator.parse_generator_config(payload) def test_manifest_embeds_inputs_and_has_no_init_container(self): config = generator.parse_generator_config(valid_config_payload()) gpg = generator.inspect_gpg_public_key(GPG_PUBLIC_KEY) mysql = generator.inspect_mysql_defaults(MYSQL_DEFAULTS) documents = generator.build_manifest( config, gpg.armored, MYSQL_DEFAULTS, LARK_WEBHOOK, ) self.assertEqual( [document["kind"] for document in documents], ["Secret", "Secret", "ConfigMap", "ConfigMap", "CronJob"], ) mysql_secret, lark_secret, gpg_config_map, workflow_config_map, cron_job = ( documents ) self.assertEqual( base64.b64decode(mysql_secret["data"]["mysql.cnf"]), MYSQL_DEFAULTS, ) self.assertEqual( base64.b64decode(lark_secret["data"]["webhook"]).decode("utf-8"), LARK_WEBHOOK, ) self.assertEqual(gpg_config_map["data"]["pubkey.asc"], gpg.armored) self.assertEqual(gpg.keys[0].fingerprint, GPG_FINGERPRINT) self.assertEqual(gpg.keys[0].user_ids, (GPG_OWNER,)) self.assertEqual(mysql.host, "mysql.example.test") self.assertEqual(mysql.port, 4000) embedded_workflow = yaml.safe_load( workflow_config_map["data"]["workflow.yaml"] ) self.assertEqual(embedded_workflow, config.workflow.as_dict()) self.assertEqual(embedded_workflow["nodeGroup"], "general") self.assertEqual(embedded_workflow["ttlSecondsAfterFinished"], 123) self.assertEqual( embedded_workflow["mysql"], { "secretName": "on-demand-packed-export-mysql", "secretKey": "mysql.cnf", }, ) pod_spec = cron_job["spec"]["jobTemplate"]["spec"]["template"]["spec"] self.assertEqual( cron_job["spec"]["jobTemplate"]["spec"][ "ttlSecondsAfterFinished" ], 123, ) self.assertNotIn("initContainers", pod_spec) self.assertEqual(cron_job["spec"]["concurrencyPolicy"], "Forbid") self.assertEqual( pod_spec["nodeSelector"], { "kubernetes.io/arch": "arm64", "serverless.tidbcloud.com/node": "general", }, ) self.assertEqual( pod_spec["tolerations"], [ { "key": "serverless.tidbcloud.com/node", "operator": "Equal", "value": "general", "effect": "NoSchedule", } ], ) gpg_volume = next( volume for volume in pod_spec["volumes"] if volume["name"] == "gpg-public-key" ) self.assertEqual( gpg_volume["configMap"]["name"], "on-demand-packed-export-gpg-public-key", ) container = pod_spec["containers"][0] self.assertEqual( container["resources"], valid_config_payload()["cronJob"]["resources"], ) self.assertNotIn("--tikv-api-pod", container["args"]) self.assertNotIn("--ttl-seconds-after-finished", container["args"]) self.assertIn('prefix = f"{cron_job_name}-"', container["args"][0]) self.assertNotIn("rsplit", container["args"][0]) self.assertNotIn("RUN_HASH", container["args"][0]) plain_environment = { item["name"]: item["value"] for item in container["env"] if "value" in item } self.assertEqual(plain_environment["CRONJOB_NAME"], CRON_JOB_NAME) self.assertEqual(plain_environment["MAX_EXECUTION_ID_LENGTH"], "14") self.assertIn( "exec python3 /app/run_packed_backup_export.py", container["args"][0], ) self.assertNotIn("/app/run-packed-backup-export.py", container["args"][0]) webhook_env = next( item for item in container["env"] if item["name"] == "LARK_BOT_WEBHOOK" ) self.assertEqual( webhook_env["valueFrom"]["secretKeyRef"], {"name": "on-demand-packed-export-lark", "key": "webhook"}, ) serialized = generator.serialize_manifest(documents) parsed_again = tuple(yaml.safe_load_all(serialized)) self.assertEqual(len(parsed_again), 5) parsed_pod_spec = ( parsed_again[-1]["spec"]["jobTemplate"]["spec"]["template"]["spec"] ) self.assertEqual( parsed_pod_spec["containers"][0]["args"][0], generator.controller_script(), ) self.assertNotIn("&id", serialized) self.assertNotIn("*id", serialized) self.assertNotIn("fixture-password", serialized) def test_invalid_sensitive_inputs_do_not_appear_in_errors(self): password = "do-not-log-this-password" with self.assertRaises(generator.GeneratorError) as mysql_error: generator.validate_mysql_defaults( f"password={password}\n".encode("utf-8") ) self.assertNotIn(password, str(mysql_error.exception)) token = "do-not-log-this-webhook-token" with self.assertRaises(generator.GeneratorError) as webhook_error: generator.validate_lark_webhook(f"http://example.test/{token}") self.assertNotIn(token, str(webhook_error.exception)) def test_gpg_checksum_is_verified(self): corrupted = GPG_PUBLIC_KEY.replace(b"=JqgC", b"=AAAA") with self.assertRaisesRegex(generator.GeneratorError, "checksum does not match"): generator.inspect_gpg_public_key(corrupted) def test_mysql_default_port_is_reported(self): defaults = MYSQL_DEFAULTS.replace(b"port=4000\n", b"") mysql = generator.inspect_mysql_defaults(defaults) self.assertEqual(mysql.port, 3306) self.assertFalse(mysql.port_is_explicit) def test_unknown_config_field_is_rejected(self): payload = valid_config_payload() payload["cronJob"]["surprise"] = True with self.assertRaisesRegex(generator.GeneratorError, "unknown field"): generator.parse_generator_config(payload) def test_job_ttl_must_be_a_non_negative_integer(self): for value in (-1, True, "2592000"): with self.subTest(value=value): payload = valid_config_payload() payload["cronJob"]["ttlSecondsAfterFinished"] = value with self.assertRaisesRegex( generator.GeneratorError, "non-negative integer", ): generator.parse_generator_config(payload) def test_removed_cron_job_fields_are_rejected(self): removed_fields = { "concurrencyPolicy": "Allow", "tikvApiPod": "other-pod", "nodeSelector": {"serverless.tidbcloud.com/node": "other"}, "tolerations": [], } for field, value in removed_fields.items(): with self.subTest(field=field): payload = valid_config_payload() payload["cronJob"][field] = value with self.assertRaisesRegex(generator.GeneratorError, "unknown field"): generator.parse_generator_config(payload) def test_mysql_reference_and_stage_node_groups_are_derived(self): for field, value in ( ("mysql", {"secretName": "manual-secret"}), ("gpg", {"configMapName": "manual-key"}), ): with self.subTest(field=field): payload = valid_config_payload() payload["workflow"][field] = value with self.assertRaisesRegex(generator.GeneratorError, "unknown field"): generator.parse_generator_config(payload) for stage in ("pack", "export", "watermark"): with self.subTest(stage=stage): payload = valid_config_payload() payload["workflow"][stage]["nodeGroup"] = "manual-group" with self.assertRaisesRegex( generator.GeneratorError, "derived from cronJob.nodeGroup", ): generator.parse_generator_config(payload) def test_duplicate_generator_config_key_is_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.yaml" path.write_text( "schemaVersion: 1\nschemaVersion: 1\n", encoding="utf-8", ) with self.assertRaisesRegex(generator.GeneratorError, "duplicate key"): generator.load_generator_config(str(path)) def test_namespace_must_match_the_controller_namespace(self): payload = valid_config_payload() payload["cronJob"]["namespace"] = "other-namespace" with self.assertRaisesRegex(generator.GeneratorError, "tidb-serverless"): generator.parse_generator_config(payload) def test_config_must_leave_room_for_scheduled_execution_id(self): payload = valid_config_payload() payload["workflow"]["pack"]["shards"] = 1000 with self.assertRaisesRegex( generator.GeneratorError, "scheduled execution IDs require 10", ): generator.parse_generator_config(payload) def test_cli_writes_manifest_to_stdout_by_default_or_dash(self): for output_arguments in ([], ["--output", "-"]): with self.subTest(output_arguments=output_arguments): with tempfile.TemporaryDirectory() as directory: temporary = Path(directory) config_path = temporary / "config.yaml" gpg_path = temporary / "public.asc" mysql_path = temporary / "mysql.cnf" webhook_path = temporary / "lark-webhook" config_path.write_text( yaml.safe_dump(valid_config_payload(), sort_keys=False), encoding="utf-8", ) gpg_path.write_bytes(GPG_PUBLIC_KEY) mysql_path.write_bytes(MYSQL_DEFAULTS) webhook_path.write_text( LARK_WEBHOOK + "\n", encoding="utf-8", ) command = [ sys.executable, str(ROOT / "generate_packed_backup_cronjob.py"), "--config", str(config_path), "--gpg-public-key", str(gpg_path), "--mysql-defaults-file", str(mysql_path), "--lark-webhook-file", str(webhook_path), *output_arguments, ] result = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual( len(tuple(yaml.safe_load_all(result.stdout))), 5, ) self.assertIn( "wrote 5 Kubernetes resources to stdout", result.stderr, ) self.assertNotIn("\x1b[", result.stderr) def test_force_requires_a_file_output(self): result = subprocess.run( [ sys.executable, str(ROOT / "generate_packed_backup_cronjob.py"), "--config", "unused", "--gpg-public-key", "unused", "--mysql-defaults-file", "unused", "--lark-webhook-env", "UNUSED", "--force", ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertEqual(result.returncode, 2) self.assertEqual(result.stdout, "") self.assertIn("--force requires --output PATH", result.stderr) def test_cli_writes_private_file_and_refuses_implicit_overwrite(self): with tempfile.TemporaryDirectory() as directory: temporary = Path(directory) config_path = temporary / "config.yaml" gpg_path = temporary / "public.asc" mysql_path = temporary / "mysql.cnf" webhook_path = temporary / "lark-webhook" output_path = temporary / "cronjob.yaml" config_path.write_text( yaml.safe_dump(valid_config_payload(), sort_keys=False), encoding="utf-8", ) gpg_path.write_bytes(GPG_PUBLIC_KEY) mysql_path.write_bytes(MYSQL_DEFAULTS) webhook_path.write_text(LARK_WEBHOOK + "\n", encoding="utf-8") command = [ sys.executable, str(ROOT / "generate_packed_backup_cronjob.py"), "--config", str(config_path), "--gpg-public-key", str(gpg_path), "--mysql-defaults-file", str(mysql_path), "--lark-webhook-file", str(webhook_path), "--output", str(output_path), ] first = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertEqual(first.returncode, 0, first.stderr) self.assertEqual(first.stdout, "") self.assertIn(GPG_FINGERPRINT, first.stderr) self.assertIn(GPG_OWNER, first.stderr) self.assertIn('host="mysql.example.test"', first.stderr) self.assertIn("port=4000 (explicit)", first.stderr) self.assertNotIn("fixture-password", first.stderr) self.assertNotIn("fixture-token", first.stderr) self.assertEqual(stat.S_IMODE(output_path.stat().st_mode), 0o600) self.assertEqual(len(tuple(yaml.safe_load_all(output_path.read_text()))), 5) second = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertNotEqual(second.returncode, 0) self.assertIn("--force", second.stderr) forced = subprocess.run( command + ["--force"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) self.assertEqual(forced.returncode, 0, forced.stderr) self.assertEqual(stat.S_IMODE(output_path.stat().st_mode), 0o600) def test_explicit_webhook_environment_source_is_supported(self): environment_name = "PACKED_BACKUP_TEST_LARK_WEBHOOK" previous = os.environ.get(environment_name) os.environ[environment_name] = LARK_WEBHOOK try: self.assertEqual( generator.read_lark_webhook(None, environment_name), LARK_WEBHOOK, ) finally: if previous is None: os.environ.pop(environment_name, None) else: os.environ[environment_name] = previous if __name__ == "__main__": unittest.main()