import os
import subprocess
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Final

import pytest

GATE: Final = Path(__file__).resolve().parents[2] / ".github/e2e-stack/assert_tests_ran.py"
SECRETS_TO_ENV: Final = GATE.with_name("secrets_to_env.py")
SELECT_TESTS: Final = GATE.with_name("select_tests.py")
REDACT_OUTPUT: Final = GATE.with_name("redact_output.py")
CANARY: Final = ("tests/e2e/access_control/test_a.py", "tests/e2e/access_control/test_b.py")
SELECTED: Final = ("tests/e2e/access_control/test_a.py", "tests/e2e/access_control/test_b.py")


@pytest.mark.parametrize(
    ("second_outcome", "expected_status"),
    (("passed", 0), ("skipped", 1), ("failure", 1), ("error", 1), ("deselected", 1)),
)
def test_each_changed_file_must_run(tmp_path: Path, second_outcome: str, expected_status: int) -> None:
    suite: Final = ET.Element("testsuite")
    _ = ET.SubElement(suite, "testcase", file=SELECTED[0])
    if second_outcome != "deselected":
        second: Final = ET.SubElement(suite, "testcase", file=SELECTED[1])
        if second_outcome != "passed":
            _ = ET.SubElement(second, second_outcome)
    report: Final = tmp_path / "report.xml"
    ET.ElementTree(suite).write(report)

    result: Final = subprocess.run([sys.executable, str(GATE), str(report), *SELECTED], capture_output=True, text=True)

    assert result.returncode == expected_status, result.stdout


@pytest.mark.parametrize("outcome", ("failure", "error"))
def test_passing_case_does_not_hide_a_failure_in_the_same_file(tmp_path: Path, outcome: str) -> None:
    suite: Final = ET.Element("testsuite")
    _ = ET.SubElement(suite, "testcase", file=SELECTED[0])
    failed: Final = ET.SubElement(suite, "testcase", file=SELECTED[0])
    _ = ET.SubElement(failed, outcome)
    report: Final = tmp_path / "report.xml"
    ET.ElementTree(suite).write(report)

    result: Final = subprocess.run(
        [sys.executable, str(GATE), str(report), SELECTED[0]], capture_output=True, text=True
    )

    assert result.returncode == 1


def test_failed_cases_are_named_per_selected_file(tmp_path: Path) -> None:
    suite: Final = ET.Element("testsuite")
    _ = ET.SubElement(suite, "testcase", file=SELECTED[0], classname="tests.e2e.access_control.test_a", name="test_ok")
    failed: Final = ET.SubElement(
        suite, "testcase", file=SELECTED[0], classname="tests.e2e.access_control.test_a", name="test_boom"
    )
    _ = ET.SubElement(failed, "failure", message="secret-bearing message")
    errored: Final = ET.SubElement(
        suite, "testcase", file=SELECTED[1], classname="tests.e2e.access_control.test_b", name="test_setup"
    )
    _ = ET.SubElement(errored, "error")
    report: Final = tmp_path / "report.xml"
    ET.ElementTree(suite).write(report)

    result: Final = subprocess.run([sys.executable, str(GATE), str(report), *SELECTED], capture_output=True, text=True)

    assert result.returncode == 1
    assert "  failed: tests.e2e.access_control.test_a::test_boom\n" in result.stdout
    assert "  failed: tests.e2e.access_control.test_b::test_setup\n" in result.stdout
    assert "test_ok" not in result.stdout
    assert "secret-bearing message" not in result.stdout


@pytest.mark.parametrize("contents", ("<testsuite/>", "<testsuite", '<testsuite><testcase name="a"/></testsuite>'))
def test_missing_execution_evidence_fails(tmp_path: Path, contents: str) -> None:
    report: Final = tmp_path / "report.xml"
    _ = report.write_text(contents)

    result: Final = subprocess.run([sys.executable, str(GATE), str(report), *SELECTED], capture_output=True, text=True)

    assert result.returncode == 1


@pytest.mark.parametrize("omitted_role", ("proxy_admin", "team_member", "internal_user_viewer"))
def test_one_passing_management_case_cannot_hide_a_missing_actor(tmp_path: Path, omitted_role: str) -> None:
    suite: Final = ET.Element("testsuite")
    path: Final = "tests/e2e/management/test_jwt_management_e2e.py"
    case: Final = ET.SubElement(suite, "testcase", file=path)
    properties: Final = ET.SubElement(case, "properties")
    _ = ET.SubElement(
        properties,
        "property",
        name="management_node",
        value=f"{path}::TestJwtManagement::test_actor_subject_and_database_role[proxy_admin_viewer]",
    )
    report: Final = tmp_path / "report.xml"
    ET.ElementTree(suite).write(report)
    result: Final = subprocess.run([sys.executable, str(GATE), str(report), path], capture_output=True, text=True)
    assert result.returncode == 1
    assert f"test_actor_subject_and_database_role[{omitted_role}]" in result.stdout


def test_short_values_are_written_without_masking_every_digit_in_the_log(tmp_path: Path) -> None:
    env_path: Final = tmp_path / ".env"

    result: Final = subprocess.run(
        [sys.executable, str(SECRETS_TO_ENV), str(env_path)],
        input='{"FLAG": "1", "API_KEY": "sk-0123456789abcdef"}',
        capture_output=True,
        text=True,
    )

    assert result.returncode == 0, result.stderr
    assert result.stdout == "::add-mask::sk-0123456789abcdef\n"
    assert env_path.read_text() == "FLAG='1'\nAPI_KEY='sk-0123456789abcdef'\n"


def redact_output(tmp_path: Path, values: tuple[str, ...], text: str) -> tuple[subprocess.CompletedProcess[str], Path]:
    env_path: Final = tmp_path / ".env"
    _ = env_path.write_text("".join(f"{name}='{value}'\n" for name, value in zip(("A", "B", "C"), values)))
    stack_env: Final = tmp_path / "stack.env"
    _ = stack_env.write_text("LITELLM_MASTER_KEY=sk-e2e-master0123\nREDIS_PORT=6379\n")
    log: Final = tmp_path / "e2e-pass-1.log"
    _ = log.write_text(text)
    out_dir: Final = tmp_path / "redacted"
    result: Final = subprocess.run(  # test-quality-ok: standalone script that imports its sibling by script directory
        [
            sys.executable,
            str(REDACT_OUTPUT),
            "--values",
            str(env_path),
            "--values",
            str(stack_env),
            "--out",
            str(out_dir),
            str(log),
        ],
        capture_output=True,
        text=True,
    )
    return result, out_dir / log.name


def test_redacted_output_hides_every_masked_value_and_keeps_the_rest(tmp_path: Path) -> None:
    text: Final = (
        "FAILED key=sk-0123456789abcdef master=sk-e2e-master0123 flag=1 port=6379 message=Missing credentials\n"
    )

    result, redacted = redact_output(tmp_path, ("sk-0123456789abcdef", "1"), text)

    assert result.returncode == 0, result.stderr
    assert redacted.read_text() == "FAILED key=*** master=*** flag=1 port=6379 message=Missing credentials\n"
    assert (redacted.stat().st_mode & 0o777) == 0o600
    assert (tmp_path / "e2e-pass-1.log").read_text() == text
    assert "sk-" not in result.stdout + result.stderr


def test_a_masked_value_that_prefixes_a_longer_one_leaves_no_tail(tmp_path: Path) -> None:
    result, redacted = redact_output(tmp_path, ("sk-0123456789", "sk-0123456789abcdef"), "token sk-0123456789abcdef\n")

    assert result.returncode == 0, result.stderr
    assert redacted.read_text() == "token ***\n"


def test_a_json_secret_is_hidden_field_by_field_however_it_is_escaped(tmp_path: Path) -> None:
    credentials: Final = (
        '{"type": "service_account", "signing_key": "MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\\n'
        'c2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\\n", "client_id": "104857600000000000001"}'
    )
    text: Final = (
        "decoded MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\n"
        "c2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\n"
        "escaped MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\\nc2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\\n\n"
        "twice MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\\\\nc2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\n"
        "client 104857600000000000001 status 403\n"
    )

    result, redacted = redact_output(tmp_path, (credentials,), text)

    assert result.returncode == 0, result.stderr
    assert redacted.read_text() == "decoded ***\n***\nescaped ***\\n***\\n\ntwice ***\\\\n***\nclient *** status 403\n"


def test_a_secret_with_xml_special_characters_is_hidden_in_the_junit_file(tmp_path: Path) -> None:
    text: Final = '<failure message="got p&amp;ss&lt;w&quot;rd-1">body p&amp;ss&lt;w"rd-1</failure>\n'

    result, redacted = redact_output(tmp_path, ('p&ss<w"rd-1',), text)

    assert result.returncode == 0, result.stderr
    assert redacted.read_text() == '<failure message="got ***">body ***</failure>\n'


def select_tests(changed: tuple[str, ...]) -> tuple[str, ...]:
    result: Final = subprocess.run(
        [sys.executable, str(SELECT_TESTS), *CANARY],
        input="".join(f"{path}\n" for path in changed),
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0, result.stderr
    return tuple(result.stdout.split())


@pytest.mark.parametrize(
    ("changed", "expected"),
    (
        (("tests/e2e/logging/test_datadog_e2e.py", "litellm/router.py"), ("tests/e2e/logging/test_datadog_e2e.py",)),
        (("tests/e2e/ui/test_keys.py", "tests/e2e/claude_code/test_cli.py", "tests/e2e/load/test_burst.py"), ()),
        (("tests/e2e/migrations/test_startup.py", "tests/e2e/migrations/test_recovery.py"), ()),
        (("tests/e2e/batches/test_managed_files_enforcement_e2e.py",), ()),
        (("tests/e2e/guardrails/test_presidio_masking_e2e.py",), ()),
        (("tests/e2e/llm_translation/realtime/test_realtime_pipecat_audio_e2e.py",), ()),
        (
            ("tests/e2e/llm_translation/realtime/test_realtime_e2e.py",),
            ("tests/e2e/llm_translation/realtime/test_realtime_e2e.py",),
        ),
        (
            ("tests/e2e/guardrails/test_bedrock_guardrail_e2e.py",),
            ("tests/e2e/guardrails/test_bedrock_guardrail_e2e.py",),
        ),
        (("tests/e2e/logging/helpers.py", "docs/my-website/docs/index.md", "tests/e2e/AGENTS.md"), ()),
        (
            ("tests/e2e/logging/test_datadog_e2e.py", "tests/e2e/logging/test_datadog_e2e.py"),
            ("tests/e2e/logging/test_datadog_e2e.py",),
        ),
    ),
)
def test_changed_suite_files_are_selected_unless_the_stack_cannot_run_them(
    changed: tuple[str, ...], expected: tuple[str, ...]
) -> None:
    assert select_tests(changed) == expected


@pytest.mark.parametrize(
    "harness_file",
    (
        "tests/e2e/proxy_client.py",
        "tests/e2e/conftest.py",
        "tests/e2e/management/management_client.py",
        "tests/e2e/management/jwt_actors.py",
        "tests/e2e/management/conftest.py",
        "tests/e2e/coverage_registry/management_cases.py",
        "tests/e2e/pytest.ini",
        "tests/e2e/gateway/stage_mirror_ci_config.yml",
        ".github/e2e-stack/up.sh",
        ".github/e2e-stack/start-idp.sh",
        "tests/e2e/idp_realm.json",
        ".github/workflows/test-e2e-changed.yml",
    ),
)
def test_harness_changes_run_the_canary_suite(harness_file: str) -> None:
    assert select_tests((harness_file, "litellm/router.py")) == CANARY


def test_a_changed_canary_file_is_selected_once_alongside_a_harness_change() -> None:
    assert select_tests((CANARY[1], "tests/e2e/proxy_client.py")) == CANARY


def test_dedicated_migration_tests_do_not_suppress_shared_harness_canaries() -> None:
    assert select_tests(("tests/e2e/migrations/test_startup.py", "tests/e2e/conftest.py")) == CANARY


def test_the_canary_joins_directly_selected_files_in_sorted_order() -> None:
    assert select_tests(("tests/e2e/logging/test_datadog_e2e.py", ".github/e2e-stack/up.sh")) == (
        *CANARY,
        "tests/e2e/logging/test_datadog_e2e.py",
    )


def test_a_harness_unit_test_change_runs_itself_and_the_canary() -> None:
    assert select_tests(("tests/e2e/test_proxy_client.py",)) == (*CANARY, "tests/e2e/test_proxy_client.py")


def test_a_canary_argument_the_shell_never_expanded_fails_the_selector() -> None:
    result: Final = subprocess.run(
        [sys.executable, str(SELECT_TESTS), "tests/e2e/access_control/test_*.py"],
        input="tests/e2e/proxy_client.py\n",
        capture_output=True,
        text=True,
    )

    assert result.returncode == 1
    assert "tests/e2e/access_control/test_*.py" in result.stderr
    assert result.stdout == ""


@pytest.mark.parametrize(
    ("secrets", "offender", "unprintable"),
    (
        ('{"AWS_ACCESS_KEY_ID": "AKIAEXAMPLE", "BAD-NAME": "shibboleth"}', "BAD-NAME", "shibboleth"),
        ("""{"AWS_SECRET_ACCESS_KEY": "quote'shibboleth"}""", "AWS_SECRET_ACCESS_KEY", "shibboleth"),
        ('{"DD_API_KEY": "line\\nshibboleth"}', "DD_API_KEY", "shibboleth"),
    ),
)
def test_an_unusable_secret_is_named_without_printing_its_value(
    tmp_path: Path, secrets: str, offender: str, unprintable: str
) -> None:
    env_path: Final = tmp_path / ".env"

    result: Final = subprocess.run(
        [sys.executable, str(SECRETS_TO_ENV), str(env_path)], input=secrets, capture_output=True, text=True
    )

    assert result.returncode == 1
    assert offender in result.stderr
    assert unprintable not in result.stderr
    assert result.stdout == ""
    assert not env_path.exists()


@pytest.mark.parametrize("phase", ("setup", "call", "teardown"))
@pytest.mark.parametrize("required_count", ("1", "4"))
def test_oauth_failure_diagnostics_do_not_publish_private_payloads(
    tmp_path: Path, phase: str, required_count: str
) -> None:
    suite: Final = ET.Element("testsuite")
    case: Final = ET.SubElement(suite, "testcase", file=SELECTED[0])
    private: Final = "private-token-in-exception-message"
    failure: Final = ET.SubElement(case, "failure", message=private)
    failure.text = private
    properties: Final = ET.SubElement(case, "properties")
    for name, value in (
        ("oauth_failure_phase", phase),
        ("oauth_exception_type", "AssertionError"),
        ("oauth_frame", "oauth_gateway.py:120:start"),
        ("oauth_frame", f"injected\\n{private}"),
        ("unrelated_property", private),
    ):
        _ = ET.SubElement(properties, "property", name=name, value=value)
    report: Final = tmp_path / "report.xml"
    ET.ElementTree(suite).write(report)
    result: Final = subprocess.run(
        [sys.executable, "-I", str(GATE), str(report), SELECTED[0]],
        capture_output=True,
        text=True,
        env={**os.environ, "E2E_REQUIRED_TEST_COUNT": required_count},
    )
    assert result.returncode == 1
    assert f"oauth_failure_phase: {phase}" in result.stdout
    assert "oauth_exception_type: AssertionError" in result.stdout
    assert "oauth_frame: oauth_gateway.py:120:start" in result.stdout
    assert private not in result.stdout + result.stderr


@pytest.mark.parametrize(
    ("count", "skip", "expected"), ((0, False, 1), (3, False, 1), (4, False, 0), (5, False, 1), (4, True, 1))
)
def test_required_count_reports_cases_before_rejecting(tmp_path: Path, count: int, skip: bool, expected: int) -> None:
    suite = ET.Element("testsuite")
    for index in range(count):
        case = ET.SubElement(suite, "testcase", file=SELECTED[0], classname="OAuth", name=f"variant{index}")
        if skip and index == 0:
            ET.SubElement(case, "skipped", message="private-skip-reason")
    report = tmp_path / "report.xml"
    ET.ElementTree(suite).write(report)
    result = subprocess.run(
        [sys.executable, "-I", str(GATE), str(report), SELECTED[0]],
        env={**os.environ, "E2E_REQUIRED_TEST_COUNT": "4"},
        capture_output=True,
        text=True,
    )
    assert result.returncode == expected
    assert f"{count} collected, {int(skip)} skipped" in result.stdout
    if skip:
        assert "skipped: OAuth::variant0" in result.stdout
    assert "private-skip-reason" not in result.stdout + result.stderr