"""Cleaning up after the removed multi-host integration. The agent sidecar and everything that proxied to it were removed in 0.48.0. Upgrading installs still carry the schema it left behind, and one part of that is not merely dead weight: the ``agent`` table held each remote host's URL and its bearer token, which is full Docker control of that host. Leaving those in the database for a feature that no longer exists would be worse than dropping them, so ``_drop_removed_schema`` drops the table outright. The ``agent_id`` columns are only dropped where the SQLite build supports it (DROP COLUMN needs 3.35+); they are nullable and unread, so leaving them is harmless. """ from __future__ import annotations import pytest from sqlalchemy import create_engine, inspect, text @pytest.fixture def pre_048_db(tmp_path, monkeypatch): """A database as an install running 0.47.0 would have it.""" import database import models # noqa: F401 — populates SQLModel.metadata engine = create_engine(f"sqlite:///{tmp_path / 'legacy.db'}") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE agent ( id INTEGER NOT NULL PRIMARY KEY, name VARCHAR NOT NULL, url VARCHAR NOT NULL, token VARCHAR NOT NULL, status VARCHAR NOT NULL, hostname VARCHAR, last_seen DATETIME, created_at DATETIME NOT NULL ) """ ) ) conn.execute( text( "INSERT INTO agent (id, name, url, token, status, created_at) VALUES" " (1, 'nas', 'http://10.0.0.5:5010', 'super-secret-agent-token'," " 'online', '2026-01-01')" ) ) conn.execute( text( """ CREATE TABLE autoupdate ( id INTEGER NOT NULL PRIMARY KEY, stack_id VARCHAR NOT NULL, agent_id INTEGER, enabled BOOLEAN NOT NULL, redeploy BOOLEAN NOT NULL, last_run DATETIME, last_status VARCHAR, last_result VARCHAR, created_at DATETIME NOT NULL ) """ ) ) conn.execute( text( "INSERT INTO autoupdate (id, stack_id, agent_id, enabled, redeploy," " created_at) VALUES (1, 'jellyfin', NULL, 1, 1, '2026-01-01')" ) ) monkeypatch.setattr(database, "engine", engine) return engine def test_the_agent_table_and_its_tokens_are_dropped(pre_048_db): import database assert "agent" in inspect(pre_048_db).get_table_names() database._drop_removed_schema() assert "agent" not in inspect(pre_048_db).get_table_names(), ( "the agent table holds host URLs and bearer tokens for a feature that no " "longer exists — it must not survive the upgrade" ) def test_surviving_rows_are_untouched(pre_048_db): """Dropping the agent schema must not disturb the policies that remain.""" import database database._drop_removed_schema() with pre_048_db.begin() as conn: row = conn.execute( text("SELECT stack_id, enabled FROM autoupdate WHERE id = 1") ).one() assert row.stack_id == "jellyfin" assert row.enabled def test_the_agent_id_column_is_dropped_where_sqlite_allows_it(pre_048_db): import sqlite3 import database database._drop_removed_schema() columns = {c["name"] for c in inspect(pre_048_db).get_columns("autoupdate")} supports_drop = tuple(int(p) for p in sqlite3.sqlite_version.split(".")) >= (3, 35, 0) if supports_drop: assert "agent_id" not in columns else: # pragma: no cover - depends on the host's SQLite assert "agent_id" in columns, "the fallback must leave the column alone" def test_an_unsupported_drop_column_does_not_block_the_table_drop( pre_048_db, monkeypatch ): """A failed DDL poisons its transaction. If the table drop and the column drops shared one, a SQLite too old for DROP COLUMN would take the agent table — tokens and all — down with it. """ import database from sqlalchemy.exc import OperationalError real_execute = database.text def exploding_text(sql): if "DROP COLUMN" in sql: raise OperationalError("DROP COLUMN", {}, Exception("near \"DROP\"")) return real_execute(sql) monkeypatch.setattr(database, "text", exploding_text) database._drop_removed_schema() assert "agent" not in inspect(pre_048_db).get_table_names(), ( "the table drop must not be collateral damage of an unsupported column drop" ) def test_running_it_twice_is_harmless(pre_048_db): """It runs on every start, not just the first one after the upgrade.""" import database database._drop_removed_schema() database._drop_removed_schema() assert "agent" not in inspect(pre_048_db).get_table_names() def test_a_fresh_install_is_unaffected(db): """Nothing to drop, and no error for trying.""" import database database._drop_removed_schema() assert "agent" not in inspect(database.engine).get_table_names() # --------------------------------------------------------------------------- # # Nothing agent-shaped is left in the running application # --------------------------------------------------------------------------- # def test_no_route_mentions_agents(app): from fastapi.routing import APIRoute leftovers = [ r.path for r in app.routes if "agent" in getattr(r, "path", "").lower() ] assert not leftovers, f"remote-host routes still registered: {leftovers}" assert not any( isinstance(r, APIRoute) and "/api/agents" in r.path for r in app.routes ) def test_no_agent_model_is_registered(): from sqlmodel import SQLModel import models # noqa: F401 assert "agent" not in SQLModel.metadata.tables def test_the_agent_modules_are_gone(): """Import guards, so a stray file cannot quietly come back.""" import importlib for name in ("agent_app", "models.agent", "routers.agents", "services.agent_service"): with pytest.raises(ModuleNotFoundError): importlib.import_module(name) def test_no_agent_token_setting(): from config import settings assert not hasattr(settings, "AGENT_TOKEN")