devx/tests/unit/test_ci_cancel_superseded_runs.py

"""Unit tests for devx.ci.cancel_superseded_runs."""

from __future__ import annotations

import json
import urllib.error
from unittest.mock import MagicMock, patch

import pytest

import devx.ci.cancel_superseded_runs as mod
from devx.ci.cancel_superseded_runs import _api_request, cancel_run, list_running_runs, main

_HTTP_NO_CONTENT = mod._HTTP_NO_CONTENT
_PAGE_SIZE = mod._PAGE_SIZE


class TestConstants:
    def test_http_no_content_is_204(self) -> None:
        assert _HTTP_NO_CONTENT == 204

    def test_page_size_is_50(self) -> None:
        assert _PAGE_SIZE == 50


class TestApiRequest:
    def test_returns_empty_for_204(self) -> None:
        mock_resp = MagicMock()
        mock_resp.status = _HTTP_NO_CONTENT
        mock_resp.read.return_value = b""
        mock_resp.__enter__ = MagicMock(return_value=mock_resp)
        mock_resp.__exit__ = MagicMock(return_value=None)
        with patch("urllib.request.urlopen", return_value=mock_resp):
            result = _api_request("POST", "/repos/test/actions/runs/1/cancel", "tok", "https://x")
        assert result == {}

    def test_returns_json_for_200(self) -> None:
        mock_resp = MagicMock()
        mock_resp.status = 200
        mock_resp.read.return_value = json.dumps({"id": 1}).encode()
        mock_resp.__enter__ = MagicMock(return_value=mock_resp)
        mock_resp.__exit__ = MagicMock(return_value=None)
        with patch("urllib.request.urlopen", return_value=mock_resp):
            result = _api_request("GET", "/repos/test/actions/runs", "tok", "https://x")
        assert result == {"id": 1}

    def test_http_error_raises(self) -> None:
        err = urllib.error.HTTPError("x", 500, "err", {}, None)
        err.read = MagicMock(return_value=b"error body")
        with patch("urllib.request.urlopen", side_effect=err):
            with pytest.raises(urllib.error.HTTPError):
                _api_request("GET", "/repos/test/actions/runs", "tok", "https://x")

    def test_url_error_raises(self) -> None:
        with patch("urllib.request.urlopen", side_effect=urllib.error.URLError("fail")):
            with pytest.raises(urllib.error.URLError):
                _api_request("GET", "/repos/test/actions/runs", "tok", "https://x")


class TestListRunningRuns:
    def test_paginates_until_empty(self) -> None:
        page1 = {"workflow_runs": [{"id": 1}, {"id": 2}], "total_count": 2}
        page2 = {"workflow_runs": [], "total_count": 2}
        responses = iter([page1, page2])
        with patch.object(mod, "_api_request", side_effect=lambda *a, **k: next(responses)):
            runs = list_running_runs("owner/repo", "tok", "https://x")
        assert len(runs) == 2

    def test_empty_first_page(self) -> None:
        with patch.object(mod, "_api_request", return_value={"workflow_runs": [], "total_count": 0}):
            runs = list_running_runs("owner/repo", "tok", "https://x")
        assert runs == []

    def test_stops_at_page_size(self) -> None:
        full_page = {"workflow_runs": [{"id": i} for i in range(_PAGE_SIZE)], "total_count": _PAGE_SIZE + 1}
        half_page = {"workflow_runs": [{"id": 99}], "total_count": _PAGE_SIZE + 1}
        responses = iter([full_page, half_page])
        with patch.object(mod, "_api_request", side_effect=lambda *a, **k: next(responses)):
            runs = list_running_runs("owner/repo", "tok", "https://x")
        assert len(runs) == _PAGE_SIZE + 1

    def test_uses_in_progress_status(self) -> None:
        with patch.object(mod, "_api_request", return_value={"workflow_runs": [], "total_count": 0}) as mock_req:
            list_running_runs("owner/repo", "tok", "https://x")
        path = mock_req.call_args.args[1]
        assert "status=in_progress" in path
        assert "status=running" not in path

    def test_accepts_bare_list(self) -> None:
        with patch.object(mod, "_api_request", return_value=[{"id": 1}, {"id": 2}]):
            runs = list_running_runs("owner/repo", "tok", "https://x")
        assert len(runs) == 2


class TestCancelRun:
    def test_success_returns_true(self) -> None:
        with patch.object(mod, "_api_request", return_value={}):
            assert cancel_run("owner/repo", 123, "tok", "https://x") is True

    def test_http_error_returns_false(self) -> None:
        with patch.object(mod, "_api_request", side_effect=urllib.error.HTTPError("x", 500, "err", {}, None)):
            assert cancel_run("owner/repo", 123, "tok", "https://x") is False


class TestMain:
    def test_no_token_exits_zero(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.delenv("CI_GITEA_API_TOKEN", raising=False)
        monkeypatch.delenv("CI_GITEA_TOKEN", raising=False)
        monkeypatch.setattr("sys.argv", ["cancel", "--repo", "o/r", "--current-run-id", "1", "--head-branch", "feat"])
        assert main() == 0

    def test_no_superseded_runs(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("CI_GITEA_API_TOKEN", "tok")
        monkeypatch.setattr("sys.argv", ["cancel", "--repo", "o/r", "--current-run-id", "10", "--head-branch", "feat"])
        with patch.object(mod, "list_running_runs", return_value=[]):
            assert main() == 0

    def test_cancels_superseded(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("CI_GITEA_API_TOKEN", "tok")
        runs = [
            {"id": 5, "head_branch": "feat"},
            {"id": 8, "head_branch": "feat"},
            {"id": 12, "head_branch": "other"},
        ]
        monkeypatch.setattr("sys.argv", ["cancel", "--repo", "o/r", "--current-run-id", "10", "--head-branch", "feat"])
        with patch.object(mod, "list_running_runs", return_value=runs):
            with patch.object(mod, "cancel_run", return_value=True) as mock_cancel:
                assert main() == 0
        cancelled_ids = [call.args[1] for call in mock_cancel.call_args_list]
        assert cancelled_ids == [5, 8]

    def test_dry_run_does_not_cancel(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("CI_GITEA_API_TOKEN", "tok")
        runs = [{"id": 5, "head_branch": "feat"}]
        monkeypatch.setattr(
            "sys.argv",
            ["cancel", "--repo", "o/r", "--current-run-id", "10", "--head-branch", "feat", "--dry-run"],
        )
        with patch.object(mod, "list_running_runs", return_value=runs):
            with patch.object(mod, "cancel_run", return_value=True) as mock_cancel:
                assert main() == 0
        assert mock_cancel.call_count == 0

    def test_cancel_failure_continues(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("CI_GITEA_API_TOKEN", "tok")
        runs = [{"id": 5, "head_branch": "feat"}, {"id": 8, "head_branch": "feat"}]
        monkeypatch.setattr("sys.argv", ["cancel", "--repo", "o/r", "--current-run-id", "10", "--head-branch", "feat"])
        with patch.object(mod, "list_running_runs", return_value=runs):
            with patch.object(mod, "cancel_run", side_effect=[False, True]):
                assert main() == 0

    def test_404_returns_zero(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("CI_GITEA_API_TOKEN", "tok")
        monkeypatch.setattr("sys.argv", ["cancel", "--repo", "o/r", "--current-run-id", "10", "--head-branch", "feat"])
        err = urllib.error.HTTPError("x", 404, "Not Found", {}, None)
        with patch.object(mod, "list_running_runs", side_effect=err):
            assert main() == 0

    def test_400_returns_zero(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("CI_GITEA_API_TOKEN", "tok")
        monkeypatch.setattr("sys.argv", ["cancel", "--repo", "o/r", "--current-run-id", "10", "--head-branch", "feat"])
        err = urllib.error.HTTPError("x", 400, "Bad Request", {}, None)
        with patch.object(mod, "list_running_runs", side_effect=err):
            assert main() == 0

    def test_500_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("CI_GITEA_API_TOKEN", "tok")
        monkeypatch.setattr("sys.argv", ["cancel", "--repo", "o/r", "--current-run-id", "10", "--head-branch", "feat"])
        err = urllib.error.HTTPError("x", 500, "Server Error", {}, None)
        with patch.object(mod, "list_running_runs", side_effect=err):
            with pytest.raises(urllib.error.HTTPError):
                main()