from __future__ import annotations import unittest from pathlib import Path WORKFLOW = ( Path(__file__).resolve().parents[1] / ".github/workflows/deploy-test-l0.yml" ).read_text() RUNBOOK = ( Path(__file__).resolve().parents[1] / "docs/test-l0-only-deployment.md" ).read_text() class DeployTest1L0WorkflowTest(unittest.TestCase): def test_exposes_the_four_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("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/standalone_l0_workers.py", 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_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-test1-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-test1-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_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_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 Worker digest change requires stop first", WORKFLOW) self.assertIn('"${CURRENT_L0_RUNNING}" == false', WORKFLOW) self.assertIn('target_mode=bootstrap', WORKFLOW) self.assertIn('"${worker_count}" != "${L0_WORKER_COUNT}" || "${worker_consistent}" != true', WORKFLOW) self.assertIn('"${candidate_source}" != "${source_sha}"', 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_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_test1_l0_container(self) -> None: workflows = Path(__file__).resolve().parents[1] / ".github/workflows" for path in workflows.glob("*.yml"): if path.name == "deploy-test-l0.yml": continue self.assertNotIn("nexuscrm-l0-test1-runtime-api", path.read_text(), path.name) if __name__ == "__main__": unittest.main()