Merge branch 'main' into dev

Author: AbhiRam162105

src/gitingest/cli.py

@@ -109,7 +109,14 @@ async def async_main(
         exclude_patterns.update(ignore_patterns)
 
         # Perform the ingest operation with branch support
-        summary, *_ = await ingest(source, max_size, include_patterns, exclude_patterns, branch=branch, output=output)
+        summary, *_ = await ingest(
+            source,
+            max_size,
+            include_patterns,
+            exclude_patterns,
+            branch=branch,
+            output=output
+        )
 
         # Display results
         click.echo(f"Analysis complete! Output written to: {output}")
@@ -131,4 +138,4 @@ async def async_main(
 
 
 if __name__ == "__main__":
-    main()
+    main()

src/server/main.py

@@ -1,6 +1,7 @@
 """ Main module for the FastAPI application. """
 
 import os
+from pathlib import Path
 
 from api_analytics.fastapi import Analytics
 from dotenv import load_dotenv
@@ -24,8 +25,11 @@
 # Register the custom exception handler for rate limits
 app.add_exception_handler(RateLimitExceeded, rate_limit_exception_handler)
 
-# Mount static files to serve CSS, JS, and other static assets
-app.mount("/static", StaticFiles(directory="static"), name="static")
+
+# Mount static files dynamically to serve CSS, JS, and other static assets
+static_dir = Path(__file__).parent.parent / "static"
+app.mount("/static", StaticFiles(directory=static_dir), name="static")
+
 
 # Set up API analytics middleware if an API key is provided
 if app_analytics_key := os.getenv("API_ANALYTICS_KEY"):

tests/test_flow_integration.py

@@ -0,0 +1,159 @@
+"""
+Integration tests for GitIngest.
+These tests cover core functionalities, edge cases, and concurrency handling.
+"""
+
+import shutil
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+from src.server.main import app
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+TEMPLATE_DIR = BASE_DIR / "src" / "templates"
+
+
+@pytest.fixture(scope="module")
+def test_client():
+    """Create a test client fixture."""
+    with TestClient(app) as client_instance:
+        client_instance.headers.update({"Host": "localhost"})
+        yield client_instance
+
+
+@pytest.fixture(scope="module", autouse=True)
+def mock_static_files():
+    """Mock the static file mount to avoid directory errors."""
+    with patch("src.server.main.StaticFiles") as mock_static:
+        mock_static.return_value = None  # Mocks the StaticFiles response
+        yield mock_static
+
+
+@pytest.fixture(scope="module", autouse=True)
+def mock_templates():
+    """Mock Jinja2 template rendering to bypass actual file loading."""
+    with patch("starlette.templating.Jinja2Templates.TemplateResponse") as mock_template:
+        mock_template.return_value = "Mocked Template Response"
+        yield mock_template
+
+
+def cleanup_temp_directories():
+    temp_dir = Path("/tmp/gitingest")
+    if temp_dir.exists():
+        try:
+            shutil.rmtree(temp_dir)
+        except PermissionError as e:
+            print(f"Error cleaning up {temp_dir}: {e}")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def cleanup():
+    """Cleanup temporary directories after tests."""
+    yield
+    cleanup_temp_directories()
+
+
+@pytest.mark.asyncio
+async def test_remote_repository_analysis(request):
+    """Test the complete flow of analyzing a remote repository."""
+    client = request.getfixturevalue("test_client")
+    form_data = {
+        "input_text": "https://github.com/octocat/Hello-World",
+        "max_file_size": "243",
+        "pattern_type": "exclude",
+        "pattern": "",
+    }
+
+    response = client.post("/", data=form_data)
+    assert response.status_code == 200, f"Form submission failed: {response.text}"
+    assert "Mocked Template Response" in response.text
+
+
+@pytest.mark.asyncio
+async def test_invalid_repository_url(request):
+    """Test handling of an invalid repository URL."""
+    client = request.getfixturevalue("test_client")
+    form_data = {
+        "input_text": "https://github.com/nonexistent/repo",
+        "max_file_size": "243",
+        "pattern_type": "exclude",
+        "pattern": "",
+    }
+
+    response = client.post("/", data=form_data)
+    assert response.status_code == 200, f"Request failed: {response.text}"
+    assert "Mocked Template Response" in response.text
+
+
+@pytest.mark.asyncio
+async def test_large_repository(request):
+    """Simulate analysis of a large repository with nested folders."""
+    client = request.getfixturevalue("test_client")
+    form_data = {
+        "input_text": "https://github.com/large/repo-with-many-files",
+        "max_file_size": "243",
+        "pattern_type": "exclude",
+        "pattern": "",
+    }
+
+    response = client.post("/", data=form_data)
+    assert response.status_code == 200, f"Request failed: {response.text}"
+    assert "Mocked Template Response" in response.text
+
+
+@pytest.mark.asyncio
+async def test_concurrent_requests(request):
+    """Test handling of multiple concurrent requests."""
+    client = request.getfixturevalue("test_client")
+
+    def make_request():
+        form_data = {
+            "input_text": "https://github.com/octocat/Hello-World",
+            "max_file_size": "243",
+            "pattern_type": "exclude",
+            "pattern": "",
+        }
+        response = client.post("/", data=form_data)
+        assert response.status_code == 200, f"Request failed: {response.text}"
+        assert "Mocked Template Response" in response.text
+
+    with ThreadPoolExecutor(max_workers=5) as executor:
+        futures = [executor.submit(make_request) for _ in range(5)]
+        for future in futures:
+            future.result()
+
+
+@pytest.mark.asyncio
+async def test_large_file_handling(request):
+    """Test handling of repositories with large files."""
+    client = request.getfixturevalue("test_client")
+    form_data = {
+        "input_text": "https://github.com/octocat/Hello-World",
+        "max_file_size": "1",
+        "pattern_type": "exclude",
+        "pattern": "",
+    }
+
+    response = client.post("/", data=form_data)
+    assert response.status_code == 200, f"Request failed: {response.text}"
+    assert "Mocked Template Response" in response.text
+
+
+@pytest.mark.asyncio
+async def test_repository_with_patterns(request):
+    """Test repository analysis with include/exclude patterns."""
+    client = request.getfixturevalue("test_client")
+    form_data = {
+        "input_text": "https://github.com/octocat/Hello-World",
+        "max_file_size": "243",
+        "pattern_type": "include",
+        "pattern": "*.md",
+    }
+
+    response = client.post("/", data=form_data)
+    assert response.status_code == 200, f"Request failed: {response.text}"
+    assert "Mocked Template Response" in response.text

tests/test_repository_clone.py

@@ -6,6 +6,7 @@
 """
 
 import asyncio
+import os
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -306,3 +307,58 @@ async def test_clone_repo_with_timeout() -> None:
             mock_exec.side_effect = asyncio.TimeoutError
             with pytest.raises(AsyncTimeoutError, match="Operation timed out after"):
                 await clone_repo(clone_config)
+
+
+@pytest.mark.asyncio
+async def test_clone_specific_branch(tmp_path):
+    """
+    Test cloning a specific branch of a repository.
+
+    Given a valid repository URL and a branch name:
+    When `clone_repo` is called,
+    Then the repository should be cloned and checked out at that branch.
+    """
+    repo_url = "https://github.com/cyclotruc/gitingest.git"
+    branch_name = "main"
+    local_path = tmp_path / "gitingest"
+
+    config = CloneConfig(url=repo_url, local_path=str(local_path), branch=branch_name)
+    await clone_repo(config)
+
+    # Assertions
+    assert local_path.exists(), "The repository was not cloned successfully."
+    assert local_path.is_dir(), "The cloned repository path is not a directory."
+
+    # Check the current branch
+    current_branch = os.popen(f"git -C {local_path} branch --show-current").read().strip()
+    assert current_branch == branch_name, f"Expected branch '{branch_name}', got '{current_branch}'."
+
+
+@pytest.mark.asyncio
+async def test_clone_branch_with_slashes(tmp_path):
+    """
+    Test cloning a branch with slashes in the name.
+
+    Given a valid repository URL and a branch name with slashes:
+    When `clone_repo` is called,
+    Then the repository should be cloned and checked out at that branch.
+    """
+    repo_url = "https://github.com/user/repo"
+    branch_name = "fix/in-operator"
+    local_path = tmp_path / "gitingest"
+
+    clone_config = CloneConfig(url=repo_url, local_path=str(local_path), branch=branch_name)
+    with patch("gitingest.repository_clone._check_repo_exists", return_value=True):
+        with patch("gitingest.repository_clone._run_git_command", new_callable=AsyncMock) as mock_exec:
+            await clone_repo(clone_config)
+
+            mock_exec.assert_called_once_with(
+                "git",
+                "clone",
+                "--depth=1",
+                "--single-branch",
+                "--branch",
+                "fix/in-operator",
+                clone_config.url,
+                clone_config.local_path,
+            )