test: consolidate migration tests into a migrations/ subpackage

Group the three upgrade-path suites (legacy->designs, hardware->
properties, pre-migration DB backup) under tests/migrations/ with
clearer names. Pure relocation via git mv; each keeps its own
self-contained fixtures. No behavior change; 627 passed.

ha-relevant: no

backend/tests/migrations/test_backup.py

@@ -0,0 +1,57 @@
+"""
+Tests for automatic DB backup before migrations.
+"""
+import os
+
+os.environ.setdefault("SECRET_KEY", "test-only-secret-key-not-for-production")
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from app.db.database import _backup_db
+
+
+@pytest.fixture()
+def tmp_db(tmp_path: Path):
+    db = tmp_path / "homelab.db"
+    db.write_bytes(b"SQLite placeholder")
+    return db
+
+
+def test_backup_created_when_db_exists(tmp_db: Path):
+    with patch("app.db.database.settings") as mock_settings, \
+         patch("app.db.database.APP_VERSION", "1.9"):
+        mock_settings.sqlite_path = str(tmp_db)
+        _backup_db()
+        backup = tmp_db.parent / "homelab.db.back-1.9"
+        assert backup.exists()
+        assert backup.read_bytes() == b"SQLite placeholder"
+
+
+def test_backup_skipped_when_db_missing(tmp_path: Path):
+    with patch("app.db.database.settings") as mock_settings, \
+         patch("app.db.database.APP_VERSION", "1.9"):
+        mock_settings.sqlite_path = str(tmp_path / "nonexistent.db")
+        _backup_db()
+        assert not any(tmp_path.glob("*.back-*"))
+
+
+def test_backup_idempotent_second_call_no_overwrite(tmp_db: Path):
+    with patch("app.db.database.settings") as mock_settings, \
+         patch("app.db.database.APP_VERSION", "1.9"):
+        mock_settings.sqlite_path = str(tmp_db)
+        _backup_db()
+        backup = tmp_db.parent / "homelab.db.back-1.9"
+        backup.write_bytes(b"original backup")
+        _backup_db()
+        assert backup.read_bytes() == b"original backup"
+
+
+def test_backup_version_in_filename(tmp_db: Path):
+    with patch("app.db.database.settings") as mock_settings, \
+         patch("app.db.database.APP_VERSION", "2.0"):
+        mock_settings.sqlite_path = str(tmp_db)
+        _backup_db()
+        assert (tmp_db.parent / "homelab.db.back-2.0").exists()

backend/tests/migrations/test_legacy_designs.py

@@ -0,0 +1,156 @@
+"""Backward-compatibility tests for the legacy → multi-design migration.
+
+Simulates a database created by a pre-"designs" version of the app and asserts
+that running init_db() adopts all existing nodes/edges/canvas into a single
+default "Network Topology" design with no data loss. The rest of the test suite
+builds the *current* schema via create_all and never exercises this upgrade
+path, so this file guards real users upgrading in place.
+"""
+import os
+
+os.environ.setdefault("SECRET_KEY", "test-only-secret-key-not-for-production")
+
+import pytest
+from sqlalchemy.ext.asyncio import create_async_engine
+
+import app.db.database as database
+
+
+@pytest.fixture
+def legacy_engine(tmp_path, monkeypatch):
+    """Point the module-global engine + sqlite_path at a throwaway legacy DB."""
+    db_path = tmp_path / "legacy.db"
+    monkeypatch.setattr(database.settings, "sqlite_path", str(db_path))
+    engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
+    monkeypatch.setattr(database, "engine", engine)
+    return db_path, engine
+
+
+async def _build_legacy_schema(engine) -> None:
+    """Create the pre-designs schema (no design_id, integer canvas_state PK)."""
+    async with engine.begin() as conn:
+        await conn.exec_driver_sql(
+            "CREATE TABLE nodes (id VARCHAR PRIMARY KEY, type VARCHAR, label VARCHAR, "
+            "status VARCHAR, services JSON, pos_x FLOAT, pos_y FLOAT)"
+        )
+        await conn.exec_driver_sql(
+            "CREATE TABLE edges (id VARCHAR PRIMARY KEY, source VARCHAR, target VARCHAR, type VARCHAR)"
+        )
+        await conn.exec_driver_sql(
+            "CREATE TABLE canvas_state (id INTEGER PRIMARY KEY, viewport JSON, "
+            "custom_style JSON, saved_at DATETIME)"
+        )
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id, type, label, status, services, pos_x, pos_y) "
+            "VALUES ('n1','server','Old Server','online','[]',10,20)"
+        )
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id, type, label, status, services, pos_x, pos_y) "
+            "VALUES ('n2','router','Old Router','offline','[]',30,40)"
+        )
+        await conn.exec_driver_sql(
+            "INSERT INTO edges (id, source, target, type) VALUES ('e1','n1','n2','ethernet')"
+        )
+        await conn.exec_driver_sql(
+            "INSERT INTO canvas_state (id, viewport, custom_style, saved_at) "
+            "VALUES (1, '{\"x\":5,\"y\":6,\"zoom\":2}', NULL, '2024-01-01 00:00:00')"
+        )
+
+
+async def test_legacy_canvas_migrates_into_default_design(legacy_engine):
+    db_path, engine = legacy_engine
+    await _build_legacy_schema(engine)
+
+    await database.init_db()
+
+    check = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
+    try:
+        async with check.begin() as conn:
+            # Exactly one seeded default design.
+            designs = (await conn.exec_driver_sql(
+                "SELECT id, name, design_type, icon FROM designs"
+            )).fetchall()
+            assert len(designs) == 1
+            did, name, dtype, icon = designs[0]
+            assert name == "Network Topology"
+            assert dtype == "network"
+            assert icon == "dashboard"
+
+            # Every legacy node adopted into the default design, data preserved.
+            nodes = (await conn.exec_driver_sql(
+                "SELECT id, label, status, design_id FROM nodes ORDER BY id"
+            )).fetchall()
+            assert [(n[0], n[1], n[2]) for n in nodes] == [
+                ("n1", "Old Server", "online"),
+                ("n2", "Old Router", "offline"),
+            ]
+            assert all(n[3] == did for n in nodes)
+
+            # Legacy edge adopted too.
+            edge = (await conn.exec_driver_sql(
+                "SELECT design_id FROM edges WHERE id='e1'"
+            )).fetchone()
+            assert edge[0] == did
+
+            # canvas_state rebuilt with design_id PK; the old id=1 row maps to the
+            # default design and the viewport survives.
+            cs = (await conn.exec_driver_sql(
+                "SELECT design_id, viewport FROM canvas_state"
+            )).fetchall()
+            assert len(cs) == 1
+            assert cs[0][0] == did
+            assert "zoom" in (cs[0][1] or "")
+    finally:
+        await check.dispose()
+        await engine.dispose()
+
+
+async def test_legacy_nodes_gain_last_scan_column(legacy_engine):
+    """A legacy nodes table (no last_scan) gains the column after init_db."""
+    db_path, engine = legacy_engine
+    await _build_legacy_schema(engine)
+
+    await database.init_db()
+
+    check = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
+    try:
+        async with check.begin() as conn:
+            cols = (await conn.exec_driver_sql("PRAGMA table_info(nodes)")).fetchall()
+            assert "last_scan" in {c[1] for c in cols}
+            # Existing rows backfill to NULL (never scanned yet).
+            last_scan = (await conn.exec_driver_sql(
+                "SELECT last_scan FROM nodes WHERE id='n1'"
+            )).fetchone()
+            assert last_scan[0] is None
+    finally:
+        await check.dispose()
+        await engine.dispose()
+
+
+async def test_migration_is_idempotent(legacy_engine):
+    """Running init_db twice must not duplicate the design or drop any data."""
+    db_path, engine = legacy_engine
+    await _build_legacy_schema(engine)
+
+    await database.init_db()
+    await database.init_db()  # second boot — should be a no-op
+
+    check = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
+    try:
+        async with check.begin() as conn:
+            designs = (await conn.exec_driver_sql("SELECT id FROM designs")).fetchall()
+            assert len(designs) == 1
+            did = designs[0][0]
+
+            nodes = (await conn.exec_driver_sql(
+                "SELECT design_id FROM nodes"
+            )).fetchall()
+            assert len(nodes) == 2
+            assert all(n[0] == did for n in nodes)
+
+            cs = (await conn.exec_driver_sql("SELECT design_id FROM canvas_state")).fetchall()
+            assert len(cs) == 1
+            assert cs[0][0] == did
+    finally:
+        await check.dispose()
+        await engine.dispose()

backend/tests/migrations/test_properties.py

@@ -0,0 +1,176 @@
+"""
+Tests for the hardware → properties migration logic.
+
+We test the migration function directly against an in-memory SQLite database
+so we can set up legacy rows (with hardware columns, NULL properties) and
+verify the migration produces the expected properties JSON.
+"""
+import json
+import os
+
+os.environ.setdefault("SECRET_KEY", "test-only-secret-key-not-for-production")
+
+import pytest
+from sqlalchemy.ext.asyncio import create_async_engine
+
+TEST_DB_URL = "sqlite+aiosqlite:///:memory:"
+
+
+async def _setup_legacy_table(conn):
+    """Create a minimal nodes table that mimics the pre-migration schema."""
+    await conn.exec_driver_sql("""
+        CREATE TABLE IF NOT EXISTS nodes (
+            id TEXT PRIMARY KEY,
+            type TEXT NOT NULL DEFAULT 'generic',
+            label TEXT NOT NULL DEFAULT '',
+            cpu_model TEXT,
+            cpu_count INTEGER,
+            ram_gb REAL,
+            disk_gb REAL,
+            show_hardware BOOLEAN NOT NULL DEFAULT 0,
+            properties JSON
+        )
+    """)
+
+
+async def _run_migration(conn):
+    """Run only the properties migration portion (extracted from init_db)."""
+    rows = await conn.exec_driver_sql(
+        "SELECT id, cpu_model, cpu_count, ram_gb, disk_gb, show_hardware "
+        "FROM nodes WHERE properties IS NULL"
+    )
+    for row in rows.fetchall():
+        node_id, cpu_model, cpu_count, ram_gb, disk_gb, show_hardware = row
+        props = []
+        visible = bool(show_hardware)
+        if cpu_model:
+            props.append({"key": "CPU Model", "value": str(cpu_model), "icon": "Cpu", "visible": visible})
+        if cpu_count is not None:
+            props.append({"key": "CPU Cores", "value": str(cpu_count), "icon": "Cpu", "visible": visible})
+        if ram_gb is not None:
+            props.append({"key": "RAM", "value": f"{ram_gb} GB", "icon": "MemoryStick", "visible": visible})
+        if disk_gb is not None:
+            props.append({"key": "Disk", "value": f"{disk_gb} GB", "icon": "HardDrive", "visible": visible})
+        await conn.exec_driver_sql(
+            "UPDATE nodes SET properties = ? WHERE id = ?",
+            (json.dumps(props), node_id),
+        )
+
+
+async def _get_properties(conn, node_id: str) -> list:
+    rows = await conn.exec_driver_sql("SELECT properties FROM nodes WHERE id = ?", (node_id,))
+    raw = rows.fetchone()[0]
+    return json.loads(raw) if raw else []
+
+
+@pytest.mark.asyncio
+async def test_migration_full_hardware():
+    """Node with all 4 hardware fields → 4 property entries with correct icons."""
+    engine = create_async_engine(TEST_DB_URL)
+    async with engine.begin() as conn:
+        await _setup_legacy_table(conn)
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id, cpu_model, cpu_count, ram_gb, disk_gb, show_hardware) "
+            "VALUES (?, ?, ?, ?, ?, ?)",
+            ("node-1", "i7-12700K", 12, 32.0, 2000.0, 1),
+        )
+        await _run_migration(conn)
+        props = await _get_properties(conn, "node-1")
+
+    assert len(props) == 4
+    assert props[0] == {"key": "CPU Model", "value": "i7-12700K", "icon": "Cpu", "visible": True}
+    assert props[1] == {"key": "CPU Cores", "value": "12", "icon": "Cpu", "visible": True}
+    assert props[2] == {"key": "RAM", "value": "32.0 GB", "icon": "MemoryStick", "visible": True}
+    assert props[3] == {"key": "Disk", "value": "2000.0 GB", "icon": "HardDrive", "visible": True}
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_migration_partial_hardware():
+    """Node with only cpu_model and ram_gb → 2 property entries."""
+    engine = create_async_engine(TEST_DB_URL)
+    async with engine.begin() as conn:
+        await _setup_legacy_table(conn)
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id, cpu_model, ram_gb, show_hardware) VALUES (?, ?, ?, ?)",
+            ("node-2", "Ryzen 5 5600", 16.0, 0),
+        )
+        await _run_migration(conn)
+        props = await _get_properties(conn, "node-2")
+
+    assert len(props) == 2
+    assert props[0]["key"] == "CPU Model"
+    assert props[0]["visible"] is False
+    assert props[1]["key"] == "RAM"
+    assert props[1]["icon"] == "MemoryStick"
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_migration_no_hardware():
+    """Node with no hardware fields → empty properties array."""
+    engine = create_async_engine(TEST_DB_URL)
+    async with engine.begin() as conn:
+        await _setup_legacy_table(conn)
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id) VALUES (?)",
+            ("node-3",),
+        )
+        await _run_migration(conn)
+        props = await _get_properties(conn, "node-3")
+
+    assert props == []
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_migration_idempotent():
+    """Running migration twice does not duplicate properties."""
+    engine = create_async_engine(TEST_DB_URL)
+    async with engine.begin() as conn:
+        await _setup_legacy_table(conn)
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id, cpu_model, show_hardware) VALUES (?, ?, ?)",
+            ("node-4", "Core i5", 1),
+        )
+        await _run_migration(conn)
+        await _run_migration(conn)  # second pass — node already has properties, should be skipped
+        props = await _get_properties(conn, "node-4")
+
+    assert len(props) == 1
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_migration_show_hardware_false_sets_visible_false():
+    """show_hardware=0 means all migrated properties have visible=False."""
+    engine = create_async_engine(TEST_DB_URL)
+    async with engine.begin() as conn:
+        await _setup_legacy_table(conn)
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id, cpu_model, ram_gb, show_hardware) VALUES (?, ?, ?, ?)",
+            ("node-5", "ARM Cortex-A72", 4.0, 0),
+        )
+        await _run_migration(conn)
+        props = await _get_properties(conn, "node-5")
+
+    assert all(p["visible"] is False for p in props)
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_migration_already_migrated_node_not_touched():
+    """Node that already has properties is skipped — existing properties preserved."""
+    existing = [{"key": "GPU", "value": "RTX 4090", "icon": "Monitor", "visible": True}]
+    engine = create_async_engine(TEST_DB_URL)
+    async with engine.begin() as conn:
+        await _setup_legacy_table(conn)
+        await conn.exec_driver_sql(
+            "INSERT INTO nodes (id, cpu_model, ram_gb, show_hardware, properties) VALUES (?, ?, ?, ?, ?)",
+            ("node-6", "i9-13900K", 64.0, 1, json.dumps(existing)),
+        )
+        await _run_migration(conn)
+        props = await _get_properties(conn, "node-6")
+
+    assert props == existing
+    await engine.dispose()