temp/features/steps/postgresql_analyzer_coverage_boost_steps.py

"""Step definitions for PostgreSQLAnalyzer coverage-boost scenarios.

Targets uncovered lines in postgresql_analyzer.py:
- Lines 112-116: exception handler in analyze()
- Line 135:      duplicate CREATE SCHEMA deduplication
- Line 198:      _extract_tables with tables_seen=None
- Line 285:      CREATE VIEW without trailing semicolon
- Lines 297-302: schema-qualified CREATE VIEW with uko:contains
"""

from __future__ import annotations

from unittest.mock import patch

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

use_step_matcher("re")

from cleveragents.domain.models.acms.postgresql_analyzer import (  # noqa: E402
    PostgreSQLAnalyzer,
)

__all__: list[str] = []

_SAMPLE_URI = "uko://test/coverage-boost"


# ---------------------------------------------------------------------------
# Given
# ---------------------------------------------------------------------------


@given(r"a fresh PostgreSQLAnalyzer instance")
def step_given_fresh_postgresql_analyzer(context: Context) -> None:
    context.analyzer = PostgreSQLAnalyzer()
    context.triples = []


# ---------------------------------------------------------------------------
# When
# ---------------------------------------------------------------------------


@when(r"I analyze DDL with duplicate schemas:")
def step_when_analyze_ddl_duplicate_schemas(context: Context) -> None:
    context.triples = context.analyzer.analyze(context.text, _SAMPLE_URI)


@when(r"I analyze DDL that triggers an internal parse error")
def step_when_analyze_ddl_with_parse_error(context: Context) -> None:
    """Force an exception inside _extract_tables to exercise lines 112-116."""
    with (
        patch.object(
            context.analyzer,
            "_extract_tables",
            side_effect=RuntimeError("simulated parse failure"),
        ),
        # Capture the log output to verify the warning was emitted.
        patch(
            "cleveragents.domain.models.acms.postgresql_analyzer.logger"
        ) as mock_logger,
    ):
        context.triples = context.analyzer.analyze(
            "CREATE SCHEMA testschema;\nCREATE TABLE t (id INT);\n",
            _SAMPLE_URI,
        )
        context.mock_logger = mock_logger


@when(r"I call _extract_tables directly without tables_seen on:")
def step_when_call_extract_tables_directly(context: Context) -> None:
    """Call _extract_tables with tables_seen=None to hit line 198."""
    context.triples = context.analyzer._extract_tables(
        context.text,
        _SAMPLE_URI,
        set(),
        None,
    )


@when(r"I analyze DDL with a view missing its semicolon:")
def step_when_analyze_ddl_view_no_semicolon(context: Context) -> None:
    context.triples = context.analyzer.analyze(context.text, _SAMPLE_URI)


@when(r"I analyze DDL with a schema-qualified view:")
def step_when_analyze_ddl_schema_qualified_view(context: Context) -> None:
    context.triples = context.analyzer.analyze(context.text, _SAMPLE_URI)


# ---------------------------------------------------------------------------
# Then
# ---------------------------------------------------------------------------


@then(r"exactly (?P<count>\d+) Schema type triple should exist")
def step_then_exactly_n_schema_triples(context: Context, count: str) -> None:
    count_int = int(count)
    actual = sum(
        1
        for t in context.triples
        if t.predicate == "rdf:type" and t.object_uri == "uko-data:Schema"
    )
    assert actual == count_int, (
        f"Expected {count_int} Schema type triple(s), got {actual}.\n"
        f"Triples: {context.triples}"
    )


@then(
    r'the result triples should contain predicate "(?P<pred>[^"]+)" with value "(?P<val>[^"]+)"'
)
def step_then_result_contains_pred_value(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}.\n"
        f"Triples: {context.triples}"
    )


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


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


@then(r"the partial results should be returned without raising")
def step_then_partial_results_returned(context: Context) -> None:
    # analyze() should have caught the exception and returned a list
    assert isinstance(context.triples, list), (
        f"Expected a list, got {type(context.triples).__name__}"
    )


@then(r"the parse error should have been logged as a warning")
def step_then_parse_error_logged(context: Context) -> None:
    mock_logger = context.mock_logger
    mock_logger.warning.assert_called_once()
    call_args = mock_logger.warning.call_args
    assert "parse error" in call_args[0][0].lower(), (
        f"Expected 'parse error' in warning message, got: {call_args[0][0]!r}"
    )


@then(r'the direct extraction should produce Table triples for "(?P<table_name>[^"]+)"')
def step_then_direct_extraction_produces_table(
    context: Context, table_name: str
) -> None:
    has_table_type = any(
        t.predicate == "rdf:type" and t.object_uri == "uko-data:Table"
        for t in context.triples
    )
    has_label = any(
        t.predicate == "rdfs:label" and t.object_value == table_name
        for t in context.triples
    )
    assert has_table_type, (
        f"No rdf:type uko-data:Table triple found.\nTriples: {context.triples}"
    )
    assert has_label, (
        f"No rdfs:label={table_name!r} triple found.\nTriples: {context.triples}"
    )


@then(r'the viewDefinition value should contain "(?P<fragment>[^"]+)"')
def step_then_view_definition_contains(context: Context, fragment: str) -> None:
    found = any(
        t.predicate == "uko-data:viewDefinition" and fragment in t.object_value
        for t in context.triples
    )
    assert found, (
        f"No uko-data:viewDefinition triple containing {fragment!r}.\n"
        f"Triples: {context.triples}"
    )


@then(
    r'the result triples should contain predicate "(?P<pred>[^"]+)" '
    r'linking schema "(?P<schema>[^"]+)" to view "(?P<view>[^"]+)"'
)
def step_then_result_contains_schema_view_link(
    context: Context, pred: str, schema: str, view: str
) -> None:
    found = any(
        t.predicate == pred and schema in t.subject_uri and view in t.object_uri
        for t in context.triples
    )
    assert found, (
        f"No triple with predicate={pred!r} linking schema containing "
        f"{schema!r} to view containing {view!r}.\n"
        f"Triples: {context.triples}"
    )


# Reset step matcher to parse (default) so subsequent step files are not affected
use_step_matcher("parse")