50 lines
2.8 KiB
Python
50 lines
2.8 KiB
Python
import hashlib
|
|
import io
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.error import HTTPError,URLError
|
|
|
|
from paper_atlas.io.http import fetch
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
def __init__(self,body,expected=None):
|
|
super().__init__(body);self.status=200;self.headers={'Content-Length':str(expected if expected is not None else len(body))}
|
|
|
|
|
|
class FetchTests(unittest.TestCase):
|
|
def test_transient_connection_failure_is_retried(self):
|
|
with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.io.http.time.sleep'),patch('paper_atlas.io.http.urllib.request.urlopen',side_effect=[URLError(ConnectionResetError('reset')),Response(b'complete')]) as network:
|
|
self.assertEqual(fetch('https://example.org/metadata',tmp),b'complete')
|
|
self.assertEqual(network.call_count,2)
|
|
|
|
def test_truncated_response_is_not_cached_as_complete(self):
|
|
with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.io.http.time.sleep'),patch('paper_atlas.io.http.urllib.request.urlopen',side_effect=[Response(b'bad',expected=8),Response(b'complete')]):
|
|
self.assertEqual(fetch('https://example.org/metadata',tmp),b'complete')
|
|
path=Path(tmp)/hashlib.sha256(b'https://example.org/metadata').hexdigest()
|
|
self.assertEqual(path.read_bytes(),b'complete')
|
|
|
|
def test_forbidden_response_is_not_retried_or_bypassed(self):
|
|
body=io.BytesIO(b'Forbidden');error=HTTPError('https://example.org/metadata',403,'Forbidden',{},body)
|
|
with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.io.http.time.sleep'),patch('paper_atlas.io.http.urllib.request.urlopen',side_effect=error) as network:
|
|
with self.assertRaises(HTTPError):fetch('https://example.org/metadata',tmp)
|
|
self.assertEqual(network.call_count,1)
|
|
self.assertTrue(body.closed)
|
|
|
|
def test_transport_retries_are_bounded(self):
|
|
with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.io.http.time.sleep'),patch('paper_atlas.io.http.urllib.request.urlopen',side_effect=ConnectionResetError('reset')) as network:
|
|
with self.assertRaises(ConnectionResetError):fetch('https://example.org/metadata',tmp)
|
|
self.assertEqual(network.call_count,3)
|
|
|
|
def test_cached_response_still_obeys_the_size_limit(self):
|
|
with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.io.http.urllib.request.urlopen') as network:
|
|
path=Path(tmp)/hashlib.sha256(b'https://example.org/metadata').hexdigest();path.write_bytes(b'too large')
|
|
with self.assertRaisesRegex(RuntimeError,'exceeds byte limit'):
|
|
fetch('https://example.org/metadata',tmp,limit=2)
|
|
network.assert_not_called()
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|