104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.parse
|
|
|
|
from werkrevisor.gitea import GiteaClient, GiteaError, PAGE_LIMIT, secure_opener
|
|
|
|
TOKEN = "gitea-secret-token-123"
|
|
|
|
|
|
class FakeResponse(io.BytesIO):
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
self.close()
|
|
|
|
|
|
class FakeOpener:
|
|
def __init__(self, handler):
|
|
self.handler = handler
|
|
self.requests = []
|
|
|
|
def open(self, request, timeout=None):
|
|
self.requests.append(request)
|
|
return self.handler(request)
|
|
|
|
|
|
def client(handler):
|
|
opener = FakeOpener(handler)
|
|
return GiteaClient("https://git.example.org/", "team", "app", TOKEN, opener=opener), opener
|
|
|
|
|
|
class GiteaClientTest(unittest.TestCase):
|
|
def test_pagination_and_auth_header(self):
|
|
def handler(request):
|
|
page = int(urllib.parse.parse_qs(urllib.parse.urlparse(request.full_url).query)["page"][0])
|
|
items = [{"n": i} for i in range(PAGE_LIMIT)] if page == 1 else [{"n": "last"}]
|
|
return FakeResponse(json.dumps(items).encode())
|
|
c, opener = client(handler)
|
|
self.assertEqual(len(c.list_pr_files(7)), PAGE_LIMIT + 1)
|
|
self.assertEqual(opener.requests[0].get_header("Authorization"), f"token {TOKEN}")
|
|
self.assertTrue(opener.requests[0].full_url.startswith(
|
|
"https://git.example.org/api/v1/repos/team/app/pulls/7/files?"))
|
|
|
|
def test_open_prs_keep_the_state_on_every_page(self):
|
|
def handler(request):
|
|
query = urllib.parse.parse_qs(urllib.parse.urlparse(request.full_url).query)
|
|
items = [{"n": i} for i in range(PAGE_LIMIT)] if query["page"] == ["1"] else []
|
|
return FakeResponse(json.dumps(items).encode())
|
|
c, opener = client(handler)
|
|
self.assertEqual(len(c.list_open_prs()), PAGE_LIMIT)
|
|
for request in opener.requests:
|
|
url = urllib.parse.urlparse(request.full_url)
|
|
self.assertEqual((url.path, urllib.parse.parse_qs(url.query)["state"]),
|
|
("/api/v1/repos/team/app/pulls", ["open"]))
|
|
|
|
def test_post_status(self):
|
|
c, opener = client(lambda request: FakeResponse(b"{}"))
|
|
c.set_status("abc", "pending", "ai-review/assessment", "x" * 300, "https://u")
|
|
request = opener.requests[0]
|
|
body = json.loads(request.data)
|
|
self.assertEqual((request.get_method(), body["state"], len(body["description"])), ("POST", "pending", 250))
|
|
self.assertTrue(request.full_url.endswith("/repos/team/app/statuses/abc"))
|
|
|
|
def test_create_review(self):
|
|
c, opener = client(lambda request: FakeResponse(b"{}"))
|
|
c.create_review(7, "APPROVED", "ok", "abc")
|
|
request = opener.requests[0]
|
|
self.assertTrue(request.full_url.endswith("/repos/team/app/pulls/7/reviews"))
|
|
self.assertEqual(json.loads(request.data), {"event": "APPROVED", "body": "ok", "commit_id": "abc"})
|
|
|
|
def test_raw_file_at_ref(self):
|
|
c, opener = client(lambda request: FakeResponse(b"\xff\0bytes"))
|
|
self.assertEqual(c.get_raw_file("tools/my file.sh", "abc"), b"\xff\0bytes")
|
|
self.assertTrue(opener.requests[0].full_url.endswith("/repos/team/app/raw/tools/my%20file.sh?ref=abc"))
|
|
|
|
def test_raw_file_reads_at_most_max_bytes(self):
|
|
c, _ = client(lambda request: FakeResponse(b"0123456789"))
|
|
self.assertEqual(c.get_raw_file("logo.png", "abc", max_bytes=4), b"0123")
|
|
|
|
def test_raw_diff(self):
|
|
c, _ = client(lambda request: FakeResponse(b"diff --git a/x b/x\n"))
|
|
self.assertEqual(c.get_pr_diff(7), "diff --git a/x b/x\n")
|
|
|
|
def test_http_error_does_not_leak_the_token(self):
|
|
def handler(request):
|
|
raise urllib.error.HTTPError(request.full_url, 403, "Forbidden", {}, None)
|
|
c, _ = client(handler)
|
|
with self.assertRaises(GiteaError) as caught:
|
|
c.get_pr(7)
|
|
self.assertIn("HTTP 403", str(caught.exception))
|
|
self.assertNotIn(TOKEN, str(caught.exception))
|
|
|
|
def test_redirects_are_not_followed(self):
|
|
handlers = [type(h).__name__ for h in secure_opener().handlers]
|
|
self.assertIn("_NoRedirect", handlers)
|
|
self.assertNotIn("HTTPRedirectHandler", handlers)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|