Class FetchDiffRefsTest

examples/gitlab_ci/post_review_test.py:1258–1378

class FetchDiffRefsTest(unittest.TestCase):
    API_BASE = "https://gitlab.example/api/v4/projects/1/merge_requests/2"
    TOKEN = "test-token"
    AUTH_HEADER = "PRIVATE-TOKEN"

    def test_success(self):
        body = json.dumps([{
            "base_commit_sha": "aaa", "start_commit_sha": "bbb", "head_commit_sha": "ccc",
        }]).encode()
        with mock.patch.object(pr.urllib.request, "urlopen",
                               lambda req, timeout=None: FakeResponse(body)), \
                mock.patch.object(pr, "_sleep", lambda _s: None):
            refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG)
        self.assertEqual(refs, {"base_sha": "aaa", "start_sha": "bbb", "head_sha": "ccc"})

    def test_returns_none_on_failure(self):
        with mock.patch.object(pr.urllib.request, "urlopen",
                               lambda req, timeout=None: (_ for _ in ()).throw(http_error(404, b"not found"))), \
                mock.patch.object(pr, "_sleep", lambda _s: None):
            refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG)
        self.assertIsNone(refs)

    def test_returns_none_on_empty_versions(self):
        with mock.patch.object(pr.urllib.request, "urlopen",
                               lambda req, timeout=None: FakeResponse(b"[]")), \
                mock.patch.object(pr, "_sleep", lambda _s: None):
            refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG)
        self.assertIsNone(refs)

    def test_uses_retry_on_transient(self):
        body = json.dumps([{
            "base_commit_sha": "aaa", "start_commit_sha": "bbb", "head_commit_sha": "ccc",
        }]).encode()
        calls = {"n": 0}

        def fake_urlopen(req, timeout=None):
            calls["n"] += 1
            if calls["n"] == 1:
                raise http_error(500, b"server error")
            return FakeResponse(body)

        with mock.patch.object(pr.urllib.request, "urlopen", fake_urlopen), \
                mock.patch.object(pr, "_sleep", lambda _s: None), \
                mock.patch.object(pr.random, "random", lambda: 0.5):
            refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG)
        self.assertEqual(calls["n"], 2)
        self.assertIsNotNone(refs)

    def _versions(self, *triples):
        # triples: (base, start, head, [created_at])
        return [{
            "base_commit_sha": t[0], "start_commit_sha": t[1],
            "head_commit_sha": t[2],
            **({"created_at": t[3]} if len(t) > 3 else {}),
        } for t in triples]

    def test_picks_version_matching_expected_head(self):
        versions = self._versions(