sashatrask
77 lines
3.5 KiB
Python
77 lines
3.5 KiB
Python
from pathlib import Path
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_DIR = ROOT / 'app'
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
import scanner
|
|
|
|
|
|
class ApiDeadlinePlumbingTests(unittest.TestCase):
|
|
def test_deadline_is_consumed_internally_not_forwarded_to_requests(self):
|
|
response = mock.Mock(status_code=200)
|
|
with mock.patch.object(scanner.requests, 'request', return_value=response) as request:
|
|
result = scanner.api_request(
|
|
'GET', 'https://example.invalid', timeout=10,
|
|
deadline=time.monotonic() + 1,
|
|
)
|
|
self.assertIs(result, response)
|
|
self.assertNotIn('deadline', request.call_args.kwargs)
|
|
self.assertGreater(request.call_args.kwargs['timeout'], 0)
|
|
self.assertLessEqual(request.call_args.kwargs['timeout'], 1)
|
|
|
|
def test_expired_deadline_never_reaches_requests(self):
|
|
with mock.patch.object(scanner.requests, 'request') as request:
|
|
with self.assertRaisesRegex(scanner.ApiRequestError, 'deadline expired'):
|
|
scanner.api_request('GET', 'https://example.invalid', deadline=time.monotonic() - 1)
|
|
request.assert_not_called()
|
|
|
|
|
|
def test_deadline_bounds_unset_timeouts(self):
|
|
for timeout in (None, (None, 30), (2, None), (None, None)):
|
|
with self.subTest(timeout=timeout):
|
|
response = mock.Mock(status_code=200)
|
|
with mock.patch.object(scanner.scan_config, 'api_proxy_enabled', False), \
|
|
mock.patch.object(scanner.requests, 'request', return_value=response) as request:
|
|
self.assertIs(scanner.api_request(
|
|
'GET', 'https://example.invalid', timeout=timeout,
|
|
deadline=time.monotonic() + 1,
|
|
), response)
|
|
effective = request.call_args.kwargs['timeout']
|
|
for value in effective if isinstance(effective, tuple) else (effective,):
|
|
self.assertGreater(value, 0)
|
|
self.assertLessEqual(value, 1)
|
|
|
|
def test_response_after_deadline_is_closed_not_returned_or_retried(self):
|
|
for stream in (False, True):
|
|
for status in (200, 503):
|
|
with self.subTest(stream=stream, status=status):
|
|
clock = [0.0]
|
|
response = mock.Mock(status_code=status)
|
|
|
|
def late_response(*args, **kwargs):
|
|
clock[0] = 11.0
|
|
return response
|
|
|
|
with mock.patch.object(scanner.time, 'monotonic', side_effect=lambda: clock[0]), \
|
|
mock.patch.object(scanner.scan_config, 'api_proxy_enabled', False), \
|
|
mock.patch.object(scanner.requests, 'request', side_effect=late_response) as request, \
|
|
mock.patch.object(scanner, '_wait_or_raise_scan_slot_fatal') as wait:
|
|
with self.assertRaisesRegex(scanner.ApiRequestError, 'deadline expired after response'):
|
|
scanner.api_request(
|
|
'GET', 'https://example.invalid', timeout=30,
|
|
deadline=10, max_retries=3, stream=stream,
|
|
)
|
|
request.assert_called_once()
|
|
response.close.assert_called_once()
|
|
wait.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|