Xarray Zarr Upstream test reporter

zarr_upstream_checker.py

#!/usr/bin/env -S uv run
"""
Zarr Upstream Compatibility Checker

This script checks the most recent upstream-dev CI workflow run for xarray
and reports on Zarr compatibility status and version information.

Usage:
    uv run zarr_upstream_checker.py

Requirements:
    - gh CLI must be installed and authenticated
"""

# /// script
# requires-python = ">=3.9"
# dependencies = [
#     "rich",
# ]
# ///

import json
import re
import subprocess
import sys
from datetime import datetime
from typing import Dict, List, Optional

from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text

console = Console()


class GitHubAPIError(Exception):
    """Custom exception for GitHub API errors"""
    pass


class ZarrUpstreamChecker:
    def __init__(self):
        self.xarray_repo = "pydata/xarray"
        self.zarr_repo = "zarr-developers/zarr-python"
        self.workflow_name = "upstream-dev-ci.yaml"

    def run_gh_command(self, args: List[str]) -> Dict:
        """Run a gh CLI command and return JSON response"""
        try:
            result = subprocess.run(
                ["gh"] + args,
                capture_output=True,
                text=True,
                check=True
            )
            return json.loads(result.stdout)
        except subprocess.CalledProcessError as e:
            if "not found" in e.stderr.lower() or "command not found" in e.stderr.lower():
                raise GitHubAPIError("gh CLI not found. Please install GitHub CLI: https://cli.github.com/")
            elif "authentication" in e.stderr.lower() or "not logged in" in e.stderr.lower():
                raise GitHubAPIError("gh CLI not authenticated. Please run: gh auth login")
            else:
                raise GitHubAPIError(f"gh CLI error: {e.stderr}")
        except json.JSONDecodeError:
            raise GitHubAPIError("Invalid JSON response from gh CLI")

    def get_latest_workflow_run(self) -> Dict:
        """Get the most recent workflow run for upstream-dev-ci.yaml"""
        try:
            # Get the most recent workflow run
            runs = self.run_gh_command([
                "run", "list",
                "--repo", self.xarray_repo,
                "--workflow", self.workflow_name,
                "--limit", "1",
                "--json", "databaseId,number,headBranch,headSha,status,conclusion,createdAt,updatedAt"
            ])
            
            if not runs:
                raise GitHubAPIError("No workflow runs found")
            
            return runs[0]
        except Exception as e:
            raise GitHubAPIError(f"Failed to get workflow runs: {e}")

    def get_workflow_jobs(self, run_id: int) -> List[Dict]:
        """Get jobs for a specific workflow run"""
        try:
            jobs = self.run_gh_command([
                "run", "view", str(run_id),
                "--repo", self.xarray_repo,
                "--json", "jobs"
            ])
            
            return jobs.get("jobs", [])
        except Exception as e:
            console.print(f"[yellow]Warning: Could not get job details: {e}[/yellow]")
            return []


    def get_workflow_logs_summary(self, run_id: int) -> Optional[str]:
        """Try to get workflow logs and extract zarr version info"""
        version_patterns = [
            r'zarr\s+(\d+\.\d+\.\d+(?:\.\w+\d+)?)',
            r'zarr-python\s+(\d+\.\d+\.\d+(?:\.\w+\d+)?)',
            r'zarr==(\d+\.\d+\.\d+(?:\.\w+\d+)?)',
            r'zarr:\s*(\d+\.\d+\.\d+(?:\.\w+\d+)?)',
        ]
        
        try:
            result = subprocess.run([
                "gh", "run", "view", str(run_id),
                "--repo", self.xarray_repo,
                "--log"
            ], capture_output=True, text=True, check=True)
            
            for pattern in version_patterns:
                matches = re.findall(pattern, result.stdout, re.IGNORECASE)
                if matches:
                    return max(set(matches), key=matches.count)
                    
        except subprocess.CalledProcessError:
            pass  # Logs might not be accessible without proper permissions
        except Exception as e:
            console.print(f"[yellow]Warning: Could not extract version from logs: {e}[/yellow]")
        
        return None

    def get_zarr_latest_commit(self) -> Optional[Dict]:
        """Get the latest commit from zarr-python main branch"""
        try:
            result = subprocess.run([
                "gh", "api", f"repos/{self.zarr_repo}/commits",
                "--jq", ".[0] | {sha: .sha, date: .commit.committer.date}"
            ], capture_output=True, text=True, check=True)
            
            return json.loads(result.stdout)
        except Exception as e:
            console.print(f"[yellow]Warning: Could not get zarr latest commit: {e}[/yellow]")
            return None

    def check_upstream_compatibility(self) -> Dict:
        """Main method to check upstream compatibility"""
        with console.status("Checking workflow runs..."):
            # Get latest workflow run
            latest_run = self.get_latest_workflow_run()
            
            # Get job details
            jobs = self.get_workflow_jobs(latest_run["databaseId"])
            
            # Find upstream-dev job
            upstream_dev_job = next(
                (job for job in jobs if "upstream-dev" in job.get("name", "").lower()),
                None
            )
            
            # Try to get zarr version from logs
            zarr_version_from_logs = self.get_workflow_logs_summary(latest_run["databaseId"])
            
            # Get latest zarr commit
            zarr_latest_commit = self.get_zarr_latest_commit()
        
        return {
            "workflow_run": latest_run,
            "upstream_dev_job": upstream_dev_job,
            "zarr_version_from_logs": zarr_version_from_logs,
            "zarr_latest_commit": zarr_latest_commit,
        }

    def format_results(self, results: Dict) -> None:
        """Format and display results using rich"""
        run = results["workflow_run"]
        job = results["upstream_dev_job"]
        log_version = results["zarr_version_from_logs"]
        zarr_commit = results["zarr_latest_commit"]
        
        # Main status panel
        conclusion = run["conclusion"]
        status_color = "green" if conclusion == "success" else "red" if conclusion == "failure" else "yellow"
        status_text = Text(f"Workflow Status: {conclusion or run['status']}", style=f"bold {status_color}")
        
        console.print(Panel(status_text, title="🔄 Upstream Dev CI Status", title_align="left"))
        
        # Workflow details table
        table = Table(show_header=True, header_style="bold blue")
        table.add_column("Property", style="cyan")
        table.add_column("Value", style="white")
        
        workflow_url = f"https://github.com/{self.xarray_repo}/actions/runs/{run['databaseId']}"
        
        table.add_row("Workflow ID", str(run["databaseId"]))
        table.add_row("Run Number", str(run["number"]))
        table.add_row("Branch", run["headBranch"])
        table.add_row("Commit", run["headSha"][:8])
        table.add_row("Started", run["createdAt"])
        table.add_row("Completed", run["updatedAt"])
        table.add_row("URL", workflow_url)
        
        console.print(table)
        
        # Upstream-dev job status
        if job:
            job_conclusion = job["conclusion"]
            job_status_color = "green" if job_conclusion == "success" else "red" if job_conclusion == "failure" else "yellow"
            job_status = Text(f"✅ Found: {job_conclusion or job['status']}", style=f"bold {job_status_color}")
        else:
            job_status = Text("❌ Not found in this run", style="bold red")
        
        console.print(Panel(job_status, title="📋 Upstream-dev Job", title_align="left"))
        
        # Version info
        if log_version:
            version_text = Text(f"Zarr version tested: {log_version}", style="bold cyan")
        else:
            version_text = Text("Zarr version: Not found in logs", style="bold yellow")
        
        console.print(Panel(version_text, title="📦 Version Info", title_align="left"))
        
        # Check workflow freshness against zarr commits
        self._display_freshness_check(run, zarr_commit)
        
        # Summary
        if job and job["conclusion"] == "success":
            if log_version:
                summary = Text(f"🎉 Upstream-dev ran successfully with Zarr {log_version}", style="bold green")
            else:
                summary = Text("✅ Upstream-dev ran successfully, but version info unclear", style="bold yellow")
        elif not job:
            summary = Text("❌ Upstream-dev job was not found in the most recent workflow run", style="bold red")
        else:
            job_status = job.get('conclusion', job.get('status', 'unknown'))
            summary = Text(f"❌ Upstream-dev job failed with status: {job_status}", style="bold red")
        
        console.print(Panel(summary, title="📊 Summary", title_align="left"))

    def _display_freshness_check(self, run: Dict, zarr_commit: Optional[Dict]) -> None:
        """Display freshness check comparing workflow time to latest zarr commit"""
        if not zarr_commit:
            return
            
        try:
            workflow_time = datetime.fromisoformat(run["createdAt"].replace('Z', '+00:00'))
            zarr_commit_time = datetime.fromisoformat(zarr_commit["date"].replace('Z', '+00:00'))
            
            commit_info = f"Latest zarr commit: {zarr_commit['sha'][:8]} ({zarr_commit['date']})\nWorkflow started: {run['createdAt']}"
            
            if zarr_commit_time > workflow_time:
                warning_text = Text(
                    f"⚠️  Warning: Zarr has newer commits since this workflow ran\n{commit_info}",
                    style="bold yellow"
                )
                console.print(Panel(warning_text, title="⚠️  Outdated Check", title_align="left"))
            else:
                freshness_text = Text(
                    f"✅ Workflow is current with latest zarr commits\n{commit_info}",
                    style="bold green"
                )
                console.print(Panel(freshness_text, title="🕐 Freshness Check", title_align="left"))
        except Exception as e:
            console.print(f"[yellow]Could not compare timestamps: {e}[/yellow]")


def main():
    """Check Zarr upstream compatibility in xarray CI"""
    checker = ZarrUpstreamChecker()
    
    try:
        results = checker.check_upstream_compatibility()
        checker.format_results(results)
    except GitHubAPIError as e:
        console.print(f"[red]Error: {e}[/red]")
        sys.exit(1)
    except Exception as e:
        console.print(f"[red]Unexpected error: {e}[/red]")
        sys.exit(1)


if __name__ == "__main__":
    main()