diff --git a/admin/views/chemistry_sampleinfo.py b/admin/views/chemistry_sampleinfo.py index b28bd112..b588da03 100644 --- a/admin/views/chemistry_sampleinfo.py +++ b/admin/views/chemistry_sampleinfo.py @@ -29,7 +29,7 @@ """ from starlette.requests import Request -from starlette_admin import HasOne +from starlette_admin.fields import HasOne from admin.views.base import OcotilloModelView diff --git a/tests/integration/test_alembic_migrations.py b/tests/integration/test_alembic_migrations.py new file mode 100644 index 00000000..92036c77 --- /dev/null +++ b/tests/integration/test_alembic_migrations.py @@ -0,0 +1,373 @@ +# =============================================================================== +# 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 diff --git a/tests/test_admin_views.py b/tests/test_admin_views.py new file mode 100644 index 00000000..9696ed1b --- /dev/null +++ b/tests/test_admin_views.py @@ -0,0 +1,110 @@ +# =============================================================================== +# 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 =============================================