feat(git): add git resource

Author: Datata1

src/codesphere/resources/workspace/__init__.py

@@ -1,12 +1,13 @@
+from .git import GitHead, WorkspaceGitManager
+from .resources import WorkspacesResource
 from .schemas import (
+    CommandInput,
+    CommandOutput,
     Workspace,
     WorkspaceCreate,
-    WorkspaceUpdate,
     WorkspaceStatus,
-    CommandInput,
-    CommandOutput,
+    WorkspaceUpdate,
 )
-from .resources import WorkspacesResource
 
 __all__ = [
     "Workspace",
@@ -16,4 +17,6 @@
     "WorkspacesResource",
     "CommandInput",
     "CommandOutput",
+    "WorkspaceGitManager",
+    "GitHead",
 ]

src/codesphere/resources/workspace/git/__init__.py

@@ -0,0 +1,4 @@
+from .models import WorkspaceGitManager
+from .schema import GitHead
+
+__all__ = ["WorkspaceGitManager", "GitHead"]

src/codesphere/resources/workspace/git/models.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from ....core.handler import _APIOperationExecutor
+from ....http_client import APIHttpClient
+from .operations import (
+    _GET_HEAD_OP,
+    _PULL_OP,
+    _PULL_WITH_REMOTE_AND_BRANCH_OP,
+    _PULL_WITH_REMOTE_OP,
+)
+from .schema import GitHead
+
+log = logging.getLogger(__name__)
+
+
+class WorkspaceGitManager(_APIOperationExecutor):
+    """Manager for git operations on a workspace."""
+
+    def __init__(self, http_client: APIHttpClient, workspace_id: int):
+        self._http_client = http_client
+        self._workspace_id = workspace_id
+        self.id = workspace_id
+
+    async def get_head(self) -> GitHead:
+        return await self._execute_operation(_GET_HEAD_OP)
+
+    async def pull(
+        self,
+        remote: Optional[str] = None,
+        branch: Optional[str] = None,
+    ) -> None:
+        if remote is not None and branch is not None:
+            await self._execute_operation(
+                _PULL_WITH_REMOTE_AND_BRANCH_OP, remote=remote, branch=branch
+            )
+        elif remote is not None:
+            await self._execute_operation(_PULL_WITH_REMOTE_OP, remote=remote)
+        else:
+            await self._execute_operation(_PULL_OP)

src/codesphere/resources/workspace/git/operations.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from ....core.operations import APIOperation
+from .schema import GitHead
+
+_GET_HEAD_OP = APIOperation(
+    method="GET",
+    endpoint_template="/workspaces/{id}/git/head",
+    response_model=GitHead,
+)
+
+_PULL_OP = APIOperation(
+    method="POST",
+    endpoint_template="/workspaces/{id}/git/pull",
+    response_model=type(None),
+)
+
+_PULL_WITH_REMOTE_OP = APIOperation(
+    method="POST",
+    endpoint_template="/workspaces/{id}/git/pull/{remote}",
+    response_model=type(None),
+)
+
+_PULL_WITH_REMOTE_AND_BRANCH_OP = APIOperation(
+    method="POST",
+    endpoint_template="/workspaces/{id}/git/pull/{remote}/{branch}",
+    response_model=type(None),
+)

src/codesphere/resources/workspace/git/schema.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from ....core.base import CamelModel
+
+
+class GitHead(CamelModel):
+    head: str

src/codesphere/resources/workspace/pipeline/resources.py

@@ -9,6 +9,7 @@
 from ...core.base import CamelModel
 from ...utils import update_model_fields
 from .envVars import EnvVar, WorkspaceEnvVarManager
+from .git import WorkspaceGitManager
 from .landscape import WorkspaceLandscapeManager
 
 log = logging.getLogger(__name__)
@@ -137,3 +138,9 @@ def landscape(self) -> WorkspaceLandscapeManager:
         """Manager for landscape operations (Multi Server Deployments)."""
         http_client = self.validate_http_client()
         return WorkspaceLandscapeManager(http_client, workspace_id=self.id)
+
+    @cached_property
+    def git(self) -> WorkspaceGitManager:
+        """Manager for git operations (head, pull)."""
+        http_client = self.validate_http_client()
+        return WorkspaceGitManager(http_client, workspace_id=self.id)

src/codesphere/resources/workspace/schemas.py

@@ -0,0 +1,55 @@
+from typing import AsyncGenerator
+
+import pytest
+
+from codesphere import CodesphereSDK
+from codesphere.resources.workspace import Workspace, WorkspaceCreate
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture(scope="module")
+async def git_workspace(
+    module_sdk_client: CodesphereSDK,
+    test_team_id: int,
+    test_plan_id: int,
+) -> AsyncGenerator[Workspace, None]:
+    payload = WorkspaceCreate(
+        team_id=test_team_id,
+        name="sdk-git-integration-test",
+        plan_id=test_plan_id,
+        git_url="https://github.com/octocat/Hello-World.git",
+    )
+
+    workspace = await module_sdk_client.workspaces.create(payload=payload)
+
+    try:
+        await workspace.wait_until_running(timeout=120.0)
+        yield workspace
+    finally:
+        try:
+            await workspace.delete()
+        except Exception:
+            pass
+
+
+class TestGitIntegration:
+    @pytest.mark.asyncio
+    async def test_get_head(self, git_workspace: Workspace):
+        result = await git_workspace.git.get_head()
+
+        assert result.head is not None
+        assert len(result.head) > 0
+
+    @pytest.mark.asyncio
+    async def test_pull_default(self, git_workspace: Workspace):
+        # This should not raise an exception
+        await git_workspace.git.pull()
+
+    @pytest.mark.asyncio
+    async def test_pull_with_remote(self, git_workspace: Workspace):
+        await git_workspace.git.pull(remote="origin")
+
+    @pytest.mark.asyncio
+    async def test_pull_with_remote_and_branch(self, git_workspace: Workspace):
+        await git_workspace.git.pull(remote="origin", branch="master")

tests/integration/test_git.py

@@ -0,0 +1,123 @@
+import pytest
+
+from codesphere.resources.workspace.git import GitHead, WorkspaceGitManager
+
+
+class TestWorkspaceGitManager:
+    @pytest.fixture
+    def git_manager(self, mock_http_client_for_resource):
+        def _create(response_data):
+            mock_client = mock_http_client_for_resource(response_data)
+            manager = WorkspaceGitManager(http_client=mock_client, workspace_id=72678)
+            return manager, mock_client
+
+        return _create
+
+    @pytest.mark.asyncio
+    async def test_get_head(self, git_manager):
+        manager, mock_client = git_manager({"head": "abc123def456"})
+
+        result = await manager.get_head()
+
+        assert isinstance(result, GitHead)
+        assert result.head == "abc123def456"
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "GET"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/git/head"
+
+    @pytest.mark.asyncio
+    async def test_pull_without_arguments(self, git_manager):
+        manager, mock_client = git_manager(None)
+
+        await manager.pull()
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/git/pull"
+
+    @pytest.mark.asyncio
+    async def test_pull_with_remote(self, git_manager):
+        manager, mock_client = git_manager(None)
+
+        await manager.pull(remote="origin")
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/git/pull/origin"
+
+    @pytest.mark.asyncio
+    async def test_pull_with_remote_and_branch(self, git_manager):
+        manager, mock_client = git_manager(None)
+
+        await manager.pull(remote="origin", branch="main")
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert (
+            call_args.kwargs.get("endpoint") == "/workspaces/72678/git/pull/origin/main"
+        )
+
+    @pytest.mark.asyncio
+    async def test_pull_with_branch_only_ignores_branch(self, git_manager):
+        manager, mock_client = git_manager(None)
+
+        # Branch without remote should be ignored per the implementation
+        await manager.pull(branch="main")
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/git/pull"
+
+    @pytest.mark.asyncio
+    async def test_pull_with_custom_remote(self, git_manager):
+        """pull should work with custom remote names."""
+        manager, mock_client = git_manager(None)
+
+        await manager.pull(remote="upstream")
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/git/pull/upstream"
+
+    @pytest.mark.asyncio
+    async def test_pull_with_feature_branch(self, git_manager):
+        manager, mock_client = git_manager(None)
+
+        await manager.pull(remote="origin", branch="feature/my-feature")
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert (
+            call_args.kwargs.get("endpoint")
+            == "/workspaces/72678/git/pull/origin/feature/my-feature"
+        )
+
+
+class TestGitHeadModel:
+    def test_create_git_head(self):
+        git_head = GitHead(head="abc123def456")
+
+        assert git_head.head == "abc123def456"
+
+    def test_git_head_from_dict(self):
+        git_head = GitHead.model_validate({"head": "abc123def456"})
+
+        assert git_head.head == "abc123def456"
+
+    def test_git_head_dump(self):
+        git_head = GitHead(head="abc123def456")
+        dumped = git_head.model_dump()
+
+        assert dumped == {"head": "abc123def456"}
+
+    def test_git_head_with_full_sha(self):
+        full_sha = "a" * 40
+        git_head = GitHead(head=full_sha)
+
+        assert git_head.head == full_sha
+        assert len(git_head.head) == 40