Enforce authenticated user and login validation in workspace operations, crashing program on failure instead of fallback defaults

gitea/workspace.py

@@ -22,11 +22,13 @@ class WorkspaceManager:
         try:
             client = GiteaClient()
             user = client.get_authenticated_user()
-            username = user.login if user else "unknown-ai"
+            if not user or not user.login:
+                raise RuntimeError("No authenticated user found.")
+            username: str = user.login
 
-            auth_str = f"{username}:{GITEA_TOKEN}"
-            auth_bytes = auth_str.encode("utf-8")
-            auth_b64 = base64.b64encode(auth_bytes).decode("utf-8")
+            auth_str: str = f"{username}:{GITEA_TOKEN}"
+            auth_bytes: bytes = auth_str.encode("utf-8")
+            auth_b64: str = base64.b64encode(auth_bytes).decode("utf-8")
 
             # Configure extraHeader locally for the repo
             subprocess.run(
@@ -34,9 +36,8 @@ class WorkspaceManager:
                 check=True, capture_output=True
             )
 
-            if user:
-                name = user.full_name or user.login or "unknown-ai"
-                email = user.email or f"{user.login or 'agent'}@noreply.gitea"
+            name: str = user.full_name or user.login
+            email: str = user.email or f"{user.login}@noreply.gitea"
             subprocess.run(
                 ["git", "-C", str(repo_path), "config", "user.name", name],
                 check=True, capture_output=True
@@ -47,6 +48,7 @@ class WorkspaceManager:
             )
         except Exception as e:
             logger.error(f"Error configuring local git user: {e}")
+            raise
 
     def get_repo_path(self, repo_full_name: str) -> Path:
         parts: list[str] = repo_full_name.split("/")
@@ -109,18 +111,15 @@ class WorkspaceManager:
         logger.info(f"Cloning repository {repo_full_name} to {repo_path}...")
         auth_url = self._get_authenticated_url(repo_full_name)
 
-        username = "unknown-ai"
-        try:
         client = GiteaClient()
         user = client.get_authenticated_user()
-            if user:
-                username = user.login
-        except Exception:
-            pass
+        if not user or not user.login:
+            raise RuntimeError("No authenticated user found.")
+        username: str = user.login
 
-        auth_str = f"{username}:{GITEA_TOKEN}"
-        auth_bytes = auth_str.encode("utf-8")
-        auth_b64 = base64.b64encode(auth_bytes).decode("utf-8")
+        auth_str: str = f"{username}:{GITEA_TOKEN}"
+        auth_bytes: bytes = auth_str.encode("utf-8")
+        auth_b64: str = 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

tests/test_workspace.py

@@ -33,9 +33,19 @@ def test_workspace_manager_configure_repo_user(
 
 
 @patch("gitea.workspace.subprocess.run")
+@patch("gitea.workspace.GiteaClient")
 def test_workspace_manager_clone_repo(
+    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()
     
     with patch.object(workspace, "_configure_repo_user") as mock_configure:
@@ -51,3 +61,37 @@ def test_workspace_manager_clone_repo(
             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)
+
+
+@patch("gitea.workspace.GiteaClient")
+def test_workspace_manager_fails_if_no_authenticated_user(
+    mock_client_class: MagicMock
+) -> None:
+    mock_client = MagicMock()
+    mock_client_class.return_value = mock_client
+    mock_client.get_authenticated_user.return_value = None
+
+    workspace = WorkspaceManager()
+    with pytest.raises(RuntimeError, match="No authenticated user found."):
+        workspace.clone_repo("meeks/repo1")
+
+    with pytest.raises(RuntimeError, match="No authenticated user found."):
+        workspace._configure_repo_user(Path("/tmp/mock-repo"))
+
+
+@patch("gitea.workspace.GiteaClient")
+def test_workspace_manager_fails_if_authenticated_user_has_no_login(
+    mock_client_class: MagicMock
+) -> None:
+    mock_client = MagicMock()
+    mock_client_class.return_value = mock_client
+    mock_user = MagicMock()
+    mock_user.login = ""
+    mock_client.get_authenticated_user.return_value = mock_user
+
+    workspace = WorkspaceManager()
+    with pytest.raises(RuntimeError, match="No authenticated user found."):
+        workspace.clone_repo("meeks/repo1")
+
+    with pytest.raises(RuntimeError, match="No authenticated user found."):
+        workspace._configure_repo_user(Path("/tmp/mock-repo"))