146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
from pathlib import Path
|
|
|
|
from lai_automation.models import FailureReport
|
|
from lai_automation.removal import (
|
|
detect_already_fixed,
|
|
extract_reference_examples,
|
|
find_matching_reference,
|
|
is_removal_only,
|
|
removal_directives,
|
|
remove_sql_fragment,
|
|
sql_contains_fragment,
|
|
)
|
|
|
|
|
|
def removal_report(**overrides) -> FailureReport:
|
|
values = {
|
|
"id": 5,
|
|
"chat_message_id": 1,
|
|
"source": "user_feedback",
|
|
"category": "incorrect_answer",
|
|
"user_feedback": "Die Anzahl der Lieferscheine ist falsch",
|
|
"feedback_type": "wrong_result",
|
|
"expected_behavior": (
|
|
'Es muss " a.bStatus_gebucht = 1 " entfernt werden! '
|
|
"Lieferscheine werden nicht gebucht! Einfach nur das aus der SQL entfernen"
|
|
),
|
|
"prompt": "Gib mir eine Anzahl von Lieferscheinen pro Monat in 2024",
|
|
"response": "0 Ergebnisse",
|
|
"sql_query": (
|
|
"SELECT COUNT(*) FROM FK_Auftrag a WHERE a.AuftragsKennung = 2 "
|
|
"AND a.bStatus_gebucht = 1 AND COALESCE(a.bStatus_storniert, 0) = 0"
|
|
),
|
|
"error_message": None,
|
|
"context": {"database_type": "erp"},
|
|
}
|
|
values.update(overrides)
|
|
return FailureReport(**values)
|
|
|
|
|
|
def semantics(sql: str, question: str | None = None) -> str:
|
|
question = question or "Gib mir eine Anzahl von Lieferscheinen pro Monat in 2024"
|
|
return f"""<?php
|
|
|
|
return [
|
|
'examples' => [
|
|
[
|
|
'question' => '{question}',
|
|
'sql' => <<<'SQL'
|
|
{sql}
|
|
SQL,
|
|
],
|
|
],
|
|
];
|
|
"""
|
|
|
|
|
|
def test_extracts_exact_quoted_german_removal_contract() -> None:
|
|
directives = removal_directives(removal_report())
|
|
|
|
assert tuple(item.fragment for item in directives) == ("a.bStatus_gebucht = 1",)
|
|
assert is_removal_only(removal_report())
|
|
|
|
|
|
def test_infers_unquoted_predicate_from_observed_sql() -> None:
|
|
report = removal_report(
|
|
expected_behavior="Bitte den bStatus_gebucht Filter aus der SQL entfernen",
|
|
)
|
|
|
|
directives = removal_directives(report)
|
|
|
|
assert len(directives) == 1
|
|
assert sql_contains_fragment(directives[0].fragment, "a.bStatus_gebucht = 1")
|
|
|
|
|
|
def test_does_not_invert_do_not_remove_instruction() -> None:
|
|
report = removal_report(
|
|
expected_behavior="Do not remove `a.bStatus_gebucht = 1` from invoice SQL",
|
|
)
|
|
|
|
assert removal_directives(report) == ()
|
|
assert not is_removal_only(report)
|
|
|
|
|
|
def test_removes_only_the_requested_conjunct() -> None:
|
|
sql = (
|
|
"SELECT * FROM FK_Auftrag a WHERE a.AuftragsKennung=2 "
|
|
"AND a.bStatus_gebucht=1 AND COALESCE(a.obsolet,0)=0 GROUP BY a.SheetNr;"
|
|
)
|
|
|
|
repaired = remove_sql_fragment(sql, "a.bStatus_gebucht = 1")
|
|
|
|
assert not sql_contains_fragment(repaired, "a.bStatus_gebucht = 1")
|
|
assert "a.AuftragsKennung=2" in repaired
|
|
assert "COALESCE(a.obsolet,0)=0" in repaired
|
|
assert "GROUP BY a.SheetNr" in repaired
|
|
|
|
|
|
def test_extracts_and_uniquely_matches_reference_example() -> None:
|
|
content = semantics("SELECT 1") + semantics(
|
|
"SELECT 2",
|
|
"Welcher Kunde hatte im Januar den meisten Umsatz?",
|
|
)
|
|
|
|
examples = extract_reference_examples(content)
|
|
match, score = find_matching_reference(content, removal_report().prompt)
|
|
|
|
assert len(examples) == 2
|
|
assert match is not None
|
|
assert match.sql == "SELECT 1"
|
|
assert score == 1.0
|
|
|
|
|
|
def test_detects_already_fixed_only_with_matching_reference(tmp_path: Path) -> None:
|
|
source = tmp_path / "config/erp_semantics.php"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text(semantics(
|
|
"SELECT COUNT(*) FROM FK_Auftrag a WHERE a.AuftragsKennung = 2 "
|
|
"AND COALESCE(a.bStatus_storniert, 0) = 0;"
|
|
))
|
|
|
|
result = detect_already_fixed(
|
|
removal_report(),
|
|
tmp_path,
|
|
("config/erp_semantics.php", "tests/Unit/SQLGeneratorBusinessContextTest.php"),
|
|
)
|
|
|
|
assert result is not None
|
|
assert result.filename == "config/erp_semantics.php"
|
|
assert result.match_score == 1.0
|
|
assert result.forbidden_fragments == ("a.bStatus_gebucht = 1",)
|
|
|
|
|
|
def test_does_not_mark_additive_or_mismatched_report_already_fixed(tmp_path: Path) -> None:
|
|
source = tmp_path / "config/erp_semantics.php"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text(semantics("SELECT COUNT(*) FROM FK_Auftrag a"))
|
|
additive = removal_report(
|
|
expected_behavior=(
|
|
'"a.bStatus_gebucht = 1" entfernen und zusätzlich System_created verwenden'
|
|
),
|
|
)
|
|
mismatched = removal_report(prompt="Zeige Rechnungen im Jahr 2024")
|
|
|
|
assert detect_already_fixed(additive, tmp_path, ("config/erp_semantics.php",)) is None
|
|
assert detect_already_fixed(mismatched, tmp_path, ("config/erp_semantics.php",)) is None
|