cleveragents-core/features/steps/postgresql_helpers_coverage_steps.py

"""Step definitions for postgresql_helpers_coverage.feature.

These steps target specific uncovered lines in _postgresql_helpers.py:
- Line 50: ident_pair with quoted identifier containing escaped double-quotes
- Line 192: _skip_quoted with escaped quotes inside strings (j += 2)
- Line 197: _skip_quoted with unclosed quote (return n)
- Lines 198-204: _skip_quoted with dollar-quoting ($$ ... $$)
- Line 215: extract_body with invalid paren_start
- Line 230: extract_body with unbalanced parentheses
- Lines 384-385, 387-388: extract_fk_triples column count mismatch warning
- Line 434: parse_table_body empty entry continue
"""

import logging

from behave import given, then, when

from cleveragents.domain.models.acms._postgresql_helpers import (
    CREATE_TABLE_RE,
    FOREIGN_KEY_RE,
    _skip_quoted,
    extract_body,
    extract_fk_triples,
    ident_pair,
    parse_table_body,
    split_entries,
    table_uri,
)

# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------


@given("the postgresql helpers module is imported")
def step_module_imported(context):
    """Ensure the module is importable."""
    assert extract_body is not None
    assert split_entries is not None
    assert ident_pair is not None


# ---------------------------------------------------------------------------
# ident_pair: quoted identifier with escaped double-quotes (line 50)
# ---------------------------------------------------------------------------


@given("a regex match with a quoted identifier containing escaped double-quotes")
def step_quoted_ident_match(context):
    """Create a regex match where the quoted branch has double-double-quotes.

    We use CREATE_TABLE_RE against a CREATE TABLE statement with a quoted
    schema name containing escaped double-quotes: "my""schema"."table"
    """
    # _IDENT produces two groups each: (quoted, unquoted)
    # CREATE_TABLE_RE groups: 1,2 = schema; 3,4 = table
    sql = 'CREATE TABLE "my""schema"."table" ('
    m = CREATE_TABLE_RE.search(sql)
    assert m is not None, f"CREATE_TABLE_RE did not match: {sql!r}"
    context.ident_match = m


@when("I call ident_pair on the quoted match")
def step_call_ident_pair_quoted(context):
    """Call ident_pair targeting the quoted schema groups."""
    # Groups 1,2 are the schema pair in CREATE_TABLE_RE
    context.ident_result = ident_pair(context.ident_match, 1, 2)


@then("the result should have double-quotes unescaped")
def step_verify_unescaped(context):
    """Verify that '""' was replaced with '"'."""
    assert context.ident_result == 'my"schema', (
        f"Expected 'my\"schema', got {context.ident_result!r}"
    )


# ---------------------------------------------------------------------------
# _skip_quoted: escaped single-quotes (line 192)
# ---------------------------------------------------------------------------


@given("SQL content with parenthesized body containing escaped single-quotes")
def step_content_escaped_single_quotes(context):
    """Build content with a string containing '' (escaped single-quote).

    E.g.: (col1 TEXT DEFAULT 'it''s fine', col2 INT)
    The '' inside the string should be skipped by _skip_quoted.
    """
    context.paren_content = "(col1 TEXT DEFAULT 'it''s fine', col2 INT)"
    context.paren_start = 0


@when("I call extract_body on the content")
def step_call_extract_body(context):
    """Call extract_body with the content."""
    context.extracted_body = extract_body(context.paren_content, context.paren_start)


@then("the body should be extracted correctly with the quoted string intact")
def step_verify_body_with_quotes(context):
    """Verify the body was extracted including the full quoted string."""
    assert "col1 TEXT DEFAULT 'it''s fine'" in context.extracted_body, (
        f"Body missing escaped-quote string: {context.extracted_body!r}"
    )
    assert "col2 INT" in context.extracted_body


@given("a table body with escaped double-quotes in an identifier")
def step_body_with_escaped_double_quotes(context):
    """Build a body containing a double-quoted identifier with "".

    E.g.: "col""name" TEXT, col2 INT
    """
    context.split_body = '"col""name" TEXT, col2 INT'


@when("I call split_entries on the body")
def step_call_split_entries(context):
    """Call split_entries."""
    context.split_result = split_entries(context.split_body)


@then("the entries should be split correctly preserving the identifier")
def step_verify_split_with_double_quotes(context):
    """Verify entries are split at depth-0 commas only."""
    assert len(context.split_result) == 2, (
        f"Expected 2 entries, got {len(context.split_result)}: {context.split_result}"
    )
    assert '"col""name" TEXT' in context.split_result[0]
    assert "col2 INT" in context.split_result[1].strip()


# ---------------------------------------------------------------------------
# _skip_quoted: unclosed quote (line 197)
# ---------------------------------------------------------------------------


@given("SQL content with an unclosed single-quote inside parentheses")
def step_content_unclosed_quote(context):
    """Build content with an unclosed single-quote.

    The _skip_quoted function should advance to end-of-string (return n).
    Because the quote is never closed, the parenthesis balance is broken,
    so extract_body returns "".
    """
    context.unclosed_content = "(col1 TEXT DEFAULT 'unclosed"
    context.unclosed_start = 0


@when("I call extract_body on the unclosed-quote content")
def step_call_extract_body_unclosed(context):
    """Call extract_body on content with unclosed quote."""
    context.unclosed_result = extract_body(
        context.unclosed_content, context.unclosed_start
    )


@then("extract_body should return an empty string for unbalanced parens")
def step_verify_unclosed_quote_empty(context):
    """When a quote is never closed the body can't be balanced."""
    assert context.unclosed_result == "", (
        f"Expected empty string, got {context.unclosed_result!r}"
    )


# ---------------------------------------------------------------------------
# _skip_quoted: dollar-quoting (lines 198-203)
# ---------------------------------------------------------------------------


@given("SQL content with a dollar-quoted string inside parentheses")
def step_content_dollar_quoted(context):
    """Build content with a $$...$$ dollar-quoted string.

    E.g.: (col1 TEXT DEFAULT $$hello, world$$, col2 INT)
    """
    context.dollar_content = "(col1 TEXT DEFAULT $$hello, world$$, col2 INT)"
    context.dollar_start = 0


@when("I call extract_body on the dollar-quoted content")
def step_call_extract_body_dollar(context):
    """Call extract_body."""
    context.dollar_result = extract_body(context.dollar_content, context.dollar_start)


@then("the body should be extracted correctly ignoring dollar-quoted content")
def step_verify_dollar_body(context):
    """The body should contain both columns."""
    assert "col1 TEXT" in context.dollar_result
    assert "col2 INT" in context.dollar_result


@given("a table body containing a dollar-quoted string with commas")
def step_dollar_body_with_commas(context):
    """A body where a dollar-quoted string contains commas that should be ignored."""
    context.dollar_split_body = "col1 TEXT DEFAULT $$a,b,c$$, col2 INT"


@when("I call split_entries on the dollar-quoted body")
def step_call_split_entries_dollar(context):
    """Call split_entries."""
    context.dollar_split_result = split_entries(context.dollar_split_body)


@then("the comma inside the dollar-quoted string should not split entries")
def step_verify_dollar_split(context):
    """Only 2 entries — the commas inside $$ should be ignored."""
    assert len(context.dollar_split_result) == 2, (
        f"Expected 2 entries, got {len(context.dollar_split_result)}: "
        f"{context.dollar_split_result}"
    )


@given("SQL content with an unclosed dollar-quoted string")
def step_content_unclosed_dollar(context):
    """Build content with $$ that is never closed.

    This exercises the `else n` branch at line 203.
    """
    context.unclosed_dollar_content = "(col1 TEXT DEFAULT $$unclosed body"
    context.unclosed_dollar_start = 0


@when("I call extract_body on the unclosed dollar-quote content")
def step_call_extract_body_unclosed_dollar(context):
    """Call extract_body."""
    context.unclosed_dollar_result = extract_body(
        context.unclosed_dollar_content, context.unclosed_dollar_start
    )


@then("extract_body should handle the unclosed dollar-quote gracefully")
def step_verify_unclosed_dollar(context):
    """With unclosed dollar-quote, parens can't balance → empty string."""
    assert context.unclosed_dollar_result == "", (
        f"Expected empty string, got {context.unclosed_dollar_result!r}"
    )


# ---------------------------------------------------------------------------
# _skip_quoted: lone dollar sign (line 204)
# ---------------------------------------------------------------------------


@given("SQL content with a lone dollar sign that is not a dollar-quote tag")
def step_content_lone_dollar(context):
    """Build content with a $ that does not form a $$...$$ tag.

    A lone '$' at a position where _DOLLAR_RE.match fails exercises line 204
    (return i + 1).  We need the $ to appear in a position where it triggers
    the dollar-check but does not match the pattern $tag$.

    _DOLLAR_RE matches $tag$ patterns. A lone "$" followed by a non-word
    char (like a space) won't match. The check in extract_body is:
        ch == "$" and _DOLLAR_RE.match(content, i)
    If _DOLLAR_RE.match fails, the "and" short-circuits and the "$" is NOT
    sent to _skip_quoted. Line 204 is the fallback "return i+1" inside
    _skip_quoted when ch == "$" but the dollar-regex doesn't match at
    that position.

    We call _skip_quoted directly with a "$" where the pattern fails.
    """
    # We'll test _skip_quoted directly for line 204
    context.lone_dollar_text = "$ not a tag"
    context.lone_dollar_pos = 0


@when("I call extract_body on the lone-dollar content")
def step_call_skip_quoted_lone_dollar(context):
    """Call _skip_quoted directly to exercise line 204."""
    context.lone_dollar_result = _skip_quoted(
        context.lone_dollar_text, context.lone_dollar_pos
    )


@then("the body should be extracted correctly ignoring the lone dollar")
def step_verify_lone_dollar(context):
    """_skip_quoted should return i+1 when $ doesn't form a dollar-quote."""
    # For "$ not a tag" at position 0:
    # ch == "$", _DOLLAR_RE.match("$ not a tag", 0) tries to match \$(\w*)\$
    # at position 0 — "$ " does not have a second $, so no match → return 0+1=1
    assert context.lone_dollar_result == 1, (
        f"Expected 1, got {context.lone_dollar_result}"
    )


# ---------------------------------------------------------------------------
# extract_body: invalid paren_start (line 215)
# ---------------------------------------------------------------------------


@when("I call extract_body with paren_start beyond content length")
def step_extract_body_beyond_length(context):
    """Call extract_body with paren_start >= len(content)."""
    context.beyond_result = extract_body("(abc)", 999)


@then("extract_body should return an empty string for invalid start")
def step_verify_empty_beyond(context):
    """Line 215 returns "" for out-of-range paren_start."""
    assert context.beyond_result == "", (
        f"Expected empty string, got {context.beyond_result!r}"
    )


@when("I call extract_body with paren_start pointing at a non-paren character")
def step_extract_body_non_paren(context):
    """Call extract_body where content[paren_start] != '('."""
    context.non_paren_result = extract_body("abc(def)", 0)


@then("extract_body should return an empty string for non-paren start")
def step_verify_empty_non_paren(context):
    """Line 215 returns "" when content[paren_start] is not '('."""
    assert context.non_paren_result == "", (
        f"Expected empty string, got {context.non_paren_result!r}"
    )


# ---------------------------------------------------------------------------
# extract_body: unbalanced parens (line 230)
# ---------------------------------------------------------------------------


@given("SQL content with an opening paren but no closing paren")
def step_content_unbalanced(context):
    """Content with an opening paren that is never closed."""
    context.unbalanced_content = "(col1 TEXT, col2 INT"
    context.unbalanced_start = 0


@when("I call extract_body on the unbalanced content")
def step_call_extract_body_unbalanced(context):
    """Call extract_body."""
    context.unbalanced_result = extract_body(
        context.unbalanced_content, context.unbalanced_start
    )


@then("extract_body should return an empty string for missing close paren")
def step_verify_unbalanced_empty(context):
    """Line 230: loop exits without finding close paren → return ""."""
    assert context.unbalanced_result == "", (
        f"Expected empty string, got {context.unbalanced_result!r}"
    )


# ---------------------------------------------------------------------------
# extract_fk_triples: column count mismatch (lines 383-388)
# ---------------------------------------------------------------------------


@given("a FOREIGN KEY clause with mismatched source and target column counts")
def step_fk_mismatch(context):
    """Build a FOREIGN KEY clause where source has 2 cols but target has 1.

    FOREIGN KEY (col_a, col_b) REFERENCES other_table (col_x)
    """
    fk_sql = "FOREIGN KEY (col_a, col_b) REFERENCES other_table (col_x)"
    m = FOREIGN_KEY_RE.search(fk_sql)
    assert m is not None, f"FOREIGN_KEY_RE did not match: {fk_sql!r}"
    context.fk_mismatch_match = m
    context.fk_resource_uri = "test-resource"
    context.fk_table_name = "my_table"
    context.fk_table_uri = table_uri("test-resource", "my_table")


@when("I extract FK triples from the mismatched clause")
def step_extract_fk_mismatch(context):
    """Call extract_fk_triples and capture log output."""
    context.fk_log_records = []

    class ListHandler(logging.Handler):
        def __init__(self, records_list):
            super().__init__()
            self._records = records_list

        def emit(self, record):
            self._records.append(record)

    _handler = ListHandler(context.fk_log_records)
    # Attach to the module's logger
    pg_logger = logging.getLogger("cleveragents.domain.models.acms._postgresql_helpers")
    # Re-enable the logger in case Alembic's fileConfig() disabled it
    pg_logger.disabled = False
    pg_logger.addHandler(_handler)
    pg_logger.setLevel(logging.DEBUG)

    try:
        context.fk_mismatch_triples = extract_fk_triples(
            context.fk_mismatch_match,
            context.fk_resource_uri,
            context.fk_table_name,
            context.fk_table_uri,
        )
    finally:
        pg_logger.removeHandler(_handler)


@then("a warning should be logged about the column count mismatch")
def step_verify_fk_warning(context):
    """Lines 384-385, 387-388: warning about mismatched column counts."""
    warnings = [r for r in context.fk_log_records if r.levelno == logging.WARNING]
    assert len(warnings) >= 1, (
        f"Expected a WARNING log, got {len(warnings)} warnings; "
        f"all records: {[r.getMessage() for r in context.fk_log_records]}"
    )
    msg = warnings[0].getMessage()
    assert "mismatch" in msg.lower(), f"Unexpected warning message: {msg}"


@then("triples should be generated only for the shorter column list")
def step_verify_fk_paired(context):
    """Only 1 foreignKeyTo triple (min of 2 source, 1 target)."""
    fk_to_triples = [
        t for t in context.fk_mismatch_triples if t.predicate == "uko-data:foreignKeyTo"
    ]
    assert len(fk_to_triples) == 1, (
        f"Expected 1 foreignKeyTo triple (min pairing), got {len(fk_to_triples)}"
    )


# ---------------------------------------------------------------------------
# parse_table_body: empty entry (line 434)
# ---------------------------------------------------------------------------


@given("a table body string with empty entries between commas")
def step_body_empty_entries(context):
    """A body like '  , col1 INT,  , col2 TEXT, '  — has empty entries."""
    context.empty_entries_body = "  , col1 INT,  , col2 TEXT, "
    context.empty_entries_resource = "test-resource"
    context.empty_entries_table = "my_table"
    context.empty_entries_table_uri = table_uri("test-resource", "my_table")


@when("I call parse_table_body on the body with empty entries")
def step_call_parse_table_body_empty(context):
    """Call parse_table_body."""
    context.empty_triples = parse_table_body(
        context.empty_entries_body,
        context.empty_entries_resource,
        context.empty_entries_table,
        context.empty_entries_table_uri,
    )


@then("parse_table_body should produce triples only for valid column definitions")
def step_verify_empty_entries_triples(context):
    """Only 2 real columns (col1, col2) should produce triples; empties are skipped."""
    label_triples = [t for t in context.empty_triples if t.predicate == "rdfs:label"]
    assert len(label_triples) == 2, (
        f"Expected 2 column labels, got {len(label_triples)}: "
        f"{[t.object_value for t in label_triples]}"
    )
    names = {t.object_value for t in label_triples}
    assert names == {"col1", "col2"}, f"Unexpected column names: {names}"


# ---------------------------------------------------------------------------
# FK with quoted schema containing escaped double-quotes (line 50 via ident_pair)
# ---------------------------------------------------------------------------


@given("a FOREIGN KEY clause referencing a quoted schema with escaped double-quotes")
def step_fk_quoted_schema(context):
    """Build: FOREIGN KEY (fk_col) REFERENCES "my""schema"."ref_table" (id)"""
    fk_sql = 'FOREIGN KEY (fk_col) REFERENCES "my""schema"."ref_table" (id)'
    m = FOREIGN_KEY_RE.search(fk_sql)
    assert m is not None, f"FOREIGN_KEY_RE did not match: {fk_sql!r}"
    context.fk_quoted_match = m
    context.fk_quoted_resource = "test-resource"
    context.fk_quoted_table = "source_table"
    context.fk_quoted_table_uri = table_uri("test-resource", "source_table")


@when("I extract FK triples from the quoted-schema clause")
def step_extract_fk_quoted_schema(context):
    """Call extract_fk_triples with the quoted-schema FK match."""
    context.fk_quoted_triples = extract_fk_triples(
        context.fk_quoted_match,
        context.fk_quoted_resource,
        context.fk_quoted_table,
        context.fk_quoted_table_uri,
    )


@then("the triples should reference the correctly unescaped schema name")
def step_verify_fk_quoted_schema_uri(context):
    """The target column URI should contain the unescaped schema name."""
    fk_to_triples = [
        t for t in context.fk_quoted_triples if t.predicate == "uko-data:foreignKeyTo"
    ]
    assert len(fk_to_triples) == 1
    target_uri = fk_to_triples[0].object_uri
    # The schema "my""schema" should be unescaped to my"schema,
    # then safe_name converts " to _, so it becomes my_schema
    assert "my_schema" in target_uri, f"Expected 'my_schema' in URI, got {target_uri!r}"