zemion
889cafaf20
Module Package Release / publish-packages (push) Successful in 17s
Release v0.1.27. Coordinated integrity review: GovOPlaN/govoplan-core#298.
82 lines
3.9 KiB
Python
82 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from alembic import command
|
|
from alembic.runtime.migration import MigrationContext
|
|
from sqlalchemy import MetaData, Table, create_engine, inspect, select
|
|
|
|
from govoplan_connectors.backend.manifest import get_manifest
|
|
from govoplan_core.db.migrations import alembic_config, migrate_database
|
|
|
|
|
|
class ConnectorsMigrationTests(unittest.TestCase):
|
|
def test_csv_evidence_upgrade_preserves_legacy_snapshot_without_inventing_source(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="govoplan-connectors-csv-migration-") as directory:
|
|
url = f"sqlite:///{Path(directory) / 'connectors.db'}"
|
|
config = alembic_config(database_url=url, enabled_modules=("connectors",), manifest_factories=(get_manifest,))
|
|
command.upgrade(config, "c0f1a2b3c4d5")
|
|
engine = create_engine(url)
|
|
try:
|
|
table = Table("connector_tabular_sources", MetaData(), autoload_with=engine)
|
|
now = datetime.now(UTC)
|
|
with engine.begin() as connection:
|
|
connection.execute(table.insert().values(
|
|
id="legacy-csv", tenant_id="tenant-1", provider="snapshot",
|
|
source_name="legacy", name="Legacy", status="active", schema_version=1,
|
|
schema=[{"name": "id", "data_type": "integer", "nullable": False}],
|
|
rows=[{"id": 1}], fingerprint="a" * 64, row_count=1, byte_count=10,
|
|
metadata={"original_label": "CSV"}, created_at=now, updated_at=now,
|
|
))
|
|
before = dict(connection.execute(select(table)).mappings().one())
|
|
command.upgrade(config, "d2a4c6e8f0b1")
|
|
upgraded = Table("connector_tabular_sources", MetaData(), autoload_with=engine)
|
|
with engine.connect() as connection:
|
|
after = dict(connection.execute(select(upgraded)).mappings().one())
|
|
self.assertIsNone(after.pop("csv_source"))
|
|
self.assertEqual(before, after)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
def test_baseline_creates_connector_tables_and_head(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="govoplan-connectors-migration-") as directory:
|
|
url = f"sqlite:///{Path(directory) / 'connectors.db'}"
|
|
migrate_database(
|
|
database_url=url,
|
|
enabled_modules=("connectors",),
|
|
manifest_factories=(get_manifest,),
|
|
)
|
|
engine = create_engine(url)
|
|
try:
|
|
with engine.connect() as connection:
|
|
self.assertIn(
|
|
"d2a4c6e8f0b1",
|
|
set(MigrationContext.configure(connection).get_current_heads()),
|
|
)
|
|
self.assertTrue(
|
|
{
|
|
"connector_tabular_sources",
|
|
"connector_sanctions_snapshots",
|
|
"connector_sanctions_acquisition_runs",
|
|
"connector_definitions",
|
|
"connector_definition_revisions",
|
|
"connector_configurations",
|
|
"connector_simulation_runs",
|
|
"connector_knowledge_profiles",
|
|
"connector_knowledge_objects",
|
|
"connector_knowledge_sync_runs",
|
|
"connector_service_desk_profiles",
|
|
"connector_service_desk_objects",
|
|
"connector_service_desk_sync_runs",
|
|
}.issubset(inspect(connection).get_table_names())
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|