"""Tests for GitLab client with mocked API."""
from unittest.mock import Mock, MagicMock, patch, call
import time
from src.gitlab_client import GitLabClient, GitLabClientError
BASE_URL = "https://gitlab.example.com"
PROJECT_PATH = "group/project"
PROJECT_ID = 123
def test_get_project_id_is_cached():
"""Test that project IDs are cached after first lookup."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
client = GitLabClient(BASE_URL, "token")
first = client.get_project_id(PROJECT_PATH)
second = client.get_project_id(PROJECT_PATH)
assert first == PROJECT_ID
assert second == PROJECT_ID
# Should only call API once due to cache
assert mock_gl.projects.get.call_count == 1
def test_get_pipeline_status_by_commit():
"""Test fetching pipeline status for a specific commit."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.side_effect = lambda pid: mock_project if pid in (PROJECT_ID, PROJECT_PATH) else None
# Mock pipeline
mock_pipeline = Mock()
mock_pipeline.id = 999
mock_pipeline.status = "running"
mock_pipeline.ref = "main"
mock_pipeline.sha = "abcdef123456"
mock_pipeline.created_at = "2026-02-10T10:00:00Z"
mock_pipeline.updated_at = "2026-02-10T10:05:00Z"
mock_pipeline.started_at = "2026-02-10T10:01:00Z"
mock_pipeline.finished_at = None
mock_pipeline.duration = None
mock_pipeline.web_url = f"{BASE_URL}/{PROJECT_PATH}/-/pipelines/999"
# Mock jobs
mock_job1 = Mock()
mock_job1.id = 1
mock_job1.name = "build"
mock_job1.status = "success"
mock_job1.stage = "build"
mock_job1.started_at = "2026-02-10T10:00:30Z"
mock_job1.finished_at = "2026-02-10T10:02:00Z"
mock_job1.duration = 90
mock_job1.web_url = f"{BASE_URL}/{PROJECT_PATH}/-/jobs/1"
mock_job2 = Mock()
mock_job2.id = 2
mock_job2.name = "test"
mock_job2.status = "running"
mock_job2.stage = "test"
mock_job2.started_at = "2026-02-10T10:02:30Z"
mock_job2.finished_at = None
mock_job2.duration = None
mock_job2.web_url = f"{BASE_URL}/{PROJECT_PATH}/-/jobs/2"
mock_pipeline.jobs.list.return_value = [mock_job1, mock_job2]
mock_project.pipelines.list.return_value = [mock_pipeline]
client = GitLabClient(BASE_URL, "token")
pipeline_info = client.get_pipeline_status(PROJECT_PATH, "main", commit="abcdef123456")
assert pipeline_info["id"] == 999
assert pipeline_info["sha"] == "abcdef123456"
assert pipeline_info["status"] == "running"
assert len(pipeline_info["jobs"]) == 2
assert pipeline_info["jobs"][0]["name"] == "build"
assert pipeline_info["jobs"][1]["name"] == "test"
def test_get_job_status_by_id():
"""Test fetching job status by job ID."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
# Mock job
mock_job = Mock()
mock_job.id = 321
mock_job.name = "lint"
mock_job.status = "failed"
mock_job.stage = "test"
mock_job.started_at = "2026-02-10T10:00:30Z"
mock_job.finished_at = "2026-02-10T10:01:00Z"
mock_job.duration = 30
mock_job.web_url = f"{BASE_URL}/{PROJECT_PATH}/-/jobs/321"
mock_project.jobs.get.return_value = mock_job
client = GitLabClient(BASE_URL, "token")
job_info = client.get_job_status(PROJECT_PATH, job_id=321)
assert job_info["id"] == 321
assert job_info["name"] == "lint"
assert job_info["status"] == "failed"
assert job_info["stage"] == "test"
def test_get_job_status_by_name():
"""Test fetching job status by job name."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
# Mock job
mock_job = Mock()
mock_job.id = 555
mock_job.name = "deploy"
mock_job.status = "success"
mock_job.stage = "deploy"
mock_job.started_at = "2026-02-10T10:05:00Z"
mock_job.finished_at = "2026-02-10T10:06:00Z"
mock_job.duration = 60
mock_job.web_url = f"{BASE_URL}/{PROJECT_PATH}/-/jobs/555"
mock_project.jobs.list.return_value = [mock_job]
client = GitLabClient(BASE_URL, "token")
job_info = client.get_job_status(PROJECT_PATH, job_name="deploy")
assert job_info["id"] == 555
assert job_info["name"] == "deploy"
assert job_info["status"] == "success"
def test_get_merge_request_for_branch():
"""Test fetching merge request for a branch."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
# Mock merge request
mock_mr = Mock()
mock_mr.iid = 7
mock_mr.title = "Add new feature"
mock_mr.state = "opened"
mock_mr.web_url = f"{BASE_URL}/{PROJECT_PATH}/-/merge_requests/7"
mock_project.mergerequests.list.return_value = [mock_mr]
client = GitLabClient(BASE_URL, "token")
mr_info = client.get_merge_request_for_branch(PROJECT_PATH, "feature")
assert mr_info is not None
assert mr_info["id"] == 7
assert mr_info["title"] == "Add new feature"
assert mr_info["state"] == "opened"
def test_get_merge_request_for_branch_not_found():
"""Test that None is returned when no MR exists for branch."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
# No merge requests
mock_project.mergerequests.list.return_value = []
client = GitLabClient(BASE_URL, "token")
mr_info = client.get_merge_request_for_branch(PROJECT_PATH, "feature")
assert mr_info is None
def test_retry_request_on_transient_error(monkeypatch):
"""Test that retry logic works on transient errors."""
monkeypatch.setattr(time, "sleep", lambda _: None)
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
# Mock pipeline - first call fails, second succeeds
mock_pipeline = Mock()
mock_pipeline.id = 444
mock_pipeline.status = "success"
mock_pipeline.ref = "main"
mock_pipeline.sha = "deadbeef"
mock_pipeline.created_at = "2026-02-10T10:00:00Z"
mock_pipeline.updated_at = "2026-02-10T10:10:00Z"
mock_pipeline.started_at = "2026-02-10T10:01:00Z"
mock_pipeline.finished_at = "2026-02-10T10:10:00Z"
mock_pipeline.duration = 540
mock_pipeline.web_url = f"{BASE_URL}/{PROJECT_PATH}/-/pipelines/444"
mock_pipeline.jobs.list.return_value = []
# First call raises error, second succeeds
from gitlab.exceptions import GitlabError
mock_project.pipelines.list.side_effect = [
GitlabError("Server error"),
[mock_pipeline]
]
client = GitLabClient(BASE_URL, "token")
pipeline_info = client.get_pipeline_status(PROJECT_PATH, "main", commit="deadbeef")
assert pipeline_info["id"] == 444
assert mock_project.pipelines.list.call_count == 2
def test_project_not_found_error():
"""Test that GitLabClientError is raised when project not found."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Simulate project not found
from gitlab.exceptions import GitlabGetError
mock_gl.projects.get.side_effect = GitlabGetError("404 Not found")
client = GitLabClient(BASE_URL, "token")
try:
client.get_project_id("nonexistent/project")
assert False, "Should have raised GitLabClientError"
except GitLabClientError as e:
assert "Project not found" in str(e)
def test_job_not_found_error():
"""Test that GitLabClientError is raised when job not found."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
# No jobs match the name
mock_project.jobs.list.return_value = []
client = GitLabClient(BASE_URL, "token")
try:
client.get_job_status(PROJECT_PATH, job_name="nonexistent")
assert False, "Should have raised GitLabClientError"
except GitLabClientError as e:
assert "Job not found" in str(e)
def test_get_job_log():
"""Test fetching raw job log content."""
with patch('src.gitlab_client.gitlab.Gitlab') as MockGitlab:
mock_gl = MagicMock()
MockGitlab.return_value = mock_gl
# Mock project
mock_project = Mock()
mock_project.id = PROJECT_ID
mock_gl.projects.get.return_value = mock_project
# Mock job with trace
mock_job = Mock()
mock_job.trace.return_value = "Job log line 1\nJob log line 2\nJob log line 3\n"
mock_project.jobs.get.return_value = mock_job
client = GitLabClient(BASE_URL, "token")
log_content = client.get_job_log(PROJECT_PATH, 555)
assert "Job log line 1" in log_content
assert "Job log line 2" in log_content
assert "Job log line 3" in log_content
mock_project.jobs.get.assert_called_once_with(555)
mock_job.trace.assert_called_once()
