289 lines
10 KiB
Python
289 lines
10 KiB
Python
# 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"
|