from __future__ import annotations import unittest import os import subprocess import sys import tempfile import textwrap from pathlib import Path WORKFLOW = ( Path(__file__).resolve().parents[1] / ".github/workflows/deploy-test2-l0.yml" ).read_text() RUNBOOK = ( Path(__file__).resolve().parents[1] / "docs/test2-l0-only-deployment.md" ).read_text() class DeployTest2L0WorkflowTest(unittest.TestCase): def test_exposes_the_l0_management_operations(self) -> None: self.assertIn("operation:", WORKFLOW) self.assertIn("- stop", WORKFLOW) self.assertIn("- restart", WORKFLOW) self.assertIn("- activate", WORKFLOW) self.assertIn("- upgrade", WORKFLOW) self.assertIn("- observe", WORKFLOW) self.assertIn("docker restart", WORKFLOW) self.assertIn("L0 is not deployed; run upgrade first", WORKFLOW) def test_stop_is_direct_idempotent_and_does_not_change_image_or_state(self) -> None: stop = WORKFLOW.split("- name: Stop L0", 1)[1].split( "- name: Replace L0 process", 1 )[0] self.assertIn('docker stop "${L0_CONTAINER}"', stop) self.assertNotIn("docker pull", stop) self.assertNotIn("docker rm", stop) self.assertNotIn("docker run", stop) self.assertIn("L0_API_ABSENT=true", WORKFLOW) self.assertIn('"${REQUESTED_OPERATION}" != stop', WORKFLOW) verify = WORKFLOW.split("- name: Verify L0 stopped", 1)[1].split( "- name: Report result", 1 )[0] self.assertIn('docker inspect --format \'{{.State.Running}}\' "${worker}"', verify) def test_is_only_the_standalone_l0_path(self) -> None: self.assertIn("docker run", WORKFLOW) self.assertIn("--detach", WORKFLOW) self.assertIn('"${RUNTIME_API_IMAGE}" bootstrap', WORKFLOW) self.assertNotIn("infra/runtime/deploy.py", WORKFLOW) self.assertNotIn("docker compose", WORKFLOW) self.assertNotIn("database-env-file", WORKFLOW) self.assertNotIn("active_run_id", WORKFLOW) self.assertNotIn("resolved_bom", WORKFLOW.lower()) self.assertNotIn("apprelease", WORKFLOW.lower()) self.assertNotIn("apply-pack", WORKFLOW.lower()) self.assertIn("RUNTIME_WORKER_IMAGE", WORKFLOW) self.assertIn("python3 scripts/test2_standalone.py workers", WORKFLOW) self.assertIn('for stem in ${L0_WORKER_STEMS}; do', WORKFLOW) self.assertNotIn("INTEGRATION_SERVICE_IMAGE", WORKFLOW) def test_selects_main_or_an_exact_published_commit(self) -> None: self.assertIn("l0_source", WORKFLOW) self.assertIn("inputs.operation == 'upgrade'", WORKFLOW) self.assertIn('release_tag="${L0_RELEASE_REPOSITORY}:${source_sha}"', WORKFLOW) self.assertIn("RUNTIME_API_IMAGE=${runtime_image}", WORKFLOW) self.assertIn("l0-engine-release", WORKFLOW) self.assertIn(".RepoDigests", WORKFLOW) self.assertIn('${RUNNER_TEMP}/l0-release.json', WORKFLOW) self.assertNotIn('${L0_STATE_ROOT}/config/l0-release.json.pending', WORKFLOW) def test_missing_containers_cannot_activate_prepared_installation(self) -> None: step = WORKFLOW.split("- name: Inspect current L0", 1)[1].split("- name: Checkout selected L0 source", 1)[0] script = textwrap.dedent(step.split("run: |\n", 1)[1]) for predecessor, prepared, expected in ( ("absent", False, "bootstrap"), ("absent", True, "bootstrap"), ("bootstrap", False, "bootstrap"), ("bootstrap", True, "bootstrap"), ("serving", True, "serving"), ): with self.subTest(predecessor=predecessor, prepared=prepared), tempfile.TemporaryDirectory() as directory: root = Path(directory) if prepared: (root / "config/workers").mkdir(parents=True) (root / "config/serving.json").write_text("{}") (root / "config/workers/audit-history-fanout.json").write_text("{}") binary = root / "bin" binary.mkdir() docker = binary / "docker" docker.write_text("#!" + sys.executable + "\n" + textwrap.dedent("""\ import json, os, sys args = sys.argv[1:] image = 'ghcr.io/pingcap-inc/nexuscrm-runtime-api@sha256:' + 'a' * 64 if args[:2] == ['image', 'inspect']: print('b' * 40) elif args[0] == 'inspect': if args[-1] != 'test-api' or os.environ['PREDECESSOR'] == 'absent': sys.exit(1) if '--format' in args: fmt = args[args.index('--format') + 1] print(image if 'Config.Image' in fmt else 'false') else: print(json.dumps([{'Config': {'Cmd': ['bootstrap'] if os.environ['PREDECESSOR'] == 'bootstrap' else None}}])) else: sys.exit('unexpected Docker command') """)) docker.chmod(0o755) output = root / "github-env" env = {**os.environ, "PATH": str(binary) + os.pathsep + os.environ["PATH"], "PREDECESSOR": predecessor, "REQUESTED_OPERATION": "upgrade", "L0_CONTAINER": "test-api", "L0_STATE_ROOT": str(root), "RUNTIME_API_REPOSITORY": "ghcr.io/pingcap-inc/nexuscrm-runtime-api", "L0_WORKER_STEMS": "audit-history-fanout", "L0_WORKER_COUNT": "1", "GITHUB_ENV": str(output)} result = subprocess.run(["bash", "-c", script], env=env, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) selected = dict(line.split("=", 1) for line in output.read_text().splitlines()) self.assertEqual(selected["CURRENT_L0_MODE"], expected) self.assertEqual(selected["CURRENT_L0_RUNNING"], "false") def test_bootstrap_upgrade_also_rebinds_existing_installed_roles(self) -> None: upgrade = WORKFLOW.split("- name: Upgrade installed L0 Worker ownership", 1)[1].split( "- name: Prepare L0 transport", 1 )[0] self.assertIn('${L0_STATE_ROOT}/config/serving.json', upgrade) self.assertIn('${L0_STATE_ROOT}/config/workers', upgrade) self.assertNotIn('"${L0_TARGET_MODE}" == serving', upgrade) self.assertIn('python3 scripts/standalone_l0_release.py', upgrade) def test_bootstrap_keeps_worker_image_identity_in_stopped_containers(self) -> None: replace = WORKFLOW.split("- name: Replace L0 process", 1)[1].split( "- name: Verify L0", 1 )[0] removal_offset = replace.index('docker rm --force "${L0_CONTAINER}"') carrier_offset = replace.index('docker create --name "nexuscrm-l0-test2-worker-${stem}"') self.assertLess(removal_offset, carrier_offset) self.assertIn('echo "RUNTIME_WORKER_IMAGE=${CURRENT_RUNTIME_WORKER_IMAGE}"', WORKFLOW) def test_restart_preserves_the_existing_immutable_image(self) -> None: self.assertIn("docker inspect --format '{{.Config.Image}}'", WORKFLOW) self.assertIn("docker image inspect --format", WORKFLOW) restart = WORKFLOW.split("- name: Restart L0", 1)[1].split( "- name: Replace L0 process", 1 )[0] self.assertNotIn("docker pull", restart) self.assertNotIn("docker rm", restart) self.assertNotIn("docker run", restart) def test_activate_keeps_the_current_image_and_uses_the_l0_state_root(self) -> None: self.assertIn('echo "RUNTIME_API_IMAGE=${CURRENT_RUNTIME_API_IMAGE}"', WORKFLOW) self.assertIn('${L0_STATE_ROOT}/config/serving.json', WORKFLOW) self.assertIn('${L0_STATE_ROOT}/config/host-compatibility.json', WORKFLOW) self.assertIn('nexuscrm.runtime-host-compatibility/v1', WORKFLOW) self.assertIn('${L0_STATE_ROOT}/data/catalog', WORKFLOW) self.assertIn('--env FILE_STORAGE_ROOT=/var/lib/nexuscrm/files', WORKFLOW) self.assertIn('--env FILE_STORAGE_LEGACY_UPLOAD_ROOT=/var/lib/nexuscrm/uploads', WORKFLOW) self.assertNotIn("activation_digest", WORKFLOW) self.assertNotIn("L0_ACTIVATION_ROOT", WORKFLOW) self.assertIn(".worker_release_digest", WORKFLOW) self.assertIn(".worker_source_revision", WORKFLOW) self.assertIn("does not match the selected Worker release", WORKFLOW) def test_serving_never_starts_only_half_of_l0(self) -> None: replace = WORKFLOW.split("- name: Replace L0 process", 1)[1].split( "- name: Verify L0", 1 )[0] self.assertIn('docker run "${worker_args[@]}" "${RUNTIME_WORKER_IMAGE}"', replace) self.assertIn('"${RUNTIME_API_IMAGE}" >/dev/null', replace) self.assertIn("runtime.bulk-ingest.v1".removeprefix("runtime.").removesuffix(".v1"), replace) self.assertIn( '"${stem}" == file-upload-lifecycle || "${stem}" == bulk-ingest', replace, ) self.assertIn("trap cleanup_incomplete_l0 EXIT INT TERM", replace) self.assertIn('wait_healthy "${L0_CONTAINER}"', replace) self.assertIn('wait_healthy "nexuscrm-l0-test2-worker-${stem}"', replace) self.assertIn("replacement_ready=true", replace) self.assertNotIn("trap - EXIT INT TERM", replace) self.assertIn("Remove incomplete L0 replacement", WORKFLOW) def test_serving_starts_api_before_workers_and_allows_durable_recovery(self) -> None: replace = WORKFLOW.split("- name: Replace L0 process", 1)[1].split( "- name: Verify L0", 1 )[0] serving = replace.split('[[ "${L0_TARGET_MODE}" == serving ]]', 1)[1] api_start = serving.index('"${RUNTIME_API_IMAGE}" >/dev/null') api_ready = serving.index('wait_healthy "${L0_CONTAINER}"') worker_start = serving.index( 'docker run "${worker_args[@]}" "${RUNTIME_WORKER_IMAGE}"' ) self.assertLess(api_start, worker_start) self.assertLess(worker_start, api_ready) self.assertIn('wait_healthy "${L0_CONTAINER}" 72', serving) def test_workers_override_the_api_database_pool(self) -> None: replace = WORKFLOW.split("- name: Replace L0 process", 1)[1].split( "- name: Verify L0", 1 )[0] self.assertIn("--env NEXUSCRM_DB_MAX_OPEN_CONNS=4", replace) self.assertIn("--env NEXUSCRM_DB_MAX_IDLE_CONNS=2", replace) def test_upgrade_preserves_the_current_process_mode(self) -> None: self.assertIn('echo "L0_TARGET_MODE=${CURRENT_L0_MODE}"', WORKFLOW) self.assertIn("A failed first activation never established a serving closure", WORKFLOW) self.assertIn("a serving L0 upgrade requires stop first", WORKFLOW) self.assertIn('"${CURRENT_L0_RUNNING}" != false', WORKFLOW) self.assertIn('python3 scripts/standalone_l0_release.py', WORKFLOW) upgrade = WORKFLOW.index('- name: Upgrade installed L0 Worker ownership') check = WORKFLOW.index('- name: Check serving inputs') self.assertLess(check, upgrade) self.assertNotIn('target_mode=bootstrap', WORKFLOW) self.assertIn('"${worker_count}" != "${L0_WORKER_COUNT}" || "${worker_consistent}" != true', WORKFLOW) self.assertIn('Worker-set change requires stop before upgrade', WORKFLOW) self.assertIn('"${candidate_source}" != "${source_sha}"', WORKFLOW) def test_layout_worker_is_retired_and_only_cleaned_up(self) -> None: replace = WORKFLOW.split("- name: Replace L0 process", 1)[1].split( "- name: Verify L0", 1 )[0] self.assertNotIn('"${stem}" == layout-authoring', replace) self.assertNotIn('NEXUSCRM_RUNTIME_WORKER_CATALOG_RECONCILE_CONFIG_FILE', replace) self.assertIn('catalog:/var/lib/nexuscrm/catalog:ro', replace) self.assertNotIn('catalog:/var/lib/nexuscrm/catalog"', replace) check = WORKFLOW.split("- name: Check serving inputs", 1)[1].split( "- name: Restart L0", 1 )[0] self.assertNotIn('layout-authoring', check) self.assertIn('L0_ALL_WORKER_STEMS=${all_workers} layout-authoring', WORKFLOW) def test_has_no_rolling_backup_or_rollback_state(self) -> None: self.assertNotIn("docker rename", WORKFLOW) self.assertNotIn("previous_container", WORKFLOW) self.assertNotIn("rollback", WORKFLOW.lower()) self.assertEqual(WORKFLOW.count('--name "${L0_CONTAINER}"'), 1) def test_proves_the_bootstrap_contract(self) -> None: self.assertIn("/livez", WORKFLOW) self.assertIn("/readyz", WORKFLOW) self.assertIn("openssl s_client", WORKFLOW) self.assertIn('wc -l)" -eq 1', WORKFLOW) def test_failure_logs_are_bounded_and_redacted(self) -> None: diagnostics = WORKFLOW.split("- name: Diagnose failed L0", 1)[1].split( "- name: Remove incomplete L0 replacement", 1 )[0] self.assertIn("docker logs --tail 200", diagnostics) self.assertIn("docker logs --tail 80", diagnostics) self.assertEqual(2, diagnostics.count("| redact_logs")) self.assertIn("", diagnostics) def test_observe_reads_owned_logs_and_tidb_history_without_mutation(self) -> None: observe = WORKFLOW.split("- name: Collect L0 performance evidence", 1)[1].split( "- name: Upload L0 performance evidence", 1 )[0] self.assertIn("inputs.operation == 'observe'", observe) self.assertIn('docker logs --since "${PERFORMANCE_SINCE}"', observe) self.assertIn("scripts/redact_bounded_log.py --max-output-bytes 8388608", observe) self.assertIn('docker inspect "${INTEGRATION_CONTAINER}"', observe) self.assertIn('"${INTEGRATION_CONTAINER}" 2>&1', observe) self.assertIn("scripts/diagnose_test2_tidb_sql.py", observe) self.assertIn("outbound-queue-snapshot.tsv", observe) self.assertIn("performance window must not exceed six hours", observe) self.assertNotIn("docker run", observe) self.assertNotIn("docker stop", observe) self.assertNotIn("docker restart", observe) self.assertNotIn("docker rm", observe) self.assertIn("retention-days: 7", WORKFLOW) def test_documents_the_application_activation_boundary(self) -> None: self.assertIn("Application Activation interacts with L0 through exactly one command", RUNBOOK) self.assertIn("`restart` remains an independent L0 operator command", RUNBOOK) self.assertIn("must not select, build, pull, or upgrade an L0 image", RUNBOOK) self.assertIn("Only this L0 management CI may", RUNBOOK) def test_no_other_workflow_owns_the_test2_l0_container(self) -> None: workflows = Path(__file__).resolve().parents[1] / ".github/workflows" for path in workflows.glob("*.yml"): if path.name == "deploy-test2-l0.yml": continue self.assertNotIn("nexuscrm-l0-test2-runtime-api", path.read_text(), path.name) if __name__ == "__main__": unittest.main()