98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
import unittest
|
|
from pathlib import Path
|
|
|
|
from scripts import deploy_with_password
|
|
|
|
|
|
class DeployWithPasswordTests(unittest.TestCase):
|
|
def test_deploy_payload_includes_automation_anchor_files(self) -> None:
|
|
self.assertIn(".dockerignore", deploy_with_password.DEPLOY_PATHS)
|
|
self.assertIn("docs", deploy_with_password.DEPLOY_PATHS)
|
|
self.assertIn("SESSION.md", deploy_with_password.DEPLOY_PATHS)
|
|
self.assertIn("TASKS.md", deploy_with_password.DEPLOY_PATHS)
|
|
|
|
def test_docker_image_includes_automation_anchor_files(self) -> None:
|
|
dockerfile = Path("Dockerfile").read_text(encoding="utf-8")
|
|
|
|
self.assertIn("COPY docs /app/docs", dockerfile)
|
|
self.assertIn("COPY SESSION.md /app/SESSION.md", dockerfile)
|
|
self.assertIn("COPY TASKS.md /app/TASKS.md", dockerfile)
|
|
|
|
def test_dockerignore_does_not_exclude_automation_docs(self) -> None:
|
|
dockerignore = Path(".dockerignore").read_text(encoding="utf-8").splitlines()
|
|
|
|
self.assertNotIn("docs/", dockerignore)
|
|
|
|
def test_compose_persists_automation_monitor_state(self) -> None:
|
|
compose = Path("docker-compose.yml").read_text(encoding="utf-8")
|
|
|
|
self.assertIn("./.claude:/app/.claude", compose)
|
|
self.assertIn("host.docker.internal:host-gateway", compose)
|
|
|
|
def test_deploy_smoke_includes_p24_execution_layer_endpoints(self) -> None:
|
|
smoke_source = Path("scripts/deploy_with_password.py").read_text(encoding="utf-8")
|
|
|
|
self.assertIn("/trading/alerts/operator-digest?instrument_id=1", smoke_source)
|
|
self.assertIn("/trading/maintenance-summary?instrument_id=1", smoke_source)
|
|
self.assertIn("/trading/unattended-harness?instrument_id=1", smoke_source)
|
|
|
|
def test_deploy_command_runs_migrations_after_compose_up(self) -> None:
|
|
command = deploy_with_password.build_deploy_command(
|
|
remote_path="/root/deploy/ai-ict",
|
|
remote_archive="/root/ai-ict-deploy.tgz",
|
|
)
|
|
|
|
self.assertIn("scripts/apply_migrations.py", command)
|
|
self.assertIn("docker rm -f ai-ict", command)
|
|
|
|
def test_deploy_env_backup_stays_inside_remote_path_for_non_root_user(self) -> None:
|
|
command = deploy_with_password.build_deploy_command(
|
|
remote_path="/home/yuqei/project/ai-exchange",
|
|
remote_archive="/home/yuqei/ai-ict-deploy.tgz",
|
|
)
|
|
|
|
self.assertIn("/home/yuqei/project/ai-exchange/.env.deploy.bak", command)
|
|
self.assertNotIn("/root/ai-ict-deploy.env.bak", command)
|
|
|
|
def test_run_remote_can_wrap_remote_step_with_sudo_without_leaking_password_in_command(self) -> None:
|
|
class _FakeStdin:
|
|
def __init__(self) -> None:
|
|
self.writes = []
|
|
|
|
def write(self, value: str) -> None:
|
|
self.writes.append(value)
|
|
|
|
def flush(self) -> None:
|
|
pass
|
|
|
|
class _FakeChannel:
|
|
def recv_exit_status(self) -> int:
|
|
return 0
|
|
|
|
class _FakeStream:
|
|
channel = _FakeChannel()
|
|
|
|
def read(self) -> bytes:
|
|
return b""
|
|
|
|
class _FakeClient:
|
|
def __init__(self) -> None:
|
|
self.command = ""
|
|
self.stdin = _FakeStdin()
|
|
|
|
def exec_command(self, command: str, timeout: int):
|
|
self.command = command
|
|
return self.stdin, _FakeStream(), _FakeStream()
|
|
|
|
client = _FakeClient()
|
|
|
|
deploy_with_password.run_remote(client, "docker compose ps", sudo_password="secret")
|
|
|
|
self.assertIn("sudo -S", client.command)
|
|
self.assertNotIn("secret", client.command)
|
|
self.assertEqual(client.stdin.writes, ["secret\n"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|