import io
import pathlib
import sys
import unittest
import zipfile
from typing import final

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from junit import ArtifactError, classify_reports, read_junit_zip
from model import Report


FIXTURES = pathlib.Path(__file__).parent / "fixtures"


def zipped(name: str) -> bytes:
    data = io.BytesIO()
    with zipfile.ZipFile(data, "w") as archive:
        archive.writestr("junit.xml", (FIXTURES / name).read_bytes())
    return data.getvalue()


@final
class JunitTest(unittest.TestCase):
    def parse(self, fixture: str, attempt: int) -> Report:
        return read_junit_zip(zipped(fixture), attempt)

    def test_nextest_retry_succeeds(self):
        occurrences = classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 1, [self.parse("nextest_flaky.xml", 1)])
        self.assertEqual(1, len(occurrences))
        self.assertEqual(frozenset(("nextest retry",)), occurrences[0].modes)

    def test_nextest_retries_never_recover(self):
        self.assertEqual([], classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 1, [self.parse("nextest_failed.xml", 1)]))

    def test_failure_with_flaky_failure_still_classified_as_pass(self):
        data = io.BytesIO()
        with zipfile.ZipFile(data, "w") as archive:
            archive.writestr("junit.xml", b'<testsuite><testcase classname="pkg::tests" name="recovers"><failure message="flaky" /><flakyFailure message="first attempt" /></testcase></testsuite>')
        report = read_junit_zip(data.getvalue(), 1)
        self.assertEqual("pass", report.testcases[0].result)
        self.assertEqual(1, report.testcases[0].flaky_retries)
        occurrences = classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 1, [report])
        self.assertEqual(1, len(occurrences))
        self.assertEqual(frozenset(("nextest retry",)), occurrences[0].modes)

    def test_nextest_retry_in_earlier_workflow_attempt_is_preserved(self):
        reports = [self.parse("nextest_flaky.xml", 1), self.parse("pass.xml", 2)]
        occurrences = classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 2, reports)
        self.assertEqual(1, len(occurrences))
        self.assertEqual(1, occurrences[0].final_attempt)

    def test_workflow_rerun_succeeds(self):
        reports = [self.parse("failure.xml", 1), self.parse("pass.xml", 2)]
        occurrences = classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 2, reports)
        self.assertEqual(frozenset(("workflow rerun",)), occurrences[0].modes)

    def test_final_workflow_attempt_fails(self):
        reports = [self.parse("pass.xml", 1), self.parse("failure.xml", 2)]
        self.assertEqual([], classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 2, reports))

    def test_missing_earlier_attempt_does_not_recover(self):
        self.assertEqual([], classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 2, [self.parse("pass.xml", 2)]))

    def test_actual_latest_artifact_missing(self):
        reports = [self.parse("failure.xml", 1), self.parse("pass.xml", 2)]
        self.assertEqual([], classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 3, reports))

    def test_intermediate_artifact_missing(self):
        reports = [self.parse("failure.xml", 1), self.parse("pass.xml", 3)]
        self.assertEqual(1, len(classify_reports("owner/repo", ".github/workflows/unit-test.yml", 9, 3, reports)))

    def test_duplicate_testcase_is_ambiguous(self):
        report = self.parse("duplicate.xml", 1)
        self.assertIn(("pkg::tests", "duplicate"), report.ambiguous)

    def test_zip_rejects_traversal_and_multiple_reports(self):
        for names in (("../junit.xml",), ("a/junit.xml", "b/junit.xml")):
            data = io.BytesIO()
            with zipfile.ZipFile(data, "w") as archive:
                for name in names:
                    archive.writestr(name, b"<testsuite />")
            with self.subTest(names=names), self.assertRaises(ArtifactError):
                _ = read_junit_zip(data.getvalue(), 1)

    def test_zip_accepts_safe_directory_entries(self):
        data = io.BytesIO()
        with zipfile.ZipFile(data, "w") as archive:
            archive.writestr("reports/", b"")
            archive.writestr("reports/junit.xml", b'<testsuite><testcase classname="pkg" name="case"/></testsuite>')
        self.assertEqual(1, len(read_junit_zip(data.getvalue(), 1).testcases))

    def test_zip_rejects_symlink_and_oversized_report(self):
        symlink = zipfile.ZipInfo("junit.xml")
        symlink.create_system = 3
        symlink.external_attr = 0o120777 << 16
        for member, content in ((symlink, b"target"), ("junit.xml", b"x" * (10 * 1024 * 1024 + 1))):
            data = io.BytesIO()
            with zipfile.ZipFile(data, "w") as archive:
                archive.writestr(member, content)
            with self.subTest(member=member), self.assertRaises(ArtifactError):
                _ = read_junit_zip(data.getvalue(), 1)

    def test_xml_rejects_doctype(self):
        data = io.BytesIO()
        with zipfile.ZipFile(data, "w") as archive:
            archive.writestr("junit.xml", b'<!DOCTYPE x [<!ENTITY e SYSTEM "file:///etc/passwd">]><testsuite><testcase classname="pkg" name="case"/></testsuite>')
        with self.assertRaises(ArtifactError):
            _ = read_junit_zip(data.getvalue(), 1)

    def test_xml_rejects_utf16_doctype_and_entity(self):
        xml = '<!DOCTYPE testsuite [<!ENTITY injected "value">]><testsuite><testcase classname="pkg" name="&injected;"/></testsuite>'.encode("utf-16")
        data = io.BytesIO()
        with zipfile.ZipFile(data, "w") as archive:
            archive.writestr("junit.xml", xml)
        with self.assertRaises(ArtifactError):
            _ = read_junit_zip(data.getvalue(), 1)

    def test_zip_rejects_windows_drive_unc_and_backslash_traversal(self):
        for unsafe in ("C:\\temp\\payload", "\\\\server\\share\\payload", "reports\\..\\payload"):
            data = io.BytesIO()
            with zipfile.ZipFile(data, "w") as archive:
                archive.writestr(unsafe, b"payload")
                archive.writestr("junit.xml", b'<testsuite><testcase classname="pkg" name="case"/></testsuite>')
            with self.subTest(unsafe=unsafe), self.assertRaises(ArtifactError):
                _ = read_junit_zip(data.getvalue(), 1)

    def test_xml_rejects_zero_testcases(self):
        data = io.BytesIO()
        with zipfile.ZipFile(data, "w") as archive:
            archive.writestr("junit.xml", b"<testsuite/>")
        with self.assertRaises(ArtifactError):
            _ = read_junit_zip(data.getvalue(), 1)


if __name__ == "__main__":
    _ = unittest.main()