cleveragents-core/features/steps/domain_analyzer_steps.py

"""Step definitions for domain-specific analyzers feature.

Covers the four built-in analyzers (Python, Markdown, PostgreSQL,
Docker Compose), the ``AnalyzerRegistry``, and protocol compliance
checks.

Step patterns are prefixed or worded to avoid collisions with existing
step definitions in ``uko_analyzers_steps.py``.  Triple-assertion steps
use ``use_step_matcher("re")`` to avoid ``parse``-format ambiguity.
"""

from __future__ import annotations

from behave import given, then, use_step_matcher, when
from behave.runner import Context

from cleveragents.domain.models.acms.analyzers import (
    AnalyzerProtocol,
    AnalyzerRegistry,
    UKOTriple,
)
from cleveragents.domain.models.acms.docker_compose_analyzer import (
    DockerComposeAnalyzer,
)
from cleveragents.domain.models.acms.markdown_analyzer import MarkdownAnalyzer
from cleveragents.domain.models.acms.postgresql_analyzer import PostgreSQLAnalyzer
from cleveragents.domain.models.acms.python_analyzer import PythonAnalyzer

__all__: list[str] = []

# ---------------------------------------------------------------------------
# Analyzer type mapping
# ---------------------------------------------------------------------------

_ANALYZER_CLASSES: dict[str, type] = {
    "PythonAnalyzer": PythonAnalyzer,
    "MarkdownAnalyzer": MarkdownAnalyzer,
    "PostgreSQLAnalyzer": PostgreSQLAnalyzer,
    "DockerComposeAnalyzer": DockerComposeAnalyzer,
}

_ALL_ANALYZERS = (
    PythonAnalyzer,
    MarkdownAnalyzer,
    PostgreSQLAnalyzer,
    DockerComposeAnalyzer,
)

# Sample content for each analyzer (cross-analyzer scenarios).
_SAMPLE_CONTENT: dict[str, str] = {
    "PythonAnalyzer": "class Example:\n    pass\n",
    "MarkdownAnalyzer": "# Title\nSome text.\n",
    "PostgreSQLAnalyzer": ("CREATE TABLE sample (\n    id SERIAL PRIMARY KEY\n);\n"),
    "DockerComposeAnalyzer": "services:\n  web:\n    image: nginx\n",
}

_SAMPLE_URI = "uko://test/resource"


# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------


@given("a {analyzer_type} analyzer")
def step_given_named_analyzer(context: Context, analyzer_type: str) -> None:
    cls = _ANALYZER_CLASSES.get(analyzer_type)
    if cls is None:
        raise ValueError(
            f"Unknown analyzer type {analyzer_type!r}. "
            f"Known types: {sorted(_ANALYZER_CLASSES)}"
        )
    context.analyzer = cls()
    # Track all created analyzers for cross-analyzer scenarios.
    if not hasattr(context, "analyzers"):
        context.analyzers = {}
    context.analyzers[analyzer_type] = context.analyzer


@given("a fresh analyzer registry")
def step_given_fresh_registry(context: Context) -> None:
    context.registry = AnalyzerRegistry()


@given("a fresh analyzer registry with all domain analyzers registered")
def step_given_fresh_registry_with_all(context: Context) -> None:
    context.registry = AnalyzerRegistry()
    for cls in _ALL_ANALYZERS:
        context.registry.register(cls())


# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------


@when("I register all 4 domain analyzers")
def step_when_register_all_four(context: Context) -> None:
    for cls in _ALL_ANALYZERS:
        context.registry.register(cls())


@when("I register a PythonAnalyzer")
def step_when_register_python(context: Context) -> None:
    context.registry.register(PythonAnalyzer())


@when('I register a duplicate analyzer that also claims ".py"')
def step_when_register_duplicate_py(context: Context) -> None:
    """Register a second analyzer whose supported_extensions include .py."""

    class _DuplicatePyAnalyzer:
        @property
        def supported_extensions(self) -> frozenset[str]:
            return frozenset({".py"})

        @property
        def domain(self) -> str:
            return "duplicate"

        def analyze(self, content: str, resource_uri: str) -> list[UKOTriple]:
            return []  # pragma: no cover

    _dup: type[AnalyzerProtocol] = _DuplicatePyAnalyzer
    del _dup  # only used for static type-checking assertion
    context.registry.register(_DuplicatePyAnalyzer())


@when('I look up extension "{ext}" in the registry')
def step_when_lookup_extension_in_registry(context: Context, ext: str) -> None:
    result = context.registry.get_for_extension(ext)
    context.lookup_result = result
    if result is not None:
        context.analyzer = result


@when("I analyze the Python content:")
def step_when_analyze_python(context: Context) -> None:
    context.triples = context.analyzer.analyze(context.text, _SAMPLE_URI)


@when("I analyze the Markdown content:")
def step_when_analyze_markdown(context: Context) -> None:
    context.triples = context.analyzer.analyze(context.text, _SAMPLE_URI)


@when("I analyze the SQL content:")
def step_when_analyze_sql(context: Context) -> None:
    context.triples = context.analyzer.analyze(context.text, _SAMPLE_URI)


@when("I analyze the Docker Compose content:")
def step_when_analyze_docker_compose(context: Context) -> None:
    context.triples = context.analyzer.analyze(context.text, _SAMPLE_URI)


@when("each analyzer processes its sample content")
def step_when_each_analyzer_processes_sample(context: Context) -> None:
    all_triples: list[UKOTriple] = []
    for name, analyzer in context.analyzers.items():
        sample = _SAMPLE_CONTENT[name]
        triples = analyzer.analyze(sample, _SAMPLE_URI)
        all_triples.extend(triples)
    context.triples = all_triples


# ---------------------------------------------------------------------------
# Then steps — protocol compliance
# ---------------------------------------------------------------------------


@then("the analyzer should satisfy the AnalyzerProtocol")
def step_then_satisfies_protocol(context: Context) -> None:
    assert isinstance(context.analyzer, AnalyzerProtocol), (
        f"{type(context.analyzer).__name__} does not satisfy AnalyzerProtocol"
    )


@then('the current analyzer domain should be "{domain}"')
def step_then_current_analyzer_domain(context: Context, domain: str) -> None:
    actual = context.analyzer.domain
    assert actual == domain, f"Expected domain={domain!r}, got {actual!r}"


# ---------------------------------------------------------------------------
# Then steps — registry assertions
# ---------------------------------------------------------------------------


@then("the registry should contain {count:d} analyzers")
def step_then_registry_count(context: Context, count: int) -> None:
    actual = len(context.registry)
    assert actual == count, f"Expected {count} analyzers in registry, got {actual}"


@then("the extension lookup result should be None")
def step_then_extension_lookup_none(context: Context) -> None:
    assert context.lookup_result is None, (
        f"Expected None, got {context.lookup_result!r}"
    )


@then('the registry extensions should include "{ext}"')
def step_then_registry_extensions_include(context: Context, ext: str) -> None:
    extensions = context.registry.list_extensions()
    assert ext in extensions, (
        f"Extension {ext!r} not in registered extensions {extensions}"
    )


# ---------------------------------------------------------------------------
# Then steps — triple count assertions
# ---------------------------------------------------------------------------


@then("no triples should be returned")
def step_then_no_triples(context: Context) -> None:
    actual = len(context.triples)
    assert actual == 0, f"Expected 0 triples, got {actual}.\nTriples: {context.triples}"


@then("the number of Table type triples should be {count:d}")
def step_then_table_type_count(context: Context, count: int) -> None:
    actual = sum(
        1
        for t in context.triples
        if t.predicate == "rdf:type" and t.object_uri == "uko-data:Table"
    )
    assert actual == count, f"Expected {count} Table type triples, got {actual}"


# =========================================================================
# Switch to regex matcher for triple-assertion steps to avoid ambiguity
# that behave's ``parse`` format causes when patterns share a prefix
# (e.g. "{pred}" greedily consuming "… and object_uri …").
# =========================================================================
use_step_matcher("re")


@then(
    r'the triples should include predicate "(?P<pred>[^"]+)"'
    r' with object_uri "(?P<obj>[^"]+)"'
)
def step_then_triple_pred_obj_uri(context: Context, pred: str, obj: str) -> None:
    found = any(t.predicate == pred and t.object_uri == obj for t in context.triples)
    assert found, (
        f"No triple with predicate={pred!r} object_uri={obj!r} in {context.triples}"
    )


@then(
    r'the triples should include predicate "(?P<pred>[^"]+)"'
    r' with object_value "(?P<val>[^"]+)"'
)
def step_then_triple_pred_obj_val(context: Context, pred: str, val: str) -> None:
    found = any(t.predicate == pred and t.object_value == val for t in context.triples)
    assert found, (
        f"No triple with predicate={pred!r} object_value={val!r} in {context.triples}"
    )


@then(
    r'the triples should include predicate "(?P<pred>[^"]+)"'
    r' with object_uri containing "(?P<fragment>[^"]+)"'
)
def step_then_triple_pred_obj_uri_contains(
    context: Context, pred: str, fragment: str
) -> None:
    found = any(
        t.predicate == pred and t.object_uri is not None and fragment in t.object_uri
        for t in context.triples
    )
    assert found, (
        f"No triple with predicate={pred!r} and object_uri containing "
        f"{fragment!r} in {context.triples}"
    )


@then(
    r'the triples should include predicate "(?P<pred>[^"]+)"'
    r' linking subject "(?P<s_frag>[^"]+)"'
    r' to object "(?P<o_frag>[^"]+)"'
)
def step_then_triple_links_subject_to_object(
    context: Context, pred: str, s_frag: str, o_frag: str
) -> None:
    found = any(
        t.predicate == pred
        and s_frag in t.subject_uri
        and (
            (t.object_uri is not None and o_frag in t.object_uri)
            or o_frag in t.object_value
        )
        for t in context.triples
    )
    assert found, (
        f"No triple with predicate={pred!r} linking subject containing "
        f"{s_frag!r} to object containing {o_frag!r} in {context.triples}"
    )


@then(
    r'the triples should not include predicate "(?P<pred>[^"]+)"'
    r' with object_uri "(?P<obj>[^"]+)"'
)
def step_then_no_triple_pred_obj_uri(context: Context, pred: str, obj: str) -> None:
    found = any(t.predicate == pred and t.object_uri == obj for t in context.triples)
    assert not found, (
        f"Unexpected triple with predicate={pred!r} object_uri={obj!r} "
        f"found in {context.triples}"
    )


@then(
    r'the triples should not include predicate "(?P<pred>[^"]+)"'
    r' with object_value "(?P<val>[^"]+)"'
)
def step_then_no_triple_pred_obj_val(context: Context, pred: str, val: str) -> None:
    found = any(t.predicate == pred and t.object_value == val for t in context.triples)
    assert not found, (
        f"Unexpected triple with predicate={pred!r} object_value={val!r} "
        f"found in {context.triples}"
    )


@then(r'the triples should include predicate "(?P<pred>[^"]+)"')
def step_then_triple_has_predicate(context: Context, pred: str) -> None:
    found = any(t.predicate == pred for t in context.triples)
    assert found, f"No triple with predicate={pred!r} in {context.triples}"


# Switch back to parse matcher for remaining steps.
use_step_matcher("parse")


# ---------------------------------------------------------------------------
# Then steps — error handling
# ---------------------------------------------------------------------------


@then("analyzing empty content with this analyzer should raise ValueError")
def step_then_empty_content_raises(context: Context) -> None:
    try:
        context.analyzer.analyze("", _SAMPLE_URI)
    except ValueError:
        return  # Expected
    raise AssertionError(
        f"{type(context.analyzer).__name__}.analyze('', ...) did not raise ValueError"
    )


@then("analyzing whitespace-only content with this analyzer should raise ValueError")
def step_then_whitespace_only_raises(context: Context) -> None:
    try:
        context.analyzer.analyze("   \n\t  ", _SAMPLE_URI)
    except ValueError:
        return  # Expected
    raise AssertionError(
        f"{type(context.analyzer).__name__}.analyze('   \\n\\t  ', ...) "
        "did not raise ValueError"
    )


@then("analyzing content with empty resource_uri should raise ValueError")
def step_then_empty_resource_uri_raises(context: Context) -> None:
    # Use a minimal valid content for each analyzer domain.
    _domain_sample: dict[str, str] = {
        "python": "class X:\n    pass\n",
        "markdown": "# Title\n",
        "postgresql": "CREATE TABLE t (id SERIAL);\n",
        "docker-compose": "services:\n  web:\n    image: nginx\n",
    }
    sample = _domain_sample.get(context.analyzer.domain, "content")
    try:
        context.analyzer.analyze(sample, "")
    except ValueError:
        return  # Expected
    raise AssertionError(
        f"{type(context.analyzer).__name__}.analyze(..., '') did not raise ValueError"
    )


# ---------------------------------------------------------------------------
# Then steps — URI scheme and confidence
# ---------------------------------------------------------------------------


@then('every triple subject_uri should start with "{prefix}"')
def step_then_all_subject_uris_start_with(context: Context, prefix: str) -> None:
    for triple in context.triples:
        assert triple.subject_uri.startswith(prefix), (
            f"Triple subject_uri {triple.subject_uri!r} does not start with {prefix!r}"
        )


@then("every triple confidence should be {score:g}")
def step_then_all_confidence_scores(context: Context, score: float) -> None:
    for triple in context.triples:
        assert triple.confidence == score, (
            f"Triple {triple!r} has confidence={triple.confidence}, expected {score}"
        )