57290da1e8
Three new unit test modules: - tests/unit/test_contrib_gitea_client.py — GiteaClient (urllib mocked) and GiteaForge delegation - tests/unit/orchestrator/test_main.py — __main__ run/status commands - tests/unit/orchestrator/test_bootstrap.py — _token, BotBottleStateStore, _to_forge_state/_to_record, make_forge, make_sidecar, build Augments to existing suites: - test_events: non-"created" comment action ignored - test_lifecycle: _iso_now callable, untracked-issue comment ignored, untracked-PR closed ignored (covers _find_by_pr return-None path) - test_runner: destroy command, _default_run via subprocess mock - test_sidecar: _jsonable dataclass/list branches, OpLog.read on missing file, drain_done_events on corrupted file, socket _Handler invalid-JSON and empty-line paths, serve() with pre-existing socket path - test_watchdog: _loop body covered by patching _TICK_SECS to 0.01s - test_webhook: unknown GET path returns 404 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
159 lines
5.7 KiB
Python
159 lines
5.7 KiB
Python
"""Unit: GiteaClient and GiteaForge (urllib mocked — no network)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
import urllib.error
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bot_bottle.contrib.gitea.client import GiteaClient, GiteaForge
|
|
|
|
|
|
def _client() -> GiteaClient:
|
|
return GiteaClient(api_url="http://g/api/v1", owner="o", repo="r", token="tok")
|
|
|
|
|
|
def _mock_response(body: bytes) -> MagicMock:
|
|
resp = MagicMock()
|
|
resp.read.return_value = body
|
|
resp.__enter__ = lambda s: s
|
|
resp.__exit__ = MagicMock(return_value=False)
|
|
return resp
|
|
|
|
|
|
class GiteaClientTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = _client()
|
|
|
|
def test_request_returns_parsed_json(self):
|
|
payload = {"number": 42}
|
|
resp = _mock_response(json.dumps(payload).encode())
|
|
with patch("urllib.request.urlopen", return_value=resp):
|
|
result = self.client._request("GET", "/repos/o/r/issues/42")
|
|
self.assertEqual(payload, result)
|
|
|
|
def test_request_empty_body_returns_none(self):
|
|
resp = _mock_response(b"")
|
|
with patch("urllib.request.urlopen", return_value=resp):
|
|
result = self.client._request("POST", "/some/path", {"x": 1})
|
|
self.assertIsNone(result)
|
|
|
|
def test_is_org_member_true_on_200(self):
|
|
mock_resp = MagicMock()
|
|
mock_resp.close = MagicMock()
|
|
with patch("urllib.request.urlopen", return_value=mock_resp):
|
|
self.assertTrue(self.client.is_org_member("myorg", "alice"))
|
|
|
|
def test_is_org_member_false_on_http_error(self):
|
|
with patch(
|
|
"urllib.request.urlopen",
|
|
side_effect=urllib.error.HTTPError("url", 404, "Not Found", {}, None),
|
|
):
|
|
self.assertFalse(self.client.is_org_member("myorg", "nobody"))
|
|
|
|
def test_get_issue(self):
|
|
resp = _mock_response(json.dumps({"number": 1}).encode())
|
|
with patch("urllib.request.urlopen", return_value=resp):
|
|
result = self.client.get_issue(1)
|
|
self.assertEqual(1, result["number"])
|
|
|
|
def test_get_pull(self):
|
|
resp = _mock_response(json.dumps({"number": 7, "merged": False}).encode())
|
|
with patch("urllib.request.urlopen", return_value=resp):
|
|
result = self.client.get_pull(7)
|
|
self.assertEqual(7, result["number"])
|
|
|
|
def test_list_comments(self):
|
|
resp = _mock_response(json.dumps([{"id": 1, "body": "hi"}]).encode())
|
|
with patch("urllib.request.urlopen", return_value=resp):
|
|
result = self.client.list_comments(1)
|
|
self.assertEqual(1, len(result))
|
|
self.assertEqual(1, result[0]["id"])
|
|
|
|
def test_create_comment(self):
|
|
resp = _mock_response(b"")
|
|
with patch("urllib.request.urlopen", return_value=resp) as mock_open:
|
|
self.client.create_comment(1, "hello")
|
|
mock_open.assert_called_once()
|
|
|
|
def test_update_issue(self):
|
|
resp = _mock_response(b"")
|
|
with patch("urllib.request.urlopen", return_value=resp) as mock_open:
|
|
self.client.update_issue(1, "new body")
|
|
mock_open.assert_called_once()
|
|
|
|
def test_request_builds_correct_url(self):
|
|
captured: list[object] = []
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured.append(req)
|
|
return _mock_response(b"{}")
|
|
|
|
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
self.client.get_issue(5)
|
|
|
|
import urllib.request as ureq
|
|
req = captured[0]
|
|
assert isinstance(req, ureq.Request)
|
|
self.assertIn("/issues/5", req.full_url)
|
|
|
|
def test_request_sends_auth_header(self):
|
|
captured: list[object] = []
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured.append(req)
|
|
return _mock_response(b"{}")
|
|
|
|
with patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
self.client.get_issue(1)
|
|
|
|
import urllib.request as ureq
|
|
req = captured[0]
|
|
assert isinstance(req, ureq.Request)
|
|
self.assertEqual("token tok", req.get_header("Authorization"))
|
|
|
|
|
|
class GiteaForgeTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = MagicMock(spec=GiteaClient)
|
|
self.forge = GiteaForge(self.client)
|
|
|
|
def test_is_org_member_delegates(self):
|
|
self.client.is_org_member.return_value = True
|
|
self.assertTrue(self.forge.is_org_member("org", "alice"))
|
|
self.client.is_org_member.assert_called_once_with("org", "alice")
|
|
|
|
def test_is_org_member_false(self):
|
|
self.client.is_org_member.return_value = False
|
|
self.assertFalse(self.forge.is_org_member("org", "outsider"))
|
|
|
|
def test_read_issue_delegates(self):
|
|
self.client.get_issue.return_value = {"number": 3}
|
|
self.assertEqual({"number": 3}, self.forge.read_issue(3))
|
|
self.client.get_issue.assert_called_once_with(3)
|
|
|
|
def test_read_pr_delegates(self):
|
|
self.client.get_pull.return_value = {"number": 5, "merged": False}
|
|
result = self.forge.read_pr(5)
|
|
self.assertEqual(5, result["number"])
|
|
self.client.get_pull.assert_called_once_with(5)
|
|
|
|
def test_read_comments_delegates(self):
|
|
self.client.list_comments.return_value = [{"id": 1}]
|
|
comments = self.forge.read_comments(1)
|
|
self.assertEqual([{"id": 1}], comments)
|
|
self.client.list_comments.assert_called_once_with(1)
|
|
|
|
def test_post_comment_delegates(self):
|
|
self.forge.post_comment(1, "looks good")
|
|
self.client.create_comment.assert_called_once_with(1, "looks good")
|
|
|
|
def test_update_description_delegates(self):
|
|
self.forge.update_description(1, "updated body")
|
|
self.client.update_issue.assert_called_once_with(1, "updated body")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|