+ Configure a new database connection or migrate your data to an external database. +
+Select the database engine you want to use.
+Leave blank to use the default path ./app/database.db.
Use require for managed cloud databases (AWS RDS, Supabase, etc.).
+
+ To use this database, set the DATABASE_URL environment variable
+ (in your .env file or Docker Compose config) to the connection string above,
+ then restart DocuElevate.
+
+ Copied! +
++ Copy all your data from one database to another (e.g. SQLite → PostgreSQL). +
++ This is your current database. Pre-filled with the running configuration. +
+ ++ The new database to copy data into. Must be empty (schema will be created automatically). +
+ +| Table | +Rows | +
|---|---|
| + | + |
| Total | ++ |
+ + Warning: This will copy all data to the target database. + The target must be empty. This operation cannot be undone. +
+Migration in progress — please do not close this page…
+Migration Successful
++ Copied rows + across tables. +
+
+ Update your DATABASE_URL environment variable to the target URL and restart DocuElevate.
+
Migration Failed
+ + + + ++ + See the Database Configuration Guide for more details. +
+{{ setting.metadata.description }}
+ {% if setting.metadata.get('help_link') %} ++ + {{ setting.metadata.get('help_link_label', 'More info') }} + +
+ {% endif %} {% if setting.metadata.type == 'boolean' %} diff --git a/tests/test_db_migrate.py b/tests/test_db_migrate.py new file mode 100644 index 00000000..a368a538 --- /dev/null +++ b/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) diff --git a/tests/test_db_wizard.py b/tests/test_db_wizard.py new file mode 100644 index 00000000..d94251b9 --- /dev/null +++ b/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 == "" diff --git a/tests/test_db_wizard_api.py b/tests/test_db_wizard_api.py new file mode 100644 index 00000000..2bcf3247 --- /dev/null +++ b/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