86 lines
4.2 KiB
Python
86 lines
4.2 KiB
Python
"""Lightweight client tests: no HTTP server, compilation or real credentials."""
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
spec = importlib.util.spec_from_file_location('documents_client', Path(__file__).with_name('documents-client.py'))
|
|
client = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(client)
|
|
|
|
|
|
class ClientTests(unittest.TestCase):
|
|
def test_only_tls_or_loopback_origins_are_accepted(self):
|
|
self.assertEqual(client.server_url('https://docs.example/'), 'https://docs.example')
|
|
self.assertEqual(client.server_url('http://127.0.0.1:18135'), 'http://127.0.0.1:18135')
|
|
for url in ['http://docs.example', 'https://user:secret@docs.example', 'file:///tmp/data',
|
|
'https://docs.example/api', 'https://docs.example/?token=secret']:
|
|
with self.assertRaises(ValueError):
|
|
client.server_url(url)
|
|
|
|
def test_key_file_permissions_and_formats(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'key.json'
|
|
token = 'tdoc_' + 'x' * 43
|
|
path.write_text(json.dumps({'token': token}))
|
|
path.chmod(0o600)
|
|
self.assertEqual(client.read_token(path), token)
|
|
path.chmod(0o644)
|
|
with self.assertRaises(ValueError):
|
|
client.read_token(path)
|
|
|
|
def test_redirects_never_forward_bearer_credentials(self):
|
|
request = urllib.request.Request('https://docs.example', headers={'Authorization': 'Bearer test-only'})
|
|
self.assertIsNone(client.NoRedirect().redirect_request(request, None, 302, '', {}, 'https://other.example'))
|
|
|
|
def test_network_retry_preserves_idempotency_and_body(self):
|
|
calls = []
|
|
|
|
class Opener:
|
|
def open(self, request, timeout):
|
|
calls.append(request)
|
|
if len(calls) == 1:
|
|
raise urllib.error.URLError('temporary network failure')
|
|
return io.BytesIO(b'{"replayed":true}')
|
|
|
|
with patch.object(client.urllib.request, 'build_opener', return_value=Opener()), patch.object(client.time, 'sleep'):
|
|
result = client.request('https://docs.example', 'test-token', 'POST', '/api/documents', {'revision': 0}, 'request-1')
|
|
self.assertTrue(result['replayed'])
|
|
self.assertEqual(calls[0].data, calls[1].data)
|
|
self.assertEqual(calls[0].get_header('Idempotency-key'), 'request-1')
|
|
self.assertEqual(calls[1].get_header('Idempotency-key'), 'request-1')
|
|
|
|
def test_large_uploads_are_split_by_count_and_bytes(self):
|
|
documents = [{'sourceSlug': str(i), 'content': 'a' * 20} for i in range(201)]
|
|
batches = client.document_batches(documents)
|
|
self.assertEqual([len(batch) for batch in batches], [100, 100, 1])
|
|
self.assertEqual([doc for batch in batches for doc in batch], documents)
|
|
bounded = client.document_batches(documents[:5], max_bytes=150)
|
|
self.assertTrue(all(len(json.dumps({'documents': batch}, ensure_ascii=False).encode()) <= 150 for batch in bounded))
|
|
|
|
def test_duplicate_identifiers_and_oversized_items_fail_before_upload(self):
|
|
with self.assertRaises(ValueError):
|
|
client.document_batches([{'sourceSlug': 'same'}, {'sourceSlug': 'same'}])
|
|
with self.assertRaises(ValueError):
|
|
client.document_batches([{'sourceSlug': 'one', 'content': 'a' * 500}], max_bytes=100)
|
|
|
|
def test_conflict_is_reported_without_printing_server_content_or_secret(self):
|
|
class Opener:
|
|
def open(self, request, timeout):
|
|
raise urllib.error.HTTPError(request.full_url, 409, 'Conflict', {}, io.BytesIO(b'{"error":"private-content"}'))
|
|
|
|
with patch.object(client.urllib.request, 'build_opener', return_value=Opener()):
|
|
with self.assertRaisesRegex(client.DocumentClientError, 'HTTP 409') as result:
|
|
client.request('https://docs.example', 'private-token', 'POST', '/api/documents', {}, 'request-1')
|
|
self.assertNotIn('private-token', str(result.exception))
|
|
self.assertNotIn('private-content', str(result.exception))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|