|
10 | 10 | import urllib.request |
11 | 11 | from pathlib import Path |
12 | 12 |
|
| 13 | +import pytest |
| 14 | + |
13 | 15 |
|
14 | 16 | ROOT = Path(__file__).resolve().parent.parent |
15 | 17 |
|
@@ -55,107 +57,3 @@ def test_ci_workflow_runs_release_quality_gates() -> None: |
55 | 57 | assert "python -m compileall -q minicode tests" in content |
56 | 58 | assert "python -m pytest -q" in content |
57 | 59 | assert "tests/test_packaging.py" in content |
58 | | - |
59 | | - |
60 | | -def test_cron_runner_empty_config_exits_cleanly(tmp_path: Path) -> None: |
61 | | - missing_config = tmp_path / "missing-cron.json" |
62 | | - |
63 | | - completed = subprocess.run( |
64 | | - [ |
65 | | - sys.executable, |
66 | | - "-m", |
67 | | - "minicode.cron_runner", |
68 | | - "--once", |
69 | | - "--dry-run", |
70 | | - "--config", |
71 | | - str(missing_config), |
72 | | - ], |
73 | | - cwd=ROOT, |
74 | | - text=True, |
75 | | - encoding="utf-8", |
76 | | - errors="replace", |
77 | | - stdout=subprocess.PIPE, |
78 | | - stderr=subprocess.PIPE, |
79 | | - timeout=20, |
80 | | - check=False, |
81 | | - ) |
82 | | - |
83 | | - assert completed.returncode == 0, completed.stderr |
84 | | - assert "No cron tasks configured" in completed.stdout |
85 | | - |
86 | | - |
87 | | -def test_gateway_health_endpoint_responds() -> None: |
88 | | - from http.server import ThreadingHTTPServer |
89 | | - |
90 | | - from minicode.gateway import MiniCodeGatewayHandler |
91 | | - |
92 | | - server = ThreadingHTTPServer(("127.0.0.1", 0), MiniCodeGatewayHandler) |
93 | | - thread = threading.Thread(target=server.serve_forever, daemon=True) |
94 | | - thread.start() |
95 | | - try: |
96 | | - port = server.server_address[1] |
97 | | - with urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=5) as response: |
98 | | - payload = json.loads(response.read().decode("utf-8")) |
99 | | - assert payload == {"ok": True, "service": "minicode-gateway"} |
100 | | - finally: |
101 | | - server.shutdown() |
102 | | - server.server_close() |
103 | | - thread.join(timeout=5) |
104 | | - |
105 | | - |
106 | | -def _post_gateway_json(port: int, payload: dict) -> tuple[int, dict]: |
107 | | - request = urllib.request.Request( |
108 | | - f"http://127.0.0.1:{port}/run", |
109 | | - data=json.dumps(payload).encode("utf-8"), |
110 | | - headers={"Content-Type": "application/json"}, |
111 | | - method="POST", |
112 | | - ) |
113 | | - try: |
114 | | - with urllib.request.urlopen(request, timeout=5) as response: |
115 | | - return response.status, json.loads(response.read().decode("utf-8")) |
116 | | - except urllib.error.HTTPError as error: |
117 | | - return error.code, json.loads(error.read().decode("utf-8")) |
118 | | - |
119 | | - |
120 | | -def test_gateway_run_endpoint_returns_headless_response(monkeypatch) -> None: |
121 | | - from http.server import ThreadingHTTPServer |
122 | | - |
123 | | - import minicode.headless |
124 | | - from minicode.gateway import MiniCodeGatewayHandler |
125 | | - |
126 | | - monkeypatch.setattr(minicode.headless, "run_headless", lambda prompt: f"mock:{prompt}") |
127 | | - server = ThreadingHTTPServer(("127.0.0.1", 0), MiniCodeGatewayHandler) |
128 | | - thread = threading.Thread(target=server.serve_forever, daemon=True) |
129 | | - thread.start() |
130 | | - try: |
131 | | - status, payload = _post_gateway_json(server.server_address[1], {"prompt": "hello"}) |
132 | | - assert status == 200 |
133 | | - assert payload == {"ok": True, "response": "mock:hello"} |
134 | | - finally: |
135 | | - server.shutdown() |
136 | | - server.server_close() |
137 | | - thread.join(timeout=5) |
138 | | - |
139 | | - |
140 | | -def test_gateway_run_endpoint_converts_system_exit_to_json_error(monkeypatch) -> None: |
141 | | - from http.server import ThreadingHTTPServer |
142 | | - |
143 | | - import minicode.headless |
144 | | - from minicode.gateway import MiniCodeGatewayHandler |
145 | | - |
146 | | - def fail_headless(_prompt: str) -> str: |
147 | | - raise SystemExit("missing config") |
148 | | - |
149 | | - monkeypatch.setattr(minicode.headless, "run_headless", fail_headless) |
150 | | - server = ThreadingHTTPServer(("127.0.0.1", 0), MiniCodeGatewayHandler) |
151 | | - thread = threading.Thread(target=server.serve_forever, daemon=True) |
152 | | - thread.start() |
153 | | - try: |
154 | | - status, payload = _post_gateway_json(server.server_address[1], {"prompt": "hello"}) |
155 | | - assert status == 500 |
156 | | - assert payload["ok"] is False |
157 | | - assert "missing config" in payload["error"] |
158 | | - finally: |
159 | | - server.shutdown() |
160 | | - server.server_close() |
161 | | - thread.join(timeout=5) |
0 commit comments