diff --git a/alembic_location.ini b/alembic_location.ini deleted file mode 100644 index 9ee39f0..0000000 --- a/alembic_location.ini +++ /dev/null @@ -1,37 +0,0 @@ -[alembic] -script_location = alembic_location -prepend_sys_path = . -path_separator = os -sqlalchemy.url = sqlite:///./data/locationRecorder.db - -[loggers] -keys = root,sqlalchemy,alembic - -[handlers] -keys = console - -[formatters] -keys = generic - -[logger_root] -level = WARN -handlers = console - -[logger_sqlalchemy] -level = WARN -handlers = -qualname = sqlalchemy.engine - -[logger_alembic] -level = INFO -handlers = -qualname = alembic - -[handler_console] -class = StreamHandler -args = (sys.stderr,) -level = NOTSET -formatter = generic - -[formatter_generic] -format = %(levelname)-5.5s [%(name)s] %(message)s diff --git a/alembic_location/README b/alembic_location/README deleted file mode 100644 index 0eb160c..0000000 --- a/alembic_location/README +++ /dev/null @@ -1,2 +0,0 @@ -This directory contains the Alembic migration environment for the Python rewrite skeleton. - diff --git a/alembic_location/env.py b/alembic_location/env.py deleted file mode 100644 index 05b9217..0000000 --- a/alembic_location/env.py +++ /dev/null @@ -1,48 +0,0 @@ -from logging.config import fileConfig - -from alembic import context -from sqlalchemy import engine_from_config, pool - -from app.config import get_settings -from app.models import Location # noqa: F401 -from app.db import Base - -config = context.config - -if config.config_file_name is not None: - fileConfig(config.config_file_name) - -settings = get_settings() -configured_url = config.get_main_option("sqlalchemy.url") -if not configured_url or configured_url == "sqlite:///./data/locationRecorder.db": - config.set_main_option("sqlalchemy.url", settings.location_database_url) - -target_metadata = Base.metadata - - -def run_migrations_offline() -> None: - url = config.get_main_option("sqlalchemy.url") - context.configure(url=url, target_metadata=target_metadata, literal_binds=True) - - with context.begin_transaction(): - context.run_migrations() - - -def run_migrations_online() -> None: - connectable = engine_from_config( - config.get_section(config.config_ini_section, {}), - prefix="sqlalchemy.", - poolclass=pool.NullPool, - ) - - with connectable.connect() as connection: - context.configure(connection=connection, target_metadata=target_metadata) - - with context.begin_transaction(): - context.run_migrations() - - -if context.is_offline_mode(): - run_migrations_offline() -else: - run_migrations_online() diff --git a/alembic_location/script.py.mako b/alembic_location/script.py.mako deleted file mode 100644 index 2e8960a..0000000 --- a/alembic_location/script.py.mako +++ /dev/null @@ -1,26 +0,0 @@ -"""${message} - -Revision ID: ${up_revision} -Revises: ${down_revision | comma,n} -Create Date: ${create_date} -""" -from typing import Sequence, Union - -from alembic import op -import sqlalchemy as sa - - -# revision identifiers, used by Alembic. -revision: str = ${repr(up_revision)} -down_revision: Union[str, None] = ${repr(down_revision)} -branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} -depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} - - -def upgrade() -> None: - ${upgrades if upgrades else "pass"} - - -def downgrade() -> None: - ${downgrades if downgrades else "pass"} - diff --git a/alembic_location/versions/.gitkeep b/alembic_location/versions/.gitkeep deleted file mode 100644 index 8b13789..0000000 --- a/alembic_location/versions/.gitkeep +++ /dev/null @@ -1 +0,0 @@ - diff --git a/alembic_location/versions/20260419_01_location_baseline.py b/alembic_location/versions/20260419_01_location_baseline.py deleted file mode 100644 index cc94da5..0000000 --- a/alembic_location/versions/20260419_01_location_baseline.py +++ /dev/null @@ -1,33 +0,0 @@ -"""location baseline - -Revision ID: 20260419_01_location_baseline -Revises: -Create Date: 2026-04-19 00:00:00.000000 -""" - -from typing import Sequence, Union - -from alembic import op -import sqlalchemy as sa - - -revision: str = "20260419_01_location_baseline" -down_revision: Union[str, None] = None -branch_labels: Union[str, Sequence[str], None] = None -depends_on: Union[str, Sequence[str], None] = None - - -def upgrade() -> None: - op.create_table( - "location", - sa.Column("person", sa.Text(), nullable=False), - sa.Column("datetime", sa.Text(), nullable=False), - sa.Column("latitude", sa.Float(), nullable=False), - sa.Column("longitude", sa.Float(), nullable=False), - sa.Column("altitude", sa.Float(), nullable=True), - sa.PrimaryKeyConstraint("person", "datetime"), - ) - - -def downgrade() -> None: - op.drop_table("location") diff --git a/alembic_poo.ini b/alembic_poo.ini deleted file mode 100644 index cfe1727..0000000 --- a/alembic_poo.ini +++ /dev/null @@ -1,37 +0,0 @@ -[alembic] -script_location = alembic_poo -prepend_sys_path = . -path_separator = os -sqlalchemy.url = sqlite:///./data/pooRecorder.db - -[loggers] -keys = root,sqlalchemy,alembic - -[handlers] -keys = console - -[formatters] -keys = generic - -[logger_root] -level = WARN -handlers = console - -[logger_sqlalchemy] -level = WARN -handlers = -qualname = sqlalchemy.engine - -[logger_alembic] -level = INFO -handlers = console -qualname = alembic - -[handler_console] -class = StreamHandler -args = (sys.stderr,) -level = NOTSET -formatter = generic - -[formatter_generic] -format = %(levelname)-5.5s [%(name)s] %(message)s diff --git a/alembic_poo/env.py b/alembic_poo/env.py deleted file mode 100644 index 98b528f..0000000 --- a/alembic_poo/env.py +++ /dev/null @@ -1,48 +0,0 @@ -from logging.config import fileConfig - -from alembic import context -from sqlalchemy import engine_from_config, pool - -from app.config import get_settings -from app.models.poo import PooRecord # noqa: F401 -from app.db import Base - -config = context.config - -if config.config_file_name is not None: - fileConfig(config.config_file_name) - -settings = get_settings() -configured_url = config.get_main_option("sqlalchemy.url") -if not configured_url or configured_url == "sqlite:///./data/pooRecorder.db": - config.set_main_option("sqlalchemy.url", settings.poo_database_url) - -target_metadata = Base.metadata - - -def run_migrations_offline() -> None: - url = config.get_main_option("sqlalchemy.url") - context.configure(url=url, target_metadata=target_metadata, literal_binds=True) - - with context.begin_transaction(): - context.run_migrations() - - -def run_migrations_online() -> None: - connectable = engine_from_config( - config.get_section(config.config_ini_section, {}), - prefix="sqlalchemy.", - poolclass=pool.NullPool, - ) - - with connectable.connect() as connection: - context.configure(connection=connection, target_metadata=target_metadata) - - with context.begin_transaction(): - context.run_migrations() - - -if context.is_offline_mode(): - run_migrations_offline() -else: - run_migrations_online() diff --git a/alembic_poo/versions/20260420_01_poo_baseline.py b/alembic_poo/versions/20260420_01_poo_baseline.py deleted file mode 100644 index 7abeb90..0000000 --- a/alembic_poo/versions/20260420_01_poo_baseline.py +++ /dev/null @@ -1,32 +0,0 @@ -"""poo baseline - -Revision ID: 20260420_01_poo_baseline -Revises: -Create Date: 2026-04-20 00:00:00.000000 -""" - -from typing import Sequence, Union - -from alembic import op -import sqlalchemy as sa - - -revision: str = "20260420_01_poo_baseline" -down_revision: Union[str, None] = None -branch_labels: Union[str, Sequence[str], None] = None -depends_on: Union[str, Sequence[str], None] = None - - -def upgrade() -> None: - op.create_table( - "poo_records", - sa.Column("timestamp", sa.Text(), nullable=False), - sa.Column("status", sa.Text(), nullable=False), - sa.Column("latitude", sa.Float(), nullable=False), - sa.Column("longitude", sa.Float(), nullable=False), - sa.PrimaryKeyConstraint("timestamp"), - ) - - -def downgrade() -> None: - op.drop_table("poo_records") diff --git a/app/main.py b/app/main.py index 3e0647e..6b42820 100644 --- a/app/main.py +++ b/app/main.py @@ -21,8 +21,6 @@ from app.services.auth import AuthBootstrapError, initialize_auth_schema from app.services.config_page import seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap from app.services.public_ip import check_public_ipv4_and_notify from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db -from scripts.location_db_adopt import LocationDatabaseAdoptionError, validate_location_runtime_db -from scripts.poo_db_adopt import PooDatabaseAdoptionError, validate_poo_runtime_db def _run_scheduled_public_ip_check() -> None: @@ -50,41 +48,16 @@ def ensure_auth_db_ready() -> None: session.close() -def ensure_location_db_ready() -> None: - settings = get_settings() - if settings.location_sqlite_path is None: - return - - try: - validate_location_runtime_db(settings.location_database_url) - except LocationDatabaseAdoptionError as exc: - raise RuntimeError(str(exc)) from exc - - -def ensure_poo_db_ready() -> None: - settings = get_settings() - if settings.poo_sqlite_path is None: - return - - try: - validate_poo_runtime_db(settings.poo_database_url) - except PooDatabaseAdoptionError as exc: - raise RuntimeError(str(exc)) from exc - - def ensure_runtime_dirs() -> None: settings = get_settings() - for path in (settings.app_sqlite_path, settings.location_sqlite_path, settings.poo_sqlite_path): - if path is not None: - path.parent.mkdir(parents=True, exist_ok=True) + if settings.app_sqlite_path is not None: + settings.app_sqlite_path.parent.mkdir(parents=True, exist_ok=True) @asynccontextmanager async def lifespan(_: FastAPI): ensure_runtime_dirs() ensure_auth_db_ready() - ensure_location_db_ready() - ensure_poo_db_ready() scheduler = BackgroundScheduler(timezone="UTC") scheduler.add_job( _run_scheduled_public_ip_check, diff --git a/scripts/location_db_adopt.py b/scripts/location_db_adopt.py deleted file mode 100644 index 4424364..0000000 --- a/scripts/location_db_adopt.py +++ /dev/null @@ -1,205 +0,0 @@ -from __future__ import annotations - -import sqlite3 -import sys -from pathlib import Path - -from alembic import command -from alembic.config import Config -from alembic.script import ScriptDirectory -from alembic.util.exc import CommandError - -PROJECT_ROOT = Path(__file__).resolve().parents[1] -if str(PROJECT_ROOT) not in sys.path: - sys.path.insert(0, str(PROJECT_ROOT)) - -from app.config import get_settings - -LOCATION_BASELINE_REVISION = "20260419_01_location_baseline" -EXPECTED_USER_VERSION = 2 -EXPECTED_LOCATION_TABLE_INFO = [ - (0, "person", "TEXT", 1, None, 1), - (1, "datetime", "TEXT", 1, None, 2), - (2, "latitude", "REAL", 1, None, 0), - (3, "longitude", "REAL", 1, None, 0), - (4, "altitude", "REAL", 0, None, 0), -] - - -class LocationDatabaseAdoptionError(RuntimeError): - """Raised when a legacy location database does not match the expected baseline.""" - - -def _database_path_from_url(database_url: str) -> Path: - prefix = "sqlite:///" - if not database_url.startswith(prefix): - raise LocationDatabaseAdoptionError( - f"Only sqlite URLs are supported for location DB adoption, got: {database_url}" - ) - return Path(database_url[len(prefix) :]) - - -def _make_alembic_config(database_url: str) -> Config: - config = Config("alembic_location.ini") - config.set_main_option("sqlalchemy.url", database_url) - return config - - -def _expected_head_revision(alembic_config: Config) -> str: - script = ScriptDirectory.from_config(alembic_config) - heads = script.get_heads() - if len(heads) != 1: - raise LocationDatabaseAdoptionError( - f"Expected exactly one Alembic head for location DB, got {len(heads)}" - ) - return heads[0] - - -def _is_known_revision(alembic_config: Config, revision: str) -> bool: - script = ScriptDirectory.from_config(alembic_config) - try: - return script.get_revision(revision) is not None - except CommandError: - return False - - -def _location_table_exists(database_path: Path) -> bool: - conn = sqlite3.connect(database_path) - try: - row = conn.execute( - "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'location'" - ).fetchone() - return row is not None - finally: - conn.close() - - -def _alembic_version_table_exists(database_path: Path) -> bool: - conn = sqlite3.connect(database_path) - try: - row = conn.execute( - "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'alembic_version'" - ).fetchone() - return row is not None - finally: - conn.close() - - -def _fetch_alembic_revision(database_path: Path) -> str: - conn = sqlite3.connect(database_path) - try: - row = conn.execute("SELECT version_num FROM alembic_version").fetchone() - if row is None: - raise LocationDatabaseAdoptionError( - "Alembic version table exists but contains no revision" - ) - return row[0] - finally: - conn.close() - - -def _fetch_location_table_info(database_path: Path) -> list[tuple]: - conn = sqlite3.connect(database_path) - try: - return list(conn.execute("PRAGMA table_info(location)")) - finally: - conn.close() - - -def _fetch_user_version(database_path: Path) -> int: - conn = sqlite3.connect(database_path) - try: - return conn.execute("PRAGMA user_version").fetchone()[0] - finally: - conn.close() - - -def validate_legacy_location_db(database_url: str) -> None: - database_path = _database_path_from_url(database_url) - if not database_path.exists(): - raise LocationDatabaseAdoptionError(f"Location DB file does not exist: {database_path}") - - if not _location_table_exists(database_path): - raise LocationDatabaseAdoptionError("Expected table 'location' was not found in the DB") - - table_info = _fetch_location_table_info(database_path) - if table_info != EXPECTED_LOCATION_TABLE_INFO: - raise LocationDatabaseAdoptionError( - "Location table schema does not match the expected baseline schema" - ) - - user_version = _fetch_user_version(database_path) - if user_version != EXPECTED_USER_VERSION: - raise LocationDatabaseAdoptionError( - f"Expected PRAGMA user_version = {EXPECTED_USER_VERSION}, got {user_version}" - ) - - -def validate_location_runtime_db(database_url: str) -> None: - database_path = _database_path_from_url(database_url) - alembic_config = _make_alembic_config(database_url) - expected_revision = _expected_head_revision(alembic_config) - if not database_path.exists(): - raise LocationDatabaseAdoptionError( - "Location DB file was not found. Run 'python scripts/location_db_adopt.py' " - "first to initialize or adopt the location DB before starting the app." - ) - - if not _alembic_version_table_exists(database_path): - raise LocationDatabaseAdoptionError( - "Location DB exists but is not yet Alembic-managed. Run " - "'python scripts/location_db_adopt.py' first to adopt the legacy DB " - "before starting the app." - ) - - current_revision = _fetch_alembic_revision(database_path) - if current_revision != expected_revision: - raise LocationDatabaseAdoptionError( - "Location DB revision mismatch. Refusing to start the app: " - f"expected {expected_revision}, got {current_revision}" - ) - - -def adopt_or_initialize_location_db(database_url: str) -> str: - database_path = _database_path_from_url(database_url) - alembic_config = _make_alembic_config(database_url) - expected_revision = _expected_head_revision(alembic_config) - - if database_path.exists(): - if _alembic_version_table_exists(database_path): - current_revision = _fetch_alembic_revision(database_path) - if current_revision == expected_revision: - return "already_managed" - if not _is_known_revision(alembic_config, current_revision): - raise LocationDatabaseAdoptionError( - "Location DB is already Alembic-managed but revision does not match " - f"a known migration revision: got {current_revision}" - ) - command.upgrade(alembic_config, "head") - return "upgraded" - - validate_legacy_location_db(database_url) - command.stamp(alembic_config, LOCATION_BASELINE_REVISION) - if LOCATION_BASELINE_REVISION != expected_revision: - command.upgrade(alembic_config, "head") - return "upgraded" - return "adopted" - - database_path.parent.mkdir(parents=True, exist_ok=True) - command.upgrade(alembic_config, "head") - return "initialized" - - -def main() -> None: - settings = get_settings() - result = adopt_or_initialize_location_db(settings.location_database_url) - if result == "initialized": - print("Initialized a new location DB via Alembic upgrade head.") - elif result == "already_managed": - print("Location DB is already Alembic-managed at the expected baseline revision.") - else: - print("Validated legacy location DB and stamped Alembic baseline successfully.") - - -if __name__ == "__main__": - main() diff --git a/scripts/poo_db_adopt.py b/scripts/poo_db_adopt.py deleted file mode 100644 index 7540dce..0000000 --- a/scripts/poo_db_adopt.py +++ /dev/null @@ -1,200 +0,0 @@ -from __future__ import annotations - -import sqlite3 -import sys -from pathlib import Path - -from alembic import command -from alembic.config import Config -from alembic.script import ScriptDirectory -from alembic.util.exc import CommandError - -PROJECT_ROOT = Path(__file__).resolve().parents[1] -if str(PROJECT_ROOT) not in sys.path: - sys.path.insert(0, str(PROJECT_ROOT)) - -from app.config import get_settings - -POO_BASELINE_REVISION = "20260420_01_poo_baseline" -EXPECTED_USER_VERSION = 1 -EXPECTED_POO_TABLE_INFO = [ - (0, "timestamp", "TEXT", 1, None, 1), - (1, "status", "TEXT", 1, None, 0), - (2, "latitude", "REAL", 1, None, 0), - (3, "longitude", "REAL", 1, None, 0), -] - - -class PooDatabaseAdoptionError(RuntimeError): - """Raised when a legacy poo database does not match the expected baseline.""" - - -def _database_path_from_url(database_url: str) -> Path: - prefix = "sqlite:///" - if not database_url.startswith(prefix): - raise PooDatabaseAdoptionError( - f"Only sqlite URLs are supported for poo DB adoption, got: {database_url}" - ) - return Path(database_url[len(prefix) :]) - - -def _make_alembic_config(database_url: str) -> Config: - config = Config("alembic_poo.ini") - config.set_main_option("sqlalchemy.url", database_url) - return config - - -def _expected_head_revision(alembic_config: Config) -> str: - script = ScriptDirectory.from_config(alembic_config) - heads = script.get_heads() - if len(heads) != 1: - raise PooDatabaseAdoptionError( - f"Expected exactly one Alembic head for poo DB, got {len(heads)}" - ) - return heads[0] - - -def _is_known_revision(alembic_config: Config, revision: str) -> bool: - script = ScriptDirectory.from_config(alembic_config) - try: - return script.get_revision(revision) is not None - except CommandError: - return False - - -def _poo_table_exists(database_path: Path) -> bool: - conn = sqlite3.connect(database_path) - try: - row = conn.execute( - "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'poo_records'" - ).fetchone() - return row is not None - finally: - conn.close() - - -def _alembic_version_table_exists(database_path: Path) -> bool: - conn = sqlite3.connect(database_path) - try: - row = conn.execute( - "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'alembic_version'" - ).fetchone() - return row is not None - finally: - conn.close() - - -def _fetch_alembic_revision(database_path: Path) -> str: - conn = sqlite3.connect(database_path) - try: - row = conn.execute("SELECT version_num FROM alembic_version").fetchone() - if row is None: - raise PooDatabaseAdoptionError("Alembic version table exists but contains no revision") - return row[0] - finally: - conn.close() - - -def _fetch_poo_table_info(database_path: Path) -> list[tuple]: - conn = sqlite3.connect(database_path) - try: - return list(conn.execute("PRAGMA table_info(poo_records)")) - finally: - conn.close() - - -def _fetch_user_version(database_path: Path) -> int: - conn = sqlite3.connect(database_path) - try: - return conn.execute("PRAGMA user_version").fetchone()[0] - finally: - conn.close() - - -def validate_legacy_poo_db(database_url: str) -> None: - database_path = _database_path_from_url(database_url) - if not database_path.exists(): - raise PooDatabaseAdoptionError(f"Poo DB file does not exist: {database_path}") - - if not _poo_table_exists(database_path): - raise PooDatabaseAdoptionError("Expected table 'poo_records' was not found in the DB") - - table_info = _fetch_poo_table_info(database_path) - if table_info != EXPECTED_POO_TABLE_INFO: - raise PooDatabaseAdoptionError("Poo table schema does not match the expected baseline") - - user_version = _fetch_user_version(database_path) - if user_version != EXPECTED_USER_VERSION: - raise PooDatabaseAdoptionError( - f"Expected PRAGMA user_version = {EXPECTED_USER_VERSION}, got {user_version}" - ) - - -def validate_poo_runtime_db(database_url: str) -> None: - database_path = _database_path_from_url(database_url) - alembic_config = _make_alembic_config(database_url) - expected_revision = _expected_head_revision(alembic_config) - if not database_path.exists(): - raise PooDatabaseAdoptionError( - "Poo DB file was not found. Run 'python scripts/poo_db_adopt.py' first to " - "initialize or adopt the poo DB before starting the app." - ) - - if not _alembic_version_table_exists(database_path): - raise PooDatabaseAdoptionError( - "Poo DB exists but is not yet Alembic-managed. Run " - "'python scripts/poo_db_adopt.py' first to adopt the legacy DB " - "before starting the app." - ) - - current_revision = _fetch_alembic_revision(database_path) - if current_revision != expected_revision: - raise PooDatabaseAdoptionError( - "Poo DB revision mismatch. Refusing to start the app: " - f"expected {expected_revision}, got {current_revision}" - ) - - -def adopt_or_initialize_poo_db(database_url: str) -> str: - database_path = _database_path_from_url(database_url) - alembic_config = _make_alembic_config(database_url) - expected_revision = _expected_head_revision(alembic_config) - - if database_path.exists(): - if _alembic_version_table_exists(database_path): - current_revision = _fetch_alembic_revision(database_path) - if current_revision == expected_revision: - return "already_managed" - if not _is_known_revision(alembic_config, current_revision): - raise PooDatabaseAdoptionError( - "Poo DB is already Alembic-managed but revision does not match " - f"a known migration revision: got {current_revision}" - ) - command.upgrade(alembic_config, "head") - return "upgraded" - - validate_legacy_poo_db(database_url) - command.stamp(alembic_config, POO_BASELINE_REVISION) - if POO_BASELINE_REVISION != expected_revision: - command.upgrade(alembic_config, "head") - return "upgraded" - return "adopted" - - database_path.parent.mkdir(parents=True, exist_ok=True) - command.upgrade(alembic_config, "head") - return "initialized" - - -def main() -> None: - settings = get_settings() - result = adopt_or_initialize_poo_db(settings.poo_database_url) - if result == "initialized": - print("Initialized a new poo DB via Alembic upgrade head.") - elif result == "already_managed": - print("Poo DB is already Alembic-managed at the expected baseline revision.") - else: - print("Validated legacy poo DB and stamped Alembic baseline successfully.") - - -if __name__ == "__main__": - main() diff --git a/scripts/run_migrations.py b/scripts/run_migrations.py index ba3af4e..b213886 100644 --- a/scripts/run_migrations.py +++ b/scripts/run_migrations.py @@ -2,16 +2,12 @@ from __future__ import annotations from app.config import get_settings from scripts.app_db_adopt import adopt_or_initialize_app_db -from scripts.location_db_adopt import adopt_or_initialize_location_db -from scripts.poo_db_adopt import adopt_or_initialize_poo_db def run_all_migrations() -> dict[str, str]: settings = get_settings() return { "app": adopt_or_initialize_app_db(settings.app_database_url), - "location": adopt_or_initialize_location_db(settings.location_database_url), - "poo": adopt_or_initialize_poo_db(settings.poo_database_url), } diff --git a/tests/conftest.py b/tests/conftest.py index 270284b..da18463 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,18 +17,6 @@ def _make_app_alembic_config(database_url: str) -> Config: return config -def _make_alembic_config(database_url: str) -> Config: - config = Config("alembic_location.ini") - config.set_main_option("sqlalchemy.url", database_url) - return config - - -def _make_poo_alembic_config(database_url: str) -> Config: - config = Config("alembic_poo.ini") - config.set_main_option("sqlalchemy.url", database_url) - return config - - @pytest.fixture def test_database_urls(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): app_database_path = tmp_path / "app_test.db" @@ -61,18 +49,6 @@ def test_database_urls(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): reset_db_caches() -@pytest.fixture -def ready_location_database(test_database_urls): - command.upgrade(_make_alembic_config(test_database_urls["location_url"]), "head") - return test_database_urls - - -@pytest.fixture -def ready_poo_database(test_database_urls): - command.upgrade(_make_poo_alembic_config(test_database_urls["poo_url"]), "head") - return test_database_urls - - @pytest.fixture def auth_database(test_database_urls, monkeypatch: pytest.MonkeyPatch): database_url = test_database_urls["app_url"] @@ -84,7 +60,7 @@ def auth_database(test_database_urls, monkeypatch: pytest.MonkeyPatch): @pytest.fixture -def app(ready_location_database, ready_poo_database, auth_database): +def app(auth_database): yield create_app() @@ -95,7 +71,7 @@ def client(app): @pytest.fixture -def location_client(ready_location_database, ready_poo_database, auth_database): +def location_client(auth_database): app_url = auth_database["app_url"] engine = create_engine(app_url, connect_args={"check_same_thread": False}) fastapi_app = create_app() @@ -105,7 +81,7 @@ def location_client(ready_location_database, ready_poo_database, auth_database): @pytest.fixture -def poo_client(ready_location_database, ready_poo_database, auth_database): +def poo_client(auth_database): app_url = auth_database["app_url"] engine = create_engine(app_url, connect_args={"check_same_thread": False}) fastapi_app = create_app() diff --git a/tests/test_app.py b/tests/test_app.py index 2857032..79b611d 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -9,7 +9,7 @@ from app.db import reset_db_caches from app.config import get_settings from app.main import create_app from scripts.app_db_adopt import APP_BASELINE_REVISION, adopt_or_initialize_app_db -from tests.conftest import _make_alembic_config, _make_app_alembic_config, _make_poo_alembic_config +from tests.conftest import _make_app_alembic_config async def _run_lifespan(app) -> None: @@ -38,16 +38,10 @@ def test_status_endpoint(client: TestClient) -> None: def test_app_start_fails_when_app_db_missing(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None: missing_app_path = tmp_path / "missing_app.db" - poo_database_path = tmp_path / "poo_ready.db" - location_database_path = tmp_path / "location_ready.db" - command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head") - command.upgrade(_make_alembic_config(f"sqlite:///{location_database_path}"), "head") monkeypatch.setenv("APP_DATABASE_URL", f"sqlite:///{missing_app_path}") monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") - monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{location_database_path}") - monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}") get_settings.cache_clear() reset_db_caches() @@ -86,10 +80,6 @@ def test_app_start_seeds_missing_config_from_env_without_overwriting_existing_va tmp_path, monkeypatch: pytest.MonkeyPatch ) -> None: app_database_url = _prepare_app_db(tmp_path) - location_database_path = tmp_path / "location_ready.db" - poo_database_path = tmp_path / "poo_ready.db" - command.upgrade(_make_alembic_config(f"sqlite:///{location_database_path}"), "head") - command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head") app_database_path = tmp_path / "app_ready.db" conn = sqlite3.connect(app_database_path) @@ -105,8 +95,6 @@ def test_app_start_seeds_missing_config_from_env_without_overwriting_existing_va monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") monkeypatch.setenv("APP_NAME", "Bootstrap Name") monkeypatch.setenv("HOME_ASSISTANT_BASE_URL", "http://bootstrap-ha.local:8123") - monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{location_database_path}") - monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}") get_settings.cache_clear() reset_db_caches() @@ -131,10 +119,6 @@ def test_app_start_syncs_app_hostname_from_env_even_when_db_has_old_value( tmp_path, monkeypatch: pytest.MonkeyPatch ) -> None: app_database_url = _prepare_app_db(tmp_path) - location_database_path = tmp_path / "location_ready.db" - poo_database_path = tmp_path / "poo_ready.db" - command.upgrade(_make_alembic_config(f"sqlite:///{location_database_path}"), "head") - command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head") app_database_path = tmp_path / "app_ready.db" conn = sqlite3.connect(app_database_path) @@ -149,8 +133,6 @@ def test_app_start_syncs_app_hostname_from_env_even_when_db_has_old_value( monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") monkeypatch.setenv("APP_HOSTNAME", "new.example.com") - monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{location_database_path}") - monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}") get_settings.cache_clear() reset_db_caches() @@ -167,98 +149,3 @@ def test_app_start_syncs_app_hostname_from_env_even_when_db_has_old_value( get_settings.cache_clear() reset_db_caches() - - -def test_app_start_fails_when_location_db_missing( - tmp_path, monkeypatch: pytest.MonkeyPatch -) -> None: - app_database_url = _prepare_app_db(tmp_path) - monkeypatch.setenv("APP_DATABASE_URL", app_database_url) - monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") - monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") - poo_database_path = tmp_path / "poo_ready.db" - command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head") - - monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{tmp_path / 'missing.db'}") - monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}") - get_settings.cache_clear() - reset_db_caches() - - app = create_app() - with pytest.raises(RuntimeError, match="Run 'python scripts/location_db_adopt.py' first"): - anyio.run(_run_lifespan, app) - - get_settings.cache_clear() - reset_db_caches() - - -def test_app_start_fails_when_location_db_exists_but_is_not_adopted( - tmp_path, monkeypatch: pytest.MonkeyPatch -) -> None: - app_database_url = _prepare_app_db(tmp_path) - monkeypatch.setenv("APP_DATABASE_URL", app_database_url) - monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") - monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") - poo_database_path = tmp_path / "poo_ready.db" - command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head") - - database_path = tmp_path / "legacy_only.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE location ( - person TEXT NOT NULL, - datetime TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - altitude REAL, - PRIMARY KEY (person, datetime) - ) - """ - ) - conn.execute("PRAGMA user_version = 2") - conn.commit() - conn.close() - - monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{database_path}") - monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}") - get_settings.cache_clear() - reset_db_caches() - - app = create_app() - with pytest.raises(RuntimeError, match="is not yet Alembic-managed"): - anyio.run(_run_lifespan, app) - - get_settings.cache_clear() - reset_db_caches() - - -def test_app_start_fails_when_location_db_revision_mismatches( - tmp_path, monkeypatch: pytest.MonkeyPatch -) -> None: - app_database_url = _prepare_app_db(tmp_path) - monkeypatch.setenv("APP_DATABASE_URL", app_database_url) - monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") - monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") - poo_database_path = tmp_path / "poo_ready.db" - command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head") - - database_path = tmp_path / "wrong_revision.db" - command.upgrade(_make_alembic_config(f"sqlite:///{database_path}"), "head") - - conn = sqlite3.connect(database_path) - conn.execute("UPDATE alembic_version SET version_num = 'wrong_revision'") - conn.commit() - conn.close() - - monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{database_path}") - monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}") - get_settings.cache_clear() - reset_db_caches() - - app = create_app() - with pytest.raises(RuntimeError, match="Location DB revision mismatch"): - anyio.run(_run_lifespan, app) - - get_settings.cache_clear() - reset_db_caches() diff --git a/tests/test_auth.py b/tests/test_auth.py index 2558a5e..f39849d 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -194,9 +194,6 @@ def test_config_page_update_persists_to_database( def test_config_page_shows_ticktick_oauth_link_when_ticktick_is_configured( - test_database_urls, - ready_location_database, - ready_poo_database, auth_database, monkeypatch, ) -> None: diff --git a/tests/test_deployment.py b/tests/test_deployment.py index f2b4f28..6bc10b2 100644 --- a/tests/test_deployment.py +++ b/tests/test_deployment.py @@ -4,18 +4,12 @@ import sqlite3 import anyio import pytest import yaml -from alembic import command from app.db import reset_db_caches from app.config import get_settings from app.main import create_app from scripts.app_db_adopt import APP_BASELINE_REVISION -from scripts.location_db_adopt import EXPECTED_USER_VERSION as LOCATION_USER_VERSION -from scripts.location_db_adopt import LOCATION_BASELINE_REVISION -from scripts.poo_db_adopt import EXPECTED_USER_VERSION as POO_USER_VERSION -from scripts.poo_db_adopt import POO_BASELINE_REVISION from scripts.run_migrations import run_all_migrations -from tests.conftest import _make_alembic_config, _make_poo_alembic_config PROJECT_ROOT = Path(__file__).resolve().parents[1] @@ -31,12 +25,8 @@ async def _run_lifespan(app) -> None: def _configure_database_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Path | str]: app_path = tmp_path / "app.db" - location_path = tmp_path / "location.db" - poo_path = tmp_path / "poo.db" monkeypatch.setenv("APP_DATABASE_URL", f"sqlite:///{app_path}") - monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{location_path}") - monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_path}") monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password") monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false") @@ -46,58 +36,9 @@ def _configure_database_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> return { "app_path": app_path, "app_url": f"sqlite:///{app_path}", - "location_path": location_path, - "location_url": f"sqlite:///{location_path}", - "poo_path": poo_path, - "poo_url": f"sqlite:///{poo_path}", } -def _create_legacy_location_db(database_path: Path) -> None: - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE location ( - person TEXT NOT NULL, - datetime TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - altitude REAL, - PRIMARY KEY (person, datetime) - ) - """ - ) - conn.execute( - "INSERT INTO location (person, datetime, latitude, longitude, altitude) VALUES (?, ?, ?, ?, ?)", - ("alice", "2026-04-22T10:00:00Z", 1.23, 4.56, 7.89), - ) - conn.execute(f"PRAGMA user_version = {LOCATION_USER_VERSION}") - conn.commit() - conn.close() - - -def _create_legacy_poo_db(database_path: Path) -> None: - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE poo_records ( - timestamp TEXT NOT NULL, - status TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - PRIMARY KEY (timestamp) - ) - """ - ) - conn.execute( - "INSERT INTO poo_records (timestamp, status, latitude, longitude) VALUES (?, ?, ?, ?)", - ("2026-04-22T11:00:00Z", "complete", 9.87, 6.54), - ) - conn.execute(f"PRAGMA user_version = {POO_USER_VERSION}") - conn.commit() - conn.close() - - def test_compose_uses_migration_job_before_app() -> None: compose = _read_yaml("docker-compose.yml") override = _read_yaml("docker-compose.override.yml") @@ -131,12 +72,8 @@ def test_migration_runner_initializes_and_is_idempotent( first_run = run_all_migrations() second_run = run_all_migrations() - assert first_run == {"app": "initialized", "location": "initialized", "poo": "initialized"} - assert second_run == { - "app": "already_managed", - "location": "already_managed", - "poo": "already_managed", - } + assert first_run == {"app": "initialized"} + assert second_run == {"app": "already_managed"} conn = sqlite3.connect(database_urls["app_path"]) try: @@ -150,48 +87,9 @@ def test_migration_runner_initializes_and_is_idempotent( finally: conn.close() - assert {"auth_users", "auth_sessions", "app_config", "alembic_version"} <= tables - - conn = sqlite3.connect(database_urls["location_path"]) - try: - assert conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] == LOCATION_BASELINE_REVISION - finally: - conn.close() - - conn = sqlite3.connect(database_urls["poo_path"]) - try: - assert conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] == POO_BASELINE_REVISION - finally: - conn.close() - - get_settings.cache_clear() - reset_db_caches() - - -def test_migration_runner_adopts_legacy_sqlite_without_data_loss( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - database_urls = _configure_database_env(tmp_path, monkeypatch) - _create_legacy_location_db(database_urls["location_path"]) - _create_legacy_poo_db(database_urls["poo_path"]) - - results = run_all_migrations() - - assert results == {"app": "initialized", "location": "adopted", "poo": "adopted"} - - conn = sqlite3.connect(database_urls["location_path"]) - try: - assert conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] == LOCATION_BASELINE_REVISION - assert conn.execute("SELECT COUNT(*) FROM location").fetchone()[0] == 1 - finally: - conn.close() - - conn = sqlite3.connect(database_urls["poo_path"]) - try: - assert conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] == POO_BASELINE_REVISION - assert conn.execute("SELECT COUNT(*) FROM poo_records").fetchone()[0] == 1 - finally: - conn.close() + assert { + "auth_users", "auth_sessions", "app_config", "alembic_version", "location", "poo_records" + } <= tables get_settings.cache_clear() reset_db_caches() @@ -202,8 +100,6 @@ def test_app_startup_still_fails_closed_without_running_adoption( ) -> None: database_urls = _configure_database_env(tmp_path, monkeypatch) missing_app_path = database_urls["app_path"] - command.upgrade(_make_alembic_config(database_urls["location_url"]), "head") - command.upgrade(_make_poo_alembic_config(database_urls["poo_url"]), "head") app = create_app() with pytest.raises(RuntimeError, match="Run 'python scripts/app_db_adopt.py' first"): diff --git a/tests/test_homeassistant_inbound.py b/tests/test_homeassistant_inbound.py index 47407f7..04c2b79 100644 --- a/tests/test_homeassistant_inbound.py +++ b/tests/test_homeassistant_inbound.py @@ -156,8 +156,6 @@ def test_homeassistant_publish_rejects_invalid_ticktick_content(location_client) def test_homeassistant_publish_poo_get_latest_publishes_latest_status( - ready_location_database, - ready_poo_database, auth_database, ) -> None: from fastapi.testclient import TestClient @@ -215,8 +213,6 @@ def test_homeassistant_publish_poo_get_latest_publishes_latest_status( def test_homeassistant_publish_returns_internal_error_for_unknown_poo_action( - ready_location_database, - ready_poo_database, auth_database, ) -> None: from fastapi.testclient import TestClient diff --git a/tests/test_location.py b/tests/test_location.py index f510245..230851d 100644 --- a/tests/test_location.py +++ b/tests/test_location.py @@ -1,24 +1,8 @@ from datetime import datetime -from pathlib import Path -import sqlite3 import pytest -from alembic import command -from alembic.config import Config from sqlalchemy import text -from scripts.location_db_adopt import ( - EXPECTED_USER_VERSION, - LOCATION_BASELINE_REVISION, - LocationDatabaseAdoptionError, - adopt_or_initialize_location_db, -) - - -def _make_alembic_config(database_url: str) -> Config: - config = Config("alembic_location.ini") - config.set_main_option("sqlalchemy.url", database_url) - return config def test_location_record_endpoint_writes_row(location_client) -> None: client, engine = location_client @@ -193,136 +177,3 @@ def test_location_record_endpoint_defaults_invalid_altitude_to_zero(location_cli assert row.altitude == pytest.approx(0.0) -def test_location_db_adoption_initializes_new_db(tmp_path: Path) -> None: - database_path = tmp_path / "new_location.db" - result = adopt_or_initialize_location_db(f"sqlite:///{database_path}") - - assert result == "initialized" - assert database_path.exists() - - conn = sqlite3.connect(database_path) - try: - revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] - location_table = conn.execute( - "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'location'" - ).fetchone() - finally: - conn.close() - - assert revision == LOCATION_BASELINE_REVISION - assert location_table is not None - - -def test_location_db_adoption_validates_and_stamps_legacy_db(tmp_path: Path) -> None: - database_path = tmp_path / "legacy_location.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE location ( - person TEXT NOT NULL, - datetime TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - altitude REAL, - PRIMARY KEY (person, datetime) - ) - """ - ) - conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}") - conn.commit() - conn.close() - - result = adopt_or_initialize_location_db(f"sqlite:///{database_path}") - - assert result == "adopted" - - conn = sqlite3.connect(database_path) - try: - revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] - finally: - conn.close() - - assert revision == LOCATION_BASELINE_REVISION - - -def test_location_db_adoption_accepts_already_managed_matching_revision( - tmp_path: Path, -) -> None: - database_path = tmp_path / "managed_location.db" - command.upgrade(_make_alembic_config(f"sqlite:///{database_path}"), "head") - - result = adopt_or_initialize_location_db(f"sqlite:///{database_path}") - - assert result == "already_managed" - - -def test_location_db_adoption_fails_closed_on_alembic_revision_mismatch( - tmp_path: Path, -) -> None: - database_path = tmp_path / "wrong_revision.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE location ( - person TEXT NOT NULL, - datetime TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - altitude REAL, - PRIMARY KEY (person, datetime) - ) - """ - ) - conn.execute("CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)") - conn.execute("INSERT INTO alembic_version (version_num) VALUES ('wrong_revision')") - conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}") - conn.commit() - conn.close() - - with pytest.raises(LocationDatabaseAdoptionError, match="known migration revision"): - adopt_or_initialize_location_db(f"sqlite:///{database_path}") - - -def test_location_db_adoption_fails_closed_on_schema_mismatch(tmp_path: Path) -> None: - database_path = tmp_path / "bad_schema.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE location ( - person TEXT NOT NULL, - datetime TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - PRIMARY KEY (person, datetime) - ) - """ - ) - conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}") - conn.commit() - conn.close() - - with pytest.raises(LocationDatabaseAdoptionError, match="schema does not match"): - adopt_or_initialize_location_db(f"sqlite:///{database_path}") - - -def test_location_db_adoption_fails_closed_on_user_version_mismatch(tmp_path: Path) -> None: - database_path = tmp_path / "bad_user_version.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE location ( - person TEXT NOT NULL, - datetime TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - altitude REAL, - PRIMARY KEY (person, datetime) - ) - """ - ) - conn.execute("PRAGMA user_version = 999") - conn.commit() - conn.close() - - with pytest.raises(LocationDatabaseAdoptionError, match="Expected PRAGMA user_version"): - adopt_or_initialize_location_db(f"sqlite:///{database_path}") diff --git a/tests/test_poo.py b/tests/test_poo.py index 9e9a4d5..f3c6eb3 100644 --- a/tests/test_poo.py +++ b/tests/test_poo.py @@ -1,17 +1,8 @@ -from pathlib import Path -import sqlite3 - import pytest from sqlalchemy import text from app.config import Settings, get_settings from app.dependencies import get_app_settings, get_homeassistant_client -from scripts.poo_db_adopt import ( - EXPECTED_USER_VERSION, - POO_BASELINE_REVISION, - PooDatabaseAdoptionError, - adopt_or_initialize_poo_db, -) class _FakeHomeAssistantClient: @@ -153,96 +144,3 @@ def test_poo_latest_endpoint_returns_ok_when_no_record_exists(poo_client_with_ov assert response.text == "" -def test_poo_db_adoption_initializes_new_db(tmp_path: Path) -> None: - database_path = tmp_path / "new_poo.db" - - result = adopt_or_initialize_poo_db(f"sqlite:///{database_path}") - - assert result == "initialized" - assert database_path.exists() - - conn = sqlite3.connect(database_path) - try: - revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] - poo_table = conn.execute( - "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'poo_records'" - ).fetchone() - finally: - conn.close() - - assert revision == POO_BASELINE_REVISION - assert poo_table is not None - - -def test_poo_db_adoption_validates_and_stamps_legacy_db(tmp_path: Path) -> None: - database_path = tmp_path / "legacy_poo.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE poo_records ( - timestamp TEXT NOT NULL, - status TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - PRIMARY KEY (timestamp) - ) - """ - ) - conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}") - conn.commit() - conn.close() - - result = adopt_or_initialize_poo_db(f"sqlite:///{database_path}") - - assert result == "adopted" - - conn = sqlite3.connect(database_path) - try: - revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] - finally: - conn.close() - - assert revision == POO_BASELINE_REVISION - - -def test_poo_db_adoption_fails_closed_on_schema_mismatch(tmp_path: Path) -> None: - database_path = tmp_path / "bad_poo_schema.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE poo_records ( - timestamp TEXT NOT NULL, - status TEXT NOT NULL, - latitude REAL NOT NULL, - PRIMARY KEY (timestamp) - ) - """ - ) - conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}") - conn.commit() - conn.close() - - with pytest.raises(PooDatabaseAdoptionError, match="schema does not match"): - adopt_or_initialize_poo_db(f"sqlite:///{database_path}") - - -def test_poo_db_adoption_fails_closed_on_user_version_mismatch(tmp_path: Path) -> None: - database_path = tmp_path / "bad_poo_user_version.db" - conn = sqlite3.connect(database_path) - conn.execute( - """ - CREATE TABLE poo_records ( - timestamp TEXT NOT NULL, - status TEXT NOT NULL, - latitude REAL NOT NULL, - longitude REAL NOT NULL, - PRIMARY KEY (timestamp) - ) - """ - ) - conn.execute("PRAGMA user_version = 999") - conn.commit() - conn.close() - - with pytest.raises(PooDatabaseAdoptionError, match="Expected PRAGMA user_version"): - adopt_or_initialize_poo_db(f"sqlite:///{database_path}") diff --git a/tests/test_ticktick.py b/tests/test_ticktick.py index 036a486..f16cfab 100644 --- a/tests/test_ticktick.py +++ b/tests/test_ticktick.py @@ -209,9 +209,6 @@ def test_create_task_posts_expected_payload(monkeypatch: pytest.MonkeyPatch) -> def test_homeassistant_publish_creates_ticktick_action_task( - test_database_urls, - ready_location_database, - ready_poo_database, auth_database, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -255,9 +252,6 @@ def test_homeassistant_publish_creates_ticktick_action_task( def test_ticktick_auth_start_redirects_authenticated_user( - test_database_urls, - ready_location_database, - ready_poo_database, auth_database, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -291,9 +285,6 @@ def test_ticktick_auth_start_redirects_authenticated_user( def test_ticktick_auth_callback_persists_token( - test_database_urls, - ready_location_database, - ready_poo_database, auth_database, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -318,7 +309,7 @@ def test_ticktick_auth_callback_persists_token( assert response.status_code == 303 assert response.headers["location"] == "/config?ticktick_oauth=success" - conn = sqlite3.connect(test_database_urls["app_path"]) + conn = sqlite3.connect(auth_database["app_path"]) try: row = conn.execute( "SELECT value FROM app_config WHERE key = ?", @@ -332,9 +323,6 @@ def test_ticktick_auth_callback_persists_token( def test_ticktick_auth_callback_redirects_on_invalid_state( - test_database_urls, - ready_location_database, - ready_poo_database, auth_database, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -356,9 +344,6 @@ def test_ticktick_auth_callback_redirects_on_invalid_state( def test_ticktick_auth_callback_redirects_when_token_exchange_fails( - test_database_urls, - ready_location_database, - ready_poo_database, auth_database, monkeypatch: pytest.MonkeyPatch, ) -> None: