# Copyright (c) 2025 Patrick Motsch # All rights reserved. """Tests for ``scripts/script_migrate_feature_instance_refs.py``. The script touches the live ``poweron_graphicaleditor`` DB. Tests run against an in-memory fake psycopg2 connection so we exercise the full code path (SELECT -> migrate -> UPDATE) without requiring a real Postgres server. """ from __future__ import annotations import importlib import json import sys from pathlib import Path from typing import Any, Dict, List, Tuple import pytest _gatewayPath = Path(__file__).resolve().parents[3] _scriptsPath = _gatewayPath / "scripts" if str(_scriptsPath) not in sys.path: sys.path.insert(0, str(_scriptsPath)) migrationModule = importlib.import_module("script_migrate_feature_instance_refs") # --------------------------------------------------------------------------- # Fake psycopg2 connection / cursor # --------------------------------------------------------------------------- class _FakeCursor: """Mimics enough of psycopg2's RealDictCursor + plain cursor for the script.""" def __init__(self, rowsByTable: Dict[str, List[Dict[str, Any]]], updates: List[Tuple[str, str, Any]]): self._rowsByTable = rowsByTable self._updates = updates self._lastFetch: List[Dict[str, Any]] = [] def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def execute(self, query: str, params: Any = None): if query.strip().upper().startswith("SELECT"): for table, rows in self._rowsByTable.items(): if table in query: self._lastFetch = list(rows) return self._lastFetch = [] return if query.strip().upper().startswith("UPDATE"): for table in self._rowsByTable: if table in query: graphValue, pk = params if hasattr(graphValue, "adapted"): graphValue = graphValue.adapted self._updates.append((table, pk, graphValue)) return return def fetchall(self): return self._lastFetch class _FakeConn: def __init__(self, rowsByTable: Dict[str, List[Dict[str, Any]]]): self._rowsByTable = rowsByTable self.updates: List[Tuple[str, str, Any]] = [] self.committed = False self.closed = False def cursor(self, cursor_factory: Any = None): return _FakeCursor(self._rowsByTable, self.updates) def commit(self): self.committed = True def close(self): self.closed = True @pytest.fixture def graphsByTable() -> Dict[str, List[Dict[str, Any]]]: return { '"Automation2Workflow"': [ { "pk": "wf-legacy", "graph": { "nodes": [ { "id": "n1", "type": "trustee.processDocuments", "parameters": {"featureInstanceId": "11111111-1111-1111-1111-111111111111"}, }, { "id": "n2", "type": "redmine.createIssue", "parameters": {"featureInstanceId": "22222222-2222-2222-2222-222222222222"}, }, ], "connections": [], }, }, { "pk": "wf-already-typed", "graph": { "nodes": [ { "id": "n1", "type": "trustee.processDocuments", "parameters": { "featureInstanceId": { "$type": "FeatureInstanceRef", "id": "33333333-3333-3333-3333-333333333333", "featureCode": "trustee", } }, } ], "connections": [], }, }, { "pk": "wf-empty-graph", "graph": {}, }, { "pk": "wf-graph-as-string", "graph": json.dumps({ "nodes": [ { "id": "n1", "type": "outlook.sendMail", "parameters": {"featureInstanceId": "44444444-4444-4444-4444-444444444444"}, } ], "connections": [], }), }, ], '"AutoVersion"': [ { "pk": "ver-legacy", "graph": { "nodes": [ { "id": "n1", "type": "ai.runPrompt", "parameters": {"featureInstanceId": "55555555-5555-5555-5555-555555555555"}, } ], "connections": [], }, } ], } # --------------------------------------------------------------------------- # Helper-level tests # --------------------------------------------------------------------------- class TestLoadGraph: def test_dictPassesThrough(self): assert migrationModule._loadGraph({"a": 1}) == {"a": 1} def test_jsonStringIsParsed(self): assert migrationModule._loadGraph('{"a": 2}') == {"a": 2} def test_emptyOrInvalidYieldsEmptyDict(self): assert migrationModule._loadGraph(None) == {} assert migrationModule._loadGraph("") == {} assert migrationModule._loadGraph("not json") == {} def test_bytesStringIsParsed(self): assert migrationModule._loadGraph(b'{"a": 3}') == {"a": 3} class TestCountMigrations: def test_zeroWhenIdentical(self): g = {"nodes": [{"id": "n", "parameters": {"featureInstanceId": "uuid"}}]} assert migrationModule._countMigrations(g, g) == 0 def test_countsMigratedFields(self): before = { "nodes": [ {"id": "n1", "parameters": {"featureInstanceId": "u1"}}, {"id": "n2", "parameters": {"featureInstanceId": "u2"}}, {"id": "n3", "parameters": {"featureInstanceId": "u3"}}, ] } after = { "nodes": [ { "id": "n1", "parameters": { "featureInstanceId": {"$type": "FeatureInstanceRef", "id": "u1"} }, }, {"id": "n2", "parameters": {"featureInstanceId": "u2"}}, { "id": "n3", "parameters": { "featureInstanceId": {"$type": "FeatureInstanceRef", "id": "u3"} }, }, ] } assert migrationModule._countMigrations(before, after) == 2 # --------------------------------------------------------------------------- # End-to-end migrate() tests # --------------------------------------------------------------------------- class TestMigrate: def test_dryRunDoesNotWriteOrCommit(self, monkeypatch, graphsByTable): conn = _FakeConn(graphsByTable) monkeypatch.setattr(migrationModule, "_connect", lambda: conn) summary = migrationModule.migrate(dryRun=True) assert conn.updates == [] assert conn.committed is False assert conn.closed is True assert summary['"Automation2Workflow"']["scanned"] == 4 assert summary['"Automation2Workflow"']["rowsChanged"] == 2 assert summary['"Automation2Workflow"']["fieldsRewritten"] == 3 assert summary['"AutoVersion"']["rowsChanged"] == 1 assert summary['"AutoVersion"']["fieldsRewritten"] == 1 def test_liveRunWritesAndCommits(self, monkeypatch, graphsByTable): conn = _FakeConn(graphsByTable) monkeypatch.setattr(migrationModule, "_connect", lambda: conn) summary = migrationModule.migrate(dryRun=False) assert conn.committed is True assert conn.closed is True updatesByPk = {pk: graph for (_table, pk, graph) in conn.updates} assert set(updatesByPk.keys()) == {"wf-legacy", "wf-graph-as-string", "ver-legacy"} legacyGraph = updatesByPk["wf-legacy"] n1Param = legacyGraph["nodes"][0]["parameters"]["featureInstanceId"] n2Param = legacyGraph["nodes"][1]["parameters"]["featureInstanceId"] assert n1Param["$type"] == "FeatureInstanceRef" assert n1Param["featureCode"] == "trustee" assert n1Param["id"] == "11111111-1111-1111-1111-111111111111" assert n2Param["featureCode"] == "redmine" verParam = updatesByPk["ver-legacy"]["nodes"][0]["parameters"]["featureInstanceId"] assert verParam["featureCode"] == "ai" stringSourcedGraph = updatesByPk["wf-graph-as-string"] outlookParam = stringSourcedGraph["nodes"][0]["parameters"]["featureInstanceId"] assert outlookParam["featureCode"] == "outlook" assert summary['"Automation2Workflow"']["fieldsRewritten"] == 3 assert summary['"AutoVersion"']["fieldsRewritten"] == 1 def test_idempotency(self, monkeypatch, graphsByTable): conn1 = _FakeConn(graphsByTable) monkeypatch.setattr(migrationModule, "_connect", lambda: conn1) migrationModule.migrate(dryRun=False) firstUpdates = {pk: graph for (_t, pk, graph) in conn1.updates} nextRows = { '"Automation2Workflow"': [ {"pk": pk, "graph": graph} for pk, graph in firstUpdates.items() if pk.startswith("wf") ], '"AutoVersion"': [ {"pk": pk, "graph": graph} for pk, graph in firstUpdates.items() if pk.startswith("ver") ], } conn2 = _FakeConn(nextRows) monkeypatch.setattr(migrationModule, "_connect", lambda: conn2) summary2 = migrationModule.migrate(dryRun=False) assert conn2.updates == [] for table, counts in summary2.items(): assert counts["rowsChanged"] == 0, f"{table} not idempotent" assert counts["fieldsRewritten"] == 0, f"{table} not idempotent"