76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
"""Focused source-level contracts for legacy Agent migrations."""
|
|
|
|
import importlib.util
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
VERSIONS = Path(__file__).parents[1] / "alembic" / "versions"
|
|
|
|
|
|
def load_migration(filename):
|
|
path = VERSIONS / filename
|
|
spec = importlib.util.spec_from_file_location(filename.removesuffix(".py"), path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class EmptyResult:
|
|
def all(self):
|
|
return []
|
|
|
|
|
|
class RecordingConnection:
|
|
def __init__(self):
|
|
self.statements = []
|
|
|
|
def execute(self, statement, parameters=None):
|
|
del parameters
|
|
self.statements.append(str(statement))
|
|
return EmptyResult()
|
|
|
|
|
|
class LegacyRoutingMigrationTests(unittest.TestCase):
|
|
def test_0018_locks_every_preflight_and_update_table_before_reads(self):
|
|
migration = load_migration("0018_legacy_agent_routing.py")
|
|
connection = RecordingConnection()
|
|
with patch.object(migration.op, "get_bind", return_value=connection):
|
|
migration.upgrade()
|
|
|
|
first = " ".join(connection.statements[0].split()).lower()
|
|
self.assertTrue(first.startswith("lock table"))
|
|
self.assertIn("in share row exclusive mode", first)
|
|
for table in (
|
|
"agent_zones", "minerops_agent_discoveries",
|
|
"minerops_agent_snapshots", "minerops_site_agents", "sites",
|
|
):
|
|
with self.subTest(table=table):
|
|
self.assertIn(table, first)
|
|
|
|
first_update = next(
|
|
index for index, statement in enumerate(connection.statements)
|
|
if "UPDATE " in statement
|
|
)
|
|
self.assertEqual(first_update, 4)
|
|
|
|
def test_0019_is_the_direct_forward_only_successor(self):
|
|
migration = load_migration("0019_agent_legacy_auth_retire.py")
|
|
self.assertEqual(migration.revision, "0019_agent_legacy_auth_retire")
|
|
self.assertEqual(migration.down_revision, "0018_legacy_agent_routing")
|
|
|
|
def test_0020_adds_only_existing_identity_transition_binding(self):
|
|
migration = load_migration(
|
|
"0020_existing_agent_credential_transition.py"
|
|
)
|
|
self.assertEqual(migration.revision, "0020_existing_agent_transition")
|
|
self.assertEqual(
|
|
migration.down_revision,
|
|
"0019_agent_legacy_auth_retire",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|