File size: 2,162 Bytes
4904e85
 
 
 
 
 
 
885a0b4
 
4904e85
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e259b96
4904e85
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
"""Smoke test for Docker image health and reset endpoints."""

import subprocess
import time
import requests  # type: ignore


IMAGE_NAME = "citywide-dispatch-supervisor"
CONTAINER_NAME = "test-dispatch-smoke"
HOST = "http://localhost:8000"


def build_image() -> None:
    print("Building Docker image...")
    result = subprocess.run(
        ["docker", "build", "-t", IMAGE_NAME, "."],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Docker build failed:\n{result.stderr}")
    print("Build succeeded.")


def run_container() -> None:
    print("Running container...")
    subprocess.run(
        [
            "docker",
            "run",
            "-d",
            "-p",
            "8000:8000",
            "--name",
            CONTAINER_NAME,
            IMAGE_NAME,
        ],
        check=True,
        capture_output=True,
        text=True,
    )
    # Give the server time to start
    time.sleep(5)
    print("Container running.")


def kill_container() -> None:
    print("Killing container...")
    subprocess.run(["docker", "kill", CONTAINER_NAME], capture_output=True)
    subprocess.run(["docker", "rm", CONTAINER_NAME], capture_output=True)
    print("Container cleaned up.")


def test_health() -> None:
    print("Testing /health endpoint...")
    response = requests.get(f"{HOST}/health", timeout=10)
    assert response.status_code == 200, f"Expected 200, got {response.status_code}"
    assert response.json() == {"status": "ok"}, f"Unexpected body: {response.json()}"
    print("Health check passed.")


def test_reset() -> None:
    print("Testing /reset endpoint...")
    response = requests.post(
        f"{HOST}/reset",
        json={"task_id": "single_incident", "seed": 42},
        timeout=10,
    )
    assert response.status_code == 200, f"Expected 200, got {response.status_code}"
    print("Reset check passed.")


def main() -> None:
    try:
        build_image()
        run_container()
        test_health()
        test_reset()
        print("\nAll smoke tests passed!")
    finally:
        kill_container()


if __name__ == "__main__":
    main()