| 684 | |
| 685 | |
| 686 | class FetchDiffRefsTest(unittest.TestCase): |
| 687 | API_BASE = "https://gitlab.example/api/v4/projects/1/merge_requests/2" |
| 688 | TOKEN = "test-token" |
| 689 | AUTH_HEADER = "PRIVATE-TOKEN" |
| 690 | |
| 691 | def test_success(self): |
| 692 | body = json.dumps([{ |
| 693 | "base_commit_sha": "aaa", |
| 694 | "start_commit_sha": "bbb", |
| 695 | "head_commit_sha": "ccc", |
| 696 | }]).encode() |
| 697 | |
| 698 | with mock.patch.object(pr.urllib.request, "urlopen", |
| 699 | lambda req, timeout=None: FakeResponse(body)), \ |
| 700 | mock.patch.object(pr, "_sleep", lambda _s: None): |
| 701 | refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG) |
| 702 | |
| 703 | self.assertEqual(refs, {"base_sha": "aaa", "start_sha": "bbb", "head_sha": "ccc"}) |
| 704 | |
| 705 | def test_returns_none_on_failure(self): |
| 706 | with mock.patch.object(pr.urllib.request, "urlopen", |
| 707 | lambda req, timeout=None: (_ for _ in ()).throw(http_error(404, b"not found"))), \ |
| 708 | mock.patch.object(pr, "_sleep", lambda _s: None): |
| 709 | refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG) |
| 710 | |
| 711 | self.assertIsNone(refs) |
| 712 | |
| 713 | def test_returns_none_on_empty_versions(self): |
| 714 | with mock.patch.object(pr.urllib.request, "urlopen", |
| 715 | lambda req, timeout=None: FakeResponse(b"[]")), \ |
| 716 | mock.patch.object(pr, "_sleep", lambda _s: None): |
| 717 | refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG) |
| 718 | |
| 719 | self.assertIsNone(refs) |
| 720 | |
| 721 | def test_uses_retry_on_transient(self): |
| 722 | body = json.dumps([{ |
| 723 | "base_commit_sha": "aaa", |
| 724 | "start_commit_sha": "bbb", |
| 725 | "head_commit_sha": "ccc", |
| 726 | }]).encode() |
| 727 | calls = {"n": 0} |
| 728 | |
| 729 | def fake_urlopen(req, timeout=None): |
| 730 | calls["n"] += 1 |
| 731 | if calls["n"] == 1: |
| 732 | raise http_error(500, b"server error") |
| 733 | return FakeResponse(body) |
| 734 | |
| 735 | with mock.patch.object(pr.urllib.request, "urlopen", fake_urlopen), \ |
| 736 | mock.patch.object(pr, "_sleep", lambda _s: None), \ |
| 737 | mock.patch.object(pr.random, "random", lambda: 0.5): |
| 738 | refs = pr.fetch_diff_refs(self.API_BASE, self.TOKEN, self.AUTH_HEADER, DEFAULT_CONFIG) |
| 739 | |
| 740 | self.assertEqual(calls["n"], 2) |
| 741 | self.assertIsNotNone(refs) |
| 742 | |
| 743 |
nothing calls this directly
no outgoing calls
no test coverage detected