From 2d04430627616bce11fb717d201cb4d1b1f30354 Mon Sep 17 00:00:00 2001 From: kstyagi Date: Fri, 25 Sep 2026 23:03:54 +0530 Subject: [PATCH] feat: add unit tests for tool API endpoints with management token validation --- tests/test_tool_api.py | 114 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 114 insertions(+) create mode 100644 tests/test_tool_api.py diff --git a/tests/test_tool_api.py b/tests/test_tool_api.py new file mode 100644 index 0000000..071a1a3 --- /dev/null +++ b/tests/test_tool_api.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import tempfile +import unittest +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +from tests.test_server import FASTAPI_AVAILABLE, FakeLifecycle +from uvm.config import Settings +from uvm.server import create_api +from uvm.state import StateStore +from uvm.tool_store import ToolJobStore +from uvm.tools import ToolService + +if FASTAPI_AVAILABLE: + from fastapi.testclient import TestClient + + +@unittest.skipUnless(FASTAPI_AVAILABLE, "FastAPI and httpx are required") +class ToolApiTests(unittest.TestCase): + def setUp(self): + temporary = self.enterContext(tempfile.TemporaryDirectory()) + settings = replace(Settings(), base=Path(temporary), api_token="test-token") + state = StateStore(settings) + self.lifecycle = FakeLifecycle() + self.lifecycle.vm.runtime_owned = True + settings.vm_dir(self.lifecycle.vm.id).mkdir(parents=True, mode=0o700) + with state.transaction() as document: + document.vms[self.lifecycle.vm.id] = self.lifecycle.vm + self.tools = ToolService(settings, state, self.lifecycle) + self.tools.start = Mock() + self.tools.shutdown = Mock() + self.tools.notify = Mock() + self.application = SimpleNamespace(settings=settings, state_store=state, lifecycle=self.lifecycle, tools=self.tools) + self.api = create_api(self.application, host="127.0.0.1") + self.client = TestClient(self.api) + self.addCleanup(self.client.close) + self.headers = {"X-UVM-Token": "test-token"} + + def job(self): + with self.application.state_store.operation_lock(): + return self.tools.store.enqueue_locked("vm-test", [{"id": "git"}], request_id="request-1") + + def test_catalog_and_job_routes_require_the_management_token(self): + for method, path in (("GET", "/tools"), ("GET", "/vms/vm-test/tools"), + ("POST", "/vms/vm-test/tools/install"), + ("GET", "/vms/vm-test/tools/jobs/tools-missing")): + with self.subTest(path=path): + response = self.client.request(method, path, json={"tools": [{"id": "git"}]}) + self.assertEqual(response.status_code, 401) + self.tools.start.assert_not_called() + + def test_catalog_has_python_options_and_create_passes_normalized_selections(self): + result = self.client.get("/tools", headers=self.headers) + self.assertEqual(result.status_code, 200) + python = next(item for item in result.json() if item["id"] == "python") + self.assertEqual(python["versions"], ["3.11", "3.12", "3.13"]) + created = self.client.post("/vms", headers=self.headers, json={"tools": [ + {"id": "python", "version": "3.12"}, {"id": "git"}, {"id": "git"}, + ]}) + self.assertEqual(created.status_code, 201, created.text) + self.assertEqual([item.key for item in self.lifecycle.last_spec.tools], ["git", "python-3.12"]) + self.tools.start.assert_called_once() + self.tools.notify.assert_called_once() + self.assertNotIn("password", created.json()) + + def test_invalid_selections_and_options_are_rejected_before_submission(self): + self.tools.submit = Mock() + invalid = [ + {"tools": []}, {"tools": None}, {"tools": ["git"]}, {"tools": [{"id": "unknown"}]}, + {"tools": [{"id": "python", "version": "3.14"}]}, {"tools": [{"id": "python"}]}, + {"tools": [{"id": "git", "script": "id"}]}, {"tools": [{"id": "git"}], "start_if_stopped": "true"}, + {"tools": [{"id": "git"}], "restart": 1}, {"tools": [{"id": "git"}], "request_id": "../../x"}, + {"tools": [{"id": "git"}], "sudo": False}, + ] + for body in invalid: + with self.subTest(body=body): + response = self.client.post("/vms/vm-test/tools/install", headers=self.headers, json=body) + self.assertEqual(response.status_code, 422, response.text) + self.tools.submit.assert_not_called() + for body in ({"tools": None}, {"tools": [{"id": "not-in-catalog"}]}): + response = self.client.post("/vms", headers=self.headers, json=body) + self.assertEqual(response.status_code, 422) + self.assertIsNone(self.lifecycle.last_spec) + + def test_install_returns_accepted_job_and_whitelists_its_response(self): + job = self.job() + job.update(password="secret", host_path="/host/private") + self.tools.submit = Mock(return_value=job) + result = self.client.post("/vms/vm-test/tools/install", headers=self.headers, + json={"tools": [{"id": "git"}], "start_if_stopped": True, "request_id": "request-1"}) + self.assertEqual(result.status_code, 202, result.text) + self.assertEqual(result.json()["id"], job["id"]) + self.assertNotIn("password", result.json()) + self.assertNotIn("host_path", result.json()) + self.assertTrue(self.tools.submit.call_args.kwargs["start_if_stopped"]) + self.assertEqual(self.tools.submit.call_args.args[0], "vm-test") + + def test_job_history_remains_exact_vm_scoped(self): + job = self.job() + result = self.client.get(f"/vms/vm-test/tools/jobs/{job['id']}", headers=self.headers) + self.assertEqual(result.status_code, 200, result.text) + self.assertEqual(result.json()["items"][0]["status"], "queued") + for identifier in ("test-vm", "10.42.0.2", "vm-other"): + response = self.client.get(f"/vms/{identifier}/tools/jobs/{job['id']}", headers=self.headers) + self.assertEqual(response.status_code, 404) + + def test_server_lifespan_starts_and_stops_workers(self): + with TestClient(self.api) as client: + self.tools.start.assert_called_once() + self.assertEqual(client.get("/tools", headers=self.headers).status_code, 200) + self.tools.shutdown.assert_called_once()