fix: resolve CI/CD issues - rewrote CI workflow and fixed Python linting errors
Some checks failed
DevDash CLI CI / test (push) Failing after 9s
Some checks failed
DevDash CLI CI / test (push) Failing after 9s
This commit is contained in:
@@ -1,96 +1,285 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional
|
||||
"""Git repository status monitoring."""
|
||||
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class GitStatusError(Exception):
|
||||
"""Exception raised for git operations."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
def _run_git_command(args: List[str], cwd: str = None) -> str:
|
||||
def run_git_command(
|
||||
*args: str,
|
||||
cwd: Path | None = None,
|
||||
) -> str:
|
||||
"""Run a git command and return output.
|
||||
|
||||
Args:
|
||||
*args: Git command arguments.
|
||||
cwd: Working directory (defaults to current directory).
|
||||
|
||||
Returns:
|
||||
Command output string.
|
||||
|
||||
Raises:
|
||||
GitStatusError: If git command fails.
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git"] + args,
|
||||
["git"] + list(args),
|
||||
cwd=cwd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise GitStatusError(f"Git command failed: {result.stderr}")
|
||||
|
||||
return result.stdout.strip()
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise GitStatusError(f"Git command failed: {e}")
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
raise GitStatusError("Git command timed out") from None
|
||||
except FileNotFoundError:
|
||||
raise GitStatusError("Git is not installed") from None
|
||||
|
||||
|
||||
def _get_current_branch(cwd: str = None) -> str:
|
||||
def _get_branch_impl(cwd: Path | None = None) -> tuple[str, bool]:
|
||||
"""Get current branch name - internal implementation."""
|
||||
output = run_git_command("rev-parse", "--abbrev-ref", "HEAD", cwd=cwd)
|
||||
|
||||
if output.strip() == "HEAD":
|
||||
commit_hash = run_git_command("rev-parse", "HEAD", cwd=cwd)
|
||||
return commit_hash[:7], True
|
||||
|
||||
return output, False
|
||||
|
||||
|
||||
def get_branch(cwd: Path | None = None) -> tuple[str, bool]:
|
||||
"""Get current branch name.
|
||||
|
||||
Args:
|
||||
cwd: Working directory.
|
||||
|
||||
Returns:
|
||||
Tuple of (branch_name, is_detached).
|
||||
"""
|
||||
return _get_branch_impl(cwd)
|
||||
|
||||
|
||||
def _get_commit_hash_impl(cwd: Path | None = None) -> str:
|
||||
"""Get current commit hash - internal implementation."""
|
||||
return run_git_command("rev-parse", "HEAD", cwd=cwd)
|
||||
|
||||
|
||||
def get_commit_hash(cwd: Path | None = None) -> str:
|
||||
"""Get current commit hash.
|
||||
|
||||
Args:
|
||||
cwd: Working directory.
|
||||
|
||||
Returns:
|
||||
Full commit hash string.
|
||||
"""
|
||||
return _get_commit_hash_impl(cwd)
|
||||
|
||||
|
||||
def _get_commit_info_impl(cwd: Path | None = None) -> tuple[str, str, datetime]:
|
||||
"""Get current commit info - internal implementation."""
|
||||
message = run_git_command("log", "-1", "--format=%s", cwd=cwd)
|
||||
author_name = run_git_command("log", "-1", "format=%an", cwd=cwd)
|
||||
author_date_str = run_git_command("log", "-1", "--format=%ai", cwd=cwd)
|
||||
|
||||
try:
|
||||
return _run_git_command(["rev-parse", "--abbrev-ref", "HEAD"], cwd)
|
||||
author_date = datetime.strptime(author_date_str, "%Y-%m-%d %H:%M:%S %z")
|
||||
except ValueError:
|
||||
try:
|
||||
author_date = datetime.strptime(author_date_str, "%Y-%m-%d %H:%M:%S")
|
||||
except ValueError:
|
||||
author_date = datetime.now()
|
||||
|
||||
return message, author_name, author_date
|
||||
|
||||
|
||||
def get_commit_info(cwd: Path | None = None) -> tuple[str, str, datetime]:
|
||||
"""Get current commit information.
|
||||
|
||||
Args:
|
||||
cwd: Working directory.
|
||||
|
||||
Returns:
|
||||
Tuple of (message, author_name, author_date).
|
||||
"""
|
||||
return _get_commit_info_impl(cwd)
|
||||
|
||||
|
||||
def _get_file_counts_impl(cwd: Path | None = None) -> tuple[int, int, int]:
|
||||
"""Get file counts - internal implementation."""
|
||||
try:
|
||||
staged_output = run_git_command("diff", "--cached", "--name-only", cwd=cwd)
|
||||
staged = len(staged_output.splitlines()) if staged_output else 0
|
||||
|
||||
unstaged_output = run_git_command("diff", "--name-only", cwd=cwd)
|
||||
unstaged = len(unstaged_output.splitlines()) if unstaged_output else 0
|
||||
|
||||
untracked_output = run_git_command("ls-files", "--others", "--exclude-standard", cwd=cwd)
|
||||
untracked = len(untracked_output.splitlines()) if untracked_output else 0
|
||||
|
||||
return staged, unstaged, untracked
|
||||
|
||||
except GitStatusError:
|
||||
return "HEAD"
|
||||
return 0, 0, 0
|
||||
|
||||
|
||||
def _get_commit_hash(cwd: str = None) -> str:
|
||||
return _run_git_command(["rev-parse", "SHORT"], cwd)
|
||||
def get_file_counts(cwd: Path | None = None) -> tuple[int, int, int]:
|
||||
"""Get count of staged, unstaged, and untracked files.
|
||||
|
||||
Args:
|
||||
cwd: Working directory.
|
||||
|
||||
Returns:
|
||||
Tuple of (staged_count, unstaged_count, untracked_count).
|
||||
"""
|
||||
return _get_file_counts_impl(cwd)
|
||||
|
||||
|
||||
def _is_detached(cwd: str = None) -> bool:
|
||||
branch = _get_current_branch(cwd)
|
||||
return branch == "HEAD"
|
||||
def _get_remote_status_impl(cwd: Path | None = None) -> tuple[int, int, str | None]:
|
||||
"""Get remote status - internal implementation."""
|
||||
try:
|
||||
branch_info = run_git_command("rev-list", "--left-right", "--count", "@{upstream}...HEAD", cwd=cwd)
|
||||
|
||||
if branch_info:
|
||||
parts = branch_info.split()
|
||||
ahead = int(parts[0]) if len(parts) > 0 else 0
|
||||
behind = int(parts[1]) if len(parts) > 1 else 0
|
||||
else:
|
||||
ahead, behind = 0, 0
|
||||
|
||||
remote_name = run_git_command("rev-parse", "--abbrev-ref", "@{upstream}", cwd=cwd)
|
||||
if remote_name:
|
||||
remote_name = remote_name.split("/")[0] if "/" in remote_name else remote_name
|
||||
else:
|
||||
remote_name = None
|
||||
|
||||
return ahead, behind, remote_name
|
||||
|
||||
except GitStatusError:
|
||||
return 0, 0, None
|
||||
|
||||
|
||||
def _get_staged_count(cwd: str = None) -> int:
|
||||
output = _run_git_command(["diff", "--staged", "--name-only"], cwd)
|
||||
return len(output.splitlines()) if output else 0
|
||||
def get_remote_status(cwd: Path | None = None) -> tuple[int, int, str | None]:
|
||||
"""Get ahead/behind status compared to upstream.
|
||||
|
||||
Args:
|
||||
cwd: Working directory.
|
||||
|
||||
Returns:
|
||||
Tuple of (ahead_count, behind_count, remote_name).
|
||||
"""
|
||||
return _get_remote_status_impl(cwd)
|
||||
|
||||
|
||||
def _get_unstaged_count(cwd: str = None) -> int:
|
||||
output = _run_git_command(["diff", "--name-only"], cwd)
|
||||
return len(output.splitlines()) if output else 0
|
||||
def _get_remote_url_impl(cwd: Path | None = None) -> str | None:
|
||||
"""Get remote URL - internal implementation."""
|
||||
try:
|
||||
output = run_git_command("remote", "get-url", "origin", cwd=cwd)
|
||||
return output if output else None
|
||||
|
||||
except GitStatusError:
|
||||
return None
|
||||
|
||||
|
||||
def _get_untracked_count(cwd: str = None) -> int:
|
||||
output = _run_git_command(["ls-files", "--others", "--exclude-standard"], cwd)
|
||||
return len(output.splitlines()) if output else 0
|
||||
def get_remote_url(cwd: Path | None = None) -> str | None:
|
||||
"""Get remote URL for origin.
|
||||
|
||||
Args:
|
||||
cwd: Working directory.
|
||||
|
||||
Returns:
|
||||
Remote URL string or None.
|
||||
"""
|
||||
return _get_remote_url_impl(cwd)
|
||||
|
||||
|
||||
def _is_git_repo_impl(cwd: Path | None = None) -> bool:
|
||||
"""Check if directory is a git repository - internal implementation."""
|
||||
try:
|
||||
run_git_command("rev-parse", "--git-dir", cwd=cwd)
|
||||
return True
|
||||
except GitStatusError:
|
||||
return False
|
||||
|
||||
|
||||
def is_git_repo(cwd: Path | None = None) -> bool:
|
||||
"""Check if directory is a git repository.
|
||||
|
||||
Args:
|
||||
cwd: Directory to check.
|
||||
|
||||
Returns:
|
||||
True if directory is a git repository.
|
||||
"""
|
||||
return _is_git_repo_impl(cwd)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GitStatus:
|
||||
"""Git repository status data."""
|
||||
|
||||
branch: str
|
||||
commit: str
|
||||
staged: int = 0
|
||||
unstaged: int = 0
|
||||
untracked: int = 0
|
||||
commit_hash: str
|
||||
commit_message: str | None = None
|
||||
author_name: str | None = None
|
||||
author_date: datetime | None = None
|
||||
staged_files: int = 0
|
||||
unstaged_files: int = 0
|
||||
untracked_files: int = 0
|
||||
ahead: int = 0
|
||||
behind: int = 0
|
||||
is_detached: bool = False
|
||||
remote_name: str | None = None
|
||||
remote_url: str | None = None
|
||||
is_clean: bool = True
|
||||
is_detached: bool = False
|
||||
|
||||
|
||||
def get_git_status(cwd: str = None) -> GitStatus:
|
||||
try:
|
||||
is_detached_head = _is_detached(cwd)
|
||||
branch = "(detached)" if is_detached_head else _get_current_branch(cwd)
|
||||
commit = _get_commit_hash(cwd)
|
||||
staged = _get_staged_count(cwd)
|
||||
unstaged = _get_unstaged_count(cwd)
|
||||
untracked = _get_untracked_count(cwd)
|
||||
is_clean = (staged == 0 and unstaged == 0 and untracked == 0)
|
||||
def get_git_status(cwd: Path | None = None) -> GitStatus:
|
||||
"""Get comprehensive git status for a repository.
|
||||
|
||||
Args:
|
||||
cwd: Repository directory path.
|
||||
|
||||
Returns:
|
||||
GitStatus dataclass with all status information.
|
||||
"""
|
||||
if cwd and not _is_git_repo_impl(cwd):
|
||||
raise GitStatusError(f"Not a git repository: {cwd}")
|
||||
|
||||
branch, is_detached = _get_branch_impl(cwd)
|
||||
commit_hash = _get_commit_hash_impl(cwd)
|
||||
commit_message, author_name, author_date = _get_commit_info_impl(cwd)
|
||||
staged, unstaged, untracked = _get_file_counts_impl(cwd)
|
||||
ahead, behind, remote_name = _get_remote_status_impl(cwd)
|
||||
remote_url = _get_remote_url_impl(cwd)
|
||||
|
||||
is_clean = staged == 0 and unstaged == 0 and untracked == 0
|
||||
|
||||
return GitStatus(
|
||||
branch=branch,
|
||||
commit=commit,
|
||||
staged=staged,
|
||||
unstaged=unstaged,
|
||||
untracked=untracked,
|
||||
is_detached=is_detached_head,
|
||||
commit_hash=commit_hash,
|
||||
commit_message=commit_message,
|
||||
author_name=author_name,
|
||||
author_date=author_date,
|
||||
staged_files=staged,
|
||||
unstaged_files=unstaged,
|
||||
untracked_files=untracked,
|
||||
ahead=ahead,
|
||||
behind=behind,
|
||||
remote_name=remote_name,
|
||||
remote_url=remote_url,
|
||||
is_clean=is_clean,
|
||||
is_detached=is_detached,
|
||||
)
|
||||
except GitStatusError:
|
||||
raise GitStatusError("Failed to get git status")
|
||||
|
||||
|
||||
def is_git_repo(cwd: str = None) -> bool:
|
||||
try:
|
||||
_run_git_command(["rev-parse", "--git-dir"], cwd)
|
||||
return True
|
||||
except GitStatusError:
|
||||
return False
|
||||
|
||||
Reference in New Issue
Block a user