128 lines
3.8 KiB
Python
128 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import stat
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi import APIRouter, HTTPException
|
|
from fastapi.routing import APIRoute
|
|
|
|
from k1link.web.compute_contour_api import (
|
|
CATALOG_SCHEMA,
|
|
ComputeContourCreate,
|
|
ComputeContourPut,
|
|
ComputeContourStore,
|
|
build_compute_contour_router,
|
|
)
|
|
|
|
|
|
def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]:
|
|
for route in router.routes:
|
|
if (
|
|
isinstance(route, APIRoute)
|
|
and route.path == path
|
|
and route.methods is not None
|
|
and method in route.methods
|
|
):
|
|
return route.endpoint
|
|
raise AssertionError(f"{method} {path} route is missing")
|
|
|
|
|
|
def test_contour_store_migrates_worker_006_as_first_configuration(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = ComputeContourStore(tmp_path / "system")
|
|
contours = store.list_contours()
|
|
|
|
assert len(contours) == 1
|
|
assert contours[0].contour_id == "worker-006"
|
|
assert contours[0].telemetry_mode == "agent-mqtt"
|
|
assert not store.path.exists()
|
|
|
|
|
|
def test_contour_store_creates_and_updates_private_catalog(tmp_path: Path) -> None:
|
|
store = ComputeContourStore(tmp_path / "system")
|
|
created = store.create(
|
|
ComputeContourCreate(
|
|
display_name="Field Worker",
|
|
expected_node_id="FIELD-01",
|
|
platform="linux",
|
|
address="192.0.2.25",
|
|
mqtt_host="192.0.2.5",
|
|
)
|
|
)
|
|
updated = store.update(
|
|
created.contour_id,
|
|
ComputeContourPut(
|
|
revision=created.revision,
|
|
display_name="Field Worker 01",
|
|
expected_node_id="FIELD-01",
|
|
platform="linux",
|
|
telemetry_mode="agent-mqtt",
|
|
address="192.0.2.25",
|
|
ssh_port=22,
|
|
mqtt_host="192.0.2.5",
|
|
mqtt_port=1883,
|
|
),
|
|
)
|
|
|
|
assert updated.display_name == "Field Worker 01"
|
|
assert updated.revision == 1
|
|
assert stat.S_IMODE(store.path.stat().st_mode) == 0o600
|
|
assert len(store.list_contours()) == 2
|
|
with pytest.raises(RuntimeError, match="revision changed"):
|
|
store.update(
|
|
created.contour_id,
|
|
ComputeContourPut(
|
|
revision=0,
|
|
display_name="stale",
|
|
expected_node_id="FIELD-01",
|
|
platform="linux",
|
|
telemetry_mode="agent-mqtt",
|
|
address="",
|
|
ssh_port=22,
|
|
mqtt_host="127.0.0.1",
|
|
mqtt_port=1883,
|
|
),
|
|
)
|
|
|
|
|
|
def test_contour_router_exposes_catalog_and_safe_install_contract(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
router = build_compute_contour_router(
|
|
root_provider=lambda: tmp_path / "system"
|
|
)
|
|
list_contours = _endpoint(router, "/api/v1/system/contours", "GET")
|
|
install = _endpoint(
|
|
router,
|
|
"/api/v1/system/contours/{contour_id}/agent-install",
|
|
"GET",
|
|
)
|
|
|
|
catalog = list_contours()
|
|
assert catalog["schema_version"] == CATALOG_SCHEMA
|
|
assert catalog["contours"][0]["contour_id"] == "worker-006"
|
|
document = install("worker-006")
|
|
assert document["agent"]["distribution"] == "Telegraf"
|
|
assert "MQTT password" in document["command"]
|
|
assert "password" not in document["agent"]["environment"]
|
|
assert document["ready"] is False
|
|
|
|
|
|
def test_contour_router_returns_404_for_unknown_contour(tmp_path: Path) -> None:
|
|
router = build_compute_contour_router(
|
|
root_provider=lambda: tmp_path / "system"
|
|
)
|
|
install = _endpoint(
|
|
router,
|
|
"/api/v1/system/contours/{contour_id}/agent-install",
|
|
"GET",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
install("missing")
|
|
assert error.value.status_code == 404
|