Merge pull request #482 from christianlouis/copilot/add-database-configuration-wizard

feat(database): integrate wizard into settings page, improve accessibility and test coverage

tests/test_db_migrate.py

@@ -0,0 +1,359 @@
+"""Tests for app/utils/db_migrate.py module."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.database import Base
+from app.utils.db_migrate import (
+    _make_engine,
+    _ordered_tables,
+    _stamp_alembic_head,
+    migrate_data,
+    preview_migration,
+)
+
+
+@pytest.mark.unit
+class TestMakeEngine:
+    """Tests for _make_engine helper function."""
+
+    def test_sqlite_engine_has_check_same_thread(self):
+        """Test that SQLite engine has check_same_thread set."""
+        engine = _make_engine("sqlite:///:memory:")
+        assert engine is not None
+        engine.dispose()
+
+    def test_non_sqlite_engine_created(self):
+        """Test that non-SQLite engine can be created (even if driver is missing)."""
+        # _make_engine only creates the engine object; it doesn't connect.
+        # If the driver isn't installed, create_engine raises at creation time.
+        try:
+            engine = _make_engine("postgresql://u:p@localhost:5432/test")
+            assert engine is not None
+            engine.dispose()
+        except Exception:
+            # Driver not installed in test environment — acceptable
+            pass
+
+
+@pytest.mark.unit
+class TestOrderedTables:
+    """Tests for _ordered_tables helper function."""
+
+    def test_known_tables_come_first(self):
+        """Test that known tables from _TABLE_ORDER come first."""
+        mock_inspector = MagicMock()
+        mock_inspector.get_table_names.return_value = [
+            "webhook_configs",
+            "documents",
+            "files",
+            "custom_table",
+            "alembic_version",
+        ]
+        result = _ordered_tables(mock_inspector)
+        # alembic_version should be skipped
+        assert "alembic_version" not in result
+        # Known tables should come first in their predefined order
+        assert result.index("documents") < result.index("files")
+        assert result.index("files") < result.index("webhook_configs")
+        # custom_table is not in _TABLE_ORDER so comes after known tables
+        assert "custom_table" in result
+
+    def test_skips_alembic_version(self):
+        """Test that alembic_version table is always skipped."""
+        mock_inspector = MagicMock()
+        mock_inspector.get_table_names.return_value = ["alembic_version", "documents"]
+        result = _ordered_tables(mock_inspector)
+        assert "alembic_version" not in result
+        assert "documents" in result
+
+    def test_unknown_tables_appended_alphabetically(self):
+        """Test that tables not in _TABLE_ORDER are appended alphabetically."""
+        mock_inspector = MagicMock()
+        mock_inspector.get_table_names.return_value = ["zebra", "apple", "documents"]
+        result = _ordered_tables(mock_inspector)
+        assert result[0] == "documents"
+        # apple and zebra should be after documents, in alpha order
+        remaining = result[1:]
+        assert remaining == sorted(remaining)
+
+    def test_empty_database(self):
+        """Test with an empty database returns empty list."""
+        mock_inspector = MagicMock()
+        mock_inspector.get_table_names.return_value = []
+        result = _ordered_tables(mock_inspector)
+        assert result == []
+
+
+@pytest.mark.unit
+class TestPreviewMigration:
+    """Tests for preview_migration function."""
+
+    def test_preview_in_memory_sqlite(self):
+        """Test previewing an in-memory SQLite database."""
+        # Create a temporary source DB with some data
+        src_engine = create_engine(
+            "sqlite:///:memory:",
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool,
+        )
+        Base.metadata.create_all(bind=src_engine)
+
+        # Insert a test row
+        Session = sessionmaker(bind=src_engine)
+        session = Session()
+        session.execute(text("INSERT INTO documents (filename) VALUES ('test.pdf')"))
+        session.commit()
+        session.close()
+
+        # Preview using the engine's URL won't work for :memory:,
+        # but we can test the error path
+        result = preview_migration("sqlite:///:memory:")
+        # For :memory: this creates a new empty DB, so tables are empty
+        assert result["success"] is True
+        assert isinstance(result["tables"], list)
+
+    def test_preview_invalid_url(self):
+        """Test preview with invalid URL returns error."""
+        result = preview_migration("invalid://not-a-db")
+        assert result["success"] is False
+        assert "error" in result
+
+    def test_preview_with_patched_source_shows_tables(self):
+        """Test preview with source that has tables and data."""
+        real_src = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_src)
+
+        # Insert test data
+        Session = sessionmaker(bind=real_src)
+        session = Session()
+        session.execute(text("INSERT INTO documents (filename) VALUES ('test.pdf')"))
+        session.commit()
+        session.close()
+
+        with patch("app.utils.db_migrate._make_engine", return_value=real_src):
+            result = preview_migration("sqlite:///:memory:")
+
+        assert result["success"] is True
+        assert result["total_rows"] >= 1
+        # At least the documents table should be in results
+        table_names = [t["name"] for t in result["tables"]]
+        assert "documents" in table_names
+        doc_table = next(t for t in result["tables"] if t["name"] == "documents")
+        assert doc_table["row_count"] >= 1
+
+
+@pytest.mark.unit
+class TestMigrateData:
+    """Tests for migrate_data function."""
+
+    def test_migrate_empty_sqlite_to_sqlite(self):
+        """Test migrating an empty SQLite DB to another SQLite DB."""
+        # Both are file-based temp databases for this test
+        src_url = "sqlite:///:memory:"
+        tgt_url = "sqlite://"  # Another in-memory DB
+
+        # Create source schema
+        src_engine = create_engine(src_url, connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=src_engine)
+        src_engine.dispose()
+
+        # Run migration from empty source
+        with patch("app.utils.db_migrate._make_engine") as mock_make:
+            # Create real engines for both
+            real_src = create_engine(
+                "sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool
+            )
+            Base.metadata.create_all(bind=real_src)
+            real_tgt = create_engine(
+                "sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool
+            )
+            mock_make.side_effect = [real_src, real_tgt]
+
+            with patch("app.utils.db_migrate._stamp_alembic_head"):
+                result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:")
+
+        assert result["success"] is True
+        assert result["rows_copied"] == 0
+
+    def test_migrate_with_data(self):
+        """Test migrating a SQLite DB with actual data."""
+        real_src = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_src)
+
+        # Insert test data
+        Session = sessionmaker(bind=real_src)
+        session = Session()
+        session.execute(text("INSERT INTO documents (filename) VALUES ('invoice.pdf')"))
+        session.execute(text("INSERT INTO documents (filename) VALUES ('receipt.pdf')"))
+        session.commit()
+        session.close()
+
+        real_tgt = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+
+        with patch("app.utils.db_migrate._make_engine") as mock_make:
+            mock_make.side_effect = [real_src, real_tgt]
+            with patch("app.utils.db_migrate._stamp_alembic_head"):
+                result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:")
+
+        assert result["success"] is True
+        assert result["rows_copied"] >= 2  # At least the 2 documents rows
+
+    def test_migrate_with_progress_callback(self):
+        """Test that progress callback is invoked during migration."""
+        real_src = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_src)
+
+        Session = sessionmaker(bind=real_src)
+        session = Session()
+        session.execute(text("INSERT INTO documents (filename) VALUES ('test.pdf')"))
+        session.commit()
+        session.close()
+
+        real_tgt = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        callback = MagicMock()
+
+        with patch("app.utils.db_migrate._make_engine") as mock_make:
+            mock_make.side_effect = [real_src, real_tgt]
+            with patch("app.utils.db_migrate._stamp_alembic_head"):
+                result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:", progress_callback=callback)
+
+        assert result["success"] is True
+        # Callback should have been called at least once for the non-empty table
+        if result["rows_copied"] > 0:
+            assert callback.call_count > 0
+
+    def test_migrate_global_exception(self):
+        """Test that a global exception is caught gracefully."""
+        with patch("app.utils.db_migrate._make_engine", side_effect=Exception("boom")):
+            result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:")
+        assert result["success"] is False
+        assert len(result["errors"]) > 0
+
+    def test_migrate_stamp_failure_is_recorded(self):
+        """Test that Alembic stamp failure is recorded as an error."""
+        real_src = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_src)
+        real_tgt = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+
+        with patch("app.utils.db_migrate._make_engine") as mock_make:
+            mock_make.side_effect = [real_src, real_tgt]
+            with patch("app.utils.db_migrate._stamp_alembic_head", side_effect=Exception("stamp failed")):
+                result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:")
+
+        # Data copy succeeds but stamp fails — errors list non-empty
+        assert len(result["errors"]) > 0
+        assert any("stamp" in e.lower() for e in result["errors"])
+
+    def test_migrate_table_copy_exception(self):
+        """Test that per-table copy exception is recorded but migration continues."""
+        real_src = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_src)
+
+        # Insert data so the table isn't empty
+        Session = sessionmaker(bind=real_src)
+        session = Session()
+        session.execute(text("INSERT INTO documents (filename) VALUES ('test.pdf')"))
+        session.commit()
+        session.close()
+
+        real_tgt = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_tgt)
+
+        # Make the target reflect fail for one table to trigger the error path
+        original_reflect = MagicMock(side_effect=Exception("reflect error"))
+
+        with patch("app.utils.db_migrate._make_engine") as mock_make:
+            mock_make.side_effect = [real_src, real_tgt]
+            with patch("app.utils.db_migrate._stamp_alembic_head"):
+                # Patch MetaData so that reflecting target raises for the first table
+                with patch("app.utils.db_migrate.MetaData") as mock_meta_cls:
+                    # First MetaData() is for source reflect (should work)
+                    src_meta = MagicMock()
+                    src_table = MagicMock()
+                    src_table.columns = []
+                    src_table.select.return_value = text("SELECT 1")
+                    src_meta.tables = {"documents": src_table}
+                    src_meta.reflect = MagicMock()
+
+                    # Second MetaData() is for target reflect (should fail)
+                    tgt_meta = MagicMock()
+                    tgt_meta.reflect.side_effect = Exception("target reflect error")
+
+                    mock_meta_cls.side_effect = [src_meta, tgt_meta]
+                    result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:")
+
+        assert any("Error copying table" in e for e in result["errors"])
+
+    def test_migrate_target_table_not_found(self):
+        """Test that missing target table after reflect is recorded."""
+        real_src = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_src)
+
+        Session = sessionmaker(bind=real_src)
+        session = Session()
+        session.execute(text("INSERT INTO documents (filename) VALUES ('test.pdf')"))
+        session.commit()
+        session.close()
+
+        real_tgt = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        # Create schema in target so reflect works but returns empty
+        Base.metadata.create_all(bind=real_tgt)
+
+        with patch("app.utils.db_migrate._make_engine") as mock_make:
+            mock_make.side_effect = [real_src, real_tgt]
+            with patch("app.utils.db_migrate._stamp_alembic_head"):
+                result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:")
+
+        # With schema in target, migration should succeed normally
+        assert result["success"] is True
+
+    def test_migrate_returns_tables_and_rows_counts(self):
+        """Test that successful migration returns expected count fields."""
+        real_src = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+        Base.metadata.create_all(bind=real_src)
+        real_tgt = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+
+        with patch("app.utils.db_migrate._make_engine") as mock_make:
+            mock_make.side_effect = [real_src, real_tgt]
+            with patch("app.utils.db_migrate._stamp_alembic_head"):
+                result = migrate_data("sqlite:///:memory:", "sqlite:///:memory:")
+
+        assert "tables_copied" in result
+        assert "rows_copied" in result
+        assert "errors" in result
+        assert isinstance(result["errors"], list)
+
+
+@pytest.mark.unit
+class TestStampAlembicHead:
+    """Tests for _stamp_alembic_head helper function."""
+
+    def test_stamp_calls_alembic_command(self):
+        """Test that stamping calls alembic command.stamp with 'head'."""
+        mock_engine = MagicMock()
+        mock_connection = MagicMock()
+        mock_engine.begin.return_value.__enter__ = MagicMock(return_value=mock_connection)
+        mock_engine.begin.return_value.__exit__ = MagicMock(return_value=False)
+
+        with patch("alembic.command.stamp") as mock_stamp:
+            _stamp_alembic_head(mock_engine)
+            mock_stamp.assert_called_once()
+            # Verify it stamps to "head"
+            args = mock_stamp.call_args
+            assert args[0][1] == "head"
+
+    def test_stamp_raises_on_error(self):
+        """Test that stamp propagates exceptions."""
+        mock_engine = MagicMock()
+        mock_connection = MagicMock()
+        mock_engine.begin.return_value.__enter__ = MagicMock(return_value=mock_connection)
+        mock_engine.begin.return_value.__exit__ = MagicMock(return_value=False)
+
+        with patch("alembic.command.stamp", side_effect=Exception("stamp error")):
+            with pytest.raises(Exception, match="stamp error"):
+                _stamp_alembic_head(mock_engine)

tests/test_db_wizard.py

@@ -0,0 +1,378 @@
+"""Tests for app/utils/db_wizard.py module."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.utils.db_wizard import (
+    _get_server_version,
+    build_connection_string,
+    get_supported_backends,
+    parse_connection_string,
+    validate_url_format,
+)
+from app.utils.db_wizard import (
+    test_connection as db_test_connection,
+)
+
+
+@pytest.mark.unit
+class TestGetSupportedBackends:
+    """Tests for get_supported_backends function."""
+
+    def test_returns_list(self):
+        """Test that it returns a non-empty list."""
+        result = get_supported_backends()
+        assert isinstance(result, list)
+        assert len(result) >= 3
+
+    def test_each_backend_has_required_keys(self):
+        """Test that each backend has expected keys."""
+        required_keys = {"id", "label", "description", "requires_host"}
+        for backend in get_supported_backends():
+            assert required_keys.issubset(set(backend.keys())), f"Missing keys in {backend.get('id')}"
+
+    def test_includes_sqlite(self):
+        """Test that SQLite is included."""
+        ids = [b["id"] for b in get_supported_backends()]
+        assert "sqlite" in ids
+
+    def test_includes_postgresql(self):
+        """Test that PostgreSQL is included."""
+        ids = [b["id"] for b in get_supported_backends()]
+        assert "postgresql" in ids
+
+    def test_includes_mysql(self):
+        """Test that MySQL is included."""
+        ids = [b["id"] for b in get_supported_backends()]
+        assert "mysql" in ids
+
+    def test_sqlite_does_not_require_host(self):
+        """Test that SQLite backend does not require host."""
+        sqlite = next(b for b in get_supported_backends() if b["id"] == "sqlite")
+        assert sqlite["requires_host"] is False
+
+    def test_postgresql_requires_host(self):
+        """Test that PostgreSQL backend requires host."""
+        pg = next(b for b in get_supported_backends() if b["id"] == "postgresql")
+        assert pg["requires_host"] is True
+
+    def test_mysql_default_port(self):
+        """Test that MySQL has default port 3306."""
+        mysql = next(b for b in get_supported_backends() if b["id"] == "mysql")
+        assert mysql["default_port"] == 3306
+
+
+@pytest.mark.unit
+class TestBuildConnectionString:
+    """Tests for build_connection_string function."""
+
+    def test_sqlite_default_path(self):
+        """Test building a SQLite URL with default path."""
+        url = build_connection_string(backend="sqlite")
+        assert url == "sqlite:///./app/database.db"
+
+    def test_sqlite_custom_path(self):
+        """Test building a SQLite URL with custom path."""
+        url = build_connection_string(backend="sqlite", sqlite_path="/data/mydb.db")
+        assert url == "sqlite:////data/mydb.db"
+
+    def test_sqlite_whitespace_path(self):
+        """Test building a SQLite URL with whitespace-only path uses default."""
+        url = build_connection_string(backend="sqlite", sqlite_path="   ")
+        assert url == "sqlite:///./app/database.db"
+
+    def test_postgresql_basic(self):
+        """Test building a basic PostgreSQL URL."""
+        url = build_connection_string(
+            backend="postgresql",
+            host="localhost",
+            database="docuelevate",
+            username="user",
+            password="pass",
+        )
+        assert url == "postgresql://user:pass@localhost:5432/docuelevate"
+
+    def test_postgresql_with_ssl(self):
+        """Test building a PostgreSQL URL with SSL."""
+        url = build_connection_string(
+            backend="postgresql",
+            host="rds.amazonaws.com",
+            database="docuelevate",
+            username="admin",
+            password="secret",
+            ssl_mode="require",
+        )
+        assert "sslmode=require" in url
+        assert "postgresql://admin:secret@rds.amazonaws.com:5432/docuelevate" in url
+
+    def test_postgresql_custom_port(self):
+        """Test building a PostgreSQL URL with custom port."""
+        url = build_connection_string(
+            backend="postgresql",
+            host="localhost",
+            port=5433,
+            database="testdb",
+            username="user",
+            password="pass",
+        )
+        assert ":5433/" in url
+
+    def test_mysql_basic(self):
+        """Test building a MySQL URL."""
+        url = build_connection_string(
+            backend="mysql",
+            host="localhost",
+            database="docuelevate",
+            username="root",
+            password="password",
+        )
+        assert url.startswith("mysql+pymysql://")
+        assert "charset=utf8mb4" in url
+
+    def test_mysql_no_duplicate_charset(self):
+        """Test that charset is not duplicated when passed in extra_options."""
+        url = build_connection_string(
+            backend="mysql",
+            host="localhost",
+            database="docuelevate",
+            username="root",
+            password="pass",
+            extra_options="charset=utf8mb4",
+        )
+        assert url.count("charset=utf8mb4") == 1
+
+    def test_mysql_extra_options(self):
+        """Test MySQL URL with extra options appended."""
+        url = build_connection_string(
+            backend="mysql",
+            host="localhost",
+            database="docuelevate",
+            username="root",
+            password="pass",
+            extra_options="connect_timeout=10",
+        )
+        assert "connect_timeout=10" in url
+        assert "charset=utf8mb4" in url
+
+    def test_unsupported_backend_raises(self):
+        """Test that unsupported backend raises ValueError."""
+        with pytest.raises(ValueError, match="Unsupported backend"):
+            build_connection_string(backend="oracle")
+
+    def test_missing_host_raises(self):
+        """Test that missing host for non-SQLite raises ValueError."""
+        with pytest.raises(ValueError, match="Host is required"):
+            build_connection_string(backend="postgresql", database="db", username="u")
+
+    def test_missing_database_raises(self):
+        """Test that missing database name raises ValueError."""
+        with pytest.raises(ValueError, match="Database name is required"):
+            build_connection_string(backend="postgresql", host="localhost", username="u")
+
+    def test_missing_username_raises(self):
+        """Test that missing username raises ValueError."""
+        with pytest.raises(ValueError, match="Username is required"):
+            build_connection_string(backend="postgresql", host="localhost", database="db")
+
+    def test_no_password(self):
+        """Test building URL without password."""
+        url = build_connection_string(
+            backend="postgresql",
+            host="localhost",
+            database="db",
+            username="user",
+        )
+        assert "user@localhost" in url
+        assert ":@" not in url
+
+    def test_postgresql_with_extra_options(self):
+        """Test PostgreSQL URL with extra query options."""
+        url = build_connection_string(
+            backend="postgresql",
+            host="localhost",
+            database="db",
+            username="user",
+            extra_options="application_name=docuelevate",
+        )
+        assert "application_name=docuelevate" in url
+
+    def test_postgresql_ssl_and_extra_options(self):
+        """Test PostgreSQL URL with both SSL and extra options combined."""
+        url = build_connection_string(
+            backend="postgresql",
+            host="localhost",
+            database="db",
+            username="user",
+            ssl_mode="require",
+            extra_options="application_name=docuelevate",
+        )
+        assert "sslmode=require" in url
+        assert "application_name=docuelevate" in url
+
+
+@pytest.mark.unit
+class TestParseConnectionString:
+    """Tests for parse_connection_string function."""
+
+    def test_parse_sqlite(self):
+        """Test parsing a SQLite URL."""
+        result = parse_connection_string("sqlite:///./app/database.db")
+        assert result["valid"] is True
+        assert result["backend"] == "sqlite"
+        assert result["is_sqlite"] is True
+
+    def test_parse_postgresql(self):
+        """Test parsing a PostgreSQL URL."""
+        result = parse_connection_string("postgresql://user:pass@host:5432/mydb")
+        assert result["valid"] is True
+        assert result["backend"] == "postgresql"
+        assert result["host"] == "host"
+        assert result["port"] == 5432
+        assert result["database"] == "mydb"
+        assert result["username"] == "user"
+        assert result["is_sqlite"] is False
+
+    def test_parse_mysql(self):
+        """Test parsing a MySQL URL."""
+        result = parse_connection_string("mysql+pymysql://root:pass@localhost:3306/db")
+        assert result["valid"] is True
+        assert result["backend"] == "mysql"
+
+    def test_parse_invalid_url(self):
+        """Test parsing an invalid URL returns error."""
+        result = parse_connection_string("not-a-valid-url://")
+        # Should still return a dict (make_url may or may not raise)
+        assert isinstance(result, dict)
+
+    def test_parse_postgresql_no_password(self):
+        """Test parsing a PostgreSQL URL without password."""
+        result = parse_connection_string("postgresql://user@host:5432/mydb")
+        assert result["valid"] is True
+        assert result["password"] == ""
+
+    def test_parse_sqlite_memory(self):
+        """Test parsing a SQLite in-memory URL."""
+        result = parse_connection_string("sqlite:///:memory:")
+        assert result["valid"] is True
+        assert result["is_sqlite"] is True
+
+
+@pytest.mark.unit
+class TestValidateUrlFormat:
+    """Tests for validate_url_format function."""
+
+    def test_valid_sqlite(self):
+        """Test valid SQLite URL."""
+        result = validate_url_format("sqlite:///./db.sqlite")
+        assert result["valid"] is True
+        assert result["backend"] == "sqlite"
+
+    def test_valid_postgresql(self):
+        """Test valid PostgreSQL URL."""
+        result = validate_url_format("postgresql://u:p@host/db")
+        assert result["valid"] is True
+
+    def test_valid_mysql(self):
+        """Test valid MySQL URL."""
+        result = validate_url_format("mysql+pymysql://u:p@host/db")
+        assert result["valid"] is True
+
+    def test_unsupported_backend(self):
+        """Test that unsupported backends are flagged."""
+        result = validate_url_format("mssql://u:p@host/db")
+        assert result["valid"] is False
+        assert "Unsupported" in result.get("error", "")
+
+    def test_invalid_format(self):
+        """Test that garbage input is invalid."""
+        result = validate_url_format("")
+        assert result["valid"] is False
+
+
+@pytest.mark.unit
+class TestTestConnection:
+    """Tests for test_connection function."""
+
+    def test_sqlite_memory_succeeds(self):
+        """Test connecting to an in-memory SQLite database."""
+        result = db_test_connection("sqlite:///:memory:")
+        assert result["success"] is True
+        assert "SQLite" in result.get("server_version", "")
+
+    def test_unreachable_host_fails(self):
+        """Test that an unreachable host returns failure."""
+        result = db_test_connection("postgresql://u:p@192.0.2.1:5432/db", timeout=2)
+        assert result["success"] is False
+        assert result["message"]  # Should contain an error message
+
+    def test_returns_backend_field(self):
+        """Test that the backend field is populated on success."""
+        result = db_test_connection("sqlite:///:memory:")
+        assert result["backend"] == "sqlite"
+
+    def test_failure_returns_empty_backend(self):
+        """Test that failure returns empty backend."""
+        result = db_test_connection("postgresql://u:p@192.0.2.1:5432/db", timeout=1)
+        assert result["backend"] == ""
+        assert result["server_version"] == ""
+
+
+@pytest.mark.unit
+class TestGetServerVersion:
+    """Tests for _get_server_version internal function."""
+
+    def test_postgresql_version(self):
+        """Test PostgreSQL version retrieval."""
+        mock_conn = MagicMock()
+        mock_conn.execute.return_value.fetchone.return_value = ("PostgreSQL 16.2 on x86_64",)
+        result = _get_server_version(mock_conn, "postgresql")
+        assert result == "PostgreSQL 16.2 on x86_64"
+
+    def test_mysql_version(self):
+        """Test MySQL version retrieval."""
+        mock_conn = MagicMock()
+        mock_conn.execute.return_value.fetchone.return_value = ("8.0.36",)
+        result = _get_server_version(mock_conn, "mysql")
+        assert result == "8.0.36"
+
+    def test_sqlite_version(self):
+        """Test SQLite version retrieval."""
+        mock_conn = MagicMock()
+        mock_conn.execute.return_value.fetchone.return_value = ("3.45.1",)
+        result = _get_server_version(mock_conn, "sqlite")
+        assert result == "SQLite 3.45.1"
+
+    def test_postgresql_empty_row(self):
+        """Test PostgreSQL version with empty row returns empty string."""
+        mock_conn = MagicMock()
+        mock_conn.execute.return_value.fetchone.return_value = None
+        result = _get_server_version(mock_conn, "postgresql")
+        assert result == ""
+
+    def test_mysql_empty_row(self):
+        """Test MySQL version with empty row returns empty string."""
+        mock_conn = MagicMock()
+        mock_conn.execute.return_value.fetchone.return_value = None
+        result = _get_server_version(mock_conn, "mysql")
+        assert result == ""
+
+    def test_sqlite_empty_row(self):
+        """Test SQLite version with empty row returns empty string."""
+        mock_conn = MagicMock()
+        mock_conn.execute.return_value.fetchone.return_value = None
+        result = _get_server_version(mock_conn, "sqlite")
+        assert result == ""
+
+    def test_unknown_backend_returns_empty(self):
+        """Test that an unknown backend returns empty string."""
+        mock_conn = MagicMock()
+        result = _get_server_version(mock_conn, "oracle")
+        assert result == ""
+
+    def test_exception_returns_empty(self):
+        """Test that an exception returns empty string."""
+        mock_conn = MagicMock()
+        mock_conn.execute.side_effect = Exception("Connection lost")
+        result = _get_server_version(mock_conn, "postgresql")
+        assert result == ""

tests/test_db_wizard_api.py

@@ -0,0 +1,377 @@
+"""Tests for app/api/database.py and app/views/db_wizard.py modules."""
+
+from unittest.mock import patch
+
+import pytest
+
+
+@pytest.mark.integration
+class TestDatabaseApiEndpoints:
+    """Tests for the database API endpoints."""
+
+    def test_list_backends(self, client):
+        """Test GET /api/database/backends returns supported backends."""
+        response = client.get("/api/database/backends")
+        assert response.status_code == 200
+        data = response.json()
+        assert isinstance(data, list)
+        assert len(data) >= 3
+        ids = [b["id"] for b in data]
+        assert "sqlite" in ids
+        assert "postgresql" in ids
+
+    def test_build_url_requires_admin(self, client):
+        """Test POST /api/database/build-url requires admin."""
+        response = client.post(
+            "/api/database/build-url",
+            json={"backend": "sqlite"},
+        )
+        assert response.status_code == 403
+
+    def test_build_url_sqlite(self, client):
+        """Test building a SQLite URL as admin."""
+        # Simulate admin session
+        with client.session_transaction() if hasattr(client, "session_transaction") else _NoOpContextManager():
+            pass
+        # Use the session cookie approach
+        client.cookies.set("session", "test")
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/build-url",
+                json={"backend": "sqlite", "sqlite_path": "/data/test.db"},
+            )
+        assert response.status_code == 200
+        assert "sqlite:////data/test.db" in response.json().get("url", "")
+
+    def test_build_url_missing_host(self, client):
+        """Test building URL with missing host returns 400."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/build-url",
+                json={"backend": "postgresql", "database": "db", "username": "u"},
+            )
+        assert response.status_code == 400
+
+    def test_test_connection_sqlite(self, client):
+        """Test connection to in-memory SQLite."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/test-connection",
+                json={"url": "sqlite:///:memory:"},
+            )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["success"] is True
+
+    def test_validate_url_valid(self, client):
+        """Test validate-url with valid SQLite URL."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/validate-url",
+                json={"url": "sqlite:///test.db"},
+            )
+        assert response.status_code == 200
+        assert response.json()["valid"] is True
+
+    def test_validate_url_invalid(self, client):
+        """Test validate-url with unsupported backend."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/validate-url",
+                json={"url": "mssql://u:p@h/d"},
+            )
+        assert response.status_code == 200
+        assert response.json()["valid"] is False
+
+    def test_parse_url(self, client):
+        """Test parse-url endpoint."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/parse-url",
+                json={"url": "postgresql://user:pass@host:5432/db"},
+            )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["backend"] == "postgresql"
+        assert data["host"] == "host"
+
+    def test_preview_migration(self, client):
+        """Test preview-migration endpoint."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/preview-migration",
+                json={"url": "sqlite:///:memory:"},
+            )
+        assert response.status_code == 200
+        data = response.json()
+        assert "tables" in data
+
+    def test_migrate_invalid_source(self, client):
+        """Test migrate endpoint with invalid source URL."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/migrate",
+                json={"source_url": "mssql://bad", "target_url": "sqlite:///:memory:"},
+            )
+        assert response.status_code == 400
+
+    def test_migrate_invalid_target(self, client):
+        """Test migrate endpoint with invalid target URL."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/migrate",
+                json={"source_url": "sqlite:///:memory:", "target_url": "mssql://bad"},
+            )
+        assert response.status_code == 400
+
+    def test_migrate_success(self, client):
+        """Test migrate endpoint with successful migration."""
+        mock_result = {"success": True, "tables_copied": 5, "rows_copied": 100, "errors": []}
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            with patch("app.api.database.migrate_data", return_value=mock_result):
+                response = client.post(
+                    "/api/database/migrate",
+                    json={
+                        "source_url": "sqlite:///:memory:",
+                        "target_url": "sqlite:///:memory:",
+                    },
+                )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["success"] is True
+        assert data["rows_copied"] == 100
+
+    def test_migrate_failure_returns_500(self, client):
+        """Test migrate endpoint returns 500 on migration failure."""
+        mock_result = {
+            "success": False,
+            "tables_copied": 2,
+            "rows_copied": 50,
+            "errors": ["Table X failed", "Stamp failed"],
+        }
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            with patch("app.api.database.migrate_data", return_value=mock_result):
+                response = client.post(
+                    "/api/database/migrate",
+                    json={
+                        "source_url": "sqlite:///:memory:",
+                        "target_url": "sqlite:///:memory:",
+                    },
+                )
+        assert response.status_code == 500
+        assert "Table X failed" in response.json()["detail"]
+
+    def test_test_connection_requires_admin(self, client):
+        """Test POST /api/database/test-connection requires admin."""
+        response = client.post(
+            "/api/database/test-connection",
+            json={"url": "sqlite:///:memory:"},
+        )
+        assert response.status_code == 403
+
+    def test_parse_url_requires_admin(self, client):
+        """Test POST /api/database/parse-url requires admin."""
+        response = client.post(
+            "/api/database/parse-url",
+            json={"url": "sqlite:///:memory:"},
+        )
+        assert response.status_code == 403
+
+    def test_validate_url_requires_admin(self, client):
+        """Test POST /api/database/validate-url requires admin."""
+        response = client.post(
+            "/api/database/validate-url",
+            json={"url": "sqlite:///:memory:"},
+        )
+        assert response.status_code == 403
+
+    def test_preview_migration_requires_admin(self, client):
+        """Test POST /api/database/preview-migration requires admin."""
+        response = client.post(
+            "/api/database/preview-migration",
+            json={"url": "sqlite:///:memory:"},
+        )
+        assert response.status_code == 403
+
+    def test_migrate_requires_admin(self, client):
+        """Test POST /api/database/migrate requires admin."""
+        response = client.post(
+            "/api/database/migrate",
+            json={"source_url": "sqlite:///:memory:", "target_url": "sqlite:///:memory:"},
+        )
+        assert response.status_code == 403
+
+    def test_build_url_postgresql(self, client):
+        """Test building a PostgreSQL URL."""
+        with patch("app.api.database._require_admin", return_value={"is_admin": True}):
+            response = client.post(
+                "/api/database/build-url",
+                json={
+                    "backend": "postgresql",
+                    "host": "localhost",
+                    "port": 5432,
+                    "database": "mydb",
+                    "username": "admin",
+                    "password": "secret",
+                },
+            )
+        assert response.status_code == 200
+        url = response.json()["url"]
+        assert "postgresql://admin:secret@localhost:5432/mydb" in url
+
+
+@pytest.mark.integration
+class TestDatabaseWizardView:
+    """Tests for the database wizard view."""
+
+    def test_database_wizard_page_loads(self, client):
+        """Test GET /database-wizard returns 200."""
+        response = client.get("/database-wizard")
+        assert response.status_code == 200
+
+    def test_database_wizard_contains_title(self, client):
+        """Test that the wizard page contains expected content."""
+        response = client.get("/database-wizard")
+        assert response.status_code == 200
+        assert "Database Configuration Wizard" in response.text
+
+    def test_database_wizard_contains_tabs(self, client):
+        """Test that the wizard page contains configure and migrate tabs."""
+        response = client.get("/database-wizard")
+        assert "Configure Database" in response.text
+        assert "Migrate Data" in response.text
+
+    def test_database_wizard_has_skip_link(self, client):
+        """Test that the wizard page includes a skip-to-content link."""
+        response = client.get("/database-wizard")
+        assert "Skip to main content" in response.text
+
+    def test_database_wizard_has_main_landmark(self, client):
+        """Test that the wizard page has a main landmark."""
+        response = client.get("/database-wizard")
+        assert 'id="main-content"' in response.text
+
+    def test_database_wizard_has_tablist_role(self, client):
+        """Test that the tab navigation has proper ARIA tablist role."""
+        response = client.get("/database-wizard")
+        assert 'role="tablist"' in response.text
+        assert 'role="tab"' in response.text
+        assert 'role="tabpanel"' in response.text
+
+    def test_database_wizard_has_aria_labels_on_backend_buttons(self, client):
+        """Test that backend selection buttons have aria-label attributes."""
+        response = client.get("/database-wizard")
+        assert 'aria-label="Select SQLite"' in response.text
+        assert 'aria-label="Select PostgreSQL"' in response.text
+        assert 'aria-label="Select MySQL / MariaDB"' in response.text
+
+    def test_database_wizard_has_form_labels(self, client):
+        """Test that form inputs have associated labels."""
+        response = client.get("/database-wizard")
+        assert 'for="sqlite_path"' in response.text
+        assert 'for="db_host"' in response.text
+        assert 'for="db_port"' in response.text
+        assert 'for="db_name"' in response.text
+        assert 'for="db_user"' in response.text
+        assert 'for="db_pass"' in response.text
+        assert 'for="ssl_mode"' in response.text
+
+    def test_database_wizard_has_aria_describedby(self, client):
+        """Test that inputs have aria-describedby pointing to help text."""
+        response = client.get("/database-wizard")
+        assert 'aria-describedby="sqlite_path_help"' in response.text
+        assert 'id="sqlite_path_help"' in response.text
+        assert 'aria-describedby="ssl_mode_help"' in response.text
+        assert 'id="ssl_mode_help"' in response.text
+        assert 'aria-describedby="mig_source_help"' in response.text
+        assert 'aria-describedby="mig_target_help"' in response.text
+
+    def test_database_wizard_has_status_roles(self, client):
+        """Test that dynamic feedback areas have role=status or role=alert."""
+        response = client.get("/database-wizard")
+        assert 'role="status"' in response.text
+        assert 'role="alert"' in response.text
+
+    def test_database_wizard_has_aria_live(self, client):
+        """Test that dynamic areas have aria-live for screen reader announcements."""
+        response = client.get("/database-wizard")
+        assert 'aria-live="polite"' in response.text
+
+    def test_database_wizard_has_progressbar(self, client):
+        """Test that the migration progress indicator has role=progressbar."""
+        response = client.get("/database-wizard")
+        assert 'role="progressbar"' in response.text
+
+    def test_database_wizard_has_focus_ring_styles(self, client):
+        """Test that interactive elements have focus ring styling."""
+        response = client.get("/database-wizard")
+        assert "focus:ring-2" in response.text
+        assert "focus:outline-none" in response.text
+
+    def test_database_wizard_has_table_scope_headers(self, client):
+        """Test that migration preview table has proper scope attributes."""
+        response = client.get("/database-wizard")
+        assert 'scope="col"' in response.text
+
+    def test_database_wizard_has_dark_mode_styles(self, client):
+        """Test that the wizard includes dark mode CSS overrides."""
+        response = client.get("/database-wizard")
+        assert "html.dark" in response.text
+
+    def test_database_wizard_copy_button_has_aria_label(self, client):
+        """Test that the copy-to-clipboard button has an aria-label."""
+        response = client.get("/database-wizard")
+        assert 'aria-label="Copy to clipboard"' in response.text
+
+    def test_database_wizard_decorative_icons_hidden(self, client):
+        """Test that decorative icons have aria-hidden=true."""
+        response = client.get("/database-wizard")
+        assert 'aria-hidden="true"' in response.text
+
+    def test_database_wizard_reduced_motion(self, client):
+        """Test that wizard respects prefers-reduced-motion media query."""
+        response = client.get("/database-wizard")
+        assert "prefers-reduced-motion" in response.text
+
+
+@pytest.mark.unit
+class TestSettingsPageWizardLink:
+    """Tests for the database wizard link on the settings page."""
+
+    def test_settings_template_has_db_wizard_link(self):
+        """Test that the settings template contains a link to the database wizard."""
+        from pathlib import Path
+
+        template_path = Path(__file__).resolve().parent.parent / "frontend" / "templates" / "settings.html"
+        content = template_path.read_text()
+        assert "/database-wizard" in content
+        assert "DB Wizard" in content
+
+    def test_settings_template_has_help_link_rendering(self):
+        """Test that the settings template renders help_link metadata."""
+        from pathlib import Path
+
+        template_path = Path(__file__).resolve().parent.parent / "frontend" / "templates" / "settings.html"
+        content = template_path.read_text()
+        assert "setting.metadata.get('help_link')" in content
+        assert "help_link_label" in content
+
+    def test_database_url_metadata_has_help_link(self):
+        """Test that database_url SETTING_METADATA includes help_link to wizard."""
+        from app.utils.settings_service import SETTING_METADATA
+
+        meta = SETTING_METADATA["database_url"]
+        assert "help_link" in meta
+        assert meta["help_link"] == "/database-wizard"
+        assert "help_link_label" in meta
+
+
+# Context manager helper for tests that don't need session_transaction
+class _NoOpContextManager:
+    """Dummy context manager for tests that don't need session_transaction."""
+
+    def __enter__(self):
+        return None
+
+    def __exit__(self, *args):
+        pass