datastore/ci/praktika/issue.py

import json
import os
import re
import sys
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import TYPE_CHECKING, List, Optional

sys.path.append(os.path.dirname(__file__) + "/../")

from praktika.gh import GH
from praktika.result import Result
from praktika.s3 import S3
from praktika.settings import Settings
from praktika.utils import MetaClasses, Shell, Utils

if TYPE_CHECKING:
    pass


def fetch_github_issues(
    label: str, state: str = "open", hours_back: float = None, repo: str = ""
) -> List[dict]:
    """
    Fetch issues from GitHub using gh CLI with manual pagination.

    Args:
        label: GitHub label to filter by
        state: Issue state (open or closed)
        hours_back: For closed issues, only fetch those closed within this many hours (default: None for all)
        repo: GitHub repository in format owner/repo (default: empty string uses current repo)

    Returns:
        List of issue dictionaries
    """
    if isinstance(label, str):
        label = [label]
    all_issues = []
    repo_arg = f"--repo {repo}" if repo else ""
    limit_per_request = 1000  # Maximum we'll fetch per request

    # Build base command
    if state == "closed" and hours_back:
        date_threshold = (datetime.now() - timedelta(hours=hours_back)).strftime(
            "%Y-%m-%dT%H:%M:%S"
        )
        label_query = " ".join([f'label:"{lbl}"' for lbl in label])
        search_query = f"{label_query} is:closed closed:>{date_threshold}"
        base_cmd = f"gh issue list {repo_arg} --search '{search_query}' --json number,title,body,closedAt,labels,url --limit {limit_per_request}"
        print(
            f"Fetching {state} issues with label '{label}' closed in last {hours_back} hours (since {date_threshold})..."
        )
    else:
        label_args = " ".join([f'--label "{lbl}"' for lbl in label])
        base_cmd = f"gh issue list {repo_arg} {label_args} --state {state} --json number,title,body,closedAt,labels,url --limit {limit_per_request}"
        print(f"Fetching {state} issues with label '{label}'...")

    try:
        output = Shell.get_output(base_cmd, verbose=True)

        if not output or not output.strip():
            print(f"  No issues found for label '{label}' with state '{state}'")
            return []

        issues = json.loads(output)
        if not issues:
            print(f"  No issues found for label '{label}' with state '{state}'")
            return []

        all_issues.extend(issues)
        print(f"  Found {len(all_issues)} issues")

        # If the limit was hit, print a warning — there may be more issues to page
        if len(issues) == limit_per_request:
            print(
                f"  WARNING: Reached limit of {limit_per_request} issues. There may be more issues not fetched."
            )

        return all_issues
    except json.JSONDecodeError as e:
        print(f"ERROR: Failed to parse JSON response for label '{label}': {e}")
        return []
    except Exception as e:
        print(f"ERROR: Failed to fetch issues with label '{label}': {e}")
        return []


class IssueLabels:
    CI_ISSUE = "testing"
    FLAKY_TEST = "flaky test"
    FUZZ = "fuzz"
    INFRASTRUCTURE = "infrastructure"
    SANITIZER = "sanitizer"


@dataclass
class Issue:
    """Represents a GitHub issue found/occurred in CI"""

    test_name: str
    closed_at: str
    number: int
    url: str
    title: str
    body: str
    labels: List[str]

    failure_reason: str = ""
    failure_flags: List[str] = field(default_factory=list)
    ci_action: str = ""
    test_pattern: str = ""
    job_pattern: str = ""

    def is_infrastructure(self):
        return "infrastructure" in self.labels

    def has_failure_reason(self):
        return bool(self.failure_reason)

    def has_failure_flags(self):
        return bool(self.failure_flags)

    def has_job_pattern(self):
        return bool(self.job_pattern) and self.job_pattern != "%"

    def has_test_pattern(self):
        return bool(self.test_pattern) and self.test_pattern != "%"

    @staticmethod
    def parse_issue_body_fields(body: str) -> dict:
        fields = {
            "test_name": "",
            "failure_reason": "",
            "failure_flags": [],
            "ci_action": "",
            "test_pattern": "",
            "job_pattern": "",
        }

        if not body:
            return fields

        patterns = {
            "test_name": r"Test name:\s*([^\n]*)",
            "failure_reason": r"Failure reason:\s*([^\n]*)",
            "ci_action": r"CI action:\s*([^\n]*)",
            "test_pattern": r"Test pattern:\s*([^\n]*)",
            "job_pattern": r"Job pattern:\s*([^\n]*)",
        }

        for field_name, pattern in patterns.items():
            match = re.search(pattern, body, re.IGNORECASE)
            if match:
                fields[field_name] = match.group(1).strip()

        flags_match = re.search(
            r"^Failure flags:[ \t]*(.*)$", body, re.IGNORECASE | re.MULTILINE
        )
        fields["failure_flags"] = []
        if flags_match:
            flags_str = flags_match.group(1).strip()
            if flags_str:
                fields["failure_flags"] = [
                    flag.strip() for flag in flags_str.split(",") if flag.strip()
                ]

        return fields

    @staticmethod
    def extract_test_name(title) -> Optional[str]:
        """
        Temporary helper method until all CI issues following the same pattern with test name in the body
        """
        pattern1 = r"\b(\d{5}_\S+)"
        match1 = re.search(pattern1, title)
        if match1:
            test_name = match1.group(1)
            return test_name.rstrip("`'\",.;:!?)")

        pattern2 = r"\b(test_[^\s`'\",]+)"
        match2 = re.search(pattern2, title)
        if match2:
            test_name = match2.group(1)
            return test_name.rstrip("`'\",.;:!?)")
        return None

    def validate(self, verbose=True):
        if not self.test_name and not self.is_infrastructure():
            if verbose:
                print(f"WARNING: Invalid issue [{self.url}] must have test_name")
            return False
        if not self.url:
            if verbose:
                print(f"WARNING: Invalid issue [{self.url}] must have issue_url")
            return False
        if not self.title:
            print(f"WARNING: Invalid issue [{self.url}] must have title")
            return False
        if self.is_infrastructure():
            if (
                self.has_job_pattern()
                or self.has_test_pattern()
                or self.has_failure_reason()
            ):
                return True
            else:
                print(
                    f"WARNING: Invalid infrastructure issue [{self.url}] must have job_pattern, test_pattern or failure_reason"
                )
                return False
        return True

    def check_result(self, result: Result, job_name: str = "") -> bool:
        """
        Check if this issue matches the given result and mark it if it does.
        Recursively processes result tree.

        Args:
            result: The result to check
            job_name: The top-level job name for infrastructure job_pattern matching

        Returns:
            True if the result was marked with this issue, False otherwise
        """
        # Skip if result is OK or already marked
        if result.is_ok() or result.has_label("issue"):
            return False

        # Recursively check sub-results first.
        # If called for the workflow result, its sub-results are jobs and their names should be used
        # for job_pattern matching.
        if result.results:
            marked = False
            is_workflow_level_call = not job_name
            for sub_result in result.results:
                sub_job_name = sub_result.name if is_workflow_level_call else job_name
                if self.check_result(sub_result, job_name=sub_job_name):
                    marked = True
            return marked

        # Leaf result - check if it matches.
        return (
            self._check_infrastructure_match(result, job_name)
            if self.is_infrastructure()
            else self._check_flaky_test_match(result)
        )

    def _check_flaky_test_match(self, result: Result) -> bool:
        """Check if this flaky test issue matches the result."""
        name_in_report = result.name
        test_name = self.test_name

        # Normalize pytest parameterized names
        if ".py" in name_in_report and ".py" in test_name:
            if "[" in test_name:
                # Issue mentions a specific parametrization: keep full name for exact match
                pass
            elif "[" in name_in_report:
                # Issue mentions only the base test: compare against the base part
                name_in_report = name_in_report.split("[")[0]

            if "::" not in test_name:
                # Issue mentions only the module path without function: extract module path from report test name
                name_in_report = name_in_report.split("::")[0]

        # Check if test name matches
        if not name_in_report.endswith(test_name):
            return False

        # Check failure_reason if present
        if self.failure_reason and self.failure_reason not in result.info:
            print(
                "WARNING: matched test name, but failure reason not found in output\n"
                f"  test:   {test_name}\n"
                f"  reason: {self.failure_reason}\n"
                f"  issue:  #{self.number}\n"
                "  action: do not match"
            )
            return False

        print(
            f"Marking '{result.name}' as flaky (matched: {test_name}, issue: #{self.number})"
        )
        result.set_label(Result.Label.ISSUE, link=self.url)
        return True

    def _check_infrastructure_match(
        self, result: Result, top_level_result_name: str
    ) -> bool:
        """Check if this infrastructure issue matches the result."""
        # Check failure_reason if present
        if self.failure_reason and self.failure_reason not in result.info:
            return False

        # Check failure_flags with result labels
        if self.failure_flags:
            if any(not result.has_label(flag) for flag in self.failure_flags):
                return False

        # Check test_pattern (SQL style %name%) with result.name
        if self.test_pattern and self.test_pattern != "%":
            pattern_parts = [p for p in self.test_pattern.split("%") if p]
            if not any(part in result.name for part in pattern_parts):
                return False

        # Check job_pattern (SQL style %name%) with top level result name
        if self.job_pattern and self.job_pattern != "%":
            job_pattern_parts = [p for p in self.job_pattern.split("%") if p]
            if not any(part in top_level_result_name for part in job_pattern_parts):
                return False

        print(
            f"  Marking '{result.name}' as infrastructure issue (issue: #{self.number})"
        )
        result.set_label(Result.Label.ISSUE, link=self.url)
        return True

    @classmethod
    def create_from(
        cls,
        title,
        body,
        labels,
        closed_at="",
        number=0,
        url="",
    ):
        body_fields = cls.parse_issue_body_fields(body)
        test_name = body_fields["test_name"]
        if not test_name:
            if IssueLabels.FLAKY_TEST in labels:
                # Extract test name from title or body
                test_name = cls.extract_test_name(title)
            else:
                test_name = title
        return Issue(
            test_name=test_name,
            closed_at=closed_at if closed_at else "",
            number=int(number),
            url=url,
            title=title,
            body=body if body else "",
            labels=labels,
            failure_reason=body_fields["failure_reason"],
            failure_flags=body_fields["failure_flags"],
            ci_action=body_fields["ci_action"],
            test_pattern=body_fields["test_pattern"],
            job_pattern=body_fields["job_pattern"],
        )

    def create_on_gh(self, repo_name):
        assert self.number == 0 and self.url == ""
        self.url = GH.create_issue(
            self.title, self.body, self.labels, repo=repo_name, verbose=True
        )
        self.number = int(self.url.split("/")[-1]) if self.url else 0
        return self

    @classmethod
    def from_dict(cls, issue: dict) -> "Issue":
        """
        Process raw GitHub issue into TestCaseIssue objects.

        Args:
            issue: raw issue dictionary from GitHub

        Returns:
            TestCaseIssue object
        """

        number = int(issue.get("number", "0"))
        title = issue.get("title", "")
        body = issue.get("body", "")
        closed_at = issue.get("closedAt", "")
        url = issue.get("url", "")
        assert number > 0, f"Issue {number} has no number"
        assert title, f"Issue {number} has no title"
        # Extract label names from the labels array
        labels = [label.get("name", "") for label in issue.get("labels", [])]
        return cls.create_from(title, body, labels, closed_at, number, url=url)


@dataclass
class TestCaseIssueCatalog(MetaClasses.Serializable):
    """Catalog of all flaky test issues"""

    name: str = "issue_catalog"
    active_test_issues: List[Issue] = field(default_factory=list)
    updated_at: float = 0.0

    @classmethod
    def file_name_static(cls, name):
        return f"/tmp/{Utils.normalize_string(name)}.json"

    @classmethod
    def from_dict(cls, obj: dict):
        """Custom deserialization to handle nested TestCaseIssue objects"""
        active_issues = [
            Issue(**issue) if isinstance(issue, dict) else issue
            for issue in obj.get("active_test_issues", [])
        ]
        return cls(
            name=obj.get("name", cls.name),
            active_test_issues=active_issues,
            updated_at=obj.get("updated_at", 0.0),
        )

    @classmethod
    def process_issue(cls, issues_raw, verbose=True):
        res = []
        for issue_ in issues_raw:
            issue = Issue.from_dict(issue_)
            if issue.validate(verbose=verbose):
                res.append(issue)
        return res

    @classmethod
    def from_gh(cls, verbose=True, repo=""):
        """
        Fetch and organize all CI test issues from GitHub.

        Args:
            verbose: Enable verbose output
            repo: GitHub repository in format owner/repo (default: empty string uses current repo)

        Returns:
            TestCaseIssueCatalog with active issues (including recently closed)
        """
        catalog = cls()
        catalog.updated_at = datetime.now().timestamp()

        # Fetch open issues with label "testing" (CI_ISSUE) first
        if verbose:
            print("\n--- Fetching active testing issues ---")
        testing_issues = fetch_github_issues(
            label=IssueLabels.CI_ISSUE, state="open", repo=repo
        )
        catalog.active_test_issues = cls.process_issue(testing_issues, verbose=verbose)
        if verbose:
            print(f"Processed {len(catalog.active_test_issues)} testing issues")

        # Fetch closed issues with label "testing" from the last 8 hours and add to active issues
        if verbose:
            print("--- Fetching closed testing issues from last 8 hours ---")
        closed_testing_issues = fetch_github_issues(
            label=IssueLabels.CI_ISSUE, state="closed", hours_back=8, repo=repo
        )
        closed_testing_processed = cls.process_issue(
            closed_testing_issues, verbose=verbose
        )
        added_closed_testing = 0
        existing_issue_numbers = {issue.number for issue in catalog.active_test_issues}
        for issue in closed_testing_processed:
            if issue.number not in existing_issue_numbers:
                catalog.active_test_issues.append(issue)
                existing_issue_numbers.add(issue.number)
                added_closed_testing += 1
        if verbose:
            print(
                f"Added {added_closed_testing} closed testing issues (total: {len(catalog.active_test_issues)})"
            )

        if verbose:
            # Print summary by label for all issues
            print("=== Issues Summary by Label ===")
            label_counts = {}
            for issue in catalog.active_test_issues:
                for label in issue.labels:
                    label_counts[label] = label_counts.get(label, 0) + 1

            # Sort by count descending
            sorted_labels = sorted(
                label_counts.items(), key=lambda x: x[1], reverse=True
            )
            for label, count in sorted_labels:
                print(f"  {label}: {count}")
            print()

        return catalog

    def to_s3(self):
        """
        Upload catalog to S3.

        Returns:
            S3 URL of the uploaded catalog
        """
        local_name = self.file_name()
        compressed_name = Utils.compress_gz(local_name)
        link = S3.copy_file_to_s3(
            local_path=compressed_name,
            s3_path=f"{Settings.S3_REPORT_BUCKET}/statistics",
            content_type="application/json",
            content_encoding="gzip",
        )
        return link

    @classmethod
    def _download_catalog(cls, bucket, suffix=""):
        """Download and decompress a single catalog from an S3 bucket."""
        local_catalog_gz = cls.file_name_static(cls.name + suffix) + ".gz"
        local_catalog_json = cls.file_name_static(cls.name + suffix)
        s3_catalog_path = f"{bucket}/statistics/{Utils.normalize_string(cls.name)}.json.gz"

        if not S3.copy_file_from_s3(
            s3_catalog_path, local_catalog_gz, _skip_download_counter=True
        ):
            print(f"  WARNING: Could not download catalog from S3: {s3_catalog_path}")
            return None

        Shell.check(f"gunzip -f {local_catalog_gz}", verbose=True)

        if not Path(local_catalog_json).exists():
            print(
                f"  WARNING: Decompressed catalog file not found: {local_catalog_json}"
            )
            return None

        return cls.from_file(local_catalog_json)

    @classmethod
    def from_s3(cls):
        """
        Download catalog from S3. If S3_UPSTREAM_REPORT_BUCKET is set,
        also downloads the upstream catalog and merges issues from both.

        Returns:
            TestCaseIssueCatalog instance or None if download failed
        """
        catalog = cls._download_catalog(Settings.S3_REPORT_BUCKET)

        if Settings.S3_UPSTREAM_REPORT_BUCKET:
            upstream = cls._download_catalog(
                Settings.S3_UPSTREAM_REPORT_BUCKET, suffix="_upstream"
            )
            if upstream:
                if catalog is None:
                    catalog = upstream
                else:
                    existing = {
                        issue.number for issue in catalog.active_test_issues
                    }
                    added = 0
                    for issue in upstream.active_test_issues:
                        if issue.number not in existing:
                            catalog.active_test_issues.append(issue)
                            added += 1
                    print(
                        f"  Merged {added} issues from upstream catalog "
                        f"(total: {len(catalog.active_test_issues)})"
                    )

        return catalog


if __name__ == "__main__":
    temp_path = f"{Utils.cwd()}/ci/tmp"
    Path(temp_path).mkdir(exist_ok=True)

    def collect_and_upload():
        print("Fetching active testing issues...")
        c = TestCaseIssueCatalog.from_gh()
        c.dump()
        print("Uploading to S3...")
        url = c.to_s3()
        print(f"Catalog uploaded to S3: {url}")
        return bool(url)

    def download():
        TestCaseIssueCatalog.from_s3().dump()
        return True

    results = []

    if "--collect-and-upload" in sys.argv:
        results.append(
            Result.from_commands_run(
                name="Collect and upload flaky test issues",
                command=lambda: collect_and_upload(),
            )
        )
        # collect_and_upload()
    elif "--download" in sys.argv:
        results.append(
            Result.from_commands_run(
                name="Download flaky test issues", command=lambda: download()
            )
        )
    else:
        print("ERROR: No action specified")
        raise

    Result.create_from(results=results).complete_job()