from pathlib import Path import unittest import yaml ROOT = Path(__file__).resolve().parents[1] WORKFLOW = (ROOT / ".github/workflows/deploy-test2-l1-application.yml").read_text() D3_WORKFLOW = (ROOT / ".github/workflows/deploy-test-serviceification.yml").read_text() class DeployTest2L1WorkflowTest(unittest.TestCase): def test_workflow_is_a_standalone_l1_replacement(self) -> None: self.assertIsInstance(yaml.safe_load(WORKFLOW), dict) self.assertIn("operation:", WORKFLOW) self.assertIn("- deploy\n - stop", WORKFLOW) self.assertIn("- observe", WORKFLOW) self.assertIn("sfcrm_app_service_image:", WORKFLOW) self.assertIn("frontend_image:", WORKFLOW) self.assertIn("nexuscrm-l1-test2-sfcrm-app-service", WORKFLOW) self.assertIn("github.ref == 'refs/heads/main'", WORKFLOW) self.assertIn("nexuscrm-sfcrm-appservice@sha256", WORKFLOW) self.assertIn("nexuscrm-frontend-assets@sha256", WORKFLOW) self.assertNotIn("serviceification-bom", WORKFLOW) self.assertIn("serviceification-d3/active.json", WORKFLOW) self.assertIn('! -L "${d3_state}"', WORKFLOW) self.assertIn("/usr/share/nexuscrm/frontend/.", WORKFLOW) self.assertIn("diff --brief --recursive --no-dereference", WORKFLOW) self.assertIn("Test2 frontend release target is not a regular directory", WORKFLOW) self.assertIn('chmod 0755 "${TEST2_ENV_ROOT}/frontend-releases"', WORKFLOW) self.assertIn('chmod 0555 "${frontend_target}"', WORKFLOW) self.assertIn("invalid Test2 L1 state file", WORKFLOW) self.assertIn("os.replace(pending, path)", WORKFLOW) self.assertIn("intentionally retains no prior instance", WORKFLOW) self.assertIn('docker logs --tail 200 "${container}" 2>&1 | redact_logs', WORKFLOW) self.assertIn("", WORKFLOW) self.assertNotIn("deploy-test2-l0.yml", WORKFLOW) self.assertNotIn("manage-test2-apprelease.yml", WORKFLOW) self.assertNotIn("catalog-reconcile", WORKFLOW) def test_stop_operation_quiesces_only_the_owned_appservice(self) -> None: stop = WORKFLOW.split(" - name: Stop L1 AppService\n", 1)[1].split( " - name: Deploy L1 AppService and UI\n", 1, )[0] self.assertIn("if: ${{ inputs.operation == 'stop' }}", stop) self.assertIn("docker container ls --all", stop) self.assertIn('name=^/${L1_CONTAINER}$', stop) self.assertIn('[[ "${matched_containers}" == "${L1_CONTAINER}" ]]', stop) self.assertNotIn("if docker container inspect", stop) self.assertIn('io.nexuscrm.test2.owner', stop) self.assertIn('[[ "${owner}" == l1-application ]]', stop) self.assertIn('docker stop --time 30 "${L1_CONTAINER}"', stop) self.assertIn('shutil.which("lsof")', stop) self.assertIn('shutil.which("ss")', stop) self.assertIn('"sudo", "-n", "lsof"', stop) self.assertIn('"sudo", "-n", "ss"', stop) self.assertIn("Test2 backend port still has a listener", stop) self.assertIn("Cannot prove Test2 backend port listener state", stop) self.assertNotIn("connect_ex", stop) self.assertNotIn("docker rm", stop) self.assertNotIn("frontend-dist", stop) self.assertNotIn("deploy-test-l0", stop) self.assertNotIn("manage-test2-apprelease", stop) def test_registry_and_release_inputs_are_deploy_only(self) -> None: login = WORKFLOW.split(" - name: Log in to release registry\n", 1)[1].split( " - name: Stop L1 AppService\n", 1, )[0] deploy = WORKFLOW.split(" - name: Deploy L1 AppService and UI\n", 1)[1] self.assertIn("if: ${{ inputs.operation == 'deploy' }}", login) self.assertIn("if: ${{ inputs.operation == 'deploy' }}", deploy) self.assertIn("nexuscrm-sfcrm-appservice@sha256", deploy) self.assertIn("nexuscrm-frontend-assets@sha256", deploy) def test_observe_reads_only_the_owned_l1_log(self) -> None: observe = WORKFLOW.split("- name: Collect L1 performance evidence", 1)[1].split( "- name: Upload L1 performance evidence", 1 )[0] self.assertIn("inputs.operation == 'observe'", observe) self.assertIn('[[ "${owner}" == l1-application ]]', observe) self.assertIn('docker logs --since "${PERFORMANCE_SINCE}"', observe) self.assertIn("scripts/redact_bounded_log.py --max-output-bytes 8388608", observe) self.assertIn("performance window must not exceed six hours", observe) self.assertNotIn("docker run", observe) self.assertNotIn("docker stop", observe) self.assertNotIn("docker rm", observe) self.assertIn("retention-days: 7", WORKFLOW) def test_candidate_is_ready_before_the_simple_cutover(self) -> None: candidate_ready = WORKFLOW.index('wait_ready "${candidate}"') stable_remove = WORKFLOW.index('docker rm --force "${L1_CONTAINER}"') frontend_precondition = WORKFLOW.index( "Test2 frontend path is not a standalone release link", ) frontend_next = WORKFLOW.index('ln -s "${frontend_target}" "${frontend_next}"') stable_ready = WORKFLOW.index('wait_ready "${L1_CONTAINER}"') frontend_switch = WORKFLOW.index( 'mv -Tf "${frontend_next}" "${frontend_live}"', ) self.assertLess(candidate_ready, stable_remove) self.assertLess(frontend_precondition, stable_remove) self.assertLess(frontend_next, stable_remove) self.assertLess(stable_ready, frontend_switch) self.assertEqual(1, WORKFLOW.count("sudo nginx -t")) self.assertNotIn("docker rename", WORKFLOW) self.assertNotIn("rollback", WORKFLOW.lower()) self.assertNotIn('rm -rf "${TEST2_ENV_ROOT}/frontend-dist"', WORKFLOW) def test_legacy_d3_no_longer_offers_or_schedules_test2(self) -> None: value = yaml.safe_load(D3_WORKFLOW) trigger = value.get("on", value.get(True)) self.assertEqual(["test1", "test3"], trigger["workflow_dispatch"]["inputs"]["slot"]["options"]) operations = { "replace-sfcrm-image": "replace-sfcrm-image", "refresh-database-from-production": "refresh-database-from-production", "deploy-and-verify": "deploy", } for job, operation in operations.items(): self.assertEqual( "${{ (inputs.slot == 'test1' || inputs.slot == 'test3') && inputs.operation == '" + operation + "' }}", value["jobs"][job]["if"], ) self.assertNotIn("name=^/nexuscrm-l0-test2-", D3_WORKFLOW) self.assertNotIn("name=^/nexuscrm-l1-test2-", D3_WORKFLOW) if __name__ == "__main__": unittest.main()