Implement LocalRepoOperator and RemoteRepoOperator classes

src/codegen/git/repo_operator/local_repo_operator.py

@@ -0,0 +1,59 @@
+from typing import Self
+
+from sqlalchemy.orm import Session
+
+from codegen.git.repo_operator.repo_operator import RepoOperator
+from codegen.git.schemas.enums import SetupOption
+from codegen.shared.logging.get_logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class LocalRepoOperator(RepoOperator):
+    """A local implementation of RepoOperator that works with local repositories.
+
+    This class extends the base RepoOperator to provide functionality specific to
+    working with repositories that are already on the local filesystem.
+    """
+
+    @classmethod
+    def from_repo_id(cls, db: Session, repo_id: int, setup_option: SetupOption = SetupOption.PULL_OR_CLONE, shallow: bool = True) -> Self:
+        """Create a LocalRepoOperator from a repository ID in the database.
+
+        Args:
+            db (Session): Database session
+            repo_id (int): ID of the repository in the database
+            setup_option (SetupOption): How to set up the repository
+            shallow (bool): Whether to do a shallow clone
+
+        Returns:
+            Self: A new LocalRepoOperator instance
+        """
+        from app.db import RepoModel
+
+        repo_model = RepoModel.find_by_id(db, repo_id)
+        if not repo_model:
+            msg = f"Repository with ID {repo_id} not found"
+            raise ValueError(msg)
+
+        return cls(repo_model=repo_model, setup_option=setup_option, shallow=shallow)
+
+    def __init__(
+        self,
+        repo_model,
+        base_dir: str = "/tmp",
+        setup_option: SetupOption = SetupOption.PULL_OR_CLONE,
+        shallow: bool = True,
+    ):
+        """Initialize a LocalRepoOperator.
+
+        Args:
+            repo_model: Repository model from the database
+            base_dir (str): Base directory for the repository
+            setup_option (SetupOption): How to set up the repository
+            shallow (bool): Whether to do a shallow clone
+        """
+        self.repo_model = repo_model
+        repo_config = repo_model.repo_config
+        repo_config.base_dir = base_dir
+        super().__init__(repo_config=repo_config, setup_option=setup_option, shallow=shallow)

src/codegen/git/repo_operator/remote_repo_operator.py

@@ -0,0 +1,80 @@
+import os
+from typing import Optional, override
+
+from sqlalchemy.orm import Session
+
+from codegen.git.clients.git_repo_client import GitRepoClient
+from codegen.git.repo_operator.repo_operator import RepoOperator
+from codegen.git.schemas.enums import SetupOption
+from codegen.shared.logging.get_logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class RemoteRepoOperator(RepoOperator):
+    """A remote implementation of RepoOperator that works with repositories hosted on GitHub.
+
+    This class extends the base RepoOperator to provide functionality specific to
+    working with repositories that are hosted on GitHub or GitHub Enterprise.
+    """
+
+    db: Session
+    repo_model: any  # RepoModel
+
+    def __init__(
+        self,
+        db: Session,
+        repo_model,
+        base_dir: str = "/tmp",
+        setup_option: SetupOption = SetupOption.PULL_OR_CLONE,
+        shallow: bool = True,
+        access_token: Optional[str] = None,
+    ):
+        """Initialize a RemoteRepoOperator.
+
+        Args:
+            db (Session): Database session
+            repo_model: Repository model from the database
+            base_dir (str): Base directory for the repository
+            setup_option (SetupOption): How to set up the repository
+            shallow (bool): Whether to do a shallow clone
+            access_token (str): GitHub access token
+        """
+        self.db = db
+        self.repo_model = repo_model
+        repo_config = repo_model.repo_config
+        repo_config.base_dir = base_dir
+        super().__init__(repo_config=repo_config, setup_option=setup_option, shallow=shallow, access_token=access_token)
+
+    @property
+    @override
+    def remote_git_repo(self) -> GitRepoClient:
+        """Get the remote Git repository client.
+
+        Returns:
+            GitRepoClient: The remote Git repository client
+        """
+        if not self._remote_git_repo:
+            self._remote_git_repo = GitRepoClient(self.repo_config, access_token=self.access_token)
+        return self._remote_git_repo
+
+    @override
+    def setup_repo_dir(self, setup_option: SetupOption = SetupOption.PULL_OR_CLONE, shallow: bool = True) -> None:
+        """Set up the repository directory.
+
+        Args:
+            setup_option (SetupOption): How to set up the repository
+            shallow (bool): Whether to do a shallow clone
+        """
+        os.makedirs(self.base_dir, exist_ok=True)
+        os.chdir(self.base_dir)
+        if setup_option is SetupOption.CLONE:
+            # if repo exists delete, then clone, else clone
+            self.clone_repo(shallow=shallow)
+        elif setup_option is SetupOption.PULL_OR_CLONE:
+            # if repo exists, pull changes, else clone
+            self.clone_or_pull_repo(shallow=shallow)
+        elif setup_option is SetupOption.SKIP:
+            if not self.repo_exists():
+                logger.warning(f"Valid git repo does not exist at {self.repo_path}. Cannot skip setup with SetupOption.SKIP.")
+        os.chdir(self.repo_path)

tests/unit/codegen/git/repo_operator/test_local_repo_operator.py

@@ -0,0 +1,75 @@
+import os
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from codegen.git.repo_operator.local_repo_operator import LocalRepoOperator
+from codegen.git.schemas.enums import SetupOption
+from codegen.git.schemas.repo_config import RepoConfig
+
+
+@pytest.fixture
+def mock_repo_model():
+    repo_model = MagicMock()
+    repo_config = RepoConfig(
+        name="test-repo",
+        full_name="test-org/test-repo",
+        clone_url="https://github.com/test-org/test-repo.git",
+    )
+    repo_model.repo_config = repo_config
+    return repo_model
+
+
+@pytest.fixture
+def temp_dir():
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        yield tmp_dir
+
+
+def test_local_repo_operator_init(mock_repo_model, temp_dir):
+    """Test that LocalRepoOperator initializes correctly."""
+    # Patch the setup_repo_dir method to avoid actual Git operations
+    with patch.object(LocalRepoOperator, "setup_repo_dir"):
+        op = LocalRepoOperator(
+            repo_model=mock_repo_model,
+            base_dir=temp_dir,
+            setup_option=SetupOption.SKIP,
+        )
+
+        assert op.repo_model == mock_repo_model
+        assert op.base_dir == temp_dir
+        assert op.repo_name == "test-repo"
+        assert op.repo_path == os.path.join(temp_dir, "test-repo")
+
+
+@patch("app.db.RepoModel")
+def test_from_repo_id(mock_repo_model_class, mock_repo_model):
+    """Test the from_repo_id class method."""
+    mock_db = MagicMock()
+    mock_repo_model_class.find_by_id.return_value = mock_repo_model
+
+    # Patch the setup_repo_dir method to avoid actual Git operations
+    with patch.object(LocalRepoOperator, "setup_repo_dir"):
+        op = LocalRepoOperator.from_repo_id(
+            db=mock_db,
+            repo_id=123,
+            setup_option=SetupOption.SKIP,
+        )
+
+        mock_repo_model_class.find_by_id.assert_called_once_with(mock_db, 123)
+        assert op.repo_model == mock_repo_model
+
+
+@patch("app.db.RepoModel")
+def test_from_repo_id_not_found(mock_repo_model_class):
+    """Test the from_repo_id class method when the repo is not found."""
+    mock_db = MagicMock()
+    mock_repo_model_class.find_by_id.return_value = None
+
+    with pytest.raises(ValueError, match="Repository with ID 123 not found"):
+        LocalRepoOperator.from_repo_id(
+            db=mock_db,
+            repo_id=123,
+            setup_option=SetupOption.SKIP,
+        )

tests/unit/codegen/git/repo_operator/test_remote_repo_operator.py

@@ -0,0 +1,131 @@
+import os
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from codegen.git.repo_operator.remote_repo_operator import RemoteRepoOperator
+from codegen.git.schemas.enums import SetupOption
+from codegen.git.schemas.repo_config import RepoConfig
+
+
+@pytest.fixture
+def mock_repo_model():
+    repo_model = MagicMock()
+    repo_config = RepoConfig(
+        name="test-repo",
+        full_name="test-org/test-repo",
+        clone_url="https://github.com/test-org/test-repo.git",
+    )
+    repo_model.repo_config = repo_config
+    return repo_model
+
+
+@pytest.fixture
+def mock_db():
+    return MagicMock()
+
+
+@pytest.fixture
+def temp_dir():
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        yield tmp_dir
+
+
+def test_remote_repo_operator_init(mock_db, mock_repo_model, temp_dir):
+    """Test that RemoteRepoOperator initializes correctly."""
+    # Patch the setup_repo_dir method to avoid actual Git operations
+    with patch.object(RemoteRepoOperator, "setup_repo_dir"):
+        op = RemoteRepoOperator(
+            db=mock_db,
+            repo_model=mock_repo_model,
+            base_dir=temp_dir,
+            setup_option=SetupOption.SKIP,
+            access_token="test-token",
+        )
+
+        assert op.db == mock_db
+        assert op.repo_model == mock_repo_model
+        assert op.base_dir == temp_dir
+        assert op.repo_name == "test-repo"
+        assert op.repo_path == os.path.join(temp_dir, "test-repo")
+        assert op.access_token == "test-token"
+
+
+def test_remote_git_repo_property(mock_db, mock_repo_model, temp_dir):
+    """Test the remote_git_repo property."""
+    # Patch the setup_repo_dir method to avoid actual Git operations
+    with patch.object(RemoteRepoOperator, "setup_repo_dir"):
+        # Patch the GitRepoClient constructor
+        with patch("codegen.git.clients.git_repo_client.GitRepoClient") as mock_client:
+            mock_client_instance = MagicMock()
+            mock_client.return_value = mock_client_instance
+
+            op = RemoteRepoOperator(
+                db=mock_db,
+                repo_model=mock_repo_model,
+                base_dir=temp_dir,
+                setup_option=SetupOption.SKIP,
+                access_token="test-token",
+            )
+
+            # Access the property to trigger the lazy initialization
+            remote_repo = op.remote_git_repo
+
+            # Verify that the GitRepoClient was created with the correct arguments
+            mock_client.assert_called_once_with(op.repo_config, access_token="test-token")
+            assert remote_repo == mock_client_instance
+
+
+def test_setup_repo_dir(mock_db, mock_repo_model, temp_dir):
+    """Test the setup_repo_dir method."""
+    # Patch the os.makedirs and os.chdir functions
+    with (
+        patch("os.makedirs") as mock_makedirs,
+        patch("os.chdir") as mock_chdir,
+        patch.object(RemoteRepoOperator, "clone_repo") as mock_clone_repo,
+        patch.object(RemoteRepoOperator, "clone_or_pull_repo") as mock_clone_or_pull_repo,
+        patch.object(RemoteRepoOperator, "repo_exists", return_value=False) as mock_repo_exists,
+    ):
+        op = RemoteRepoOperator(
+            db=mock_db,
+            repo_model=mock_repo_model,
+            base_dir=temp_dir,
+            setup_option=None,  # Don't call setup_repo_dir in __init__
+        )
+
+        # Test CLONE option
+        op.setup_repo_dir(setup_option=SetupOption.CLONE, shallow=True)
+        mock_makedirs.assert_called_with(temp_dir, exist_ok=True)
+        mock_chdir.assert_any_call(temp_dir)
+        mock_clone_repo.assert_called_once_with(shallow=True)
+        mock_chdir.assert_called_with(op.repo_path)
+
+        # Reset mocks
+        mock_makedirs.reset_mock()
+        mock_chdir.reset_mock()
+        mock_clone_repo.reset_mock()
+
+        # Test PULL_OR_CLONE option
+        op.setup_repo_dir(setup_option=SetupOption.PULL_OR_CLONE, shallow=False)
+        mock_makedirs.assert_called_with(temp_dir, exist_ok=True)
+        mock_chdir.assert_any_call(temp_dir)
+        mock_clone_or_pull_repo.assert_called_once_with(shallow=False)
+        mock_chdir.assert_called_with(op.repo_path)
+
+        # Reset mocks
+        mock_makedirs.reset_mock()
+        mock_chdir.reset_mock()
+        mock_clone_or_pull_repo.reset_mock()
+
+        # Test SKIP option
+        with patch("codegen.shared.logging.get_logger.get_logger") as mock_logger:
+            mock_logger_instance = MagicMock()
+            mock_logger.return_value = mock_logger_instance
+
+            op.setup_repo_dir(setup_option=SetupOption.SKIP)
+            mock_makedirs.assert_called_with(temp_dir, exist_ok=True)
+            mock_chdir.assert_any_call(temp_dir)
+            mock_repo_exists.assert_called_once()
+            mock_logger_instance.warning.assert_called_once()
+            mock_chdir.assert_called_with(op.repo_path)