199 lines
6.9 KiB
Python
199 lines
6.9 KiB
Python
# Copyright (c) 2026 Patrick Motsch
|
|
# All rights reserved.
|
|
"""Unit tests for the trustee ontology and the ontology-to-prompt compiler.
|
|
|
|
Verifies:
|
|
|
|
* the descriptor passes Pydantic validation
|
|
* `constraintsForTable` correctly scopes by table/field prefix
|
|
* the compiler emits a stable header + every entity name + every
|
|
constraint message
|
|
* the QueryValidator picks up ontology constraints (NEVER_AGGREGATE on
|
|
closingBalance) over the convention-based defaults
|
|
* the `getAgentOntology()` hook on `mainTrustee` returns the descriptor
|
|
* `_buildValidatorForFeature("trustee")` wires the validator with the
|
|
ontology
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from modules.features.trustee.mainTrustee import getAgentOntology
|
|
from modules.features.trustee.trusteeOntology import getTrusteeOntology
|
|
from modules.serviceCenter.services.serviceAgent.datamodelOntology import (
|
|
ConstraintRule,
|
|
OntologyDescriptor,
|
|
SemanticType,
|
|
ValidationErrorCode,
|
|
)
|
|
from modules.serviceCenter.services.serviceAgent.featureDataAgent import (
|
|
_buildValidatorForFeature,
|
|
_loadFeatureOntologyBlock,
|
|
)
|
|
from modules.serviceCenter.services.serviceAgent.ontologyToPromptCompiler import (
|
|
compileOntologyToPrompt,
|
|
)
|
|
from modules.serviceCenter.services.serviceAgent.queryValidator import QueryValidator
|
|
from modules.shared import fkRegistry
|
|
|
|
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def _ensureModels():
|
|
fkRegistry._ensureModelsLoaded()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OntologyDescriptor structure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_trusteeOntology_returnsValidDescriptor():
|
|
ont = getTrusteeOntology()
|
|
assert isinstance(ont, OntologyDescriptor)
|
|
assert ont.featureCode == "trustee"
|
|
assert ont.entities and ont.relations and ont.constraints and ont.canonicalPatterns
|
|
|
|
|
|
def test_trusteeOntology_hasBankAccountSpecialization():
|
|
ont = getTrusteeOntology()
|
|
bank = next((e for e in ont.entities if e.name == "BankAccount"), None)
|
|
assert bank is not None
|
|
assert bank.parentEntity == "Account"
|
|
assert bank.semanticType == SemanticType.ACCOUNT
|
|
|
|
|
|
def test_trusteeOntology_closingBalanceIsNeverAggregate():
|
|
ont = getTrusteeOntology()
|
|
constraints = ont.constraintsForTable("TrusteeDataAccountBalance")
|
|
matching = [
|
|
c for c in constraints
|
|
if c.rule == ConstraintRule.NEVER_AGGREGATE
|
|
and c.appliesTo == "TrusteeDataAccountBalance.closingBalance"
|
|
]
|
|
assert matching, "Expected NEVER_AGGREGATE constraint on closingBalance"
|
|
|
|
|
|
def test_trusteeOntology_requiresPeriodFilterOnBalanceTable():
|
|
ont = getTrusteeOntology()
|
|
constraints = ont.constraintsForTable("TrusteeDataAccountBalance")
|
|
table_level = [c for c in constraints if c.rule == ConstraintRule.REQUIRES_FILTER_ON]
|
|
assert table_level, "Expected at least one REQUIRES_FILTER_ON constraint"
|
|
required = table_level[0].params.get("requiredFields") or []
|
|
assert "periodYear" in required
|
|
assert "periodMonth" in required
|
|
|
|
|
|
def test_constraintsForTable_filtersScopeCorrectly():
|
|
ont = getTrusteeOntology()
|
|
bal = ont.constraintsForTable("TrusteeDataAccountBalance")
|
|
journal = ont.constraintsForTable("TrusteeDataJournalLine")
|
|
for c in bal:
|
|
assert c.appliesTo.startswith("TrusteeDataAccountBalance")
|
|
for c in journal:
|
|
assert c.appliesTo.startswith("TrusteeDataJournalLine")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Prompt compiler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_compiler_emitsExpectedHeader():
|
|
block = compileOntologyToPrompt(getTrusteeOntology())
|
|
assert block.startswith("DOMAIN ONTOLOGY (trustee):"), block.splitlines()[0]
|
|
|
|
|
|
def test_compiler_includesAllEntityNames():
|
|
ont = getTrusteeOntology()
|
|
block = compileOntologyToPrompt(ont)
|
|
for e in ont.entities:
|
|
assert e.name in block, f"Entity {e.name} missing from compiled prompt"
|
|
|
|
|
|
def test_compiler_includesAllConstraintMessages():
|
|
ont = getTrusteeOntology()
|
|
block = compileOntologyToPrompt(ont)
|
|
for c in ont.constraints:
|
|
assert c.message.split(".")[0] in block, f"Constraint message missing: {c.message[:40]}"
|
|
|
|
|
|
def test_compiler_includesCanonicalPatternTools():
|
|
ont = getTrusteeOntology()
|
|
block = compileOntologyToPrompt(ont)
|
|
for p in ont.canonicalPatterns:
|
|
assert p.intent in block
|
|
assert p.pattern["tool"] in block
|
|
|
|
|
|
def test_compiler_deterministic():
|
|
block1 = compileOntologyToPrompt(getTrusteeOntology())
|
|
block2 = compileOntologyToPrompt(getTrusteeOntology())
|
|
assert block1 == block2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# QueryValidator x ontology integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_validator_picksUpOntologyNeverAggregate():
|
|
validator = QueryValidator(ontology=getTrusteeOntology())
|
|
err = validator.validateAggregateQuery(
|
|
"TrusteeDataAccountBalance",
|
|
{"aggregate": "SUM", "field": "closingBalance"},
|
|
)
|
|
assert err is not None
|
|
assert err.code == ValidationErrorCode.INVALID_AGGREGATE_TARGET
|
|
assert err.field == "closingBalance"
|
|
|
|
|
|
def test_validator_ontologyConstraintFiresOnDebitTotal():
|
|
validator = QueryValidator(ontology=getTrusteeOntology())
|
|
err = validator.validateAggregateQuery(
|
|
"TrusteeDataAccountBalance",
|
|
{"aggregate": "SUM", "field": "debitTotal"},
|
|
)
|
|
assert err is not None
|
|
assert err.code == ValidationErrorCode.INVALID_AGGREGATE_TARGET
|
|
|
|
|
|
def test_validator_allowsLegitimateAggregateOnJournalLine():
|
|
validator = QueryValidator(ontology=getTrusteeOntology())
|
|
err = validator.validateAggregateQuery(
|
|
"TrusteeDataJournalLine",
|
|
{"aggregate": "SUM", "field": "debitAmount"},
|
|
)
|
|
assert err is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# featureDataAgent integration hooks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_mainTrustee_getAgentOntology_returnsDescriptor():
|
|
ont = getAgentOntology()
|
|
assert isinstance(ont, OntologyDescriptor)
|
|
assert ont.featureCode == "trustee"
|
|
|
|
|
|
def test_loadFeatureOntologyBlock_returnsCompiledBlock():
|
|
block = _loadFeatureOntologyBlock("trustee")
|
|
assert block.startswith("DOMAIN ONTOLOGY (trustee):")
|
|
assert "BankAccount" in block
|
|
|
|
|
|
def test_loadFeatureOntologyBlock_unknownFeatureReturnsEmpty():
|
|
assert _loadFeatureOntologyBlock("doesNotExist") == ""
|
|
|
|
|
|
def test_buildValidatorForFeature_trustee_hasOntology():
|
|
validator = _buildValidatorForFeature("trustee")
|
|
assert validator._ontology is not None
|
|
assert validator._ontology.featureCode == "trustee"
|
|
|
|
|
|
def test_buildValidatorForFeature_unknownFeature_noOntology():
|
|
validator = _buildValidatorForFeature("doesNotExist")
|
|
assert validator._ontology is None
|