# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import itertools
import logging
from abc import ABC, ABCMeta
from dataclasses import dataclass
from enum import Enum
from pathlib import PurePath
from typing import Any, ClassVar, TypeVar, cast

from pants.core.goals.package import BuiltPackage, PackageFieldSet
from pants.core.util_rules.distdir import DistDir
from pants.engine.addresses import Address, UnparsedAddressInputs
from pants.engine.collection import Collection
from pants.engine.console import Console
from pants.engine.desktop import OpenFiles, OpenFilesRequest
from pants.engine.engine_aware import EngineAwareReturnType
from pants.engine.environment import Environment, EnvironmentRequest
from pants.engine.fs import EMPTY_FILE_DIGEST, Digest, FileDigest, MergeDigests, Snapshot, Workspace
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.internals.session import RunId
from pants.engine.process import (
    FallibleProcessResult,
    InteractiveProcess,
    InteractiveProcessResult,
    ProcessResultMetadata,
)
from pants.engine.rules import Effect, Get, MultiGet, collect_rules, goal_rule, rule
from pants.engine.target import (
    FieldSet,
    FieldSetsPerTarget,
    FieldSetsPerTargetRequest,
    NoApplicableTargetsBehavior,
    SourcesField,
    SpecialCasedDependencies,
    TargetRootsToFieldSets,
    TargetRootsToFieldSetsRequest,
    Targets,
)
from pants.engine.unions import UnionMembership, union
from pants.option.option_types import BoolOption, EnumOption, StrListOption, StrOption
from pants.util.docutil import bin_name
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TestResult(EngineAwareReturnType):
    exit_code: int | None
    stdout: str
    stdout_digest: FileDigest
    stderr: str
    stderr_digest: FileDigest
    address: Address
    output_setting: ShowOutput
    result_metadata: ProcessResultMetadata | None

    coverage_data: CoverageData | None = None
    xml_results: Snapshot | None = None
    # Any extra output (such as from plugins) that the test runner was configured to output.
    extra_output: Snapshot | None = None

    # Prevent this class from being detected by pytest as a test class.
    __test__ = False

    @classmethod
    def skip(cls, address: Address, output_setting: ShowOutput) -> TestResult:
        return cls(
            exit_code=None,
            stdout="",
            stderr="",
            stdout_digest=EMPTY_FILE_DIGEST,
            stderr_digest=EMPTY_FILE_DIGEST,
            address=address,
            output_setting=output_setting,
            result_metadata=None,
        )

    @classmethod
    def from_fallible_process_result(
        cls,
        process_result: FallibleProcessResult,
        address: Address,
        output_setting: ShowOutput,
        *,
        coverage_data: CoverageData | None = None,
        xml_results: Snapshot | None = None,
        extra_output: Snapshot | None = None,
    ) -> TestResult:
        return cls(
            exit_code=process_result.exit_code,
            stdout=process_result.stdout.decode(),
            stdout_digest=process_result.stdout_digest,
            stderr=process_result.stderr.decode(),
            stderr_digest=process_result.stderr_digest,
            address=address,
            output_setting=output_setting,
            result_metadata=process_result.metadata,
            coverage_data=coverage_data,
            xml_results=xml_results,
            extra_output=extra_output,
        )

    @property
    def skipped(self) -> bool:
        return self.exit_code is None or self.result_metadata is None

    def __lt__(self, other: Any) -> bool:
        """We sort first by status (skipped vs failed vs succeeded), then alphanumerically within
        each group."""
        if not isinstance(other, TestResult):
            return NotImplemented
        if self.exit_code == other.exit_code:
            return self.address.spec < other.address.spec
        if self.skipped or self.exit_code is None:
            return True
        if other.skipped or other.exit_code is None:
            return False
        return abs(self.exit_code) < abs(other.exit_code)

    def artifacts(self) -> dict[str, FileDigest | Snapshot] | None:
        output: dict[str, FileDigest | Snapshot] = {
            "stdout": self.stdout_digest,
            "stderr": self.stderr_digest,
        }
        if self.xml_results:
            output["xml_results"] = self.xml_results
        return output

    def level(self) -> LogLevel:
        if self.skipped:
            return LogLevel.DEBUG
        return LogLevel.INFO if self.exit_code == 0 else LogLevel.ERROR

    def message(self) -> str:
        if self.skipped:
            return f"{self.address} skipped."
        status = "succeeded" if self.exit_code == 0 else f"failed (exit code {self.exit_code})"
        message = f"{self.address} {status}."
        if self.output_setting == ShowOutput.NONE or (
            self.output_setting == ShowOutput.FAILED and self.exit_code == 0
        ):
            return message
        output = ""
        if self.stdout:
            output += f"\n{self.stdout}"
        if self.stderr:
            output += f"\n{self.stderr}"
        if output:
            output = f"{output.rstrip()}\n\n"
        return f"{message}{output}"

    def metadata(self) -> dict[str, Any]:
        return {"address": self.address.spec}

    def cacheable(self) -> bool:
        """Is marked uncacheable to ensure that it always renders."""
        return False


class ShowOutput(Enum):
    """Which tests to emit detailed output for."""