lai_automatisierung/tests/test_prompt.py

from lai_automation.models import FailureReport
from lai_automation.prompt import build_retry_task, build_task, redact


def test_redact_removes_common_secrets_and_personal_markers() -> None:
    value = "password=hunter2 token: abc123 admin@example.com 10.2.3.4 Bearer secret.jwt"
    result = redact(value)
    assert "hunter2" not in result
    assert "abc123" not in result
    assert "admin@example.com" not in result
    assert "10.2.3.4" not in result
    assert "secret.jwt" not in result


def test_task_contains_constraints_and_redacted_report() -> None:
    report = FailureReport(
        id=7,
        chat_message_id=91,
        source="user_feedback",
        category="incorrect_answer",
        user_feedback="password=secret-value",
        feedback_type="wrong_join",
        expected_behavior="Join FK_Artikel with FK_AuftragPos and filter Auftragkennung = 3.",
        prompt="List locations for admin@example.com",
        response="The wrong product count was returned.",
        sql_query="SELECT * FROM company",
        error_message=None,
        context={"model": "mistral", "database_type": "customer"},
    )
    task = build_task(report)
    assert "failure report #7" in task
    assert "secret-value" not in task
    assert "admin@example.com" not in task
    assert "Do not commit or push" in task
    assert "never mock or stub the behavior being tested" in task
    assert "Repository ownership map:" in task
    assert "config/erp_semantics.php owns ERP joins" in task
    assert "Tests must contain assertions that execute" in task
    assert "Do not expose protected production methods publicly" in task
    assert "Feedback type: wrong_join" in task
    assert "Join FK_Artikel with FK_AuftragPos and filter Auftragkennung = 3." in task
    assert "The wrong product count was returned." in task


def test_redact_removes_extended_credentials() -> None:
    private_key = "-----BEGIN PRIVATE KEY-----\nprivate-material\n-----END PRIVATE KEY-----"
    value = (
        "AKIAIOSFODNN7EXAMPLE "
        "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0In0.signature "
        "mysql://admin:database-password@db.internal/lai "
        f"{private_key}"
    )
    result = redact(value)
    assert "AKIAIOSFODNN7EXAMPLE" not in result
    assert "eyJhbGci" not in result
    assert "database-password" not in result
    assert "private-material" not in result


def test_retry_task_contains_failure_without_secrets() -> None:
    report = FailureReport(
        id=8,
        chat_message_id=1,
        source="user_feedback",
        category="incorrect_answer",
        user_feedback="Wrong result",
        feedback_type="wrong_result",
        expected_behavior="Use a weighted average",
        prompt="Average sale price",
        response="Zero",
        sql_query="SELECT AVG(list_price)",
        error_message=None,
        context={"database_type": "erp"},
    )
    task = build_retry_task(report, RuntimeError("Missing import; token=secret-value"))

    assert "Missing import" in task
    assert "secret-value" not in task
    assert "Do not remove or weaken tests" in task
    assert "unrelated production routing" in task
    assert "Use a weighted average" in task
    assert "SELECT AVG(list_price)" in task


def test_incorrect_answer_task_guides_semantics_level_repair() -> None:
    report = FailureReport(
        id=42,
        chat_message_id=1,
        source="user_feedback",
        category="incorrect_answer",
        user_feedback="Wrong result",
        feedback_type="wrong_result",
        expected_behavior="Sum net values and divide by quantity",
        prompt="Average sale price for article 9031",
        response="0.00",
        sql_query="SELECT AVG(Vk_preis) FROM FK_Artikel",
        error_message=None,
        context={"database_type": "erp"},
    )

    task = build_task(report)

    assert "config/*_semantics.php" in task
    assert "Do not hardcode" in task
    assert "protected schema/business-" in task
    assert "context builder" in task
    assert "known-bad SQL" in task
    assert "compare every alias.column" in task
    assert "Never replace one missing column" in task


def test_retry_task_keeps_failure_from_start_of_long_output() -> None:
    report = FailureReport(
        id=43,
        chat_message_id=1,
        source="automatic",
        category="sql_execution",
        user_feedback=None,
        feedback_type=None,
        expected_behavior=None,
        prompt="Top products",
        response=None,
        sql_query="SELECT a.missing FROM orders a",
        error_message="Unknown column",
        context={"database_type": "erp"},
    )

    task = build_retry_task(report, RuntimeError("DECISIVE ASSERTION\n" + "x" * 11_000))

    assert "DECISIVE ASSERTION" in task


def test_customer_database_task_explains_tenant_scope() -> None:
    report = FailureReport(
        id=13,
        chat_message_id=1,
        source="automatic",
        category="sql_execution",
        user_feedback=None,
        feedback_type=None,
        expected_behavior=None,
        prompt="What did this customer buy most?",
        response=None,
        sql_query="SELECT * FROM invoice_row WHERE InvoiceCreator = 'tenant'",
        error_message="Unknown column InvoiceCreator",
        context={"database_type": "customer"},
    )

    task = build_task(report)

    assert "already selected one tenant database" in task
    assert "must not" in task
    assert "guessed column" in task
    assert "tenant_type" in task


def test_unresolved_placeholder_task_guides_real_answer_path() -> None:
    report = FailureReport(
        id=49,
        chat_message_id=1,
        source="user_feedback",
        category="incorrect_answer",
        user_feedback="Token was not restored",
        feedback_type="wrong_result",
        expected_behavior="Return the database value",
        prompt="Average price",
        response="Price: __LAI_UNKNOWN_VALUE__",
        sql_query="SELECT AVG(price)",
        error_message=None,
        context={"database_type": "erp"},
    )

    task = build_task(report)

    assert "AIService::generateDatabaseResponse" in task
    assert "exact token map" in task
    assert "HTTP-backed AIService test" in task
    assert "local database-result fallback" in task