From 9053f6a5428899ac84404971917876752143af38 Mon Sep 17 00:00:00 2001 From: Kimball Bighorse Date: Wed, 4 Feb 2026 09:14:23 -0800 Subject: [PATCH 1/6] feat: add integration tests for Alembic migrations Add comprehensive test coverage for Alembic migrations to catch migration errors before deployment. Tests include: - Migration history validation (no branching, chain integrity) - Schema verification (core tables, NMA legacy tables, columns) - Foreign key integrity (data model relationships) - Index verification (spatial indexes) - Downgrade capability (skipped by default for safety) Fixes #356 Co-Authored-By: Claude Opus 4.5 --- tests/integration/test_alembic_migrations.py | 381 +++++++++++++++++++ 1 file changed, 381 insertions(+) create mode 100644 tests/integration/test_alembic_migrations.py diff --git a/tests/integration/test_alembic_migrations.py b/tests/integration/test_alembic_migrations.py new file mode 100644 index 00000000..f9f2c200 --- /dev/null +++ b/tests/integration/test_alembic_migrations.py @@ -0,0 +1,381 @@ +# =============================================================================== +# Copyright 2026 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Integration tests for Alembic migrations. + +Tests that: +1. Migrations run successfully (upgrade head) +2. Expected tables and columns exist after migration +3. Migration history is consistent +4. Downgrade paths work (optional, selected migrations) + +These tests ensure CI catches migration errors before merge and that +schema drift between models and migrations is detected. + +Related: GitHub Issue #356 +""" + +import os + +import pytest +from alembic import command +from alembic.config import Config +from alembic.script import ScriptDirectory +from sqlalchemy import inspect, text + +from db.engine import engine, session_ctx + + +def _alembic_config() -> Config: + """Get Alembic configuration pointing to project root.""" + root = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) + cfg = Config(os.path.join(root, "alembic.ini")) + cfg.set_main_option("script_location", os.path.join(root, "alembic")) + return cfg + + +# ============================================================================= +# Migration History Tests +# ============================================================================= + + +class TestMigrationHistory: + """Tests for migration script consistency.""" + + def test_migrations_have_no_multiple_heads(self): + """ + Migration history should have a single head (no branching). + + Multiple heads indicate parallel migrations that need to be merged. + """ + config = _alembic_config() + script = ScriptDirectory.from_config(config) + heads = script.get_heads() + + assert len(heads) == 1, ( + f"Multiple migration heads detected: {heads}. " + "Run 'alembic merge heads' to resolve." + ) + + def test_all_migrations_have_down_revision(self): + """ + All migrations except the first should have a down_revision. + + This ensures the migration chain is unbroken. + """ + config = _alembic_config() + script = ScriptDirectory.from_config(config) + + revisions_without_down = [] + base_found = False + + for rev in script.walk_revisions(): + if rev.down_revision is None: + if base_found: + revisions_without_down.append(rev.revision) + base_found = True + + assert not revisions_without_down, ( + f"Migrations missing down_revision (besides base): {revisions_without_down}" + ) + + def test_current_revision_matches_head(self): + """ + Database should be at the latest migration head. + + This verifies that test setup ran migrations successfully. + """ + config = _alembic_config() + script = ScriptDirectory.from_config(config) + head = script.get_current_head() + + with engine.connect() as conn: + result = conn.execute( + text("SELECT version_num FROM alembic_version") + ) + current = result.scalar() + + assert current == head, ( + f"Database at revision {current}, expected head {head}. " + "Run 'alembic upgrade head'." + ) + + +# ============================================================================= +# Schema Verification Tests +# ============================================================================= + + +class TestSchemaAfterMigration: + """Tests that verify expected schema exists after migrations.""" + + @pytest.fixture(autouse=True) + def inspector(self): + """Provide SQLAlchemy inspector for schema introspection.""" + self._inspector = inspect(engine) + yield + self._inspector = None + + def test_core_tables_exist(self): + """Core application tables should exist after migration.""" + expected_tables = [ + "location", + "thing", + "observation", + "sample", + "sensor", + "contact", + "field_event", + "field_activity", + "group", + "asset", + "parameter", + "lexicon_term", + "lexicon_category", + ] + + existing_tables = self._inspector.get_table_names() + + missing = [t for t in expected_tables if t not in existing_tables] + assert not missing, f"Missing core tables: {missing}" + + def test_legacy_nma_tables_exist(self): + """Legacy NMA tables should exist for data migration support.""" + expected_nma_tables = [ + "NMA_Chemistry_SampleInfo", + "NMA_MajorChemistry", + "NMA_MinorTraceChemistry", + "NMA_FieldParameters", + "NMA_HydraulicsData", + "NMA_Stratigraphy", + "NMA_Radionuclides", + "NMA_AssociatedData", + "NMA_WeatherData", + ] + + existing_tables = self._inspector.get_table_names() + + missing = [t for t in expected_nma_tables if t not in existing_tables] + assert not missing, f"Missing NMA legacy tables: {missing}" + + def test_thing_table_has_required_columns(self): + """Thing table should have all required columns.""" + columns = {c["name"] for c in self._inspector.get_columns("thing")} + + required_columns = [ + "id", + "name", + "thing_type", + "release_status", + "created_at", + "nma_pk_welldata", + "nma_pk_location", + ] + + missing = [c for c in required_columns if c not in columns] + assert not missing, f"Thing table missing columns: {missing}" + + def test_location_table_has_geometry_column(self): + """Location table should have PostGIS geometry column.""" + columns = {c["name"] for c in self._inspector.get_columns("location")} + + assert "point" in columns, "Location table missing 'point' geometry column" + + def test_observation_table_has_required_columns(self): + """Observation table should have all required columns.""" + columns = {c["name"] for c in self._inspector.get_columns("observation")} + + required_columns = [ + "id", + "observation_datetime", + "value", + "unit", + "sample_id", + "release_status", + ] + + missing = [c for c in required_columns if c not in columns] + assert not missing, f"Observation table missing columns: {missing}" + + def test_alembic_version_table_exists(self): + """Alembic version tracking table should exist.""" + tables = self._inspector.get_table_names() + assert "alembic_version" in tables, "alembic_version table missing" + + def test_postgis_extension_enabled(self): + """PostGIS extension should be enabled.""" + with session_ctx() as session: + result = session.execute( + text("SELECT extname FROM pg_extension WHERE extname = 'postgis'") + ) + postgis = result.scalar() + + assert postgis == "postgis", "PostGIS extension not enabled" + + +# ============================================================================= +# Foreign Key Integrity Tests +# ============================================================================= + + +class TestForeignKeyIntegrity: + """Tests that verify FK relationships are properly defined.""" + + @pytest.fixture(autouse=True) + def inspector(self): + """Provide SQLAlchemy inspector for schema introspection.""" + self._inspector = inspect(engine) + yield + self._inspector = None + + def test_observation_has_sample_fk(self): + """Observation should have FK to Sample.""" + fks = self._inspector.get_foreign_keys("observation") + fk_tables = {fk["referred_table"] for fk in fks} + + assert "sample" in fk_tables, "Observation missing FK to sample" + + def test_sample_has_field_activity_fk(self): + """Sample should have FK to FieldActivity.""" + fks = self._inspector.get_foreign_keys("sample") + fk_tables = {fk["referred_table"] for fk in fks} + + assert "field_activity" in fk_tables, "Sample missing FK to field_activity" + + def test_field_activity_has_field_event_fk(self): + """FieldActivity should have FK to FieldEvent.""" + fks = self._inspector.get_foreign_keys("field_activity") + fk_tables = {fk["referred_table"] for fk in fks} + + assert "field_event" in fk_tables, "FieldActivity missing FK to field_event" + + def test_field_event_has_thing_fk(self): + """FieldEvent should have FK to Thing.""" + fks = self._inspector.get_foreign_keys("field_event") + fk_tables = {fk["referred_table"] for fk in fks} + + assert "thing" in fk_tables, "FieldEvent missing FK to thing" + + def test_nma_chemistry_has_thing_fk(self): + """NMA_Chemistry_SampleInfo should have FK to Thing.""" + fks = self._inspector.get_foreign_keys("NMA_Chemistry_SampleInfo") + fk_tables = {fk["referred_table"] for fk in fks} + + assert "thing" in fk_tables, ( + "NMA_Chemistry_SampleInfo missing FK to thing" + ) + + +# ============================================================================= +# Index Tests +# ============================================================================= + + +class TestIndexes: + """Tests that verify important indexes exist.""" + + @pytest.fixture(autouse=True) + def inspector(self): + """Provide SQLAlchemy inspector for schema introspection.""" + self._inspector = inspect(engine) + yield + self._inspector = None + + def test_location_has_spatial_index(self): + """Location table should have spatial index on point column.""" + indexes = self._inspector.get_indexes("location") + index_columns = [] + for idx in indexes: + index_columns.extend(idx.get("column_names", [])) + + # Spatial indexes may be named differently, check for point column + # or gist index type + has_point_index = "point" in index_columns or any( + "point" in str(idx.get("name", "")).lower() or + "gist" in str(idx.get("name", "")).lower() + for idx in indexes + ) + + # Also check via pg_indexes for GIST indexes + if not has_point_index: + with session_ctx() as session: + result = session.execute( + text(""" + SELECT indexname FROM pg_indexes + WHERE tablename = 'location' + AND indexdef LIKE '%gist%' + """) + ) + gist_indexes = result.fetchall() + has_point_index = len(gist_indexes) > 0 + + assert has_point_index, "Location table missing spatial index on point" + + +# ============================================================================= +# Downgrade Tests (Selective) +# ============================================================================= + + +class TestMigrationDowngrade: + """ + Tests for migration downgrade capability. + + Note: These tests are more expensive as they modify schema. + Only test critical migrations. + """ + + @pytest.mark.skip(reason="Downgrade tests modify schema - run manually") + def test_can_downgrade_one_revision(self): + """ + Should be able to downgrade one revision and upgrade back. + + This is a destructive test - skipped by default. + """ + config = _alembic_config() + script = ScriptDirectory.from_config(config) + head = script.get_current_head() + + # Get the revision before head + head_script = script.get_revision(head) + if head_script.down_revision is None: + pytest.skip("Cannot downgrade from base revision") + + previous = head_script.down_revision + if isinstance(previous, tuple): + previous = previous[0] + + # Downgrade + command.downgrade(config, previous) + + # Verify we're at previous revision + with engine.connect() as conn: + result = conn.execute( + text("SELECT version_num FROM alembic_version") + ) + current = result.scalar() + assert current == previous + + # Upgrade back + command.upgrade(config, "head") + + # Verify we're back at head + with engine.connect() as conn: + result = conn.execute( + text("SELECT version_num FROM alembic_version") + ) + current = result.scalar() + assert current == head From 1684768727a92454c4e9acec4aabca9be2a19879 Mon Sep 17 00:00:00 2001 From: kbighorse Date: Wed, 4 Feb 2026 17:14:11 +0000 Subject: [PATCH 2/6] Formatting changes --- admin/views/chemistry_sampleinfo.py | 1 - tests/integration/test_alembic_migrations.py | 32 +++++++------------- 2 files changed, 11 insertions(+), 22 deletions(-) diff --git a/admin/views/chemistry_sampleinfo.py b/admin/views/chemistry_sampleinfo.py index 9aa6654e..c941d086 100644 --- a/admin/views/chemistry_sampleinfo.py +++ b/admin/views/chemistry_sampleinfo.py @@ -28,7 +28,6 @@ - thing_id: Integer FK to Thing.id """ - from admin.views.base import OcotilloModelView diff --git a/tests/integration/test_alembic_migrations.py b/tests/integration/test_alembic_migrations.py index f9f2c200..99e7b89a 100644 --- a/tests/integration/test_alembic_migrations.py +++ b/tests/integration/test_alembic_migrations.py @@ -88,9 +88,9 @@ def test_all_migrations_have_down_revision(self): revisions_without_down.append(rev.revision) base_found = True - assert not revisions_without_down, ( - f"Migrations missing down_revision (besides base): {revisions_without_down}" - ) + assert ( + not revisions_without_down + ), f"Migrations missing down_revision (besides base): {revisions_without_down}" def test_current_revision_matches_head(self): """ @@ -103,9 +103,7 @@ def test_current_revision_matches_head(self): head = script.get_current_head() with engine.connect() as conn: - result = conn.execute( - text("SELECT version_num FROM alembic_version") - ) + result = conn.execute(text("SELECT version_num FROM alembic_version")) current = result.scalar() assert current == head, ( @@ -274,9 +272,7 @@ def test_nma_chemistry_has_thing_fk(self): fks = self._inspector.get_foreign_keys("NMA_Chemistry_SampleInfo") fk_tables = {fk["referred_table"] for fk in fks} - assert "thing" in fk_tables, ( - "NMA_Chemistry_SampleInfo missing FK to thing" - ) + assert "thing" in fk_tables, "NMA_Chemistry_SampleInfo missing FK to thing" # ============================================================================= @@ -304,21 +300,19 @@ def test_location_has_spatial_index(self): # Spatial indexes may be named differently, check for point column # or gist index type has_point_index = "point" in index_columns or any( - "point" in str(idx.get("name", "")).lower() or - "gist" in str(idx.get("name", "")).lower() + "point" in str(idx.get("name", "")).lower() + or "gist" in str(idx.get("name", "")).lower() for idx in indexes ) # Also check via pg_indexes for GIST indexes if not has_point_index: with session_ctx() as session: - result = session.execute( - text(""" + result = session.execute(text(""" SELECT indexname FROM pg_indexes WHERE tablename = 'location' AND indexdef LIKE '%gist%' - """) - ) + """)) gist_indexes = result.fetchall() has_point_index = len(gist_indexes) > 0 @@ -363,9 +357,7 @@ def test_can_downgrade_one_revision(self): # Verify we're at previous revision with engine.connect() as conn: - result = conn.execute( - text("SELECT version_num FROM alembic_version") - ) + result = conn.execute(text("SELECT version_num FROM alembic_version")) current = result.scalar() assert current == previous @@ -374,8 +366,6 @@ def test_can_downgrade_one_revision(self): # Verify we're back at head with engine.connect() as conn: - result = conn.execute( - text("SELECT version_num FROM alembic_version") - ) + result = conn.execute(text("SELECT version_num FROM alembic_version")) current = result.scalar() assert current == head From 7a44936299267b6283683e1e2284abd2bd09f798 Mon Sep 17 00:00:00 2001 From: Kimball Bighorse Date: Wed, 4 Feb 2026 09:17:54 -0800 Subject: [PATCH 3/6] style: fix SQL query string indentation Address Copilot review comment about inconsistent indentation. Co-Authored-By: Claude Opus 4.5 --- tests/integration/test_alembic_migrations.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/integration/test_alembic_migrations.py b/tests/integration/test_alembic_migrations.py index 99e7b89a..92036c77 100644 --- a/tests/integration/test_alembic_migrations.py +++ b/tests/integration/test_alembic_migrations.py @@ -308,11 +308,13 @@ def test_location_has_spatial_index(self): # Also check via pg_indexes for GIST indexes if not has_point_index: with session_ctx() as session: - result = session.execute(text(""" - SELECT indexname FROM pg_indexes - WHERE tablename = 'location' - AND indexdef LIKE '%gist%' - """)) + result = session.execute( + text( + "SELECT indexname FROM pg_indexes " + "WHERE tablename = 'location' " + "AND indexdef LIKE '%gist%'" + ) + ) gist_indexes = result.fetchall() has_point_index = len(gist_indexes) > 0 From bbc80e93390897dc54141c683a5c0c7b18c0c8fb Mon Sep 17 00:00:00 2001 From: Kimball Bighorse Date: Wed, 4 Feb 2026 09:24:24 -0800 Subject: [PATCH 4/6] fix: add missing imports to ChemistrySampleInfoAdmin Add missing `Request` and `HasOne` imports that were causing NameError during test collection. Co-Authored-By: Claude Opus 4.5 --- admin/views/chemistry_sampleinfo.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/admin/views/chemistry_sampleinfo.py b/admin/views/chemistry_sampleinfo.py index c941d086..b588da03 100644 --- a/admin/views/chemistry_sampleinfo.py +++ b/admin/views/chemistry_sampleinfo.py @@ -28,6 +28,9 @@ - thing_id: Integer FK to Thing.id """ +from starlette.requests import Request +from starlette_admin.fields import HasOne + from admin.views.base import OcotilloModelView From 0dfabec41e1220071914b4cad74e2728dcef6144 Mon Sep 17 00:00:00 2001 From: Kimball Bighorse Date: Wed, 4 Feb 2026 09:29:22 -0800 Subject: [PATCH 5/6] test: add admin views import tests Add tests that verify all admin view modules can be imported successfully. This catches missing imports (like Request, HasOne) during CI before they cause runtime errors. Tests include: - Package-level import verification - Individual module import verification - Core view module parametrized tests - Configuration attribute validation Co-Authored-By: Claude Opus 4.5 --- tests/test_admin_views.py | 109 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 tests/test_admin_views.py diff --git a/tests/test_admin_views.py b/tests/test_admin_views.py new file mode 100644 index 00000000..d5e1ce98 --- /dev/null +++ b/tests/test_admin_views.py @@ -0,0 +1,109 @@ +# =============================================================================== +# Copyright 2026 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Tests for admin views module. + +These tests ensure admin views can be imported without errors, +catching missing imports and syntax issues early in CI. +""" + +import importlib +import pkgutil + +import pytest + + +class TestAdminViewsImport: + """Tests that verify all admin views can be imported successfully.""" + + def test_admin_package_imports(self): + """ + Admin package should import without errors. + + This catches missing imports like Request, HasOne, etc. + """ + import admin # noqa: F401 + + def test_admin_views_package_imports(self): + """Admin views subpackage should import without errors.""" + import admin.views # noqa: F401 + + def test_all_view_modules_import(self): + """ + All individual admin view modules should import successfully. + + Iterates through all modules in admin.views and verifies each can be imported. + """ + import admin.views + + failed_imports = [] + + for importer, modname, ispkg in pkgutil.iter_modules(admin.views.__path__): + if modname.startswith("_"): + continue + full_name = f"admin.views.{modname}" + try: + importlib.import_module(full_name) + except Exception as e: + failed_imports.append((full_name, str(e))) + + assert not failed_imports, ( + f"Failed to import admin view modules:\n" + + "\n".join(f" {name}: {err}" for name, err in failed_imports) + ) + + @pytest.mark.parametrize( + "view_module", + [ + "base", + "thing", + "location", + "observation", + "sample", + "contact", + "chemistry_sampleinfo", + "major_chemistry", + "minor_trace_chemistry", + ], + ) + def test_core_view_modules_import(self, view_module: str): + """Core admin view modules should import without errors.""" + importlib.import_module(f"admin.views.{view_module}") + + +class TestAdminViewsConfiguration: + """Tests for admin view configuration validity.""" + + def test_all_exported_views_have_required_attributes(self): + """All exported admin views should have required attributes.""" + import admin.views + + for name in admin.views.__all__: + view_class = getattr(admin.views, name) + + # All views should have a name attribute + assert hasattr( + view_class, "name" + ), f"{view_class.__name__} missing 'name' attribute" + + # All views inheriting from ModelView should have pk_attr + if hasattr(view_class, "model"): + assert hasattr( + view_class, "pk_attr" + ), f"{view_class.__name__} missing 'pk_attr' attribute" + + +# ============= EOF ============================================= From 0606552fb997904f0bf90816f442be8ffd4fc048 Mon Sep 17 00:00:00 2001 From: kbighorse Date: Wed, 4 Feb 2026 17:29:02 +0000 Subject: [PATCH 6/6] Formatting changes --- tests/test_admin_views.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/test_admin_views.py b/tests/test_admin_views.py index d5e1ce98..9696ed1b 100644 --- a/tests/test_admin_views.py +++ b/tests/test_admin_views.py @@ -60,9 +60,10 @@ def test_all_view_modules_import(self): except Exception as e: failed_imports.append((full_name, str(e))) - assert not failed_imports, ( - f"Failed to import admin view modules:\n" - + "\n".join(f" {name}: {err}" for name, err in failed_imports) + assert ( + not failed_imports + ), f"Failed to import admin view modules:\n" + "\n".join( + f" {name}: {err}" for name, err in failed_imports ) @pytest.mark.parametrize(