anti
a27e3f5e0f
Two related fixes that came out of running the W5 tests locally:
1. tests/__init__.py — empty file, makes 'tests/' a package so pytest
stops inserting it into sys.path. Without it, 'tests/docker/'
(the docker-image test category) shadowed the installed docker SDK
on every engine-touching test in the repo:
module 'docker' has no attribute 'DockerClient'
Pytest's default --import-mode=prepend was the culprit; making
tests/ a package is the cheapest fix and doesn't change
--import-mode for the whole tree.
2. delete_topology_decky / delete_topology_edge / delete_lan grow an
'enforce_pending: bool = True' kwarg. Default preserves the HTTP
CRUD guard (api_decky_crud / api_edge_crud / api_lan_crud get the
409 for free). apply_remove_decky / apply_detach_decky /
apply_remove_lan now pass enforce_pending=False — the mutator
queue is the live-editing surface and has its own active-topology
gating; the repo's pending-only guard was for design-time CRUD
that mustn't bypass it. Without this, apply_remove_decky was
silently broken on active topologies pre-W5; W5's new test
surfaced it on first run.
10/10 new W5 tests pass; 58/58 across mutator + topology suites.
126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
"""LAN CRUD within a topology."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from sqlalchemy import asc, select, text, update
|
|
|
|
from decnet.web.db.models import LAN, TopologyEdge
|
|
|
|
|
|
class LansMixin:
|
|
"""``self._assert_pending`` / ``self._check_and_bump_version`` resolve
|
|
through ``TopologyCoreMixin`` via MRO."""
|
|
|
|
async def add_lan(
|
|
self,
|
|
data: dict[str, Any],
|
|
*,
|
|
expected_version: Optional[int] = None,
|
|
) -> str:
|
|
async with self._session() as session:
|
|
await self._check_and_bump_version(
|
|
session, data["topology_id"], expected_version
|
|
)
|
|
row = LAN(**data)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row.id
|
|
|
|
async def update_lan(
|
|
self,
|
|
lan_id: str,
|
|
fields: dict[str, Any],
|
|
*,
|
|
expected_version: Optional[int] = None,
|
|
enforce_pending: bool = False,
|
|
) -> None:
|
|
if not fields:
|
|
return
|
|
async with self._session() as session:
|
|
result = await session.execute(
|
|
select(LAN).where(LAN.id == lan_id)
|
|
)
|
|
lan = result.scalar_one_or_none()
|
|
if lan is None:
|
|
raise ValueError(f"lan {lan_id!r} not found")
|
|
if enforce_pending:
|
|
await self._assert_pending(session, lan.topology_id)
|
|
if expected_version is not None:
|
|
await self._check_and_bump_version(
|
|
session, lan.topology_id, expected_version
|
|
)
|
|
await session.execute(
|
|
update(LAN).where(LAN.id == lan_id).values(**fields)
|
|
)
|
|
await session.commit()
|
|
|
|
async def delete_lan(
|
|
self,
|
|
lan_id: str,
|
|
*,
|
|
expected_version: Optional[int] = None,
|
|
enforce_pending: bool = True,
|
|
) -> None:
|
|
"""Cascade-delete a LAN.
|
|
|
|
Rejects if any decky declares this LAN as its home (i.e. has a
|
|
non-bridge edge to it — the only LAN that decky lives in). The
|
|
caller must delete or reassign the home-deckies first.
|
|
|
|
``enforce_pending=True`` by default keeps the HTTP CRUD guard
|
|
intact; the mutator's ``apply_remove_lan`` opts out (it has
|
|
already gated on topology status and the live-LAN docker
|
|
materialisation runs after).
|
|
"""
|
|
from decnet.topology.status import TopologyNotEditable # noqa: F401
|
|
|
|
async with self._session() as session:
|
|
result = await session.execute(select(LAN).where(LAN.id == lan_id))
|
|
lan = result.scalar_one_or_none()
|
|
if lan is None:
|
|
return
|
|
if enforce_pending:
|
|
await self._assert_pending(session, lan.topology_id)
|
|
|
|
# Home-decky check: any decky whose only edge lands here?
|
|
edges_result = await session.execute(
|
|
select(TopologyEdge).where(TopologyEdge.lan_id == lan_id)
|
|
)
|
|
edges_here = edges_result.scalars().all()
|
|
decky_uuids_on_this_lan = {e.decky_uuid for e in edges_here}
|
|
for decky_uuid in decky_uuids_on_this_lan:
|
|
other = await session.execute(
|
|
select(TopologyEdge).where(
|
|
TopologyEdge.decky_uuid == decky_uuid,
|
|
TopologyEdge.lan_id != lan_id,
|
|
)
|
|
)
|
|
if other.scalars().first() is None:
|
|
raise ValueError(
|
|
f"cannot delete LAN {lan.name!r}: decky "
|
|
f"{decky_uuid} has no other LAN (would be orphaned)"
|
|
)
|
|
|
|
if expected_version is not None:
|
|
await self._check_and_bump_version(
|
|
session, lan.topology_id, expected_version
|
|
)
|
|
# Cascade edges → LAN.
|
|
await session.execute(
|
|
text("DELETE FROM topology_edges WHERE lan_id = :l"),
|
|
{"l": lan_id},
|
|
)
|
|
await session.execute(text("DELETE FROM lans WHERE id = :l"), {"l": lan_id})
|
|
await session.commit()
|
|
|
|
async def list_lans_for_topology(
|
|
self, topology_id: str
|
|
) -> list[dict[str, Any]]:
|
|
async with self._session() as session:
|
|
result = await session.execute(
|
|
select(LAN).where(LAN.topology_id == topology_id).order_by(asc(LAN.name))
|
|
)
|
|
return [r.model_dump(mode="json") for r in result.scalars().all()]
|