78 lines
4.2 KiB
Python
78 lines
4.2 KiB
Python
from contextlib import redirect_stdout
|
|
from dataclasses import replace
|
|
from io import StringIO
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
|
|
from gitea_service import GiteaConfig, GiteaError
|
|
import gitea_preflight
|
|
|
|
|
|
class GiteaPreflightTests(unittest.TestCase):
|
|
config = GiteaConfig('https://gitea.invalid', 'synthetic-private-value', 'kai', 'pingu-concerts')
|
|
|
|
def test_only_get_requests_with_existing_scopes_and_no_payload_logging(self):
|
|
requests = []
|
|
def handler(request):
|
|
requests.append(request)
|
|
self.assertEqual(request.method, 'GET')
|
|
self.assertNotIn(self.config.token, str(request.url))
|
|
self.assertEqual(request.headers['Authorization'], 'token ' + self.config.token)
|
|
if request.url.path.endswith('/user'):
|
|
return httpx.Response(200, json={'login': 'metalcircle-bot'})
|
|
self.assertEqual(request.url.path, '/api/v1/repos/kai/pingu-concerts/issues')
|
|
self.assertEqual(request.url.params['limit'], '1')
|
|
return httpx.Response(200, json=[{'body': 'private report contents'}])
|
|
output = StringIO()
|
|
with redirect_stdout(output):
|
|
gitea_preflight.check(self.config, httpx.MockTransport(handler))
|
|
self.assertEqual(output.getvalue(), '')
|
|
self.assertEqual(len(requests), 2)
|
|
|
|
def test_missing_gitea_values_and_invalid_configuration_never_contact_network(self):
|
|
for field in ('url', 'token', 'owner', 'repo'):
|
|
with self.subTest(field=field), self.assertRaisesRegex(GiteaError, '^configuration$'):
|
|
gitea_preflight.check(replace(self.config, **{field: ''}))
|
|
for field, value in (('url', 'file:///tmp/test'), ('url', 'https://user:pass@gitea.invalid'),
|
|
('url', 'https://gitea.invalid:invalid'), ('url', 'https://bad host'),
|
|
('url', 'https://gitea.invalid?token=secret'), ('owner', '../owner'),
|
|
('repo', '..'), ('repo', 'some/repo')):
|
|
with self.subTest(field=field, value=value), self.assertRaisesRegex(GiteaError, '^configuration$'):
|
|
gitea_preflight.check(replace(self.config, **{field: value}))
|
|
|
|
def test_wrong_account_stops_before_repository_access(self):
|
|
def handler(request):
|
|
self.assertTrue(request.url.path.endswith('/user'))
|
|
return httpx.Response(200, json={'login': 'kai'})
|
|
with self.assertRaisesRegex(GiteaError, '^wrong_account$'):
|
|
gitea_preflight.check(self.config, httpx.MockTransport(handler))
|
|
|
|
def test_http_failures_are_sanitized_for_both_requests(self):
|
|
for endpoint in ('/user', '/issues'):
|
|
for status in (401, 403, 404, 500, 503, 302):
|
|
def handler(request):
|
|
if request.url.path.endswith(endpoint):
|
|
return httpx.Response(status, text=self.config.token + ' private response')
|
|
return httpx.Response(200, json={'login': 'metalcircle-bot'})
|
|
with self.subTest(endpoint=endpoint, status=status):
|
|
with self.assertRaisesRegex(GiteaError, '^http_' + str(status) + '$'):
|
|
gitea_preflight.check(self.config, httpx.MockTransport(handler))
|
|
|
|
def test_timeout_network_and_invalid_response_are_sanitized(self):
|
|
for error in (httpx.ConnectError(self.config.token), httpx.ReadTimeout(self.config.token)):
|
|
def handler(request):
|
|
raise error
|
|
with self.assertRaisesRegex(GiteaError, '^network_or_timeout$'):
|
|
gitea_preflight.check(self.config, httpx.MockTransport(handler))
|
|
for response in (httpx.Response(200, text='invalid json'), httpx.Response(200, json=[])):
|
|
with self.assertRaisesRegex(GiteaError, '^invalid_response$'):
|
|
gitea_preflight.check(self.config, httpx.MockTransport(lambda request: response))
|
|
|
|
def test_cli_output_only_contains_safe_categories(self):
|
|
output = StringIO()
|
|
with patch.object(gitea_preflight, 'check', side_effect=GiteaError('http_403')), redirect_stdout(output):
|
|
self.assertEqual(gitea_preflight.main(), 1)
|
|
self.assertEqual(output.getvalue(), 'FAIL: gitea_http_403\n')
|