platform-core/tests/unit/services/test_inheritFlags.py
2026-05-18 07:56:53 +02:00

330 lines
15 KiB
Python

"""Unit tests for `_inheritFlags` cascade-inherit helpers.
Verifies:
- getEffectiveFlag walks ancestors via path-prefix matching
- root default is False (or 'personal' for scope) when nothing explicit in chain
- only same-connectionId AND same-sourceType ancestors are considered
- cascadeResetDescendants only touches descendants with explicit values for THAT flag
- '/' is treated as ancestor of every non-root path
- '/foo' is NOT ancestor of '/foobar' (must require '/' separator)
"""
from __future__ import annotations
import unittest
from typing import List
from unittest.mock import MagicMock
from modules.serviceCenter.services.serviceKnowledge import _inheritFlags
def _ds(idVal: str, path: str, **flags) -> dict:
"""Build a DataSource dict with sensible defaults for a fixture."""
base = {
"id": idVal,
"connectionId": "conn-1",
"sourceType": "sharepointFolder",
"path": path,
"neutralize": None,
"ragIndexEnabled": None,
"scope": None,
}
base.update(flags)
return base
class TestEffectiveFlag(unittest.TestCase):
def test_explicit_own_value_wins(self):
root = _ds("r", "/", neutralize=False)
leaf = _ds("l", "/folder/sub", neutralize=True)
self.assertTrue(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [root, leaf]))
def test_inherits_from_root_when_own_is_none(self):
root = _ds("r", "/", neutralize=True)
leaf = _ds("l", "/folder/sub")
self.assertTrue(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [root, leaf]))
def test_default_false_when_chain_empty(self):
leaf = _ds("l", "/folder/sub")
self.assertFalse(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [leaf]))
def test_nearest_ancestor_wins_over_distant(self):
root = _ds("r", "/", neutralize=False)
mid = _ds("m", "/folder", neutralize=True)
leaf = _ds("l", "/folder/sub")
self.assertTrue(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [root, mid, leaf]))
def test_different_connection_ignored(self):
otherConn = _ds("o", "/", connectionId="conn-2", neutralize=True)
leaf = _ds("l", "/folder")
self.assertFalse(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [otherConn, leaf]))
def test_different_sourcetype_ignored(self):
otherType = _ds("o", "/", sourceType="outlookFolder", neutralize=True)
leaf = _ds("l", "/folder")
self.assertFalse(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [otherType, leaf]))
def test_path_separator_required(self):
"""`/foo` must NOT be ancestor of `/foobar` (no shared `/` boundary)."""
notAncestor = _ds("a", "/foo", neutralize=True)
leaf = _ds("l", "/foobar")
self.assertFalse(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [notAncestor, leaf]))
def test_root_is_ancestor_of_everything(self):
root = _ds("r", "/", neutralize=True)
leaf = _ds("l", "/anything/anywhere")
self.assertTrue(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [root, leaf]))
def test_scope_inheritance_with_string_default(self):
root = _ds("r", "/", scope="mandate")
leaf = _ds("l", "/folder")
self.assertEqual(_inheritFlags.getEffectiveFlag(leaf, "scope", [root, leaf]), "mandate")
def test_scope_default_personal_when_empty(self):
leaf = _ds("l", "/folder")
self.assertEqual(_inheritFlags.getEffectiveFlag(leaf, "scope", [leaf]), "personal")
def test_unknown_flag_raises(self):
leaf = _ds("l", "/")
with self.assertRaises(ValueError):
_inheritFlags.getEffectiveFlag(leaf, "unknownFlag", [leaf])
def test_explicit_false_overrides_inherited_true(self):
"""Explicit False on a child must NOT cascade up to True from an ancestor."""
root = _ds("r", "/", neutralize=True)
leaf = _ds("l", "/folder", neutralize=False)
self.assertFalse(_inheritFlags.getEffectiveFlag(leaf, "neutralize", [root, leaf]))
def test_connection_root_inherits_cross_sourcetype(self):
"""Connection-root (sourceType=authority, path='/') is ancestor of all DS in that connection."""
connRoot = _ds("conn", "/", sourceType="msft", neutralize=True)
spService = _ds("sp", "/", sourceType="sharepointFolder")
olService = _ds("ol", "/", sourceType="outlookFolder")
self.assertTrue(_inheritFlags.getEffectiveFlag(spService, "neutralize", [connRoot, spService, olService]))
self.assertTrue(_inheritFlags.getEffectiveFlag(olService, "neutralize", [connRoot, spService, olService]))
def test_same_sourcetype_ancestor_wins_over_connection_root(self):
"""A same-sourceType service-root ancestor beats the connection-root."""
connRoot = _ds("conn", "/", sourceType="msft", neutralize=True)
spRoot = _ds("sp", "/", sourceType="sharepointFolder", neutralize=False)
spLeaf = _ds("spl", "/sites/x", sourceType="sharepointFolder")
self.assertFalse(_inheritFlags.getEffectiveFlag(spLeaf, "neutralize", [connRoot, spRoot, spLeaf]))
def test_connection_root_does_not_self_inherit(self):
"""Connection-root has no ancestor — does not infinite-loop on itself."""
connRoot = _ds("conn", "/", sourceType="msft")
self.assertFalse(_inheritFlags.getEffectiveFlag(connRoot, "neutralize", [connRoot]))
class TestCascadeReset(unittest.TestCase):
def _makeRootIf(self, dataSources: List[dict]):
rootIf = MagicMock()
rootIf.db.getRecordset = MagicMock(return_value=dataSources)
modified = []
def _modify(model, recordId, fields):
modified.append((recordId, fields))
rootIf.db.recordModify = MagicMock(side_effect=_modify)
return rootIf, modified
def test_resets_only_explicit_descendants(self):
parent = _ds("p", "/sites", neutralize=True)
explicitChild = _ds("c1", "/sites/folder1", neutralize=False)
inheritChild = _ds("c2", "/sites/folder2") # inherit -> not touched
sibling = _ds("s", "/other", neutralize=True) # NOT a descendant
rootIf, modified = self._makeRootIf([parent, explicitChild, inheritChild, sibling])
affected = _inheritFlags.cascadeResetDescendants(rootIf, parent, "neutralize")
self.assertEqual(affected, 1)
self.assertEqual(modified, [("c1", {"neutralize": None})])
def test_does_not_touch_other_flags(self):
parent = _ds("p", "/sites", neutralize=True)
child = _ds("c", "/sites/sub", neutralize=False, ragIndexEnabled=True)
rootIf, modified = self._makeRootIf([parent, child])
_inheritFlags.cascadeResetDescendants(rootIf, parent, "neutralize")
self.assertEqual(modified, [("c", {"neutralize": None})])
# ragIndexEnabled and scope on the child must remain untouched.
def test_does_not_cross_sourcetype(self):
"""Non-connection-root parents stay within their sourceType for cascade."""
parent = _ds("p", "/", neutralize=True, sourceType="sharepointFolder")
otherTypeDescendant = _ds("o", "/anything", neutralize=False, sourceType="outlookFolder")
rootIf, modified = self._makeRootIf([parent, otherTypeDescendant])
affected = _inheritFlags.cascadeResetDescendants(rootIf, parent, "neutralize")
self.assertEqual(affected, 0)
self.assertEqual(modified, [])
def test_connection_root_cascades_cross_sourcetype(self):
"""Toggle on connection-root cascades into every explicit DS of that connection."""
connRoot = _ds("conn", "/", sourceType="msft", neutralize=True)
spExplicit = _ds("sp", "/", sourceType="sharepointFolder", neutralize=False)
olInherit = _ds("ol", "/", sourceType="outlookFolder")
spLeafExplicit = _ds("sp-leaf", "/sites/x", sourceType="sharepointFolder", neutralize=True)
rootIf, modified = self._makeRootIf([connRoot, spExplicit, olInherit, spLeafExplicit])
affected = _inheritFlags.cascadeResetDescendants(rootIf, connRoot, "neutralize")
# spExplicit and spLeafExplicit had explicit values → reset. olInherit untouched.
self.assertEqual(affected, 2)
self.assertEqual({m[0] for m in modified}, {"sp", "sp-leaf"})
for _, fields in modified:
self.assertEqual(fields, {"neutralize": None})
def test_unknown_flag_raises(self):
parent = _ds("p", "/", neutralize=True)
rootIf, _ = self._makeRootIf([parent])
with self.assertRaises(ValueError):
_inheritFlags.cascadeResetDescendants(rootIf, parent, "unknownFlag")
def _fds(idVal: str, *, tableName: str, recordFilter=None, **flags) -> dict:
"""Build a FeatureDataSource dict fixture."""
base = {
"id": idVal,
"workspaceInstanceId": "ws-1",
"tableName": tableName,
"recordFilter": recordFilter,
"neutralize": None,
"scope": None,
}
base.update(flags)
return base
class TestFdsClassifyAndAncestry(unittest.TestCase):
def test_classify_workspace_wildcard(self):
self.assertEqual(_inheritFlags._fdsClassify(_fds("a", tableName="*")), "workspace")
def test_classify_table_wildcard(self):
self.assertEqual(_inheritFlags._fdsClassify(_fds("a", tableName="Pos")), "table")
def test_classify_record_specific(self):
rec = _fds("a", tableName="Pos", recordFilter={"id": "r-1"})
self.assertEqual(_inheritFlags._fdsClassify(rec), "record")
def test_workspace_is_ancestor_of_table_and_record(self):
ws = _fds("ws", tableName="*")
tbl = _fds("t", tableName="Pos")
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"})
self.assertTrue(_inheritFlags._fdsIsAncestor(ws, tbl))
self.assertTrue(_inheritFlags._fdsIsAncestor(ws, rec))
def test_table_is_ancestor_of_record_same_table_only(self):
tbl = _fds("t", tableName="Pos")
recSame = _fds("r1", tableName="Pos", recordFilter={"id": "1"})
recOther = _fds("r2", tableName="Other", recordFilter={"id": "1"})
self.assertTrue(_inheritFlags._fdsIsAncestor(tbl, recSame))
self.assertFalse(_inheritFlags._fdsIsAncestor(tbl, recOther))
def test_record_has_no_descendants(self):
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"})
tbl = _fds("t", tableName="Pos")
self.assertFalse(_inheritFlags._fdsIsAncestor(rec, tbl))
def test_no_cross_workspace_ancestry(self):
ws = _fds("ws", tableName="*", workspaceInstanceId="ws-A")
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"}, workspaceInstanceId="ws-B")
self.assertFalse(_inheritFlags._fdsIsAncestor(ws, rec))
class TestFdsEffectiveFlag(unittest.TestCase):
def test_own_explicit_wins(self):
ws = _fds("ws", tableName="*", neutralize=False)
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"}, neutralize=True)
self.assertTrue(_inheritFlags.getEffectiveFlagFds(rec, "neutralize", [ws, rec]))
def test_inherits_from_table_wildcard(self):
tbl = _fds("t", tableName="Pos", neutralize=True)
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"})
self.assertTrue(_inheritFlags.getEffectiveFlagFds(rec, "neutralize", [tbl, rec]))
def test_table_wildcard_beats_workspace_wildcard(self):
ws = _fds("ws", tableName="*", neutralize=False)
tbl = _fds("t", tableName="Pos", neutralize=True)
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"})
self.assertTrue(_inheritFlags.getEffectiveFlagFds(rec, "neutralize", [ws, tbl, rec]))
def test_workspace_wildcard_inherits_when_no_table(self):
ws = _fds("ws", tableName="*", neutralize=True)
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"})
self.assertTrue(_inheritFlags.getEffectiveFlagFds(rec, "neutralize", [ws, rec]))
def test_default_false_when_chain_empty(self):
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"})
self.assertFalse(_inheritFlags.getEffectiveFlagFds(rec, "neutralize", [rec]))
def test_unknown_flag_raises(self):
rec = _fds("r", tableName="*")
with self.assertRaises(ValueError):
_inheritFlags.getEffectiveFlagFds(rec, "ragIndexEnabled", [rec])
class TestFdsCascadeReset(unittest.TestCase):
def _makeRootIf(self, fdses):
rootIf = MagicMock()
rootIf.db.getRecordset = MagicMock(return_value=fdses)
modified = []
def _modify(model, recordId, fields):
modified.append((recordId, fields))
rootIf.db.recordModify = MagicMock(side_effect=_modify)
return rootIf, modified
def test_workspace_cascades_to_all_explicit_descendants(self):
ws = _fds("ws", tableName="*", neutralize=True)
tblExplicit = _fds("t", tableName="Pos", neutralize=False)
tblInherit = _fds("t2", tableName="Other")
recExplicit = _fds("r", tableName="Pos", recordFilter={"id": "1"}, neutralize=True)
rootIf, modified = self._makeRootIf([ws, tblExplicit, tblInherit, recExplicit])
affected = _inheritFlags.cascadeResetDescendantsFds(rootIf, ws, "neutralize")
self.assertEqual(affected, 2)
self.assertEqual({m[0] for m in modified}, {"t", "r"})
def test_table_cascades_only_to_same_table_records(self):
tbl = _fds("t", tableName="Pos", neutralize=True)
recSame = _fds("r1", tableName="Pos", recordFilter={"id": "1"}, neutralize=False)
recOther = _fds("r2", tableName="Other", recordFilter={"id": "1"}, neutralize=False)
rootIf, modified = self._makeRootIf([tbl, recSame, recOther])
affected = _inheritFlags.cascadeResetDescendantsFds(rootIf, tbl, "neutralize")
self.assertEqual(affected, 1)
self.assertEqual(modified, [("r1", {"neutralize": None})])
def test_record_has_no_cascade(self):
rec = _fds("r", tableName="Pos", recordFilter={"id": "1"}, neutralize=True)
rootIf, modified = self._makeRootIf([rec])
affected = _inheritFlags.cascadeResetDescendantsFds(rootIf, rec, "neutralize")
self.assertEqual(affected, 0)
self.assertEqual(modified, [])
def test_unknown_flag_raises(self):
ws = _fds("ws", tableName="*", neutralize=True)
rootIf, _ = self._makeRootIf([ws])
with self.assertRaises(ValueError):
_inheritFlags.cascadeResetDescendantsFds(rootIf, ws, "ragIndexEnabled")
class TestPathNormalization(unittest.TestCase):
def test_empty_path_normalises_to_root(self):
self.assertEqual(_inheritFlags._normalisePath(""), "/")
self.assertEqual(_inheritFlags._normalisePath(None), "/")
def test_trailing_slash_stripped(self):
self.assertEqual(_inheritFlags._normalisePath("/foo/"), "/foo")
self.assertEqual(_inheritFlags._normalisePath("/"), "/")
def test_leading_slash_added(self):
self.assertEqual(_inheritFlags._normalisePath("foo/bar"), "/foo/bar")
if __name__ == "__main__":
unittest.main()