"""Console contracts: immutable uploads, isolation and durable queue semantics.""" import json from pathlib import Path import tempfile import unittest from fastapi.testclient import TestClient from console.app import create_app from console.settings import Settings from console.store import Store class ConsoleTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.source = self.root / "original" / "scripts" self.source.mkdir(parents=True) self.settings = Settings(self.root / "runtime", [ {"id": "trial", "path": str(self.source)}], allowed_hosts=["testserver"], allowed_networks=["127.0.0.0/8"], test_mode=True) self.settings.test_definitions = [{"key": name, "source_group": "trial", "entry": name, "name": name, "kind": "test"} for name in ("sample.py", "test.py", "stop.sh")] self.app = create_app(self.settings) self.client = TestClient(self.app, client=("127.0.0.1", 5555)) self.headers = {"X-XAS-Console": "1"} def tearDown(self): self.client.close() self.temp.cleanup() def upload(self, content=b"print('hello')\n", name="test.py"): r = self.client.post("/api/scripts/upload", files={"file": (name, content)}, data={"name": "Sample"}, headers=self.headers) self.assertEqual(r.status_code, 200, r.text) return r.json() def queue(self, script, key="9ad0745c-0d5e-4d1e-88e8-000000000001"): return self.client.post(f"/api/scripts/{script['id']}/runs", json={"request_id": key}, headers=self.headers) def test_upload_versions_are_immutable_and_runs_pin_version(self): script = self.upload() run = self.queue(script).json() new = self.client.post(f"/api/scripts/{script['id']}/versions", files={"file": ("test.py", b"print('v2')\n")}, headers=self.headers) self.assertEqual(new.status_code, 200, new.text) old = self.client.get(f"/api/scripts/{script['id']}/code?version=1").json() self.assertEqual(old["code"], "print('hello')\n") self.assertEqual(self.client.get(f"/api/runs/{run['id']}").json()["version"], 1) self.assertEqual(new.json()["version"], 2) def test_code_cannot_be_patched_and_path_cannot_be_replaced(self): s = self.upload() for patch in ({"code": "bad"}, {"source_path": "/etc/passwd"}, {"config": {"profile": "legacy_smoke"}}): r = self.client.patch(f"/api/scripts/{s['id']}", json=patch, headers=self.headers) self.assertEqual(r.status_code, 422, r.text) def test_upload_rejects_traversal_secrets_and_oversize(self): for name, content in [("../x.py", b"pass"), ("x.env", b"pass"), ("x.py", b"API_KEY = 'abc-secret-key-value-01234'"), ("x.py", b"x" * (1024 * 1024 + 1))]: r = self.client.post("/api/scripts/upload", files={"file": (name, content)}, headers=self.headers) self.assertIn(r.status_code, (400, 413, 422), r.text) def test_cross_site_mutations_and_host_spoofing_rejected(self): r = self.client.post("/api/discover", headers={**self.headers, "Origin": "https://evil.test"}) self.assertEqual(r.status_code, 403) self.assertEqual(self.client.post("/api/discover").status_code, 403) self.assertEqual(self.client.get("/api/summary", headers={"Host": "evil.test"}).status_code, 403) def test_duplicate_submission_and_single_claim(self): s = self.upload() a, b = self.queue(s), self.queue(s) self.assertEqual(a.json()["id"], b.json()["id"]) store = Store(self.settings) self.assertEqual(store.claim()["id"], a.json()["id"]) self.assertIsNone(store.claim()) self.queue(s, "9ad0745c-0d5e-4d1e-88e8-000000000002") self.assertIsNone(store.claim()) def test_pending_cancellation_does_not_execute(self): run = self.queue(self.upload()).json() r = self.client.post(f"/api/runs/{run['id']}/cancel", headers=self.headers) self.assertEqual(r.json()["state"], "cancelled") self.assertIsNone(Store(self.settings).claim()) def test_external_inventory_readonly_and_missing_detection(self): p = self.source / "sample.py" p.write_text("print('original')", encoding="utf-8") self.client.post("/api/discover", headers=self.headers) s = self.client.get("/api/scripts").json()["items"][0] self.assertEqual(s["kind"], "external") self.assertFalse(s["enabled"]) self.assertEqual(p.read_text(), "print('original')") p.unlink() self.client.post("/api/discover", headers=self.headers) self.assertEqual(self.client.get(f"/api/scripts/{s['id']}").json()["status"], "missing") def test_service_management_script_cannot_be_enabled(self): (self.source / "stop.sh").write_text("docker stop other-service\n") self.client.post("/api/discover", headers=self.headers) s = self.client.get("/api/scripts").json()["items"][0] r = self.client.patch(f"/api/scripts/{s['id']}", json={"enabled": True}, headers=self.headers) self.assertEqual(r.status_code, 400, r.text) def test_external_output_path_cannot_escape_source_family(self): (self.source / "test.py").write_text("print(1)\n") self.client.post("/api/discover", headers=self.headers) s = self.client.get("/api/scripts").json()["items"][0] for path in ["/etc", str(self.source), str(self.source.parent / "configs")]: r = self.client.patch(f"/api/scripts/{s['id']}", json={"config": {"output_paths": [path]}}, headers=self.headers) self.assertEqual(r.status_code, 400, r.text) def test_log_offset_unicode_and_artifact_path_boundary(self): run = self.queue(self.upload()).json() d = self.settings.runs / run["id"] d.mkdir(exist_ok=True) (d / "console.log").write_bytes("你好\n".encode("utf-8")) a = self.client.get(f"/api/runs/{run['id']}/log?offset=0").json() self.assertEqual(a["text"], "你好\n") b = self.client.get(f"/api/runs/{run['id']}/log?offset={a['next_offset']}").json() self.assertEqual(b["text"], "") self.assertEqual(self.client.get(f"/api/runs/{run['id']}/artifacts/nope").status_code, 404) def test_finished_log_utf8_character_across_chunk_boundary(self): run = self.queue(self.upload()).json() directory = self.settings.runs / run["id"] directory.mkdir() content = "x" * 65535 + "你" + "tail" (directory / "console.log").write_bytes(content.encode("utf-8")) Store(self.settings).finish(run["id"], "succeeded") first = self.client.get(f"/api/runs/{run['id']}/log").json() second = self.client.get(f"/api/runs/{run['id']}/log?offset={first['next_offset']}").json() self.assertEqual(first["text"] + second["text"], content) def test_runtime_parent_link_rejected_before_writes(self): from deploy.dedicated import validate_runtime outside = self.root / "outside" outside.mkdir() project = self.root / "project" project.mkdir() try: (project / "runtime").symlink_to(outside, target_is_directory=True) except OSError: self.skipTest("OS account cannot create directory symlinks") with self.assertRaises(ValueError): validate_runtime(project) with self.assertRaises(ValueError): Settings(project / "runtime", []).prepare() self.assertEqual(list(outside.iterdir()), []) if __name__ == "__main__": unittest.main()