platform-core/tests/unit/services/test_udbNodes.py
ValueOn AG ebc4b2a080
Some checks failed
Deploy Plattform-Core (Int) / test (push) Failing after 12s
Deploy Plattform-Core (Int) / deploy (push) Has been skipped
cp adapted to 2026 poweron 2
2026-06-09 09:58:05 +02:00

451 lines
21 KiB
Python

# Copyright (c) 2026 PowerOn AG
# All rights reserved.
"""Unit tests for the polymorphic UDB node hierarchy (udbNodes.py).
Each concrete node class is exercised for:
- `supportsFlag` returns the right set per kind
- `canEdit` enforces DS-owner vs FDS-feature-admin
- `getEffectiveFlag` resolves walk + aggregate correctly
- `setFlag` writes the right record and (where applicable) cascades
- `toDict` produces the expected wire shape
"""
from __future__ import annotations
import unittest
from unittest.mock import MagicMock, patch
from modules.serviceCenter.services.serviceKnowledge.udbNodes import (
UdbNode,
SyntheticContainerNode,
MandateGroupNode,
ConnectionNode,
ServiceNode,
FolderNode,
FileNode,
FdsWorkspaceNode,
FdsTableNode,
FdsFieldNode,
isFeatureAdmin,
)
class _FakeUser:
def __init__(self, userId: str = "user-1"):
self.id = userId
class _FakeContext:
def __init__(self, userId: str = "user-1"):
self.user = _FakeUser(userId)
self.mandateId = "m1"
class TestSupportsFlag(unittest.TestCase):
def test_synthetic_container_supports_nothing(self):
n = SyntheticContainerNode("k", "label", icon="x")
self.assertFalse(n.supportsFlag("neutralize"))
self.assertFalse(n.supportsFlag("scope"))
self.assertFalse(n.supportsFlag("ragIndexEnabled"))
def test_connection_supports_neutralize_and_rag(self):
n = ConnectionNode("c1", "msft", label="m", parentKey="personalRoot", rec=None)
self.assertTrue(n.supportsFlag("neutralize"))
self.assertFalse(n.supportsFlag("scope"))
self.assertTrue(n.supportsFlag("ragIndexEnabled"))
def test_fds_table_supports_neutralize_and_rag_but_not_scope(self):
n = FdsTableNode(
featureInstanceId="fi1", featureCode="trustee", tableName="Pos",
objectKey="data.feature.trustee.Pos", label="Positions",
parentKey="feat|m1|trustee|fi1", rec=None, hasFields=False,
)
self.assertTrue(n.supportsFlag("neutralize"))
self.assertTrue(n.supportsFlag("ragIndexEnabled"))
self.assertFalse(n.supportsFlag("scope"))
def test_fds_field_supports_only_neutralize(self):
n = FdsFieldNode(
featureInstanceId="fi1", tableName="Pos", fieldName="amount",
parentKey="fdstbl|fi1|Pos", tableRec=None, featureCode="trustee",
)
self.assertTrue(n.supportsFlag("neutralize"))
self.assertFalse(n.supportsFlag("scope"))
self.assertFalse(n.supportsFlag("ragIndexEnabled"))
class TestCanEditDataSourceOwner(unittest.TestCase):
def test_owner_can_edit(self):
rec = {"id": "ds1", "userId": "user-1"}
node = ConnectionNode("c1", "msft", "m", "personalRoot", rec=rec)
self.assertTrue(node.canEdit(_FakeContext("user-1"), MagicMock()))
def test_non_owner_cannot_edit(self):
rec = {"id": "ds1", "userId": "user-other"}
node = ConnectionNode("c1", "msft", "m", "personalRoot", rec=rec)
self.assertFalse(node.canEdit(_FakeContext("user-1"), MagicMock()))
def test_virtual_node_own_connection_can_edit(self):
rootIf = MagicMock()
rootIf.db.getRecord.return_value = {"id": "c1", "userId": "user-1"}
node = ConnectionNode("c1", "msft", "m", "personalRoot", rec=None)
self.assertTrue(node.canEdit(_FakeContext("user-1"), rootIf))
def test_virtual_node_other_connection_cannot_edit(self):
rootIf = MagicMock()
rootIf.db.getRecord.return_value = {"id": "c1", "userId": "user-other"}
node = ConnectionNode("c1", "msft", "m", "personalRoot", rec=None)
self.assertFalse(node.canEdit(_FakeContext("user-1"), rootIf))
def test_virtual_node_missing_connection_cannot_edit(self):
rootIf = MagicMock()
rootIf.db.getRecord.return_value = None
node = ConnectionNode("c1", "msft", "m", "personalRoot", rec=None)
self.assertFalse(node.canEdit(_FakeContext("user-1"), rootIf))
class TestCanEditFdsFeatureAdmin(unittest.TestCase):
def _buildRootIfWithAdminRole(self, hasAdmin: bool):
rootIf = MagicMock()
access = MagicMock(id="acc1", enabled=True)
rootIf.getFeatureAccess.return_value = access
rootIf.getRoleIdsForFeatureAccess.return_value = ["role-1"]
rootIf.db.getRecord.return_value = {
"id": "role-1",
"roleLabel": "trustee-admin" if hasAdmin else "trustee-user",
}
return rootIf
def test_admin_can_edit_fds_table(self):
rootIf = self._buildRootIfWithAdminRole(hasAdmin=True)
node = FdsTableNode("fi1", "trustee", "Pos", "key", "Positions",
"feat|m1|trustee|fi1", rec={"id": "fds1"}, hasFields=False)
self.assertTrue(node.canEdit(_FakeContext(), rootIf))
def test_non_admin_cannot_edit_fds_table(self):
rootIf = self._buildRootIfWithAdminRole(hasAdmin=False)
node = FdsTableNode("fi1", "trustee", "Pos", "key", "Positions",
"feat|m1|trustee|fi1", rec={"id": "fds1"}, hasFields=False)
self.assertFalse(node.canEdit(_FakeContext(), rootIf))
def test_fds_field_uses_feature_admin_check(self):
rootIf = self._buildRootIfWithAdminRole(hasAdmin=True)
field = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec={"id": "fds1"}, featureCode="trustee")
self.assertTrue(field.canEdit(_FakeContext(), rootIf))
class TestGetEffectiveFlag(unittest.TestCase):
def test_ds_walk_inherits_from_authority_root(self):
root = {
"id": "r", "connectionId": "c", "sourceType": "msft", "path": "/",
"userId": "user-1", "neutralize": True, "ragIndexEnabled": None,
}
node = FolderNode(
connectionId="c", service="sharepoint", sourceType="sharepointFolder",
path="/sites/x", label="x", parentKey="svc|c|sharepoint",
rec=None, hasChildren=True,
)
self.assertTrue(node.getEffectiveFlag("neutralize", [root], [], "walk"))
def test_fds_field_neutralize_from_neutralize_fields(self):
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralizeFields": ["amount"],
}
node = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
self.assertTrue(node.getEffectiveFlag("neutralize", [], [rec], "aggregate"))
other = FdsFieldNode("fi1", "Pos", "currency", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
# currency is not in the override list and the table has no
# explicit neutralize -> inherits the default (False).
self.assertFalse(other.getEffectiveFlag("neutralize", [], [rec], "aggregate"))
def test_fds_field_inherits_true_from_table(self):
"""Field without explicit override inherits the table's explicit
neutralize. Regression: previously fell through to False, so
toggling the table left the field icon unchanged."""
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": True, "neutralizeFields": None,
}
node = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
self.assertTrue(node.getEffectiveFlag("neutralize", [], [rec], "aggregate"))
def test_fds_field_inherits_from_workspace_via_table(self):
"""Field walks the whole FDS ancestor chain: table -> workspace."""
ws = {
"id": "fds-ws", "featureInstanceId": "fi1", "tableName": "*",
"recordFilter": None, "neutralize": True,
}
tbl = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": None, "neutralizeFields": None,
}
node = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec=tbl, featureCode="trustee")
self.assertTrue(node.getEffectiveFlag("neutralize", [], [ws, tbl], "aggregate"))
def test_fds_field_explicit_override_beats_table_false(self):
"""Per-column override (True via list entry) beats an explicit
table False -- this is the one case where the two-source model
diverges intentionally and produces the 'mixed' aggregate."""
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": False, "neutralizeFields": ["amount"],
}
amount = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
other = FdsFieldNode("fi1", "Pos", "currency", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
self.assertTrue(amount.getEffectiveFlag("neutralize", [], [rec], "aggregate"))
# currency inherits from table -> False
self.assertFalse(other.getEffectiveFlag("neutralize", [], [rec], "aggregate"))
def test_fds_table_mixed_when_field_and_table_disagree(self):
# table.neutralize=False, field "amount" is in neutralizeFields => mixed
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": False, "ragIndexEnabled": None,
"neutralizeFields": ["amount"],
}
table = FdsTableNode("fi1", "trustee", "Pos", "key", "Positions",
"feat|m1|trustee|fi1", rec=rec, hasFields=True)
# FdsTableNode.getEffectiveFlag itself only consults FDS records,
# not field nodes. The aggregation across field nodes is wired
# in _buildTree via `_wireTableFieldsAsLogicalChildren`. So we
# exercise the explicit FDS walk here:
val = table.getEffectiveFlag("neutralize", [], [rec], "walk")
self.assertFalse(val)
class TestSetFlag(unittest.TestCase):
def test_setflag_writes_value_on_ds(self):
rec = {"id": "ds1", "connectionId": "c", "sourceType": "msft", "path": "/",
"userId": "user-1"}
node = ConnectionNode("c", "msft", "m", "personalRoot", rec=rec)
rootIf = MagicMock()
rootIf.db.getRecordset.return_value = [] # no siblings -> no cascade
node.setFlag("neutralize", True, rootIf)
rootIf.db.recordModify.assert_called()
args = rootIf.db.recordModify.call_args[0]
self.assertEqual(args[1], "ds1")
self.assertEqual(args[2], {"neutralize": True})
def test_setflag_virtual_ds_auto_creates_record(self):
"""Toggling a flag on a virtual DS node must auto-create the
DataSource record so the flag can be persisted."""
node = FolderNode(
connectionId="c1", service="sharepoint",
sourceType="sharepointFolder", path="/sites/x/docs",
label="docs", parentKey="svc|c1|sharepoint",
rec=None, hasChildren=True,
)
rootIf = MagicMock()
rootIf.db.getRecordset.return_value = []
rootIf.db.getRecord.return_value = {"id": "c1", "userId": "user-1"}
createdRec = {"id": "ds-new", "connectionId": "c1",
"sourceType": "sharepointFolder", "path": "/sites/x/docs",
"userId": "user-1"}
rootIf.db.recordCreate.return_value = createdRec
node.setFlag("neutralize", True, rootIf)
rootIf.db.recordCreate.assert_called_once()
rootIf.db.recordModify.assert_called()
args = rootIf.db.recordModify.call_args[0]
self.assertEqual(args[1], "ds-new")
self.assertEqual(args[2], {"neutralize": True})
def test_fds_field_setflag_mutates_neutralize_fields(self):
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": False, "neutralizeFields": None,
}
node = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
rootIf = MagicMock()
node.setFlag("neutralize", True, rootIf)
rootIf.db.recordModify.assert_called()
# last call: set neutralizeFields to ["amount"]
args = rootIf.db.recordModify.call_args[0]
self.assertEqual(args[1], "fds-tbl")
self.assertEqual(args[2], {"neutralizeFields": ["amount"]})
def test_fds_field_setflag_removes_field_when_toggled_off(self):
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralizeFields": ["amount", "currency"],
}
node = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
rootIf = MagicMock()
node.setFlag("neutralize", False, rootIf)
args = rootIf.db.recordModify.call_args[0]
self.assertEqual(args[2], {"neutralizeFields": ["currency"]})
def test_fds_field_setflag_roundtrip(self):
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralizeFields": None,
}
node = FdsFieldNode("fi1", "Pos", "amount", "fdstbl|fi1|Pos",
tableRec=rec, featureCode="trustee")
rootIf = MagicMock()
node.setFlag("neutralize", True, rootIf)
self.assertTrue(node.getEffectiveFlag("neutralize", [], [rec], "aggregate"))
node.setFlag("neutralize", False, rootIf)
self.assertFalse(node.getEffectiveFlag("neutralize", [], [rec], "aggregate"))
def test_fds_table_explicit_neutralize_wipes_own_neutralize_fields(self):
"""Setting an explicit neutralize on a table must clear its own
`neutralizeFields` list. Otherwise the table's aggregate stays
'mixed' because field children walk to True via that list and the
UI shows no change after the toggle."""
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": None,
"neutralizeFields": ["amount", "currency"],
}
node = FdsTableNode("fi1", "trustee", "Pos", "key", "Positions",
"feat|m1|trustee|fi1", rec=rec, hasFields=True)
rootIf = MagicMock()
rootIf.db.getRecordset.return_value = [rec] # no descendants
node.setFlag("neutralize", False, rootIf)
rootIf.db.recordModify.assert_called()
args = rootIf.db.recordModify.call_args[0]
self.assertEqual(args[1], "fds-tbl")
self.assertEqual(
args[2], {"neutralize": False, "neutralizeFields": None},
)
def test_fds_table_setflag_inherit_keeps_neutralize_fields(self):
"""`value=None` (reset to inherit) must NOT cascade and must NOT
wipe `neutralizeFields`; that matches the cascade-reset spec
(only explicit toggles clear descendants)."""
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": True,
"neutralizeFields": ["amount"],
}
node = FdsTableNode("fi1", "trustee", "Pos", "key", "Positions",
"feat|m1|trustee|fi1", rec=rec, hasFields=True)
rootIf = MagicMock()
rootIf.db.getRecordset.return_value = [rec]
node.setFlag("neutralize", None, rootIf)
args = rootIf.db.recordModify.call_args[0]
self.assertEqual(args[2], {"neutralize": None})
def test_fds_table_setflag_rag_does_not_touch_neutralize_fields(self):
"""A RAG toggle on the table must leave `neutralizeFields` alone
(it is neutralize-only field state)."""
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "ragIndexEnabled": None,
"neutralizeFields": ["amount"],
}
node = FdsTableNode("fi1", "trustee", "Pos", "key", "Positions",
"feat|m1|trustee|fi1", rec=rec, hasFields=True)
rootIf = MagicMock()
rootIf.db.getRecordset.return_value = [rec]
node.setFlag("ragIndexEnabled", True, rootIf)
args = rootIf.db.recordModify.call_args[0]
self.assertEqual(args[2], {"ragIndexEnabled": True})
def test_fds_workspace_neutralize_clears_descendant_neutralize_fields(self):
"""Workspace toggle must clear per-column overrides on descendant
tables; otherwise the table aggregate stays 'mixed' because some
field children still read True from the list."""
wsRec = {
"id": "fds-ws", "featureInstanceId": "fi1", "tableName": "*",
"recordFilter": None, "neutralize": None,
}
tblRec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "neutralize": None,
"neutralizeFields": ["amount", "currency"],
}
node = FdsWorkspaceNode("m1", "trustee", "fi1", label="Trustee",
icon="trustee", parentKey="mgrp|m1", rec=wsRec)
rootIf = MagicMock()
rootIf.db.getRecordset.return_value = [wsRec, tblRec]
node.setFlag("neutralize", True, rootIf)
calls = rootIf.db.recordModify.call_args_list
modifyMap = {c[0][1]: c[0][2] for c in calls}
self.assertEqual(modifyMap["fds-tbl"], {"neutralizeFields": None})
self.assertEqual(modifyMap["fds-ws"], {"neutralize": True})
def test_fds_workspace_rag_does_not_clear_descendant_neutralize_fields(self):
"""A RAG toggle on the workspace must not touch descendant
`neutralizeFields`."""
wsRec = {
"id": "fds-ws", "featureInstanceId": "fi1", "tableName": "*",
"recordFilter": None, "ragIndexEnabled": None,
}
tblRec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"recordFilter": None, "ragIndexEnabled": None,
"neutralizeFields": ["amount"],
}
node = FdsWorkspaceNode("m1", "trustee", "fi1", label="Trustee",
icon="trustee", parentKey="mgrp|m1", rec=wsRec)
rootIf = MagicMock()
rootIf.db.getRecordset.return_value = [wsRec, tblRec]
node.setFlag("ragIndexEnabled", True, rootIf)
calls = rootIf.db.recordModify.call_args_list
modifyIds = [c[0][1] for c in calls]
self.assertNotIn("fds-tbl", modifyIds)
class TestToDict(unittest.TestCase):
def test_fds_table_dict_has_neutralize_fields(self):
rec = {
"id": "fds-tbl", "featureInstanceId": "fi1", "tableName": "Pos",
"neutralizeFields": ["amount"],
}
node = FdsTableNode("fi1", "trustee", "Pos", "key", "Positions",
"feat|m1|trustee|fi1", rec=rec, hasFields=True)
out = node.toDict([], [rec])
self.assertEqual(out["neutralizeFields"], ["amount"])
self.assertEqual(out["kind"], "fdsTable")
self.assertEqual(out["modelType"], "FeatureDataSource")
self.assertEqual(out["effectiveScope"], "personal") # FDS has no scope
def test_synthetic_container_has_no_dataSourceId(self):
n = SyntheticContainerNode("personalRoot", "Personal", icon="person",
defaultExpanded=True)
d = n.toDict([], [])
self.assertIsNone(d["dataSourceId"])
self.assertEqual(d["effectiveNeutralize"], False)
class TestIsFeatureAdmin(unittest.TestCase):
def test_no_access_returns_false(self):
rootIf = MagicMock()
rootIf.getFeatureAccess.return_value = None
self.assertFalse(isFeatureAdmin(rootIf, "user-1", "fi1"))
def test_no_roles_returns_false(self):
rootIf = MagicMock()
rootIf.getFeatureAccess.return_value = MagicMock(id="acc1", enabled=True)
rootIf.getRoleIdsForFeatureAccess.return_value = []
self.assertFalse(isFeatureAdmin(rootIf, "user-1", "fi1"))
def test_non_admin_role_returns_false(self):
rootIf = MagicMock()
rootIf.getFeatureAccess.return_value = MagicMock(id="acc1", enabled=True)
rootIf.getRoleIdsForFeatureAccess.return_value = ["r1"]
rootIf.db.getRecord.return_value = {"id": "r1", "roleLabel": "trustee-user"}
self.assertFalse(isFeatureAdmin(rootIf, "user-1", "fi1"))
def test_admin_role_returns_true(self):
rootIf = MagicMock()
rootIf.getFeatureAccess.return_value = MagicMock(id="acc1", enabled=True)
rootIf.getRoleIdsForFeatureAccess.return_value = ["r1"]
rootIf.db.getRecord.return_value = {"id": "r1", "roleLabel": "workspace-admin"}
self.assertTrue(isFeatureAdmin(rootIf, "user-1", "fi1"))
if __name__ == "__main__":
unittest.main()