import subprocess from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] def test_teleo_agent_template_supports_optional_per_agent_env_file(): unit = (REPO_ROOT / "systemd" / "teleo-agent@.service").read_text() assert "Environment=PYTHONUNBUFFERED=1" in unit assert "EnvironmentFile=-/opt/teleo-eval/secrets/teleo-agent-%i.env" in unit def test_teleo_agent_template_allows_transcript_dumps_under_strict_fs(): unit = (REPO_ROOT / "systemd" / "teleo-agent@.service").read_text() assert "ProtectSystem=strict" in unit assert "ReadWritePaths=/opt/teleo-eval/transcripts" in unit def test_deploy_scripts_restart_wallet_test_agent_when_present(): auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text() manual_deploy = (REPO_ROOT / "deploy" / "deploy.sh").read_text() assert "add_restart_if_unit_exists teleo-agent@leo-wallet-test" in auto_deploy assert 'systemctl list-units --all --full "$1.service"' in auto_deploy assert "teleo-agent@leo-wallet-test" in manual_deploy def test_deploy_scripts_do_not_start_inactive_legacy_leo_agent(): auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text() manual_deploy = (REPO_ROOT / "deploy" / "deploy.sh").read_text() assert "add_restart_if_unit_active teleo-agent@leo" in auto_deploy assert "add_restart teleo-agent@leo" not in auto_deploy assert "systemctl is-active --quiet teleo-agent@leo.service" in manual_deploy assert "teleo-pipeline teleo-diagnostics teleo-agent@leo" not in manual_deploy def test_auto_deploy_prefers_github_remote_when_present(): auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text() assert 'DEPLOY_REMOTE="${TELEO_DEPLOY_REMOTE:-}"' in auto_deploy assert "remote get-url github" in auto_deploy assert 'git fetch "$DEPLOY_REMOTE" main' in auto_deploy assert 'git rev-parse "$DEPLOY_REMOTE/main"' in auto_deploy assert 'git merge --ff-only "$DEPLOY_REMOTE/main"' in auto_deploy def test_auto_deploy_executable_safety_net_stays_inside_synced_dirs(): auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text() assert 'for dir in "$PIPELINE_DIR" "$TELEGRAM_DIR" "$DIAGNOSTICS_DIR" "$AGENT_STATE_DIR"' in auto_deploy assert "find /opt/teleo-eval -maxdepth 3 -name '*.sh'" not in auto_deploy assert "backups" in auto_deploy def test_agent_healthcheck_timer_runs_both_live_telegram_agents(): service = (REPO_ROOT / "systemd" / "teleo-agent-healthcheck.service").read_text() timer = (REPO_ROOT / "systemd" / "teleo-agent-healthcheck.timer").read_text() assert "/opt/teleo-eval/telegram/agent_healthcheck.py" in service assert "--agents leo leo-wallet-test" in service assert '--since "20 min ago"' in service assert "OnUnitActiveSec=5min" in timer assert "WantedBy=timers.target" in timer def test_deploy_scripts_install_systemd_units_and_enable_agent_healthcheck(): auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text() manual_deploy = (REPO_ROOT / "deploy" / "deploy.sh").read_text() assert 'SYSTEMD_DIR="/etc/systemd/system"' in auto_deploy assert 'sudo install -m 0644 "$unit" "$SYSTEMD_DIR/$(basename "$unit")"' in auto_deploy assert "sudo systemctl daemon-reload" in auto_deploy assert "sudo systemctl enable --now teleo-agent-healthcheck.timer" in auto_deploy assert 'git diff --name-only "$OLD_SHA" "$NEW_SHA" -- systemd/teleo-agent@.service' in auto_deploy assert 'VPS_SYSTEMD="/etc/systemd/system"' in manual_deploy assert "teleo-agent-healthcheck.timer" in manual_deploy def _leoclean_gateway_restart_required(*changed_paths: str) -> bool: result = subprocess.run( ["bash", str(REPO_ROOT / "deploy" / "leoclean-gateway-restart-required.sh")], input="\n".join(changed_paths), capture_output=True, check=False, text=True, ) assert result.returncode in {0, 1}, result.stderr return result.returncode == 0 def test_deploy_scripts_sync_leoclean_skills_and_restart_gateway_for_runtime_changes(): auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text() manual_deploy = (REPO_ROOT / "deploy" / "deploy.sh").read_text() sudoers = (REPO_ROOT / "deploy" / "sudoers" / "teleo-auto-deploy").read_text() assert 'LEOCLEAN_BIN_DIR="/home/teleo/.hermes/profiles/leoclean/bin"' in auto_deploy assert 'LEOCLEAN_SKILLS_DIR="/home/teleo/.hermes/profiles/leoclean/skills"' in auto_deploy assert 'LEOCLEAN_PLUGINS_DIR="/home/teleo/.hermes/profiles/leoclean/plugins"' in auto_deploy assert "hermes-agent/leoclean-bin/" in auto_deploy assert "hermes-agent/leoclean-skills/vps/" in auto_deploy assert "hermes-agent/leoclean-plugins/vps/" in auto_deploy assert 'HERMES_PATCH_DIR="/home/teleo/.hermes/teleo-runtime-patches"' in auto_deploy assert '"status": "installed_now"' in auto_deploy assert "Hermes response-transform drift repaired" in auto_deploy assert "deploy/leoclean-gateway-restart-required.sh" in auto_deploy assert "TELEO_AUTO_DEPLOY_REEXECED" in auto_deploy assert 'exec bash "$DEPLOY_CHECKOUT/deploy/auto-deploy.sh"' in auto_deploy assert "add_restart_if_unit_active leoclean-gateway" in auto_deploy assert 'sudo systemctl restart "$svc"' in auto_deploy assert "sudo systemctl restart $RESTART" not in auto_deploy assert "NOPASSWD: /usr/bin/systemctl restart leoclean-gateway" in sudoers assert "NOPASSWD: /bin/systemctl restart leoclean-gateway" in sudoers assert 'VPS_LEOCLEAN_BIN="/home/teleo/.hermes/profiles/leoclean/bin"' in manual_deploy assert 'VPS_LEOCLEAN_SKILLS="/home/teleo/.hermes/profiles/leoclean/skills"' in manual_deploy assert 'VPS_LEOCLEAN_PLUGINS="/home/teleo/.hermes/profiles/leoclean/plugins"' in manual_deploy assert "hermes-agent/leoclean-bin/" in manual_deploy assert "hermes-agent/leoclean-skills/vps/" in manual_deploy assert "hermes-agent/leoclean-plugins/vps/" in manual_deploy assert 'VPS_HERMES_PATCHES="/home/teleo/.hermes/teleo-runtime-patches"' in manual_deploy assert "systemctl is-active --quiet leoclean-gateway.service" in manual_deploy def test_deploy_scripts_syntax_check_leoclean_database_runtime_before_restart(): auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text() manual_deploy = (REPO_ROOT / "deploy" / "deploy.sh").read_text() assert "hermes-agent/leoclean-bin/*.py" in auto_deploy assert "scripts/leo_behavior_manifest.py" in auto_deploy assert '"$REPO_ROOT/hermes-agent/leoclean-bin/"*.py' in manual_deploy assert '"$REPO_ROOT/scripts/leo_behavior_manifest.py"' in manual_deploy def test_auto_deploy_hot_syncs_leoclean_markdown_without_gateway_restart(): assert not _leoclean_gateway_restart_required("hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md") assert not _leoclean_gateway_restart_required("hermes-agent/leoclean-skills/vps/live-leo-telegram/SKILL.md") def test_auto_deploy_restarts_gateway_for_leoclean_runtime_changes(): assert _leoclean_gateway_restart_required("hermes-agent/leoclean-bin/kb_tool.py") assert _leoclean_gateway_restart_required("hermes-agent/leoclean-skills/vps/teleo-kb-bridge/runtime_helper.py") assert _leoclean_gateway_restart_required("hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py") assert _leoclean_gateway_restart_required("hermes-agent/patches/apply_response_transform_hook.py") def test_auto_deploy_sudoers_installer_is_root_only_and_visudo_checked(): installer = (REPO_ROOT / "deploy" / "install-auto-deploy-sudoers.sh").read_text() assert 'TARGET="/etc/sudoers.d/teleo-auto-deploy"' in installer assert "id -u" in installer assert 'install -m 0440 "$SOURCE" "$TARGET"' in installer assert 'visudo -cf "$TARGET"' in installer def test_gcp_leoclean_skill_sync_targets_parallel_runtime_without_secrets(): script_path = REPO_ROOT / "deploy" / "sync-gcp-leoclean-skills.sh" script = script_path.read_text() assert 'PROJECT="${PROJECT:-teleo-501523}"' in script assert 'ZONE="${ZONE:-europe-west6-a}"' in script assert 'INSTANCE="${INSTANCE:-teleo-prod-1}"' in script assert 'SERVICE="${SERVICE:-leoclean-gcp-prod-parallel.service}"' in script assert "hermes-agent/leoclean-skills/gcp" in script assert "--tunnel-through-iap" in script assert "/home/teleo/.hermes/profiles/leoclean/skills" in script assert "/kb/claims/|wrap claim IDs, proposal IDs" in script assert 'COPYFILE_DISABLE=1 tar --format=ustar -C "$SOURCE_DIR" -cf - .' in script assert 'sudo tar -C "\\$remote_skills" -xf -' in script assert "sudo systemctl restart" in script assert "sudo rsync" not in script assert "TELEGRAM_BOT_TOKEN" not in script assert "CLOUDSQL_PASSWORD" not in script def test_gcp_runtime_sync_is_source_bound_and_scoped(): script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() dropin = (REPO_ROOT / "systemd" / "leoclean-gcp-prod-parallel-cloudsql.conf").read_text() assert 'git -C "$REPO_ROOT" rev-parse HEAD' in script assert 'git -C "$REPO_ROOT" diff --quiet HEAD' in script assert "merge-base --is-ancestor" in script assert "Refusing to deploy unmerged revision" in script assert "Refusing to install runtime files without --restart" in script assert 'DROPIN_NAME="10-cloudsql-memory.conf"' in script assert "20-canonical-kb.conf" not in script assert "sha256sum" in script assert "--preflight-only" in script assert "--restart" in script assert '/usr/bin/python3 "$tool_path" status --format json --redacted' in script assert "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com" in script assert "169.254.169.254/computeMetadata/v1/instance/service-accounts/default/email" in script assert "ProxyHandler({})" in script assert "env -i" in script assert 'run_scoped_status pre "$payload_dir/$TOOL_REL"' in script assert 'run_permission_verifier pre "$payload_dir/$PERMISSION_VERIFIER_REL"' in script assert 'sudo -n -u teleo mkdir -p -- "$receipt_dir"' in script assert 'if [ "$PREFLIGHT_ONLY" = true ]; then' in script assert "run_service_wrapper_status post" in script assert "run_service_wrapper_status verify" in script assert '"$SERVICE_PROFILE_BIN/teleo-kb" status --format json --redacted' in script wrapper_probe = script.split("run_service_wrapper_status() {", maxsplit=1)[1].split( "\n}\n\nservice_fingerprint", maxsplit=1 )[0] assert 'nsenter -t "$main_pid" -m --env' in wrapper_probe assert "/usr/bin/setpriv" in wrapper_probe assert "--reuid=teleo" in wrapper_probe assert "TELEO_CLOUDSQL_" not in wrapper_probe assert "/usr/bin/env -i" not in wrapper_probe assert 'run_permission_verifier post "$runtime_bin/verify_gcp_leoclean_runtime_permissions.py"' in script assert 'SERVICE_ENV_VERIFIER_REL="ops/verify_gcp_leoclean_service_environment.py"' in script assert 'assert_runtime_environment "$runtime_bin/verify_gcp_leoclean_service_environment.py"' in script assert 'assert_runtime_environment "$REMOTE_RUNTIME_BIN/verify_gcp_leoclean_service_environment.py"' in script assert "REMOTE_CLOUDSDK_CONFIG" not in script assert "--property=DropInPaths" in script assert "--property=EnvironmentFiles" in script assert "Service has an effective EnvironmentFile" in script assert "Service has a later drop-in" in script assert "assert_administrator_secret_denied" in script assert '"stdout_discarded": True' in script assert '"credential_context": "systemd_main_pid_environment"' in script administrator_probe = script.split("assert_administrator_secret_denied() {", maxsplit=1)[1].split( "\n}\n\nrun_scoped_status", maxsplit=1 )[0] assert 'nsenter -t "$main_pid" -m --env' in administrator_probe assert "/usr/bin/setpriv" in administrator_probe assert "env -i" not in administrator_probe assert 'preflight_service_before="$(service_fingerprint)"' in script assert 'assert_service_fingerprint "$preflight_service_before"' in script assert "backup_path" in script assert "backup_directory_state" in script assert "restore_path" in script assert "restore_directory_state" in script assert 'runtime_grandparent="$(dirname "$runtime_parent")"' in script assert 'runtime_anchor="$(dirname "$runtime_grandparent")"' in script assert 'dropin_parent="$(dirname "$dropin_dir")"' in script assert 'backup_directory_state "$runtime_grandparent" runtime_grandparent' in script assert 'restore_directory_state "$runtime_grandparent" runtime_grandparent' in script assert 'sudo test -d "$runtime_anchor"' in script assert 'sudo test -d "$dropin_parent"' in script assert 'sudo rmdir -- "$target"' in script assert "rollback_runtime" in script assert "trap 'on_exit $?' EXIT" in script assert 'backup_path "$dropin_dir/$DROPIN_NAME" dropin' in script assert 'dropin_stage="$dropin_dir/.${DROPIN_NAME}.livingip-new.$$"' in script assert 'sudo mv -f -- "$dropin_stage" "$dropin_dir/$DROPIN_NAME"' in script assert 'active_state" != "active"' in script assert 'sub_state" != "running"' in script assert 'sudo systemctl stop "$SERVICE"' in script assert 'sudo systemctl start "$SERVICE"' in script assert 'REMOTE_RUNTIME_BIN="${REMOTE_RUNTIME_BIN:-/usr/local/libexec/livingip/leoclean-kb}"' in script assert 'sudo install -d -o root -g root -m 0755 "$runtime_parent"' in script assert 'sudo install -d -o root -g root -m 0755 "$runtime_bin"' in script assert 'remote_tmp="$(sudo mktemp -d /var/tmp/teleo-gcp-leoclean-runtime.XXXXXX)"' in script assert 'sudo install -d -o root -g root -m 0755 "$payload_dir"' in script assert 'sudo install -d -o root -g root -m 0700 "$backup_dir"' in script assert 'sudo install -d -o teleo -g teleo -m 0700 "$work_dir"' in script assert "sudo tar --no-same-owner --no-same-permissions" in script assert 'sudo chown -R root:root "$payload_dir"' in script assert 'sudo find "$payload_dir" -type d -exec chmod 0555 {} +' in script assert 'sudo find "$payload_dir" -type l -print -quit' in script assert 'sudo -n -u teleo test ! -w "$payload_dir"' in script assert 'sudo -n -u teleo test ! -w "$backup_dir"' in script assert 'sudo touch "$backup_dir/$name.present"' in script assert 'sudo test -f "$backup_dir/$name.present"' in script assert 'sudo install -o root -g root -m 0755 "$payload_dir/$WRAPPER_REL" "$wrapper_stage"' in script assert 'sudo install -o root -g root -m 0755 "$payload_dir/$TOOL_REL" "$tool_stage"' in script assert ( 'sudo install -o root -g root -m 0755 "$payload_dir/$PERMISSION_VERIFIER_REL" ' '"$permission_verifier_stage"' in script ) assert ( 'sudo install -o root -g root -m 0755 "$payload_dir/$SERVICE_ENV_VERIFIER_REL" ' '"$service_env_verifier_stage"' in script ) assert 'assert_installed_files\nsudo systemctl daemon-reload' in script assert 'tar -C "$remote_tmp" -xf -' not in script assert 'mkdir -p "$backup_dir"' not in script assert 'for protected_file in \\\n "$wrapper" \\' in script assert '"$permission_verifier" \\' in script assert '"$service_env_verifier" \\' in script assert 'sudo -n -u teleo test ! -w "$protected_file"' in script assert 'for protected_dir in \\\n /usr \\' in script assert "/usr/local/libexec/livingip" in script assert 'sudo -n -u teleo test ! -w "$protected_dir"' in script assert 'for executable_dir in \\\n /usr/local/sbin \\' in script assert 'sudo stat -Lc \'%U:%G\' "$executable_dir"' in script assert 'sudo -n -u teleo test ! -w "$executable_dir"' in script assert 'for protected_file in /bin/bash /usr/bin/curl /usr/bin/python3' in script assert '/usr/bin/curl -q --noproxy' in script assert 'test "$(sudo cat "$revision")" = "$SOURCE_REVISION"' in script assert 'assert_existing_root_directory "$existing_directory"' in script assert 'assert_existing_root_file "$existing_file"' in script assert 'validate_restored_path "$runtime_bin/teleo-kb" wrapper' in script assert 'validate_restored_directory "$runtime_bin" runtime_bin' in script rollback = script.split("rollback_runtime() {", maxsplit=1)[1].split("\n}\n\non_exit()", maxsplit=1)[0] assert rollback.index("validate_restored_path") < rollback.index("systemctl daemon-reload") assert rollback.index("systemctl daemon-reload") < rollback.index('systemctl start "$SERVICE"') assert rollback.count('if [ "$rollback_rc" -eq 0 ]; then') >= 3 deploy_tail = script.split("mutation_started=true", maxsplit=1)[1] assert deploy_tail.index("assert_installed_files") < deploy_tail.index("systemctl daemon-reload") assert deploy_tail.index("systemctl daemon-reload") < deploy_tail.index("assert_unit_configuration") assert deploy_tail.index("assert_unit_configuration") < deploy_tail.index('systemctl start "$SERVICE"') for install_line in ( 'sudo install -o root -g root -m 0755 "$payload_dir/$WRAPPER_REL"', 'sudo install -o root -g root -m 0755 "$payload_dir/$TOOL_REL"', 'sudo install -o root -g root -m 0755 "$payload_dir/$PERMISSION_VERIFIER_REL"', 'sudo install -o root -g root -m 0755 "$payload_dir/$SERVICE_ENV_VERIFIER_REL"', 'sudo install -o root -g root -m 0644 "$payload_dir/$DROPIN_REL"', ): assert install_line in deploy_tail assert 'sudo install -o root -g root -m 0755 "$work_dir/' not in script assert 'nsenter -t "$main_pid" -m -- sha256sum "$SERVICE_PROFILE_BIN/teleo-kb"' in script assert 'findmnt -T "$SERVICE_PROFILE_BIN"' in script assert 'findmnt -M "$SERVICE_HERMES_ROOT"' in script assert 'findmnt -M "$SERVICE_PROFILE_ROOT/state"' in script assert 'findmnt -M "$SERVICE_PROFILE_ROOT/workspace"' in script assert "cap_sys_admin" in script assert '"current_tier": "T3_live_readonly"' in script assert '"status": "pass"' in script assert "UnsetEnvironment=PGPASSWORD" in dropin assert "UnsetEnvironment=GOOGLE_APPLICATION_CREDENTIALS CLOUDSDK_CONFIG" in dropin assert "UnsetEnvironment=BASH_ENV ENV BASHOPTS SHELLOPTS" in dropin assert "UnsetEnvironment=LD_PRELOAD LD_LIBRARY_PATH LD_AUDIT" in dropin assert "UnsetEnvironment=PYTHONHOME PYTHONPATH PYTHONSTARTUP PYTHONINSPECT" in dropin assert "UnsetEnvironment=HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY" in dropin assert "UnsetEnvironment=SSL_CERT_FILE SSL_CERT_DIR REQUESTS_CA_BUNDLE CURL_CA_BUNDLE SSLKEYLOGFILE" in dropin assert ( "BindReadOnlyPaths=/usr/local/libexec/livingip/leoclean-kb:/home/teleo/.hermes/profiles/leoclean/bin" in dropin ) assert "ReadOnlyPaths=/home/teleo/.hermes" in dropin assert ( "ReadWritePaths=/home/teleo/.hermes/profiles/leoclean/state " "/home/teleo/.hermes/profiles/leoclean/workspace" in dropin ) assert "CapabilityBoundingSet=~CAP_SYS_ADMIN" in dropin assert "RuntimeDirectory=" not in dropin assert "Environment=CLOUDSDK_" not in dropin assert "Environment=PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" in dropin assert "TELEO_KB_MODE=cloudsql" in dropin assert "TELEO_GCP_METADATA_ONLY=1" in dropin assert "TELEO_CLOUDSQL_USER=leoclean_kb_runtime" in dropin assert "gcp-teleo-pgvector-standby-leoclean-kb-runtime-password" in dropin assert "TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK=0" in dropin assert not any(line.startswith("Environment=PGPASSWORD") for line in dropin.splitlines()) assert "gcp-teleo-pgvector-standby-postgres-password" not in dropin def test_gcp_runtime_embedded_remote_programs_are_bash_syntax_valid() -> None: script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() blocks: dict[str, str] = {} for variable in ("remote_helpers", "verify_body", "sync_body"): marker = f"{variable}=$(cat <<'REMOTE'\n" assert script.count(marker) == 1 start = script.index(marker) + len(marker) end = script.index("\nREMOTE\n)", start) blocks[variable] = script[start:end] for name, body in ( *blocks.items(), ("composed_verify", blocks["remote_helpers"] + "\n" + blocks["verify_body"]), ("composed_sync", blocks["remote_helpers"] + "\n" + blocks["sync_body"]), ): completed = subprocess.run( ["bash", "-n"], input=body, capture_output=True, check=False, text=True, ) assert completed.returncode == 0, f"{name}: {completed.stderr}" embedded_python = [part.split("\nPY", maxsplit=1)[0] for part in blocks["remote_helpers"].split("<<'PY'\n")[1:]] assert len(embedded_python) == 2 for index, body in enumerate(embedded_python): completed = subprocess.run( ["python3", "-c", "import sys; compile(sys.stdin.read(), '', 'exec')"], input=body, capture_output=True, check=False, text=True, ) assert completed.returncode == 0, f"embedded_python_{index}: {completed.stderr}" def test_gcp_runtime_rejects_preexisting_symlink_before_mutation(tmp_path: Path) -> None: script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() function_start = script.index("assert_existing_root_directory() {") function_end = script.index("\n\nbackup_path() {", function_start) trust_functions = script[function_start:function_end] real_directory = tmp_path / "real-runtime" real_directory.mkdir() symlink = tmp_path / "runtime-link" symlink.symlink_to(real_directory, target_is_directory=True) harness = f""" set -euo pipefail sudo() {{ if [ "${{1:-}}" = -n ]; then shift if [ "${{1:-}}" = -u ]; then shift 2; fi fi "$@" }} {trust_functions} assert_existing_root_directory "$1" """ completed = subprocess.run( ["bash", "-s", "--", str(symlink)], input=harness, capture_output=True, check=False, text=True, ) assert completed.returncode != 0 def test_gcp_runtime_executable_configuration_override_gate() -> None: script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() function_start = script.index("assert_unit_configuration() {") function_end = script.index("\n\nassert_runtime_environment() {", function_start) unit_configuration_function = script[function_start:function_end] harness = f""" set -euo pipefail SERVICE=leoclean.service REMOTE_DROPIN_DIR=/etc/systemd/system/leoclean.service.d DROPIN_NAME=10-cloudsql-memory.conf systemctl() {{ case "$*" in *--property=DropInPaths*) printf '%s\\n' "${{FAKE_DROPINS:-}}" ;; *--property=EnvironmentFiles*) printf '%s\\n' "${{FAKE_ENVIRONMENT_FILES:-}}" ;; *) return 90 ;; esac }} {unit_configuration_function} assert_unit_configuration """ reviewed = "/etc/systemd/system/leoclean.service.d/10-cloudsql-memory.conf" cases = ( { "FAKE_DROPINS": f"{reviewed} /etc/systemd/system/leoclean.service.d/99-conflict.conf", "FAKE_ENVIRONMENT_FILES": "", }, {"FAKE_DROPINS": reviewed, "FAKE_ENVIRONMENT_FILES": "/etc/leoclean-runtime.env"}, ) for environment in cases: completed = subprocess.run( ["bash", "-s"], input=harness, capture_output=True, check=False, env=environment, text=True, ) assert completed.returncode != 0 allowed = subprocess.run( ["bash", "-s"], input=harness, capture_output=True, check=False, env={"FAKE_DROPINS": reviewed, "FAKE_ENVIRONMENT_FILES": ""}, text=True, ) assert allowed.returncode == 0, allowed.stderr def test_gcp_runtime_absence_rollback_removes_every_created_directory(tmp_path: Path) -> None: script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() function_start = script.index("backup_directory_state() {") function_end = script.index("\nrestore_path() {", function_start) directory_functions = script[function_start:function_end] harness = f""" set -euo pipefail sudo() {{ "$@"; }} root="$1" backup_dir="$root/backup" mkdir -p "$backup_dir" "$root/usr/local" "$root/etc/systemd/system" runtime_grandparent="$root/usr/local/libexec" runtime_parent="$runtime_grandparent/livingip" runtime_bin="$runtime_parent/leoclean-kb" dropin_dir="$root/etc/systemd/system/service.d" {directory_functions} backup_directory_state "$runtime_grandparent" runtime_grandparent backup_directory_state "$runtime_parent" runtime_parent backup_directory_state "$runtime_bin" runtime_bin backup_directory_state "$dropin_dir" dropin_dir mkdir -p "$runtime_bin" "$dropin_dir" restore_directory_state "$runtime_bin" runtime_bin restore_directory_state "$runtime_parent" runtime_parent restore_directory_state "$runtime_grandparent" runtime_grandparent restore_directory_state "$dropin_dir" dropin_dir test ! -e "$runtime_grandparent" test ! -e "$dropin_dir" test -d "$root/usr/local" test -d "$root/etc/systemd/system" """ completed = subprocess.run( ["bash", "-s", "--", str(tmp_path)], input=harness, capture_output=True, check=False, text=True, ) assert completed.returncode == 0, completed.stderr def test_gcp_runtime_rollback_never_reloads_or_restarts_after_validation_failure(tmp_path: Path) -> None: script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() function_start = script.index("rollback_runtime() {") function_end = script.index("\n\non_exit() {", function_start) rollback_function = script[function_start:function_end] action_log = tmp_path / "systemctl-actions" harness = f""" set -u SERVICE=leoclean.service runtime_bin=/usr/local/libexec/livingip/leoclean-kb runtime_parent=/usr/local/libexec/livingip runtime_grandparent=/usr/local/libexec dropin_dir=/etc/systemd/system/leoclean.service.d DROPIN_NAME=10-cloudsql-memory.conf action_log="$1" sudo() {{ if [ "${{1:-}}" = systemctl ]; then printf '%s\\n' "$2" >>"$action_log" return 0 fi "$@" }} restore_path() {{ return 0; }} restore_directory_state() {{ return 0; }} validate_restored_path() {{ [ "$2" != wrapper ]; }} validate_restored_directory() {{ return 0; }} assert_unit_configuration() {{ printf '%s\\n' unit-configuration >>"$action_log"; }} assert_service_running() {{ printf '%s\\n' assert-running >>"$action_log"; }} {rollback_function} rollback_runtime """ completed = subprocess.run( ["bash", "-s", "--", str(action_log)], input=harness, capture_output=True, check=False, text=True, ) assert completed.returncode != 0 assert action_log.read_text(encoding="utf-8").splitlines() == ["stop"]