Public Access
emil
81dc30ecff
Post-merge / detect-type (push) Successful in 9s
Post-merge / validate-commit-msg (push) Successful in 11s
Post-merge / vikunja (push) Successful in 13s
Post-merge / sync-wiki (push) Successful in 17s
Post-merge / configure-repo (push) Successful in 10s
Post-merge / release (push) Successful in 30s
Build Images / detect-type (push) Successful in 48s
Post-merge / badges (push) Successful in 41s
Post-merge / publish (push) Successful in 16s
Build Images / build-and-push (push) Successful in 4m41s
Build Images / cleanup (push) Successful in 2m23s
313 lines
13 KiB
Python
313 lines
13 KiB
Python
"""Unit tests for devx.tools.pr_logs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from devx.api_clients import APIError, GiteaClient
|
|
from devx.tools.pr_logs import (
|
|
_find_failed_jobs,
|
|
_find_job_by_name,
|
|
_find_latest_run_by_sha,
|
|
_get_pr_sha,
|
|
_print_failed_steps,
|
|
_print_job_summary,
|
|
_print_logs,
|
|
cli,
|
|
)
|
|
|
|
|
|
class TestGetPrSha:
|
|
def test_returns_sha(self) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
assert _get_pr_sha(client, 42) == "abc123"
|
|
|
|
def test_returns_empty_when_missing(self) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.get_pr.return_value = {"head": {}}
|
|
assert _get_pr_sha(client, 42) == ""
|
|
|
|
|
|
class TestFindLatestRunBySha:
|
|
def test_returns_matching_run(self) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.list_action_runs.return_value = {
|
|
"workflow_runs": [
|
|
{"id": 2, "head_sha": "def456"},
|
|
{"id": 1, "head_sha": "abc123def"},
|
|
],
|
|
}
|
|
result = _find_latest_run_by_sha(client, "abc123")
|
|
assert result is not None
|
|
assert result["id"] == 1
|
|
|
|
def test_returns_none_when_no_match(self) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.list_action_runs.return_value = {
|
|
"workflow_runs": [{"id": 1, "head_sha": "def456"}],
|
|
}
|
|
result = _find_latest_run_by_sha(client, "abc123")
|
|
assert result is None
|
|
|
|
def test_returns_none_when_empty(self) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.list_action_runs.return_value = {"workflow_runs": []}
|
|
result = _find_latest_run_by_sha(client, "abc123")
|
|
assert result is None
|
|
|
|
|
|
class TestFindFailedJobs:
|
|
def test_returns_failed(self) -> None:
|
|
jobs = [
|
|
{"id": 1, "name": "quality", "conclusion": "failure"},
|
|
{"id": 2, "name": "lint", "conclusion": "success"},
|
|
]
|
|
result = _find_failed_jobs(jobs)
|
|
assert len(result) == 1
|
|
assert result[0]["name"] == "quality"
|
|
|
|
def test_empty_when_none_failed(self) -> None:
|
|
jobs = [{"id": 1, "name": "quality", "conclusion": "success"}]
|
|
assert _find_failed_jobs(jobs) == []
|
|
|
|
|
|
class TestFindJobByName:
|
|
def test_case_insensitive_partial(self) -> None:
|
|
jobs = [{"id": 1, "name": "CI / quality (pull_request)"}]
|
|
result = _find_job_by_name(jobs, "QUALITY")
|
|
assert result is not None
|
|
assert result["id"] == 1
|
|
|
|
def test_returns_none_when_not_found(self) -> None:
|
|
jobs = [{"id": 1, "name": "quality"}]
|
|
assert _find_job_by_name(jobs, "molecule") is None
|
|
|
|
|
|
class TestPrintJobSummary:
|
|
def test_prints_all_jobs(self, capsys: pytest.CaptureFixture) -> None:
|
|
jobs = [
|
|
{"id": 1, "name": "quality", "conclusion": "failure", "status": "completed"},
|
|
{"id": 2, "name": "lint", "conclusion": "success", "status": "completed"},
|
|
]
|
|
_print_job_summary(jobs)
|
|
out = capsys.readouterr().out
|
|
assert "[FAIL]" in out
|
|
assert "[OK]" in out
|
|
assert "quality" in out
|
|
assert "lint" in out
|
|
|
|
|
|
class TestPrintFailedSteps:
|
|
def test_prints_failed_steps(self, capsys: pytest.CaptureFixture) -> None:
|
|
job = {
|
|
"steps": [
|
|
{"name": "checkout", "number": 1, "conclusion": "success"},
|
|
{"name": "Unit tests", "number": 3, "conclusion": "failure"},
|
|
]
|
|
}
|
|
result = _print_failed_steps(job)
|
|
assert result == [3]
|
|
out = capsys.readouterr().out
|
|
assert "FAILED step #3" in out
|
|
assert "Unit tests" in out
|
|
|
|
def test_no_failed_steps(self, capsys: pytest.CaptureFixture) -> None:
|
|
job = {"steps": [{"name": "checkout", "number": 1, "conclusion": "success"}]}
|
|
result = _print_failed_steps(job)
|
|
assert result == []
|
|
|
|
def test_no_steps_key(self, capsys: pytest.CaptureFixture) -> None:
|
|
result = _print_failed_steps({})
|
|
assert result == []
|
|
|
|
|
|
class TestPrintLogs:
|
|
def test_prints_all_lines(self, capsys: pytest.CaptureFixture) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.get_action_job_logs.return_value = "line 1\nline 2\nline 3"
|
|
_print_logs(client, 100, tail=0)
|
|
out = capsys.readouterr().out
|
|
assert "line 1" in out
|
|
assert "line 3" in out
|
|
|
|
def test_tail_truncates(self, capsys: pytest.CaptureFixture) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.get_action_job_logs.return_value = "\n".join(f"line {i}" for i in range(100))
|
|
_print_logs(client, 100, tail=10)
|
|
out = capsys.readouterr().out
|
|
assert "line 99" in out
|
|
assert "line 0" not in out
|
|
assert "showing last 10" in out
|
|
|
|
def test_api_error_handled(self, capsys: pytest.CaptureFixture) -> None:
|
|
client = MagicMock(spec=GiteaClient)
|
|
client.get_action_job_logs.side_effect = APIError(404, "not found")
|
|
_print_logs(client, 100, tail=0)
|
|
out = capsys.readouterr().out
|
|
assert "Could not fetch logs" in out
|
|
|
|
|
|
class TestCli:
|
|
def test_no_token_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("CI_GITEA_TOKEN", raising=False)
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42"])
|
|
assert result.exit_code != 0
|
|
assert "CI_GITEA_TOKEN" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.REPO_OWNER", "")
|
|
def test_no_owner_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42"])
|
|
assert result.exit_code != 0
|
|
assert "owner" in result.output.lower()
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
@patch("devx.tools.pr_status.subprocess.run")
|
|
def test_auto_detect_pr(
|
|
self, mock_subprocess: MagicMock, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
mock_subprocess.return_value = MagicMock(returncode=0, stdout="feature-branch\n")
|
|
client = mock_client_cls.return_value
|
|
client.list_prs.return_value = [{"number": 42, "head": {"ref": "feature-branch"}}]
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
client.list_action_runs.return_value = {"workflow_runs": []}
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, [])
|
|
assert result.exit_code != 0
|
|
assert "Fetching logs for PR #42" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
def test_no_runs_found(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
client = mock_client_cls.return_value
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
client.list_action_runs.return_value = {"workflow_runs": []}
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42"])
|
|
assert result.exit_code != 0
|
|
assert "No workflow runs" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
def test_no_jobs(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
client = mock_client_cls.return_value
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
client.list_action_runs.return_value = {
|
|
"workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}],
|
|
}
|
|
client.get_action_run_jobs.return_value = []
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42"])
|
|
assert result.exit_code == 0
|
|
assert "No jobs" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
def test_no_failed_jobs(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
client = mock_client_cls.return_value
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
client.list_action_runs.return_value = {
|
|
"workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}],
|
|
}
|
|
client.get_action_run_jobs.return_value = [
|
|
{"id": 100, "name": "quality", "conclusion": "success", "status": "completed", "steps": []}
|
|
]
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42"])
|
|
assert result.exit_code == 0
|
|
assert "No failed jobs" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
def test_failed_job_logs(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
client = mock_client_cls.return_value
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
client.list_action_runs.return_value = {
|
|
"workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}],
|
|
}
|
|
client.get_action_run_jobs.return_value = [
|
|
{
|
|
"id": 100,
|
|
"name": "quality",
|
|
"conclusion": "failure",
|
|
"status": "completed",
|
|
"steps": [
|
|
{"name": "checkout", "number": 1, "conclusion": "success"},
|
|
{"name": "Unit tests", "number": 3, "conclusion": "failure"},
|
|
],
|
|
}
|
|
]
|
|
client.get_action_job_logs.return_value = "error: test failed"
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42", "--tail", "0"])
|
|
assert result.exit_code == 0
|
|
assert "FAILED step #3" in result.output
|
|
assert "error: test failed" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
def test_specific_job(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
client = mock_client_cls.return_value
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
client.list_action_runs.return_value = {
|
|
"workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}],
|
|
}
|
|
client.get_action_run_jobs.return_value = [
|
|
{"id": 100, "name": "quality", "conclusion": "success", "status": "completed", "steps": []},
|
|
{"id": 101, "name": "lint", "conclusion": "success", "status": "completed", "steps": []},
|
|
]
|
|
client.get_action_job_logs.return_value = "lint output here"
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42", "--job", "lint", "--tail", "0"])
|
|
assert result.exit_code == 0
|
|
assert "lint output here" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
def test_job_not_found(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
client = mock_client_cls.return_value
|
|
client.get_pr.return_value = {"head": {"sha": "abc123"}}
|
|
client.list_action_runs.return_value = {
|
|
"workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}],
|
|
}
|
|
client.get_action_run_jobs.return_value = [
|
|
{"id": 100, "name": "quality", "conclusion": "success", "status": "completed", "steps": []}
|
|
]
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42", "--job", "nonexistent"])
|
|
assert result.exit_code != 0
|
|
assert "No job matching" in result.output
|
|
|
|
@patch("devx.tools.pr_logs.GiteaClient")
|
|
def test_no_sha_raises(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CI_GITEA_TOKEN", "tok")
|
|
monkeypatch.setenv("DEVX_REPO_OWNER", "owner")
|
|
monkeypatch.setenv("DEVX_REPO_NAME", "repo")
|
|
client = mock_client_cls.return_value
|
|
client.get_pr.return_value = {"head": {}}
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--pr", "42"])
|
|
assert result.exit_code != 0
|
|
assert "SHA" in result.output
|