from __future__ import annotations import unittest from fastapi import HTTPException from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.tabular_sources import ( TabularReadRequest, TabularSnapshotInput, TabularSourceAccessError, TabularSourceValidationError, ) from govoplan_core.db.base import Base from govoplan_connectors.backend.db.models import ConnectorTabularSource from govoplan_connectors.backend.router import api_create_tabular_snapshot from govoplan_connectors.backend.schemas import SnapshotCreateRequest from govoplan_connectors.backend.tabular_sources import ( READ_SCOPE, WRITE_SCOPE, SqlTabularSourceProvider, parse_csv_snapshot, ) def principal( tenant_id: str = "tenant-1", *, scopes: tuple[str, ...] = (READ_SCOPE, WRITE_SCOPE), ) -> ApiPrincipal: return ApiPrincipal( principal=PrincipalRef( account_id="account-1", membership_id="membership-1", tenant_id=tenant_id, scopes=frozenset(scopes), ), account=object(), user=object(), ) class ConnectorsTabularSourceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(self.engine, tables=[ConnectorTabularSource.__table__]) self.Session = sessionmaker(bind=self.engine) self.session = self.Session() self.provider = SqlTabularSourceProvider() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all(self.engine, tables=[ConnectorTabularSource.__table__]) self.engine.dispose() def test_snapshot_round_trip_preserves_schema_fingerprint_and_bounds(self) -> None: created = self.provider.create_snapshot( self.session, principal(), snapshot=TabularSnapshotInput( name="Monthly cases", source_name="monthly_cases_2026_07", rows=( {"case_id": "A-1", "amount": 12, "active": True}, {"case_id": "A-2", "amount": None, "active": False}, ), ), ) self.session.commit() listed = self.provider.list_sources(self.session, principal()) preview = self.provider.read_source( self.session, principal(), request=TabularReadRequest( source_ref=created.ref, limit=1, expected_fingerprint=created.fingerprint, ), ) self.assertEqual((created.ref,), tuple(source.ref for source in listed)) self.assertEqual( ["case_id", "amount", "active"], [column.name for column in created.schema], ) self.assertEqual(2, preview.total_rows) self.assertEqual(1, len(preview.rows)) self.assertTrue(preview.truncated) self.assertEqual(created.fingerprint, preview.source.fingerprint) def test_tenant_and_scope_isolation_are_enforced(self) -> None: created = self.provider.create_snapshot( self.session, principal(), snapshot=TabularSnapshotInput( name="Private", source_name="private_source", rows=({"id": 1},), ), ) self.session.commit() self.assertEqual((), self.provider.list_sources(self.session, principal("tenant-2"))) self.assertIsNone( self.provider.get_source( self.session, principal("tenant-2"), source_ref=created.ref, ) ) with self.assertRaises(TabularSourceAccessError): self.provider.list_sources( self.session, principal(scopes=()), ) def test_duplicate_source_name_and_stale_fingerprint_are_rejected(self) -> None: snapshot = TabularSnapshotInput( name="Cases", source_name="cases", rows=({"id": 1},), ) created = self.provider.create_snapshot(self.session, principal(), snapshot=snapshot) self.session.commit() with self.assertRaises(TabularSourceValidationError): self.provider.create_snapshot(self.session, principal(), snapshot=snapshot) with self.assertRaises(TabularSourceValidationError): self.provider.read_source( self.session, principal(), request=TabularReadRequest( source_ref=created.ref, expected_fingerprint="stale", ), ) def test_csv_parser_infers_scalar_values_and_rejects_duplicate_headers(self) -> None: rows = parse_csv_snapshot( "\ufeffid;amount;active;note\n0012;12.5;true;\n2;7;false;ok\n", delimiter=";", ) self.assertEqual( ( {"id": "0012", "amount": 12.5, "active": True, "note": None}, {"id": 2, "amount": 7, "active": False, "note": "ok"}, ), rows, ) with self.assertRaises(TabularSourceValidationError): parse_csv_snapshot("id,id\n1,2\n", delimiter=",") with self.assertRaises(TabularSourceValidationError): parse_csv_snapshot("id,name\n1,Ada,extra\n", delimiter=",") def test_malformed_csv_api_request_is_reported_as_validation_error(self) -> None: payload = SnapshotCreateRequest( name="Malformed", source_name="malformed", format="csv", csv_text="id,name\n1,Ada,extra\n", ) with self.assertRaises(HTTPException) as raised: api_create_tabular_snapshot( payload, session=self.session, principal=principal(), ) self.assertEqual(422, raised.exception.status_code) if __name__ == "__main__": unittest.main()