nest/tests/test_tool_api.py

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.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()