feat(api/topology): POST/DELETE/deploy endpoints for MazeNET topologies

This commit is contained in:
2026-04-20 19:34:35 -04:00
parent 6db5842a28
commit 999113e3c3
5 changed files with 380 additions and 0 deletions

View File

@@ -10,6 +10,9 @@ live one-per-file and are aggregated here.
from fastapi import APIRouter
from .api_catalog import router as _catalog_router
from .api_create_topology import router as _create_router
from .api_delete_topology import router as _delete_router
from .api_deploy_topology import router as _deploy_router
from .api_get_topology import router as _get_router
from .api_list_topologies import router as _list_router
@@ -22,6 +25,9 @@ topology_router = APIRouter(prefix="/topologies", tags=["topologies"])
# parameterized fallback.
topology_router.include_router(_catalog_router)
topology_router.include_router(_list_router)
topology_router.include_router(_create_router)
topology_router.include_router(_deploy_router)
topology_router.include_router(_delete_router)
topology_router.include_router(_get_router)

View File

@@ -0,0 +1,60 @@
"""POST /topologies — generate and persist a new MazeNET topology."""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, status
from decnet.telemetry import traced as _traced
from decnet.topology.allocator import reserved_subnets
from decnet.topology.config import TopologyConfig
from decnet.topology.generator import generate
from decnet.topology.persistence import persist
from decnet.web.db.models import TopologyGenerateRequest, TopologySummary
from decnet.web.dependencies import repo, require_admin
router = APIRouter()
@router.post(
"/",
tags=["MazeNET Topologies"],
response_model=TopologySummary,
status_code=status.HTTP_201_CREATED,
responses={
400: {"description": "Malformed or invalid generation parameters"},
401: {"description": "Missing or invalid credentials"},
403: {"description": "Insufficient permissions"},
409: {"description": "Generator could not allocate subnets (exhausted pool)"},
},
)
@_traced("api.topology.create")
async def api_create_topology(
body: TopologyGenerateRequest,
_admin: dict = Depends(require_admin),
) -> TopologySummary:
try:
config = TopologyConfig(
name=body.name,
depth=body.depth,
branching_factor=body.branching_factor,
deckies_per_lan_min=body.deckies_per_lan_min,
deckies_per_lan_max=body.deckies_per_lan_max,
bridge_forward_probability=body.bridge_forward_probability,
cross_edge_probability=body.cross_edge_probability,
services_explicit=body.services_explicit,
randomize_services=body.randomize_services,
seed=body.seed,
)
except (ValueError, TypeError) as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
try:
plan = generate(config, reserved_subnets=await reserved_subnets(repo))
except RuntimeError as exc:
# Subnet allocator exhaustion or similar planner-level failure.
raise HTTPException(status_code=409, detail=str(exc)) from exc
except (ValueError, TypeError) as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
topology_id = await persist(repo, plan)
row = await repo.get_topology(topology_id)
return TopologySummary(**row)

View File

@@ -0,0 +1,51 @@
"""DELETE /topologies/{id} — cascade-delete a pending or torn-down topology."""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Response, status
from decnet.telemetry import traced as _traced
from decnet.topology.status import TopologyStatus
from decnet.web.dependencies import repo, require_admin
router = APIRouter()
# Only allow delete when containers are guaranteed not to be running.
# ACTIVE / DEPLOYING / DEGRADED / TEARING_DOWN must teardown first.
_DELETABLE: frozenset[str] = frozenset(
{TopologyStatus.PENDING, TopologyStatus.TORN_DOWN, TopologyStatus.FAILED}
)
@router.delete(
"/{topology_id}",
tags=["MazeNET Topologies"],
status_code=status.HTTP_204_NO_CONTENT,
responses={
400: {"description": "Malformed path parameters"},
401: {"description": "Missing or invalid credentials"},
403: {"description": "Insufficient permissions"},
404: {"description": "Topology not found"},
409: {"description": "Topology has running resources; teardown first"},
},
)
@_traced("api.topology.delete")
async def api_delete_topology(
topology_id: str,
_admin: dict = Depends(require_admin),
) -> Response:
topo = await repo.get_topology(topology_id)
if topo is None:
raise HTTPException(status_code=404, detail="Topology not found")
if topo["status"] not in _DELETABLE:
raise HTTPException(
status_code=409,
detail=(
f"Topology is {topo['status']!r}; teardown to 'torn_down' "
f"before delete."
),
)
deleted = await repo.delete_topology_cascade(topology_id)
if not deleted:
# Race: row vanished between the status check and the cascade.
raise HTTPException(status_code=404, detail="Topology not found")
return Response(status_code=status.HTTP_204_NO_CONTENT)

View File

@@ -0,0 +1,72 @@
"""POST /topologies/{id}/deploy — transition pending → deploying and fire
the background deploy.
The actual Docker work happens in a BackgroundTask so the HTTP caller
returns quickly with ``202 Accepted``. Status transitions
(``deploying`` → ``active`` | ``failed``) are written by
:func:`decnet.engine.deployer.deploy_topology` itself.
"""
from __future__ import annotations
import asyncio
import logging
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status
from decnet.engine.deployer import deploy_topology
from decnet.telemetry import traced as _traced
from decnet.topology.status import TopologyStatus
from decnet.web.db.models import TopologySummary
from decnet.web.dependencies import repo, require_admin
log = logging.getLogger(__name__)
router = APIRouter()
async def _run_deploy(topology_id: str) -> None:
"""BackgroundTask body: deploy, swallow + log any exception so the
task runner doesn't crash. Status on failure is marked by
:func:`deploy_topology` via its own exception handler.
"""
try:
await deploy_topology(repo, topology_id)
except asyncio.CancelledError: # pragma: no cover — shutdown
raise
except Exception as exc: # noqa: BLE001
log.error("background deploy of %s failed: %s", topology_id, exc)
@router.post(
"/{topology_id}/deploy",
tags=["MazeNET Topologies"],
response_model=TopologySummary,
status_code=status.HTTP_202_ACCEPTED,
responses={
400: {"description": "Malformed path parameters"},
401: {"description": "Missing or invalid credentials"},
403: {"description": "Insufficient permissions"},
404: {"description": "Topology not found"},
409: {"description": "Topology is not in 'pending' status"},
},
)
@_traced("api.topology.deploy")
async def api_deploy_topology(
topology_id: str,
background: BackgroundTasks,
_admin: dict = Depends(require_admin),
) -> TopologySummary:
topo = await repo.get_topology(topology_id)
if topo is None:
raise HTTPException(status_code=404, detail="Topology not found")
if topo["status"] != TopologyStatus.PENDING:
raise HTTPException(
status_code=409,
detail=(
f"Topology is {topo['status']!r}; only 'pending' topologies "
f"can be deployed."
),
)
background.add_task(_run_deploy, topology_id)
return TopologySummary(**topo)

View File

@@ -0,0 +1,191 @@
"""Phase 3 Step 3 — write endpoints: create / delete / deploy."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
import pytest
from decnet.topology.config import TopologyConfig
from decnet.topology.generator import generate
from decnet.topology.persistence import persist, transition_status
from decnet.topology.status import TopologyStatus
from decnet.web.dependencies import repo as _repo
_V1 = "/api/v1/topologies"
def _generate_payload(name: str = "from-api") -> dict:
return {
"name": name,
"depth": 1,
"branching_factor": 1,
"deckies_per_lan_min": 1,
"deckies_per_lan_max": 1,
"services_explicit": ["ssh"],
"randomize_services": False,
"seed": 1,
}
def _cfg(name: str = "draft") -> TopologyConfig:
return TopologyConfig(
name=name,
depth=1,
branching_factor=1,
deckies_per_lan_min=1,
deckies_per_lan_max=1,
services_explicit=["ssh"],
randomize_services=False,
seed=0,
)
async def _seed(name: str = "draft") -> str:
return await persist(_repo, generate(_cfg(name)))
# ── POST /topologies ──────────────────────────────────────────────
@pytest.mark.anyio
async def test_create_ok(client, auth_token):
r = await client.post(
f"{_V1}/",
json=_generate_payload(),
headers={"Authorization": f"Bearer {auth_token}"},
)
assert r.status_code == 201, r.text
body = r.json()
assert body["status"] == TopologyStatus.PENDING
assert body["name"] == "from-api"
# Children were persisted.
lans = await _repo.list_lans_for_topology(body["id"])
assert len(lans) >= 1
@pytest.mark.anyio
async def test_create_requires_admin(client, viewer_token):
r = await client.post(
f"{_V1}/",
json=_generate_payload(),
headers={"Authorization": f"Bearer {viewer_token}"},
)
assert r.status_code == 403
@pytest.mark.anyio
async def test_create_requires_auth(client):
r = await client.post(f"{_V1}/", json=_generate_payload())
assert r.status_code == 401
@pytest.mark.anyio
async def test_create_bad_body(client, auth_token):
r = await client.post(
f"{_V1}/",
json={"name": "x"}, # missing required fields
headers={"Authorization": f"Bearer {auth_token}"},
)
# Project-wide validation handler: missing fields → 400 (not 422).
assert r.status_code == 400
# ── DELETE /topologies/{id} ───────────────────────────────────────
@pytest.mark.anyio
async def test_delete_pending_ok(client, auth_token):
topology_id = await _seed("for-delete")
r = await client.delete(
f"{_V1}/{topology_id}",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert r.status_code == 204
assert await _repo.get_topology(topology_id) is None
@pytest.mark.anyio
async def test_delete_active_blocked(client, auth_token):
topology_id = await _seed("for-delete-active")
await transition_status(_repo, topology_id, TopologyStatus.DEPLOYING)
await transition_status(_repo, topology_id, TopologyStatus.ACTIVE)
r = await client.delete(
f"{_V1}/{topology_id}",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert r.status_code == 409
assert await _repo.get_topology(topology_id) is not None
@pytest.mark.anyio
async def test_delete_missing_404(client, auth_token):
r = await client.delete(
f"{_V1}/does-not-exist",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert r.status_code == 404
@pytest.mark.anyio
async def test_delete_requires_admin(client, viewer_token):
topology_id = await _seed("viewer-delete")
r = await client.delete(
f"{_V1}/{topology_id}",
headers={"Authorization": f"Bearer {viewer_token}"},
)
assert r.status_code == 403
# ── POST /topologies/{id}/deploy ──────────────────────────────────
@pytest.mark.anyio
async def test_deploy_accepts_pending(client, auth_token):
topology_id = await _seed("for-deploy")
with patch(
"decnet.web.router.topology.api_deploy_topology.deploy_topology",
new=AsyncMock(return_value=None),
) as mock_deploy:
r = await client.post(
f"{_V1}/{topology_id}/deploy",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert r.status_code == 202, r.text
body = r.json()
assert body["id"] == topology_id
# BackgroundTasks run after the response, so the mock must have been invoked
# by the time the client context exits.
mock_deploy.assert_called_once()
@pytest.mark.anyio
async def test_deploy_non_pending_blocked(client, auth_token):
topology_id = await _seed("for-deploy-blocked")
await transition_status(_repo, topology_id, TopologyStatus.DEPLOYING)
r = await client.post(
f"{_V1}/{topology_id}/deploy",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert r.status_code == 409
@pytest.mark.anyio
async def test_deploy_missing_404(client, auth_token):
r = await client.post(
f"{_V1}/missing/deploy",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert r.status_code == 404
@pytest.mark.anyio
async def test_deploy_requires_admin(client, viewer_token):
topology_id = await _seed("viewer-deploy")
r = await client.post(
f"{_V1}/{topology_id}/deploy",
headers={"Authorization": f"Bearer {viewer_token}"},
)
assert r.status_code == 403