aaronrene / scooling-lab public
test_e2e.py python
173 lines 7.1 KB
Raw
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ breaking 41 days ago
1 """End-to-end tier tests for Scooling Lab HTTP training routes."""
2
3 from __future__ import annotations
4
5 import json
6 import threading
7 import unittest
8 from http.server import ThreadingHTTPServer
9 from urllib.error import HTTPError
10 from urllib.request import Request, urlopen
11
12 from scooling_lab_helpers import PROJECT_ROOT, valid_payload
13
14 from scooling_lab.api import make_handler
15 from scooling_lab.contracts import TrainingJobRequest, TrainingJobStatus
16 from scooling_lab.service import TrainingApiService
17 from scooling_lab.store import TrainingJobStore
18
19
20 class ScoolingLabEndToEndTests(unittest.TestCase):
21 """E2E tests across the local dependency-free HTTP API surface."""
22
23 def test_e2e_create_fetch_provenance_delete_and_verify_absence(self) -> None:
24 """A completed artifact can be fetched, deleted, and verified absent."""
25
26 service = TrainingApiService(TrainingJobStore())
27 server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(service))
28 thread = threading.Thread(target=server.serve_forever, daemon=True)
29 thread.start()
30 base_url = f"http://127.0.0.1:{server.server_port}"
31 try:
32 created = self._json_request(
33 f"{base_url}/training/jobs",
34 "POST",
35 valid_payload("e2e-provenance-delete"),
36 )
37 job_id = str(created["id"])
38 artifacts = self._json_request(
39 f"{base_url}/training/jobs/{job_id}/artifacts", "GET"
40 )
41 artifact = artifacts["artifacts"][0]
42 artifact_id = str(artifact["id"])
43 provenance = self._json_request(
44 f"{base_url}/training/jobs/{job_id}/provenance", "GET"
45 )
46 deleted_hashes = (
47 str(artifact["datasetHash"]),
48 str(artifact["artifactHash"]),
49 str(provenance["trainingConfigHash"]),
50 )
51
52 deletion = self._json_request(
53 f"{base_url}/training/jobs/{job_id}/artifacts/{artifact_id}", "DELETE"
54 )
55 job_after_delete = self._json_request(
56 f"{base_url}/training/jobs/{job_id}", "GET"
57 )
58 artifacts_after_delete = self._json_request(
59 f"{base_url}/training/jobs/{job_id}/artifacts", "GET"
60 )
61
62 self.assertTrue(deletion["verified"])
63 self.assertTrue(service.verify_deleted_artifact_absence(deleted_hashes))
64 self.assertEqual(job_after_delete["status"], "deleted")
65 self.assertEqual(artifacts_after_delete["artifacts"], [])
66 with self.assertRaises(HTTPError) as raised:
67 self._json_request(
68 f"{base_url}/training/jobs/{job_id}/provenance", "GET"
69 )
70 self.assertEqual(raised.exception.code, 404)
71 raised.exception.close()
72 finally:
73 server.shutdown()
74 server.server_close()
75 thread.join(timeout=2)
76 thread.join(timeout=2)
77
78 def _json_request(
79 self, url: str, method: str, payload: dict[str, object] | None = None
80 ) -> dict[str, object]:
81 body = None
82 headers = {"Content-Type": "application/json"}
83 if payload is not None:
84 body = json.dumps(payload).encode("utf-8")
85 request = Request(url, data=body, headers=headers, method=method)
86 with urlopen(request, timeout=5) as response:
87 decoded = json.loads(response.read().decode("utf-8"))
88 if not isinstance(decoded, dict):
89 raise AssertionError("expected JSON object")
90 return decoded
91
92 def test_e2e_http_create_poll_completed_and_list_artifacts(self) -> None:
93 """The dependency-free HTTP API completes the fake-worker fixture flow."""
94
95 service = TrainingApiService(TrainingJobStore())
96 server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(service))
97 thread = threading.Thread(target=server.serve_forever, daemon=True)
98 thread.start()
99 base_url = f"http://127.0.0.1:{server.server_port}"
100 try:
101 created = self._json_request(
102 f"{base_url}/training/jobs", "POST", valid_payload("e2e")
103 )
104 self.assertEqual(created["status"], "succeeded")
105 job_id = str(created["id"])
106
107 fetched = self._json_request(f"{base_url}/training/jobs/{job_id}", "GET")
108 self.assertEqual(fetched["status"], "succeeded")
109
110 artifacts = self._json_request(
111 f"{base_url}/training/jobs/{job_id}/artifacts", "GET"
112 )
113 self.assertEqual(len(artifacts["artifacts"]), 1)
114 self.assertEqual(artifacts["artifacts"][0]["jobId"], job_id)
115 finally:
116 server.shutdown()
117 server.server_close()
118
119 def test_e2e_t4_cancel_retry_to_success_and_success_retry_refusal(self) -> None:
120 """HTTP routes cover cancellation, retry success, and success retry refusal."""
121
122 store = TrainingJobStore(queue_limit=6)
123 cancelled_request = TrainingJobRequest.from_mapping(valid_payload("e2e-cancel"))
124 failed_request = TrainingJobRequest.from_mapping(valid_payload("e2e-failed"))
125 cancellable = store.create(cancelled_request)
126 failed = store.create(failed_request)
127 store.update_status(failed.id, TrainingJobStatus.FAILED)
128 service = TrainingApiService(store)
129 server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(service))
130 thread = threading.Thread(target=server.serve_forever, daemon=True)
131 thread.start()
132 base_url = f"http://127.0.0.1:{server.server_port}"
133 try:
134 cancelled = self._json_request(
135 f"{base_url}/training/jobs/{cancellable.id}/cancel", "POST"
136 )
137 retried = self._json_request(
138 f"{base_url}/training/jobs/{failed.id}/retry", "POST"
139 )
140 succeeded = self._json_request(
141 f"{base_url}/training/jobs",
142 "POST",
143 valid_payload("e2e-retry-refused"),
144 )
145
146 self.assertEqual(cancelled["status"], "cancelled")
147 self.assertEqual(retried["status"], "succeeded")
148 self.assertEqual(retried["retryOfJobId"], failed.id)
149 self.assertNotEqual(retried["id"], failed.id)
150 with self.assertRaises(HTTPError) as raised:
151 self._json_request(
152 f"{base_url}/training/jobs/{succeeded['id']}/retry", "POST"
153 )
154 self.assertEqual(raised.exception.code, 409)
155 raised.exception.close()
156 finally:
157 server.shutdown()
158 server.server_close()
159 thread.join(timeout=2)
160
161 def test_e2e_ci_workflow_runs_lab_chain(self) -> None:
162 """The CI workflow contains unittest, secret scan, and BOM audit steps."""
163
164 workflow = (PROJECT_ROOT / ".github/workflows/ci.yml").read_text(
165 encoding="utf-8"
166 )
167 self.assertIn("python -m unittest discover", workflow)
168 self.assertIn("gitleaks detect", workflow)
169 self.assertIn("python -m scooling_lab.bom", workflow)
170
171
172 if __name__ == "__main__":
173 unittest.main()
File History 1 commit
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ 41 days ago