Skip to content

feat: refactor is_valid_expression to support AhbContext #751

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hf-kklein merged 1 commit into from
Apr 12, 2026
Merged

feat: refactor is_valid_expression to support AhbContext #751

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 45 additions & 17 deletions src/ahbicht/content_evaluation/expression_check.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,37 +10,59 @@
from lark import Token, Tree
from lark.exceptions import UnexpectedCharacters, VisitError

from ahbicht.content_evaluation.ahb_context import AhbContext
from ahbicht.expressions import InvalidExpressionError
from ahbicht.expressions.ahb_expression_evaluation import evaluate_ahb_expression_tree
from ahbicht.expressions.condition_expression_parser import extract_categorized_keys_from_tree
from ahbicht.expressions.expression_resolver import parse_expression_including_unresolved_subexpressions
from ahbicht.models.content_evaluation_result import ContentEvaluationResult

if TYPE_CHECKING:
from ahbicht.content_evaluation.ahb_context import AhbContext
from efoli import EdifactFormat, EdifactFormatVersion


async def is_valid_expression(
async def is_valid_expression( # pylint: disable=too-many-locals
expression_or_tree: Union[str, Tree[Token]],
content_evaluation_result_setter: Callable[[ContentEvaluationResult], Any],
content_evaluation_result_setter: Optional[Callable[[ContentEvaluationResult], Any]] = None,
ahb_context: Optional[AhbContext] = None,
edifact_format: Optional[EdifactFormat] = None,
edifact_format_version: Optional[EdifactFormatVersion] = None,
) -> tuple[bool, Optional[str]]:
"""
Returns true iff the given expression is both well-formed and valid.
An expression is valid if and only if all possible content evaluations lead to a meaningful results.
⚠ This only works if the provided content_evaluation_result_setter writes the EvaluatableData in such a way, that
the injected Evaluators (FC/RC/Hints/Package) can work with it.
This is easiest for the ContentEvaluationResultBased FC/RC/Hints/Package token logic providers.
:param content_evaluation_result_setter: a threadsafe method that writes the given Content Evaluation Result into
the evaluatable data

There are two ways to use this function:

1. (New, recommended) Pass ``edifact_format`` and ``edifact_format_version``. A fresh ``AhbContext`` is built
for each possible CER automatically. No inject setup needed.

2. (Legacy, deprecated) Pass a ``content_evaluation_result_setter`` callback that writes the CER into the
evaluatable data for the injected evaluators. Requires a configured inject container.

:param expression_or_tree: "Muss [1] U [2]" (returns True) "Muss ([61]u [584]) o[583]" (returns False)
:param content_evaluation_result_setter: (deprecated) a threadsafe method that writes the given Content Evaluation
Result into the evaluatable data
:param ahb_context: optional AhbContext; if provided, its edifact_format/version are used to build per-CER contexts
:param edifact_format: the EDIFACT format for building per-CER AhbContexts (used if ahb_context is None)
:param edifact_format_version: the EDIFACT format version for building per-CER AhbContexts
:return: (True,None) iff the expression is valid; (False, error message) otherwise
"""
# Determine format/version for AhbContext construction
_edifact_format = edifact_format
_edifact_format_version = edifact_format_version
if ahb_context is not None:
_edifact_format = ahb_context.evaluatable_data.edifact_format
_edifact_format_version = ahb_context.evaluatable_data.edifact_format_version

tree: Tree[Token]
parse_context_kwargs: dict = {}
if ahb_context is not None:
parse_context_kwargs["ahb_context"] = ahb_context
if isinstance(expression_or_tree, str):
try:
tree = await parse_expression_including_unresolved_subexpressions(
expression_or_tree, ahb_context=ahb_context
expression_or_tree, **parse_context_kwargs
)
except SyntaxError as syntax_error:
return False, str(syntax_error)
Expand All @@ -53,19 +75,25 @@ async def is_valid_expression(
else:
raise ValueError(f"{expression_or_tree} is neither a string nor a Tree")
categorized_key_extract = extract_categorized_keys_from_tree(tree, sanitize=True)
context_kwargs: dict = {}
if ahb_context is not None:
context_kwargs["ahb_context"] = ahb_context
evaluation_tasks: list[Awaitable] = []
for content_evaluation_result in categorized_key_extract.generate_possible_content_evaluation_results():
# create (but do not await) the evaluation tasks for all possible content evaluation results
# the idea is, that if _any_ evaluation task raises an uncatched exception this can be interpreted as:
# "the expression is invalid"

async def evaluate_with_cer(cer: ContentEvaluationResult):
content_evaluation_result_setter(cer)
eval_kwargs: dict = {}
if _edifact_format is not None and _edifact_format_version is not None:
# New path: build a fresh AhbContext per CER
cer_context = AhbContext.from_content_evaluation_result(cer, _edifact_format, _edifact_format_version)
eval_kwargs["ahb_context"] = cer_context
elif content_evaluation_result_setter is not None:
# Legacy path: use the setter callback + inject
content_evaluation_result_setter(cer)
else:
raise ValueError(
"is_valid_expression requires either (edifact_format + edifact_format_version) "
"or a content_evaluation_result_setter callback."
)
try:
await evaluate_ahb_expression_tree(tree, **context_kwargs)
await evaluate_ahb_expression_tree(tree, **eval_kwargs)
except NotImplementedError as not_implemented_error:
# we can ignore some specific errors
if "due to missing information" in str(not_implemented_error):
Expand Down
59 changes: 13 additions & 46 deletions unittests/test_validity_check.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,57 +1,15 @@
from contextvars import ContextVar
from typing import Optional

import inject
import pytest

from ahbicht.content_evaluation.evaluationdatatypes import EvaluatableData, EvaluatableDataProvider
from ahbicht.content_evaluation.evaluator_factory import create_content_evaluation_result_based_evaluators
from ahbicht.content_evaluation.expression_check import is_valid_expression
from ahbicht.content_evaluation.token_logic_provider import SingletonTokenLogicProvider, TokenLogicProvider
from ahbicht.expressions.expression_resolver import parse_expression_including_unresolved_subexpressions
from ahbicht.models.content_evaluation_result import ContentEvaluationResult
from unittests.defaults import default_test_format, default_test_version

_content_evaluation_result: ContextVar[Optional[ContentEvaluationResult]] = ContextVar(
"_content_evaluation_result", default=None
)


def _get_evaluatable_data():
"""
returns the _content_evaluation_result context var value wrapped in a EvaluatableData container.
This is the kind of data that the ContentEvaluationResultBased RC/FC Evaluators, HintsProvider and Package Resolver
require.
:return:
"""
cer = _content_evaluation_result.get()
return EvaluatableData(
body=cer.model_dump(mode="json"),
edifact_format=default_test_format,
edifact_format_version=default_test_version,
)


class TestValidityCheck:
"""
a test class for the expression validation feature
"""

@pytest.fixture
def inject_cer_evaluators(self):
def configure(binder):
binder.bind(
TokenLogicProvider,
SingletonTokenLogicProvider(
[*create_content_evaluation_result_based_evaluators(default_test_format, default_test_version)]
),
)
binder.bind_to_provider(EvaluatableDataProvider, _get_evaluatable_data)

inject.configure_once(configure)
yield
inject.clear()

@pytest.mark.parametrize(
"ahb_expression,expected_result",
[
Expand All @@ -65,23 +23,32 @@ def configure(binder):
pytest.param(
"([446] ∧ ([465] ∨ [466]) ∧ [467] ∧ ([468] ⊻ ([469] ∧ [470])) ⊻ [448]", False
), # unbalanced brackets
pytest.param("Muss [15] ∧ [2050]", True), # contains a 'Geschütztes' Leerzeichen
pytest.param("Muss [15] ∧ [2050]", True), # contains a 'Geschütztes' Leerzeichen
pytest.param("Muss [15]🙄∧ [2050]", False),
pytest.param(
"X ([950] [509] ∧ ([64] V [70])) V ([960] [522] ∧ [71] ∧ [53])", True
), # nur echt mit 'V' statt LOR
pytest.param("X [1P0..n]", True),
],
)
async def test_is_valid_expression(self, ahb_expression: str, expected_result: bool, inject_cer_evaluators):
actual_str = await is_valid_expression(ahb_expression, lambda cer: _content_evaluation_result.set(cer))
async def test_is_valid_expression(self, ahb_expression: str, expected_result: bool):
"""Tests validity using AhbContext (no inject setup needed)."""
actual_str = await is_valid_expression(
ahb_expression,
edifact_format=default_test_format,
edifact_format_version=default_test_version,
)
assert actual_str[0] == expected_result
# check the tree as argument, too
try:
tree = await parse_expression_including_unresolved_subexpressions(ahb_expression)
except SyntaxError:
return # ok, the syntax error is actually raised on parsing already
actual_tree = await is_valid_expression(tree, lambda cer: _content_evaluation_result.set(cer))
actual_tree = await is_valid_expression(
tree,
edifact_format=default_test_format,
edifact_format_version=default_test_version,
)
assert actual_tree[0] == expected_result

async def test_is_valid_expression_value_error(self):
Expand Down
Loading