test_e2e.py
python
sha256:2313b6a6353294a0ca08ff579624198da250b99163a86215aed31a905912e360
Land Scooling Lab T0/T2 training contract and seven-tier tests
Human
minor
⚠ breaking
44 days ago
| 1 | """End-to-end tier tests for the Scooling Lab HTTP API contract.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import threading |
| 7 | import unittest |
| 8 | import urllib.request |
| 9 | from http.server import ThreadingHTTPServer |
| 10 | |
| 11 | from scooling_lab_helpers import PROJECT_ROOT, valid_payload |
| 12 | |
| 13 | from scooling_lab.api import make_handler |
| 14 | from scooling_lab.service import TrainingApiService |
| 15 | from scooling_lab.store import TrainingJobStore |
| 16 | |
| 17 | |
| 18 | class ScoolingLabE2ETests(unittest.TestCase): |
| 19 | """E2E tests for create -> poll -> completed -> listArtifacts.""" |
| 20 | |
| 21 | def test_e2e_http_create_poll_completed_and_list_artifacts(self) -> None: |
| 22 | """The dependency-free HTTP API completes the fake-worker fixture flow.""" |
| 23 | |
| 24 | service = TrainingApiService(TrainingJobStore()) |
| 25 | server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(service)) |
| 26 | thread = threading.Thread(target=server.serve_forever, daemon=True) |
| 27 | thread.start() |
| 28 | base_url = f"http://127.0.0.1:{server.server_port}" |
| 29 | try: |
| 30 | created = post_json(f"{base_url}/training/jobs", valid_payload("e2e")) |
| 31 | self.assertEqual(created["status"], "succeeded") |
| 32 | job_id = str(created["id"]) |
| 33 | |
| 34 | fetched = get_json(f"{base_url}/training/jobs/{job_id}") |
| 35 | self.assertEqual(fetched["status"], "succeeded") |
| 36 | |
| 37 | artifacts = get_json(f"{base_url}/training/jobs/{job_id}/artifacts") |
| 38 | self.assertEqual(len(artifacts["artifacts"]), 1) |
| 39 | self.assertEqual(artifacts["artifacts"][0]["jobId"], job_id) |
| 40 | finally: |
| 41 | server.shutdown() |
| 42 | server.server_close() |
| 43 | |
| 44 | def test_e2e_ci_workflow_runs_lab_chain(self) -> None: |
| 45 | """The CI workflow contains unittest, secret scan, and BOM audit steps.""" |
| 46 | |
| 47 | workflow = (PROJECT_ROOT / ".github/workflows/ci.yml").read_text( |
| 48 | encoding="utf-8" |
| 49 | ) |
| 50 | self.assertIn("python -m unittest discover", workflow) |
| 51 | self.assertIn("gitleaks detect", workflow) |
| 52 | self.assertIn("python -m scooling_lab.bom", workflow) |
| 53 | |
| 54 | |
| 55 | def post_json(url: str, payload: dict[str, object]) -> dict[str, object]: |
| 56 | """POST JSON and return decoded JSON without external network use.""" |
| 57 | |
| 58 | request = urllib.request.Request( |
| 59 | url, |
| 60 | data=json.dumps(payload).encode("utf-8"), |
| 61 | headers={"Content-Type": "application/json"}, |
| 62 | method="POST", |
| 63 | ) |
| 64 | with urllib.request.urlopen(request, timeout=5) as response: |
| 65 | result = json.loads(response.read().decode("utf-8")) |
| 66 | if not isinstance(result, dict): |
| 67 | raise AssertionError("expected JSON object") |
| 68 | return result |
| 69 | |
| 70 | |
| 71 | def get_json(url: str) -> dict[str, object]: |
| 72 | """GET JSON from the local fixture server.""" |
| 73 | |
| 74 | with urllib.request.urlopen(url, timeout=5) as response: |
| 75 | result = json.loads(response.read().decode("utf-8")) |
| 76 | if not isinstance(result, dict): |
| 77 | raise AssertionError("expected JSON object") |
| 78 | return result |
| 79 | |
| 80 | |
| 81 | if __name__ == "__main__": |
| 82 | unittest.main() |
File History
1 commit
sha256:2313b6a6353294a0ca08ff579624198da250b99163a86215aed31a905912e360
Land Scooling Lab T0/T2 training contract and seven-tier tests
Human
minor
⚠
44 days ago