144 lines
8.9 KiB
Python
144 lines
8.9 KiB
Python
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
from starlette.requests import Request
|
|
|
|
from bug_reporter import browser_summary, issue_body, safe_route
|
|
from gitea_service import GiteaConfig, GiteaError, GiteaService, redact
|
|
|
|
|
|
class GiteaTests(unittest.TestCase):
|
|
def service(self, handler):
|
|
return GiteaService(GiteaConfig('http://gitea.invalid', 'test-secret-never-print', 'kai', 'pingu-concerts'), httpx.MockTransport(handler))
|
|
|
|
def test_existing_labels_bot_and_payload(self):
|
|
requests = []
|
|
def handler(request):
|
|
requests.append(request)
|
|
if request.url.path.endswith('/user'):
|
|
return httpx.Response(200, json={'login': 'metalcircle-bot'})
|
|
if request.url.path.endswith('/labels'):
|
|
return httpx.Response(200, json=[{'id': 7, 'name': 'reported-from-metalcircle'}, {'id': 9, 'name': 'android'}, {'id': 99, 'name': 'unrelated'}])
|
|
return httpx.Response(201, json={'number': 123, 'user': {'login': 'metalcircle-bot'}})
|
|
self.assertEqual(self.service(handler).create_issue('A test title', 'A test body', 'android'), 123)
|
|
payload = json.loads(requests[-1].content)
|
|
self.assertEqual(payload['labels'], [7, 9])
|
|
self.assertEqual(payload['title'], '[MetalCircle Bug] A test title')
|
|
self.assertEqual(requests[-1].url.path, '/api/v1/repos/kai/pingu-concerts/issues')
|
|
self.assertNotIn('test-secret', str(requests[-1].url))
|
|
|
|
def test_screenshot_is_uploaded_as_issue_asset_and_linked_into_issue_body(self):
|
|
requests = []
|
|
image = b"sanitized-raster-image-bytes"
|
|
def handler(request):
|
|
requests.append(request)
|
|
if request.url.path.endswith('/user'):
|
|
return httpx.Response(200, json={'login':'metalcircle-bot'})
|
|
if request.url.path.endswith('/labels'):
|
|
return httpx.Response(200, json=[])
|
|
if request.method == 'POST' and request.url.path.endswith('/issues'):
|
|
return httpx.Response(201, json={'number':123,'user':{'login':'metalcircle-bot'}})
|
|
if request.method == 'POST' and request.url.path.endswith('/issues/123/assets'):
|
|
self.assertEqual(request.url.params['name'], 'screenshot-safe.png')
|
|
self.assertIn('multipart/form-data', request.headers['content-type'])
|
|
self.assertIn(image, request.read())
|
|
self.assertIn('filename="screenshot-safe.png"', request.read().decode('latin1'))
|
|
return httpx.Response(201, json={'browser_download_url':'http://gitea.invalid/attachments/safe-id','id':5})
|
|
if request.method == 'PATCH' and request.url.path.endswith('/issues/123'):
|
|
self.assertIn('', json.loads(request.content)['body'])
|
|
return httpx.Response(200, json={'number':123})
|
|
raise AssertionError(f'unexpected request {request.method} {request.url.path}')
|
|
result=self.service(handler).create_issue('Screenshot test','safe body','android',
|
|
screenshot=('screenshot-safe.png',image,'image/png'))
|
|
self.assertEqual(result,123)
|
|
self.assertEqual([request.method for request in requests],['GET','GET','POST','POST','PATCH'])
|
|
self.assertNotIn(b'test-secret-never-print', b''.join(request.content for request in requests))
|
|
|
|
def test_screenshot_upload_failure_is_uncertain_and_secret_free(self):
|
|
def handler(request):
|
|
if request.url.path.endswith('/user'): return httpx.Response(200,json={'login':'metalcircle-bot'})
|
|
if request.url.path.endswith('/labels'): return httpx.Response(200,json=[])
|
|
if request.method == 'POST' and request.url.path.endswith('/issues'): return httpx.Response(201,json={'number':9,'user':{'login':'metalcircle-bot'}})
|
|
return httpx.Response(413,text='private upload failure details')
|
|
with self.assertLogs('gitea_service','WARNING') as logs:
|
|
with self.assertRaises(GiteaError) as caught:
|
|
self.service(handler).create_issue('Title','Body','general',screenshot=('screenshot.jpg',b'image','image/jpeg'))
|
|
self.assertTrue(caught.exception.uncertain)
|
|
self.assertNotIn('private upload failure details','\n'.join(logs.output))
|
|
|
|
def test_missing_or_inaccessible_labels_do_not_block(self):
|
|
for response in (httpx.Response(200, json=[]), httpx.Response(403), httpx.Response(200, json={'invalid': True})):
|
|
def handler(request):
|
|
if request.url.path.endswith('/user'): return httpx.Response(200, json={'login': 'metalcircle-bot'})
|
|
if request.url.path.endswith('/labels'): return response
|
|
self.assertEqual(json.loads(request.content)['labels'], [])
|
|
return httpx.Response(201, json={'number': 4, 'user': {'login': 'metalcircle-bot'}})
|
|
self.assertEqual(self.service(handler).create_issue('Test title', 'body', 'general'), 4)
|
|
|
|
def test_personal_account_is_rejected_before_post(self):
|
|
def handler(request):
|
|
self.assertEqual(request.method, 'GET')
|
|
return httpx.Response(200, json={'login': 'kai'})
|
|
with self.assertRaisesRegex(GiteaError, 'wrong_account'):
|
|
self.service(handler).create_issue('title', 'body', 'general')
|
|
|
|
def test_http_errors_and_secret_free_logs(self):
|
|
for code in (401, 403, 404, 500, 502):
|
|
def handler(request):
|
|
if request.url.path.endswith('/user'): return httpx.Response(200, json={'login': 'metalcircle-bot'})
|
|
if request.url.path.endswith('/labels'): return httpx.Response(200, json=[])
|
|
return httpx.Response(code, text='test-secret-never-print')
|
|
with self.subTest(code=code), self.assertLogs('gitea_service', 'WARNING') as logs:
|
|
with self.assertRaises(GiteaError) as caught:
|
|
self.service(handler).create_issue('title', 'body', 'general')
|
|
self.assertEqual(caught.exception.uncertain, code >= 500)
|
|
self.assertNotIn('test-secret-never-print', '\n'.join(logs.output))
|
|
|
|
def test_timeout_unreachable_and_invalid_response(self):
|
|
for failure in ('timeout', 'unreachable', 'invalid_json', 'invalid_issue'):
|
|
def handler(request):
|
|
if failure == 'unreachable': raise httpx.ConnectError('test-secret-never-print')
|
|
if request.url.path.endswith('/user'): return httpx.Response(200, json={'login': 'metalcircle-bot'})
|
|
if request.url.path.endswith('/labels'): return httpx.Response(200, json=[])
|
|
if failure == 'timeout': raise httpx.ReadTimeout('test-secret-never-print')
|
|
if failure == 'invalid_json': return httpx.Response(201, text='not json')
|
|
return httpx.Response(201, json={'number': 'x'})
|
|
with self.subTest(failure=failure), self.assertLogs('gitea_service', 'WARNING') as logs:
|
|
with self.assertRaises(GiteaError) as caught:
|
|
self.service(handler).create_issue('title', 'body', 'general')
|
|
self.assertEqual(caught.exception.uncertain, failure != 'unreachable')
|
|
self.assertNotIn('test-secret', '\n'.join(logs.output))
|
|
|
|
def test_context_allowlist_and_redaction(self):
|
|
secret = 'session-secret-never-send'
|
|
request = Request({'type': 'http', 'headers': [
|
|
(b'cookie', ('pingu_session='+secret).encode()),
|
|
(b'authorization', b'Bearer authorization-secret'),
|
|
(b'user-agent', b'Chrome/128.0 secret-header'),
|
|
]})
|
|
data = dict(description='This is '+secret, expected='Expected result', steps='', category='android',
|
|
severity='normal', technical=True, platform='Android', app_version='1.1.0-debug',
|
|
route='/concerts/1?token=secret-query')
|
|
with patch.dict('os.environ', {'GITEA_TOKEN': 'bot-secret-never-send'}):
|
|
body = issue_body(data, {'id': 12, 'username': 'real-user'}, request)
|
|
for excluded in (secret, 'authorization-secret', 'secret-header', 'secret-query', 'bot-secret-never-send'):
|
|
self.assertNotIn(excluded, body)
|
|
self.assertIn('real-user', body)
|
|
self.assertIn('12', body)
|
|
self.assertIn('/concerts/1', body)
|
|
self.assertIn('Chrome 128.0', body)
|
|
self.assertEqual(safe_route('/password-reset/sensitive'), '/')
|
|
self.assertEqual(safe_route('/register/sensitive'), '/')
|
|
self.assertEqual(browser_summary('Cookie: evil'), 'Unknown')
|
|
data['technical'] = False
|
|
self.assertNotIn('Chrome', issue_body(data, {'id': 12, 'username': 'real-user'}, request))
|
|
self.assertNotIn('/concerts/1', issue_body(data, {'id': 12, 'username': 'real-user'}, request))
|
|
self.assertNotIn('super-secret', redact('password=super-secret'))
|
|
self.assertNotIn('private bytes', redact('-----BEGIN PRIVATE KEY-----\nprivate bytes\n-----END PRIVATE KEY-----'))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|