diff --git a/gitea/workspace.py b/gitea/workspace.py index d650063..295f211 100644 --- a/gitea/workspace.py +++ b/gitea/workspace.py @@ -1,9 +1,12 @@ import os import logging import subprocess +import base64 +import shutil from pathlib import Path from urllib.parse import urlparse from .config import GITEA_REPOS_ROOT, GITEA_URL, GITEA_TOKEN +from gitea.client import GiteaClient logger: logging.Logger = logging.getLogger("gitea-workspace") @@ -14,49 +17,25 @@ class WorkspaceManager: def __init__(self) -> None: self.root_dir: Path = Path(GITEA_REPOS_ROOT).expanduser().resolve() self.root_dir.mkdir(parents=True, exist_ok=True) - self._configure_git_credentials() - - def _configure_git_credentials(self) -> None: - try: - # Unset any global configs we might have set previously - subprocess.run( - ["git", "config", "--global", "--unset", "credential.helper"], - capture_output=True - ) - subprocess.run( - ["git", "config", "--global", "--unset", "user.name"], - capture_output=True - ) - subprocess.run( - ["git", "config", "--global", "--unset", "user.email"], - capture_output=True - ) - except Exception as e: - logger.error(f"Error unsetting global configs: {e}") def _configure_repo_user(self, repo_path: Path) -> None: try: - # Configure credential helper locally for the repo - subprocess.run( - ["git", "-C", str(repo_path), "config", "credential.helper", "store"], - check=True, capture_output=True - ) - # Write to ~/.git-credentials - parsed = urlparse(GITEA_URL.rstrip("/")) - cred_line = f"{parsed.scheme}://meeks-ai:{GITEA_TOKEN}@{parsed.netloc}\n" - cred_file = Path("~/.git-credentials").expanduser() - if cred_file.exists(): - content = cred_file.read_text() - if cred_line.strip() not in content: - cred_file.write_text(content + cred_line) - else: - cred_file.write_text(cred_line) - - from gitea.client import GiteaClient client = GiteaClient() user = client.get_authenticated_user() + username = user.login if user else "unknown-ai" + + auth_str = f"{username}:{GITEA_TOKEN}" + auth_bytes = auth_str.encode("utf-8") + auth_b64 = base64.b64encode(auth_bytes).decode("utf-8") + + # Configure extraHeader locally for the repo + subprocess.run( + ["git", "-C", str(repo_path), "config", "http.extraHeader", f"Authorization: Basic {auth_b64}"], + check=True, capture_output=True + ) + if user: - name = user.full_name or user.login or "meeks-ai" + name = user.full_name or user.login or "unknown-ai" email = user.email or f"{user.login or 'agent'}@noreply.gitea" subprocess.run( ["git", "-C", str(repo_path), "config", "user.name", name], @@ -123,13 +102,28 @@ class WorkspaceManager: if not (repo_path / ".git").exists(): new_path: Path = repo_path.parent / f"{repo_full_name.replace('/', '_')}_old" if new_path.exists(): - import shutil shutil.rmtree(new_path) repo_path.rename(new_path) return repo_path logger.info(f"Cloning repository {repo_full_name} to {repo_path}...") auth_url = self._get_authenticated_url(repo_full_name) - subprocess.run(["git", "clone", auth_url, str(repo_path)], check=True, capture_output=True) + + username = "unknown-ai" + try: + client = GiteaClient() + user = client.get_authenticated_user() + if user: + username = user.login + except Exception: + pass + + auth_str = f"{username}:{GITEA_TOKEN}" + auth_bytes = auth_str.encode("utf-8") + auth_b64 = base64.b64encode(auth_bytes).decode("utf-8") + subprocess.run( + ["git", "clone", "-c", f"http.extraHeader=Authorization: Basic {auth_b64}", auth_url, str(repo_path)], + check=True, capture_output=True + ) self._configure_repo_user(repo_path) return repo_path diff --git a/tests/test_workspace.py b/tests/test_workspace.py new file mode 100644 index 0000000..0d6259f --- /dev/null +++ b/tests/test_workspace.py @@ -0,0 +1,53 @@ +from pathlib import Path +from unittest.mock import MagicMock, patch +import pytest + +from gitea.workspace import WorkspaceManager + + +@patch("gitea.workspace.subprocess.run") +@patch("gitea.workspace.GiteaClient") +def test_workspace_manager_configure_repo_user( + mock_client_class: MagicMock, + mock_run: MagicMock +) -> None: + mock_client = MagicMock() + mock_client_class.return_value = mock_client + mock_user = MagicMock() + mock_user.full_name = "Agent Tester" + mock_user.login = "agent-test" + mock_user.email = "agent-test@example.com" + mock_client.get_authenticated_user.return_value = mock_user + + workspace = WorkspaceManager() + repo_path = Path("/tmp/mock-repo") + + workspace._configure_repo_user(repo_path) + + assert mock_run.call_count >= 3 + calls = [c[0][0] for c in mock_run.call_args_list] + + assert any("http.extraHeader" in call for call in calls) + assert any("user.name" in call for call in calls) + assert any("user.email" in call for call in calls) + + +@patch("gitea.workspace.subprocess.run") +def test_workspace_manager_clone_repo( + mock_run: MagicMock +) -> None: + workspace = WorkspaceManager() + + with patch.object(workspace, "_configure_repo_user") as mock_configure: + with patch.object(workspace, "get_repo_path") as mock_get_path: + mock_repo_path = MagicMock(spec=Path) + mock_repo_path.exists.return_value = False + mock_get_path.return_value = mock_repo_path + + workspace.clone_repo("meeks/repo1") + + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + assert "clone" in args + assert any("http.extraHeader=Authorization: Basic" in arg for arg in args) + mock_configure.assert_called_once_with(mock_repo_path)