import hashlib
import io
import pathlib
import stat
import xml.etree.ElementTree as ET
import zipfile
from collections import Counter
from typing import Dict, List, Mapping, Optional, Sequence, Set, Tuple

from model import Occurrence, Report, Testcase


MAX_XML_BYTES = 10 * 1024 * 1024


class ArtifactError(Exception):
    pass


def _tag(element: ET.Element) -> str:
    return element.tag.rsplit("}", 1)[-1]


def read_junit_zip(data: bytes, attempt: int) -> Report:
    try:
        with zipfile.ZipFile(io.BytesIO(data)) as archive:
            matches: List[zipfile.ZipInfo] = []
            for member in archive.infolist():
                path = pathlib.PurePosixPath(member.filename.replace("\\", "/"))
                windows_path = pathlib.PureWindowsPath(member.filename)
                mode = member.external_attr >> 16
                if path.is_absolute() or windows_path.is_absolute() or windows_path.drive or ".." in path.parts or stat.S_ISLNK(mode):
                    raise ArtifactError("unsafe artifact archive")
                if member.is_dir():
                    continue
                if path.name == "junit.xml":
                    matches.append(member)
            if len(matches) != 1:
                raise ArtifactError("artifact must contain exactly one junit.xml")
            member = matches[0]
            if member.file_size <= 0 or member.file_size > MAX_XML_BYTES:
                raise ArtifactError("invalid junit.xml size")
            xml_data = archive.read(member)
    except (zipfile.BadZipFile, OSError, KeyError) as error:
        raise ArtifactError("unreadable artifact archive") from error
    return parse_junit(xml_data, attempt)


def parse_junit(data: bytes, attempt: int) -> Report:
    class RejectingTreeBuilder(ET.TreeBuilder):
        def doctype(self, _name: str, _public_id: Optional[str], _system_id: Optional[str]) -> None:
            raise ArtifactError("DTD and entities are not allowed")

    try:
        root = ET.fromstring(data, parser=ET.XMLParser(target=RejectingTreeBuilder()))
    except ET.ParseError as error:
        raise ArtifactError("malformed junit XML") from error
    if _tag(root) not in ("testsuite", "testsuites"):
        raise ArtifactError("unsupported junit root")
    parsed: List[Testcase] = []
    identities: List[Tuple[str, str]] = []
    for element in root.iter():
        if _tag(element) != "testcase":
            continue
        classname = element.get("classname", "")
        name = element.get("name", "")
        if not classname or not name:
            raise ArtifactError("testcase identity fields must be non-empty")
        identity = (classname, name)
        identities.append(identity)
        child_tags = [_tag(child) for child in list(element)]
        flaky_retries = sum(tag in ("flakyFailure", "flakyError") for tag in child_tags)
        failed_retries = sum(tag in ("rerunFailure", "rerunError") for tag in child_tags)
        # A testcase with flakyFailure/flakyError recovered within the run.
        # Classify it as "pass" regardless of any <failure> tag that nextest may
        # emit alongside when flaky-fail-status = "failure" is configured.
        if flaky_retries:
            result = "pass"
        elif "failure" in child_tags or "error" in child_tags:
            result = "failure"
        elif "skipped" in child_tags:
            result = "skipped"
        else:
            result = "pass"
        parsed.append(Testcase(classname, name, result, flaky_retries, failed_retries))
    duplicates = frozenset(identity for identity, count in Counter(identities).items() if count > 1)
    if not identities:
        raise ArtifactError("junit report contains no testcases")
    return Report(attempt, tuple(item for item in parsed if (item.classname, item.name) not in duplicates), duplicates)


def _fingerprint(repository: str, workflow_path: str, run_id: int, attempt: int, identity: Tuple[str, str]) -> str:
    values = (repository, workflow_path, str(run_id), str(attempt), identity[0], identity[1])
    return hashlib.sha256("\0".join(values).encode("utf-8")).hexdigest()


def classify_reports(
    repository: str,
    workflow_path: str,
    run_id: int,
    latest_attempt: int,
    reports: Sequence[Report],
    workflow_name: str = "",
    run_url: str = "",
    occurred_at: str = "",
    attempt_times: Optional[Mapping[int, str]] = None,
) -> List[Occurrence]:
    by_attempt = {report.attempt: report for report in reports}
    latest = by_attempt.get(latest_attempt)
    modes: Dict[Tuple[int, str, str], Set[str]] = {}
    for report in reports:
        for testcase in report.testcases:
            if testcase.result == "pass" and testcase.flaky_retries:
                modes.setdefault((report.attempt, testcase.classname, testcase.name), set()).add("nextest retry")
    if latest is not None:
        for testcase in latest.testcases:
            identity = (testcase.classname, testcase.name)
            lower_failed = any(
                candidate.result == "failure"
                for attempt, report in by_attempt.items() if attempt < latest_attempt
                for candidate in report.testcases
                if (candidate.classname, candidate.name) == identity
            )
            if testcase.result == "pass" and lower_failed:
                modes.setdefault((latest_attempt, identity[0], identity[1]), set()).add("workflow rerun")
    occurrences: List[Occurrence] = []
    for identity, detected in modes.items():
        attempt, classname, name = identity
        occurrence_time = (attempt_times or {}).get(attempt, occurred_at)
        occurrences.append(Occurrence(repository, workflow_path, workflow_name, run_id, attempt, run_url, occurrence_time, classname, name, frozenset(detected), _fingerprint(repository, workflow_path, run_id, attempt, (classname, name))))
    return sorted(occurrences, key=lambda item: item.fingerprint)