217 lines
11 KiB
Python
217 lines
11 KiB
Python
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/<claim_id>|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 "--restart" in script
|
|
assert 'python3 "$tool_path" status --format json --redacted' in script
|
|
assert "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com" in script
|
|
assert "metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/email" in script
|
|
assert "env -u PGPASSWORD" in script
|
|
assert 'run_scoped_status pre "$remote_tmp/$TOOL_REL"' in script
|
|
assert 'run_scoped_status post "$profile_bin/cloudsql_memory_tool.py"' in script
|
|
assert "backup_path" in script
|
|
assert "restore_path" 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 "UnsetEnvironment=PGPASSWORD" in dropin
|
|
assert "RuntimeDirectory=leoclean-gcp-prod-parallel" in dropin
|
|
assert "RuntimeDirectoryMode=0700" in dropin
|
|
assert "Environment=CLOUDSDK_CONFIG=/run/leoclean-gcp-prod-parallel" in dropin
|
|
assert "TELEO_KB_MODE=cloudsql" 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
|