jpmschweitzerandClaude Opus 4.5
64574bcc39
Build and Push / build (release) Failing after 17s
Extracted standalone scheduler service with: - FastAPI REST API for task management - APScheduler-based task execution - PostgreSQL persistence - Docker container support - Gitea Actions CI/CD workflow 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
241 lines
9.2 KiB
Python
241 lines
9.2 KiB
Python
"""
|
|
Tests for the documentation sync executor.
|
|
"""
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
from src.executors import doc_sync_executor
|
|
from src.config import Settings
|
|
|
|
|
|
@pytest.mark.executor
|
|
@pytest.mark.unit
|
|
class TestDocSyncExecutor:
|
|
"""Tests for doc_sync_executor module."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_requires_all_config_fields(self, test_settings: Settings):
|
|
"""Test that executor validates required config fields."""
|
|
incomplete_config = {
|
|
"project": "test",
|
|
# Missing other required fields
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="Missing required config"):
|
|
await doc_sync_executor.execute(incomplete_config, test_settings)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_validates_project_name(self, test_settings: Settings):
|
|
"""Test that executor requires project name."""
|
|
config = {
|
|
"project": "",
|
|
"upstream_repo": "https://github.com/test/repo.git",
|
|
"gitea_repo": "library/test"
|
|
}
|
|
|
|
with pytest.raises(ValueError):
|
|
await doc_sync_executor.execute(config, test_settings)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_successful_sync_entire_repo(
|
|
self,
|
|
test_settings: Settings,
|
|
sample_doc_sync_config: dict,
|
|
temp_work_dir: Path
|
|
):
|
|
"""Test successful documentation sync of entire repository."""
|
|
# Modify config to sync entire repo
|
|
config = {**sample_doc_sync_config, "docs_paths": []}
|
|
|
|
with patch('src.executors.doc_sync_executor._run_command', new_callable=AsyncMock) as mock_run, \
|
|
patch('src.executors.doc_sync_executor._get_git_commit', new_callable=AsyncMock) as mock_get_commit, \
|
|
patch('src.executors.doc_sync_executor.shutil') as mock_shutil, \
|
|
patch('src.executors.doc_sync_executor.Path') as mock_path:
|
|
|
|
# Mock git commit hash
|
|
mock_get_commit.return_value = "abc123def456"
|
|
|
|
# Mock path operations
|
|
mock_work_dir = MagicMock()
|
|
mock_upstream_dir = MagicMock()
|
|
mock_gitea_dir = MagicMock()
|
|
|
|
# Setup directory mocking
|
|
mock_upstream_dir.iterdir.return_value = [
|
|
MagicMock(name=".git", is_dir=lambda: True),
|
|
MagicMock(name="README.md", is_dir=lambda: False),
|
|
MagicMock(name="docs", is_dir=lambda: True),
|
|
]
|
|
mock_gitea_dir.iterdir.return_value = [
|
|
MagicMock(name=".git", is_dir=lambda: True)
|
|
]
|
|
|
|
mock_path.return_value = mock_work_dir
|
|
mock_work_dir.__truediv__.side_effect = [mock_upstream_dir, mock_gitea_dir]
|
|
|
|
# Mock git status to show changes
|
|
mock_run.side_effect = [
|
|
"", # git clone upstream
|
|
"", # git rev-parse HEAD
|
|
"", # git clone gitea
|
|
"", # git add
|
|
"M README.md\n", # git status --porcelain (has changes)
|
|
"", # git commit
|
|
"", # git tag
|
|
"", # git push branch
|
|
"", # git push tag
|
|
]
|
|
|
|
result = await doc_sync_executor.execute(config, test_settings)
|
|
|
|
assert "Successfully synced" in result
|
|
assert "test-project" in result.lower()
|
|
assert "entire repository" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_sync_specific_paths(
|
|
self,
|
|
test_settings: Settings,
|
|
sample_doc_sync_config: dict
|
|
):
|
|
"""Test syncing only specific paths."""
|
|
config = {**sample_doc_sync_config, "docs_paths": ["/docs", "/examples"]}
|
|
|
|
with patch('src.executors.doc_sync_executor._run_command', new_callable=AsyncMock) as mock_run, \
|
|
patch('src.executors.doc_sync_executor._get_git_commit', new_callable=AsyncMock) as mock_get_commit, \
|
|
patch('src.executors.doc_sync_executor.shutil') as mock_shutil, \
|
|
patch('src.executors.doc_sync_executor.Path') as mock_path:
|
|
|
|
mock_get_commit.return_value = "abc123"
|
|
|
|
# Mock path operations
|
|
mock_upstream_dir = MagicMock()
|
|
mock_gitea_dir = MagicMock()
|
|
mock_docs = MagicMock(name="docs")
|
|
mock_docs.exists.return_value = True
|
|
mock_docs.is_dir.return_value = True
|
|
mock_docs.name = "docs"
|
|
|
|
mock_examples = MagicMock(name="examples")
|
|
mock_examples.exists.return_value = True
|
|
mock_examples.is_dir.return_value = True
|
|
mock_examples.name = "examples"
|
|
|
|
mock_upstream_dir.__truediv__.side_effect = [mock_docs, mock_examples]
|
|
mock_gitea_dir.iterdir.return_value = []
|
|
|
|
# Mock git status to show changes
|
|
async def run_command_side_effect(cmd, **kwargs):
|
|
if 'status' in cmd:
|
|
return "M docs/README.md\n"
|
|
return ""
|
|
|
|
mock_run.side_effect = run_command_side_effect
|
|
|
|
result = await doc_sync_executor.execute(config, test_settings)
|
|
|
|
# Should mention the specific paths
|
|
assert "docs" in result.lower() or "Successfully synced" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_handles_no_changes(
|
|
self,
|
|
test_settings: Settings,
|
|
sample_doc_sync_config: dict
|
|
):
|
|
"""Test that executor handles case where there are no changes."""
|
|
with patch('src.executors.doc_sync_executor._run_command', new_callable=AsyncMock) as mock_run, \
|
|
patch('src.executors.doc_sync_executor._get_git_commit', new_callable=AsyncMock) as mock_get_commit, \
|
|
patch('src.executors.doc_sync_executor.shutil'), \
|
|
patch('src.executors.doc_sync_executor.Path') as mock_path:
|
|
|
|
mock_get_commit.return_value = "abc123"
|
|
|
|
# Mock empty git status (no changes)
|
|
async def run_command_side_effect(cmd, cwd=None, capture=False):
|
|
if 'status' in cmd and '--porcelain' in cmd:
|
|
return "" # No changes
|
|
return ""
|
|
|
|
mock_run.side_effect = run_command_side_effect
|
|
|
|
# Setup minimal mocking
|
|
mock_work_dir = MagicMock()
|
|
mock_upstream_dir = MagicMock()
|
|
mock_gitea_dir = MagicMock()
|
|
mock_gitea_dir.iterdir.return_value = []
|
|
mock_upstream_dir.iterdir.return_value = []
|
|
|
|
result = await doc_sync_executor.execute(sample_doc_sync_config, test_settings)
|
|
|
|
assert "already up to date" in result.lower() or "no changes" in result.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_cleanup_on_error(
|
|
self,
|
|
test_settings: Settings,
|
|
sample_doc_sync_config: dict
|
|
):
|
|
"""Test that executor cleans up on error."""
|
|
with patch('src.executors.doc_sync_executor._run_command', new_callable=AsyncMock) as mock_run, \
|
|
patch('src.executors.doc_sync_executor.shutil') as mock_shutil, \
|
|
patch('src.executors.doc_sync_executor.Path') as mock_path:
|
|
|
|
# Make git clone fail
|
|
mock_run.side_effect = Exception("Git clone failed")
|
|
|
|
mock_work_dir = MagicMock()
|
|
mock_work_dir.exists.return_value = True
|
|
mock_path.return_value = mock_work_dir
|
|
|
|
with pytest.raises(Exception, match="Git clone failed"):
|
|
await doc_sync_executor.execute(sample_doc_sync_config, test_settings)
|
|
|
|
# Should have attempted cleanup
|
|
mock_shutil.rmtree.assert_called()
|
|
|
|
|
|
@pytest.mark.executor
|
|
@pytest.mark.unit
|
|
class TestDocSyncHelpers:
|
|
"""Tests for doc_sync_executor helper functions."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_command_success(self):
|
|
"""Test that _run_command executes successfully."""
|
|
with patch('asyncio.create_subprocess_exec', new_callable=AsyncMock) as mock_exec:
|
|
mock_proc = AsyncMock()
|
|
mock_proc.returncode = 0
|
|
mock_proc.communicate.return_value = (b"output", b"")
|
|
mock_exec.return_value = mock_proc
|
|
|
|
result = await doc_sync_executor._run_command(
|
|
["echo", "test"],
|
|
capture=True
|
|
)
|
|
|
|
assert result == "output"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_command_failure(self):
|
|
"""Test that _run_command raises on failure."""
|
|
with patch('asyncio.create_subprocess_exec', new_callable=AsyncMock) as mock_exec:
|
|
mock_proc = AsyncMock()
|
|
mock_proc.returncode = 1
|
|
mock_proc.communicate.return_value = (b"", b"error message")
|
|
mock_exec.return_value = mock_proc
|
|
|
|
with pytest.raises(Exception, match="Command failed"):
|
|
await doc_sync_executor._run_command(["false"])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_git_commit(self, temp_work_dir: Path):
|
|
"""Test getting git commit hash."""
|
|
with patch('src.executors.doc_sync_executor._run_command', new_callable=AsyncMock) as mock_run:
|
|
mock_run.return_value = "abc123def456\n"
|
|
|
|
commit = await doc_sync_executor._get_git_commit(temp_work_dir)
|
|
|
|
assert commit == "abc123def456"
|
|
mock_run.assert_called_once()
|