From 549c24514e524bda31041b0c386d7e557865251d Mon Sep 17 00:00:00 2001 From: aidankhogg Date: Tue, 30 Jun 2026 00:29:49 +0100 Subject: [PATCH] Implement org app deployment layer --- netengine/handlers/app_handler.py | 364 ++++++++++++++++++++---------- tests/test_app_handler.py | 200 ++++++++++++++++ 2 files changed, 443 insertions(+), 121 deletions(-) create mode 100644 tests/test_app_handler.py diff --git a/netengine/handlers/app_handler.py b/netengine/handlers/app_handler.py index 1285a51..f848ef2 100644 --- a/netengine/handlers/app_handler.py +++ b/netengine/handlers/app_handler.py @@ -1,4 +1,9 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass, field from datetime import UTC, datetime +from pathlib import Path from typing import Any, cast from netengine.core.pgmq_client import PGMQClient @@ -13,6 +18,82 @@ from netengine.spec.models import NetEngineSpec +@dataclass(frozen=True) +class AppOIDCSettings: + enabled: bool = True + public: bool = True + realm: str | None = None + client_id_template: str = "{org}-{app}" + redirect_uri_template: str = "https://{domain}/*" + + +@dataclass(frozen=True) +class AppDNSSettings: + enabled: bool = True + ttl: int = 300 + zone_template: str = "{org}.internal" + domain_template: str = "{subdomain}.{org}.internal" + + +@dataclass(frozen=True) +class AppCatalogEntry: + name: str + image: str + ports: list[int] = field(default_factory=list) + environment: dict[str, str] = field(default_factory=dict) + volumes: dict[str, Any] = field(default_factory=dict) + healthcheck_path: str = "/" + oidc: AppOIDCSettings = field(default_factory=AppOIDCSettings) + dns: AppDNSSettings = field(default_factory=AppDNSSettings) + + +class AppCatalog: + """Catalog of deployable org applications with runtime deployment metadata.""" + + _BUILTINS: dict[str, AppCatalogEntry] = { + "gitea": AppCatalogEntry("gitea", "gitea/gitea:latest", [3000], healthcheck_path="/"), + "wordpress": AppCatalogEntry("wordpress", "wordpress:latest", [80], healthcheck_path="/"), + "nextcloud": AppCatalogEntry( + "nextcloud", "nextcloud:latest", [80], healthcheck_path="/status.php" + ), + } + + def __init__(self, entries: list[Any] | None = None) -> None: + self._entries = dict(self._BUILTINS) + for entry in entries or []: + parsed = self._from_spec(entry) + self._entries[parsed.name] = parsed + + def get(self, app_name: str) -> AppCatalogEntry: + return self._entries.get(app_name, AppCatalogEntry(app_name, app_name, [])) + + def _from_spec(self, entry: Any) -> AppCatalogEntry: + data = entry.model_dump() if hasattr(entry, "model_dump") else dict(entry) + oidc_raw = data.get("oidc") or data.get("oidc_settings") or {} + dns_raw = data.get("dns") or data.get("dns_settings") or {} + return AppCatalogEntry( + name=data["name"], + image=data["image"], + ports=list(data.get("ports") or ([data["port"]] if data.get("port") else [])), + environment=dict(data.get("environment") or data.get("env") or {}), + volumes=dict(data.get("volumes") or {}), + healthcheck_path=data.get("healthcheck_path") or data.get("healthcheck") or "/", + oidc=AppOIDCSettings( + enabled=bool(data.get("oidc_integration", oidc_raw.get("enabled", True))), + public=bool(oidc_raw.get("public", True)), + realm=oidc_raw.get("realm"), + client_id_template=oidc_raw.get("client_id_template", "{org}-{app}"), + redirect_uri_template=oidc_raw.get("redirect_uri_template", "https://{domain}/*"), + ), + dns=AppDNSSettings( + enabled=bool(dns_raw.get("enabled", True)), + ttl=int(dns_raw.get("ttl", 300)), + zone_template=dns_raw.get("zone_template", "{org}.internal"), + domain_template=dns_raw.get("domain_template", "{subdomain}.{org}.internal"), + ), + ) + + class AppHandler: def __init__( self, @@ -23,6 +104,7 @@ def __init__( state: Any, context: PhaseContext | None = None, pgmq: PGMQAdapterProtocol | None = None, + catalog: AppCatalog | None = None, ) -> None: self.context = context or PhaseContext( spec=cast(NetEngineSpec, {}), runtime_state=state, logger=get_logger(__name__) @@ -32,6 +114,10 @@ def __init__( self.pki = pki self.oidc = oidc self.state = state + self.catalog = catalog or AppCatalog( + getattr(getattr(self.context, "spec", None), "org_apps", None) + and getattr(self.context.spec.org_apps, "catalog", []) + ) self._db = None self.pgmq = pgmq or self.context.pgmq_client or PGMQClient() @@ -45,31 +131,70 @@ async def _get_db(self) -> Any: async def deploy_app( self, org: str, app_name: str, subdomain: str, config: dict[str, Any] | None = None ) -> dict[str, Any]: - if config is None: - config = {} - """4‑step deployment: container → DNS → cert → OIDC.""" - # Step 1: Determine AND bridge for this org - and_name = f"{org.replace('_', '-')}-net" - # Step 2: Start container inside the org's AND bridge + config = config or {} + entry = self.catalog.get(app_name) + now = datetime.now(UTC).isoformat() container_name = f"netengines_{org}_{app_name}" - # Use the appropriate image from catalog - image = self._get_app_image(app_name) - # We need to attach to the AND bridge. - # Start container on the AND bridge (not core). - container_id = await self._start_app_container(container_name, image, and_name, config) - - # Step 3: Register DNS (e.g., gitea.acme.internal) - domain = f"{subdomain}.{org}.internal" - # Get gateway IP for this AND (from address_leases) - gateway_ip = await self._get_gateway_ip(and_name) - await self.dns.add_zone_record( - self.context, f"{org}.internal", "A", subdomain, gateway_ip, 300 - ) + domain = entry.dns.domain_template.format(org=org, app=app_name, subdomain=subdomain) + client_id = entry.oidc.client_id_template.format(org=org, app=app_name, domain=domain) + deployment = { + "org": org, + "app": app_name, + "domain": domain, + "container_id": None, + "client_id": client_id if entry.oidc.enabled else None, + "status": "deploying", + "failure_detail": None, + "deployed_at": now, + "updated_at": now, + } + await self._persist_deployment(deployment) + try: + cert, key = await self.pki.issue_cert(domain, [f"*.{org}.internal"]) + cert_mount = await self._prepare_cert_mount(container_name, domain, cert, key) + and_name = f"{org.replace('_', '-')}-net" + container_id = await self._start_app_container( + container_name, entry, and_name, config, cert_mount + ) + deployment["container_id"] = container_id + if entry.dns.enabled: + gateway_ip = await self._get_gateway_ip(and_name) + zone = entry.dns.zone_template.format(org=org, app=app_name, subdomain=subdomain) + await self.dns.add_zone_record( + self.context, zone, "A", subdomain, gateway_ip, entry.dns.ttl + ) + self._record_certificate(domain, org, cert) + if entry.oidc.enabled: + await self.oidc.create_client( + realm=self._oidc_realm(org, entry), + client_id=client_id, + name=f"{org} {app_name}", + redirect_uris=[ + entry.oidc.redirect_uri_template.format( + domain=domain, org=org, app=app_name + ) + ], + public=entry.oidc.public, + ) + deployment.update({"status": "deployed", "updated_at": datetime.now(UTC).isoformat()}) + await self._persist_deployment(deployment) + return deployment + except Exception as exc: + deployment.update( + { + "status": "failed", + "failure_detail": str(exc), + "updated_at": datetime.now(UTC).isoformat(), + } + ) + await self._persist_deployment(deployment) + raise - # Step 4: Issue TLS certificate via PKI - cert, key = await self.pki.issue_cert(domain, [f"*.{org}.internal"]) + async def _persist_deployment(self, deployment: dict[str, Any]) -> None: + db = await self._get_db() + await db.table("app_deployments").upsert(dict(deployment)).execute() - # Track issued certificate in RuntimeState + def _record_certificate(self, domain: str, org: str, cert: str) -> None: expiry = self.pki.extract_cert_expiry(cert) self.context.runtime_state.issued_certificates[domain] = { "cert_type": "app", @@ -81,113 +206,95 @@ async def deploy_app( } self.context.runtime_state.save() - # Mount cert into container (via volume or exec write) - await self._inject_cert(container_id, domain, cert, key) - - # Step 5: Create OIDC client in in‑world Keycloak realm - client_id = f"{org}-{app_name}" - await self.oidc.create_client( - realm="inworld", - client_id=client_id, - name=f"{org} {app_name}", - redirect_uris=[f"https://{domain}/*"], - public=True, + async def _prepare_cert_mount( + self, container_name: str, domain: str, cert: str, key: str + ) -> dict[str, Any]: + cert_dir = ( + Path(os.environ.get("NETENGINE_APP_CERT_DIR", "/var/lib/netengines/certs")) + / container_name ) - - # Store deployment metadata - deployment = { - "org": org, - "app": app_name, - "domain": domain, - "container_id": container_id, - "client_id": client_id, - "deployed_at": datetime.now(UTC).isoformat(), - } - db = await self._get_db() - await db.table("app_deployments").upsert(deployment).execute() - - return deployment + cert_dir.mkdir(parents=True, exist_ok=True) + (cert_dir / "tls.crt").write_text(cert) + (cert_dir / "tls.key").write_text(key) + return {str(cert_dir): {"bind": "/certs", "mode": "ro"}} async def _start_app_container( - self, name: str, image: str, network: str, config: dict[str, Any] + self, + name: str, + entry: AppCatalogEntry, + network: str, + config: dict[str, Any], + cert_mount: dict[str, Any], ) -> str: - """Start a container on the given AND bridge.""" - # The container should not be attached to core – only to its AND. - # We'll create it with no network, then attach. + volumes = {**entry.volumes, **config.get("volumes", {}), **cert_mount} + env = {**entry.environment, **config.get("environment", {})} container_id = await self.docker.start_container( name=name, - image=image, + image=entry.image, command=config.get("command", []), - volumes=config.get("volumes", {}), - network=None, # not attached yet + volumes=volumes, + network=None, ip=None, - environment=config.get("environment", {}), + environment=env, + ports=entry.ports, + healthcheck_path=entry.healthcheck_path, ) - # Attach to the AND bridge - bridge_name = f"netengines_and_{network}" - # Assign an IP from the bridge's subnet (we can let Docker assign dynamically) - await self.docker.connect_network(container_id, bridge_name, ip=None) + await self.docker.connect_network(container_id, f"netengines_and_{network}", ip=None) return container_id + def _oidc_realm(self, org: str, entry: AppCatalogEntry) -> str: + if entry.oidc.realm: + return entry.oidc.realm.format(org=org) + output = self.context.runtime_state.identity_inworld_output or {} + if "realm_name" in output: + return str(output["realm_name"]) + realms = output.get("realms_created") or [] + org_realm = f"{org}-realm" + return ( + org_realm + if org_realm in realms + else getattr(self.context.spec.identity_inworld, "realm_name", "inworld") + ) + + async def update_app( + self, deployment: dict[str, Any], config: dict[str, Any] | None = None + ) -> dict[str, Any]: + return await self.deploy_app( + deployment["org"], deployment["app"], deployment["domain"].split(".")[0], config or {} + ) + + async def restart_app(self, container_id: str) -> None: + await self.docker.stop_container(container_id) + await self.docker.start_container( + name=container_id, + image="", + command=[], + volumes={}, + network=None, + ip=None, + environment={}, + ) + + async def teardown_app(self, deployment: dict[str, Any]) -> None: + if deployment.get("container_id"): + await self.docker.stop_container(deployment["container_id"]) + deployment.update({"status": "torn_down", "updated_at": datetime.now(UTC).isoformat()}) + await self._persist_deployment(deployment) + + async def healthcheck_app(self, container_id: str, path: str = "/") -> bool: + code, _ = await self.docker.exec_command( + container_id, ["wget", "-q", "-O", "-", f"http://127.0.0.1{path}"] + ) + return code == 0 + async def _get_gateway_ip(self, and_name: str) -> str: - """Query DB for the CIDR of this AND and derive gateway IP.""" import ipaddress db = await self._get_db() result = await db.table("address_leases").select("cidr").eq("and_name", and_name).execute() if not result.data: raise ServicesError(f"AND {and_name} not found") - cidr = result.data[0]["cidr"] - # Gateway IP is the first usable IP in the CIDR block - network = ipaddress.ip_network(cidr, strict=False) - gateway_ip = str(network.network_address + 1) - return gateway_ip - - async def _inject_cert(self, container_id: str, domain: str, cert: str, key: str) -> None: - """Write cert and key into the container (via volume or exec).""" - # For simplicity, we use a volume mount shared between app and host. - # Or we can write via `docker exec` with `cat` redirection. - # We'll use exec to write to /etc/ssl/certs. - # Create temporary files and copy them. - # We'll use a shared directory on the host mounted into the container. - # For MVP, we'll just rely on the container image to fetch certs via volume. - # We'll store certs in a volume named `netengines_app_certs_{container_id}` - volume_name = f"netengines_certs_{container_id}" - await self.docker.ensure_volume(volume_name) - # Write cert/key to the volume using a temporary container - # Or we can mount the volume and write using host filesystem. - # Simpler: we'll mount the volume to the app container and put certs there. - # But we need the container to be restarted. - # For now, we assume the app container mounts /certs and reads from there. - # We'll write the files now. - import os - - cert_dir = f"/var/lib/netengines/certs/{container_id}" - os.makedirs(cert_dir, exist_ok=True) - with open(f"{cert_dir}/tls.crt", "w") as f: - f.write(cert) - with open(f"{cert_dir}/tls.key", "w") as f: - f.write(key) - # Now we need to mount this directory into the running container. - # Docker doesn't allow mounting new volumes to a running container. - # Option A: Stop container, re‑create with mount, start again. - # Option B: Use `docker cp` to copy files into the running container. - # We'll use `docker cp`. - await self.docker.copy_to_container( - container_id, f"{cert_dir}/tls.crt", "/etc/ssl/certs/tls.crt" - ) - await self.docker.copy_to_container( - container_id, f"{cert_dir}/tls.key", "/etc/ssl/private/tls.key" - ) - - def _get_app_image(self, app_name: str) -> str: - """Map catalog app names to Docker images.""" - catalog = { - "gitea": "gitea/gitea:latest", - "wordpress": "wordpress:latest", - "nextcloud": "nextcloud:latest", - } - return catalog.get(app_name, app_name) + return str(ipaddress.ip_network(result.data[0]["cidr"], strict=False).network_address + 1) class OrgAppsPhaseHandler(BasePhaseHandler): @@ -199,26 +306,41 @@ async def execute(self, context: PhaseContext) -> None: context.logger.info("Phase 9: org_apps not enabled, skipping deployments") context.runtime_state.org_apps_output = cast(Any, {"deployments": []}) return - docker = context.docker_client - dns = DNSHandler() if docker is None: raise ServicesError("docker_client is required for org app deployments") - pki = PKIHandler(docker, context.runtime_state, context.spec) + inworld_spec = context.spec.identity_inworld + identity_output = context.runtime_state.identity_inworld_output or {} + keycloak_url = ( + identity_output.get("keycloak_url") + or identity_output.get("issuer", "").split("/realms/")[0] + or f"https://{inworld_spec.canonical_name}" + ) + admin_password = ( + identity_output.get("admin_password") + or context.runtime_state.inworld_admin_password + or "" + ) oidc = OIDCHandler( - keycloak_url="https://auth.platform.internal", - admin_username="admin", - admin_password=context.runtime_state.inworld_admin_password or "", + keycloak_url=keycloak_url, + admin_username=identity_output.get("admin_username", "admin"), + admin_password=admin_password, ) - handler = AppHandler(docker, dns, pki, oidc, context.runtime_state, context) - - deployments: list[dict[str, Any]] = [] + handler = AppHandler( + docker, + DNSHandler(), + PKIHandler(docker, context.runtime_state, context.spec), + oidc, + context.runtime_state, + context, + catalog=AppCatalog(org_apps_spec.catalog), + ) + deployments = [] for dep in org_apps_spec.deployments: subdomain = dep.subdomain or dep.app result = await handler.deploy_app(dep.org, dep.app, subdomain, {}) deployments.append(result) context.logger.info(f"Phase 9: deployed {dep.app} for {dep.org} at {result['domain']}") - context.runtime_state.org_apps_output = cast(Any, {"deployments": deployments}) async def healthcheck(self, context: PhaseContext) -> bool: diff --git a/tests/test_app_handler.py b/tests/test_app_handler.py new file mode 100644 index 0000000..76d78ad --- /dev/null +++ b/tests/test_app_handler.py @@ -0,0 +1,200 @@ +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from netengine.handlers.app_handler import AppCatalog, AppHandler, OrgAppsPhaseHandler +from netengine.handlers.context import PhaseContext, RuntimeState +from netengine.logs import get_logger +from netengine.spec.models import AppCatalogEntry as SpecCatalogEntry +from netengine.spec.models import AppDeployment, IdentityInWorldPhase, OrgAppsPhase + + +class Result: + data = [{"cidr": "10.42.0.0/24"}] + + +def mock_db(): + db = MagicMock() + table = MagicMock() + db.table.return_value = table + table.upsert.return_value.execute = AsyncMock(return_value=SimpleNamespace(data=[])) + table.select.return_value.eq.return_value.execute = AsyncMock(return_value=Result()) + return db + + +def context(tmp_path): + state = RuntimeState() + state.identity_inworld_output = { + "issuer": "https://auth.acme.internal/realms/inworld", + "realm_name": "inworld", + "admin_username": "kc-admin", + "admin_password": "from-state", + } + spec = SimpleNamespace( + identity_inworld=IdentityInWorldPhase(canonical_name="auth.internal", realm_name="inworld"), + org_apps=OrgAppsPhase( + catalog=[ + SpecCatalogEntry(name="demo", image="demo:1", port=8080, oidc_integration=True) + ], + deployments=[AppDeployment(org="acme", app="demo", subdomain="demo")], + ), + ) + return PhaseContext( + spec=spec, runtime_state=state, logger=get_logger("test"), docker_client=AsyncMock() + ) + + +def deps(tmp_path, ctx): + docker = AsyncMock() + docker.start_container = AsyncMock(return_value="ctr-1") + docker.connect_network = AsyncMock() + docker.exec_command = AsyncMock(return_value=(0, "ok")) + dns = AsyncMock() + pki = AsyncMock() + pki.issue_cert = AsyncMock(return_value=("CERT", "KEY")) + pki.extract_cert_expiry = MagicMock(return_value=datetime.now(UTC) + timedelta(days=1)) + oidc = AsyncMock() + oidc.create_client = AsyncMock() + handler = AppHandler( + docker, + dns, + pki, + oidc, + ctx.runtime_state, + ctx, + catalog=AppCatalog(ctx.spec.org_apps.catalog), + ) + handler._db = mock_db() + return handler, docker, dns, pki, oidc + + +@pytest.mark.asyncio +async def test_deploy_app_uses_catalog_and_persists_status(tmp_path, monkeypatch): + monkeypatch.setenv("NETENGINE_APP_CERT_DIR", str(tmp_path)) + ctx = context(tmp_path) + handler, docker, dns, _pki, oidc = deps(tmp_path, ctx) + + result = await handler.deploy_app("acme", "demo", "demo") + + assert result["status"] == "deployed" + docker.start_container.assert_awaited_once() + assert docker.start_container.call_args.kwargs["image"] == "demo:1" + assert docker.start_container.call_args.kwargs["ports"] == [8080] + dns.add_zone_record.assert_awaited_once() + oidc.create_client.assert_awaited_once_with( + realm="inworld", + client_id="acme-demo", + name="acme demo", + redirect_uris=["https://demo.acme.internal/*"], + public=True, + ) + upserts = handler._db.table.return_value.upsert.call_args_list + assert [c.args[0]["status"] for c in upserts] == ["deploying", "deployed"] + + +@pytest.mark.asyncio +async def test_certificate_is_mounted_before_container_start(tmp_path, monkeypatch): + monkeypatch.setenv("NETENGINE_APP_CERT_DIR", str(tmp_path)) + ctx = context(tmp_path) + handler, docker, *_ = deps(tmp_path, ctx) + + await handler.deploy_app("acme", "demo", "demo") + + volumes = docker.start_container.call_args.kwargs["volumes"] + cert_dir = tmp_path / "netengines_acme_demo" + assert (cert_dir / "tls.crt").read_text() == "CERT" + assert volumes[str(cert_dir)] == {"bind": "/certs", "mode": "ro"} + assert not hasattr(docker, "copy_to_container") or docker.copy_to_container.await_count == 0 + + +@pytest.mark.asyncio +async def test_dns_failure_persists_failure_detail(tmp_path, monkeypatch): + monkeypatch.setenv("NETENGINE_APP_CERT_DIR", str(tmp_path)) + ctx = context(tmp_path) + handler, _docker, dns, *_ = deps(tmp_path, ctx) + dns.add_zone_record.side_effect = RuntimeError("dns down") + + with pytest.raises(RuntimeError, match="dns down"): + await handler.deploy_app("acme", "demo", "demo") + + failed = handler._db.table.return_value.upsert.call_args_list[-1].args[0] + assert failed["status"] == "failed" + assert failed["failure_detail"] == "dns down" + + +@pytest.mark.asyncio +async def test_oidc_client_creation_failure_persists_failure(tmp_path, monkeypatch): + monkeypatch.setenv("NETENGINE_APP_CERT_DIR", str(tmp_path)) + ctx = context(tmp_path) + handler, _docker, _dns, _pki, oidc = deps(tmp_path, ctx) + oidc.create_client.side_effect = RuntimeError("oidc failed") + + with pytest.raises(RuntimeError, match="oidc failed"): + await handler.deploy_app("acme", "demo", "demo") + + failed = handler._db.table.return_value.upsert.call_args_list[-1].args[0] + assert failed["status"] == "failed" + assert failed["failure_detail"] == "oidc failed" + + +@pytest.mark.asyncio +async def test_idempotent_redeploy_uses_stable_name_and_domain(tmp_path, monkeypatch): + monkeypatch.setenv("NETENGINE_APP_CERT_DIR", str(tmp_path)) + ctx = context(tmp_path) + handler, docker, *_ = deps(tmp_path, ctx) + + first = await handler.deploy_app("acme", "demo", "demo") + second = await handler.deploy_app("acme", "demo", "demo") + + assert first["domain"] == second["domain"] == "demo.acme.internal" + names = [call.kwargs["name"] for call in docker.start_container.call_args_list] + assert names == ["netengines_acme_demo", "netengines_acme_demo"] + + +@pytest.mark.asyncio +async def test_org_apps_phase_uses_runtime_identity_outputs(tmp_path, monkeypatch): + monkeypatch.setenv("NETENGINE_APP_CERT_DIR", str(tmp_path)) + ctx = context(tmp_path) + ctx.docker_client = AsyncMock() + ctx.docker_client.start_container = AsyncMock(return_value="ctr-1") + ctx.docker_client.connect_network = AsyncMock() + + with ( + patch("netengine.handlers.app_handler.OIDCHandler") as oidc_cls, + patch("netengine.handlers.app_handler.PKIHandler") as pki_cls, + patch("netengine.handlers.app_handler.DNSHandler") as dns_cls, + patch( + "netengine.handlers.app_handler.AppHandler._get_db", AsyncMock(return_value=mock_db()) + ), + ): + pki = AsyncMock() + pki.issue_cert = AsyncMock(return_value=("CERT", "KEY")) + pki.extract_cert_expiry = MagicMock(return_value=datetime.now(UTC) + timedelta(days=1)) + pki_cls.return_value = pki + dns_cls.return_value = AsyncMock() + oidc_cls.return_value = AsyncMock() + + await OrgAppsPhaseHandler().execute(ctx) + + oidc_cls.assert_called_once_with( + keycloak_url="https://auth.acme.internal", + admin_username="kc-admin", + admin_password="from-state", + ) + assert ctx.runtime_state.org_apps_output["deployments"][0]["status"] == "deployed" + + +@pytest.mark.asyncio +async def test_lifecycle_helpers(tmp_path, monkeypatch): + monkeypatch.setenv("NETENGINE_APP_CERT_DIR", str(tmp_path)) + ctx = context(tmp_path) + handler, docker, *_ = deps(tmp_path, ctx) + + assert await handler.healthcheck_app("ctr-1", "/health") is True + await handler.teardown_app({"org": "acme", "app": "demo", "container_id": "ctr-1"}) + + docker.exec_command.assert_awaited_once() + docker.stop_container.assert_awaited_once_with("ctr-1") + assert handler._db.table.return_value.upsert.call_args.args[0]["status"] == "torn_down"