lai_automatisierung/src/lai_automation/git.py

from __future__ import annotations

import os
import pwd
import shlex
import shutil
import subprocess
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlparse

from .config import Settings


@dataclass(frozen=True)
class ChangeSummary:
    files: tuple[str, ...]
    added_lines: int
    deleted_lines: int

    @property
    def changed_lines(self) -> int:
        return self.added_lines + self.deleted_lines


class GitWorkspace:
    def __init__(self, settings: Settings, report_id: int) -> None:
        self.settings = settings
        self.path = settings.workspace_root / f"failure-{report_id}"
        self.branch = f"ai/failure-{report_id}"

    def prepare(self) -> None:
        repository = urlparse(self.settings.source_repository)
        gitea = urlparse(self.settings.gitea_url)
        if repository.scheme != "https" or repository.hostname != gitea.hostname:
            raise ValueError("SOURCE_REPOSITORY must use HTTPS on the configured Gitea host")
        shutil.rmtree(self.path, ignore_errors=True)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._git("clone", "--branch", self.settings.source_branch, "--single-branch", self.settings.source_repository, str(self.path), cwd=self.path.parent)
        self._git("config", "user.name", "LAI Automation")
        self._git("config", "user.email", "automation@localhost")
        self._git("checkout", "-b", self.branch)

    def run(
        self,
        command: str,
        check: bool = True,
        username: str | None = None,
        extra_environment: Mapping[str, str] | None = None,
    ) -> subprocess.CompletedProcess[str]:
        return self.run_args(
            shlex.split(command),
            check=check,
            username=username,
            extra_environment=extra_environment,
        )

    def run_args(
        self,
        command: Sequence[str],
        check: bool = True,
        username: str | None = None,
        extra_environment: Mapping[str, str] | None = None,
    ) -> subprocess.CompletedProcess[str]:
        environment = dict(os.environ)
        drop_privileges = None
        if username:
            account = pwd.getpwnam(username)
            environment = {
                "HOME": account.pw_dir,
                "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
            }

            def drop_privileges() -> None:
                os.setgroups([])
                os.setgid(account.pw_gid)
                os.setuid(account.pw_uid)

            environment.update(extra_environment or {})

        return subprocess.run(
            command,
            cwd=self.path,
            env=environment,
            text=True,
            capture_output=True,
            check=check,
            timeout=1800,
            preexec_fn=drop_privileges,
        )

    def discard_changes(self) -> None:
        self._git("reset", "--hard", "HEAD")
        self._git("clean", "-fd")

    def summary(self) -> ChangeSummary:
        self._git("add", "-N", "--all")
        output = self._git("diff", "--numstat").stdout
        files: list[str] = []
        added = deleted = 0
        for line in output.splitlines():
            add, remove, filename = line.split("\t", 2)
            files.append(filename)
            added += int(add) if add.isdigit() else 0
            deleted += int(remove) if remove.isdigit() else 0
        return ChangeSummary(tuple(files), added, deleted)

    def commit(self, report_id: int) -> str:
        self._git("add", "--all")
        self._git("commit", "-m", f"Fix failure report #{report_id}")
        return self._git("rev-parse", "HEAD").stdout.strip()

    def push_branch(self) -> None:
        self._git_authenticated("push", "origin", f"HEAD:{self.branch}")

    def push_main(self) -> None:
        self._git_authenticated("push", "origin", f"HEAD:{self.settings.source_branch}")

    def cleanup(self) -> None:
        if self.path.exists():
            shutil.rmtree(self.path)

    def _git(self, *args: str, cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            ["git", *args],
            cwd=cwd or self.path,
            text=True,
            capture_output=True,
            check=True,
            timeout=300,
        )

    def _git_authenticated(self, *args: str) -> subprocess.CompletedProcess[str]:
        askpass = self.path / ".git" / "lai-askpass.sh"
        username_file = self.path / ".git" / "lai-bot-user"
        token_file = self.path / ".git" / "lai-bot-token"
        try:
            username_file.write_text(self.settings.gitea_username)
            token_file.write_text(self.settings.gitea_token)
            username_file.chmod(0o600)
            token_file.chmod(0o600)
            askpass.write_text(
                "#!/bin/sh\ncase \"$1\" in *Username*) cat \"$GIT_BOT_USER_FILE\";; *) cat \"$GIT_BOT_TOKEN_FILE\";; esac\n"
            )
            askpass.chmod(0o700)
            env = {
                "HOME": os.environ.get("HOME", "/root"),
                "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
                "GIT_ASKPASS": str(askpass),
                "GIT_TERMINAL_PROMPT": "0",
                "GIT_BOT_USER_FILE": str(username_file),
                "GIT_BOT_TOKEN_FILE": str(token_file),
            }
            return subprocess.run(
                ["git", *args],
                cwd=self.path,
                env=env,
                text=True,
                capture_output=True,
                check=True,
                timeout=300,
            )
        finally:
            askpass.unlink(missing_ok=True)
            username_file.unlink(missing_ok=True)
            token_file.unlink(missing_ok=True)