113 lines
5.3 KiB
Python
113 lines
5.3 KiB
Python
"""Real HTTP checks of the standalone Java demo service; no external providers."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
class MockTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.temp = tempfile.TemporaryDirectory()
|
|
cls.work = Path(cls.temp.name)
|
|
cls.java = str(Path(os.environ['JAVA_HOME']) / 'bin' / 'java')
|
|
javac = str(Path(os.environ['JAVA_HOME']) / 'bin' / 'javac')
|
|
subprocess.run([javac, '--release', '21', '-d', str(cls.work), str(HERE / 'MockServer.java')], check=True)
|
|
with socket.socket() as s:
|
|
s.bind(('127.0.0.1', 0))
|
|
cls.port = s.getsockname()[1]
|
|
cls.base = f'http://localhost:{cls.port}'
|
|
cls.env = dict(os.environ, MOCK_ADMIN_TOKEN='synthetic-test-token', MOCK_DATA=str(cls.work / 'data'),
|
|
MOCK_PORT=str(cls.port), MOCK_BIND='127.0.0.1')
|
|
cls.start()
|
|
|
|
@classmethod
|
|
def start(cls):
|
|
cls.proc = subprocess.Popen([cls.java, '-cp', str(cls.work), 'MockServer'], env=cls.env,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
for _ in range(100):
|
|
try:
|
|
with urllib.request.urlopen(cls.base + '/health', timeout=1) as r:
|
|
if r.status == 200:
|
|
return
|
|
except (OSError, urllib.error.URLError):
|
|
time.sleep(.1)
|
|
cls.proc.terminate()
|
|
cls.proc.wait(timeout=10)
|
|
raise AssertionError('Mock failed to start')
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.proc.terminate()
|
|
cls.proc.wait(timeout=10)
|
|
cls.temp.cleanup()
|
|
|
|
def call(self, path, method='POST', body=b'{}', headers=None):
|
|
req = urllib.request.Request(self.base + path, data=body if method != 'GET' else None,
|
|
headers=headers or {}, method=method)
|
|
try:
|
|
r = urllib.request.urlopen(req, timeout=3)
|
|
except urllib.error.HTTPError as e:
|
|
r = e
|
|
with r:
|
|
return r.status, r.read(), r.headers
|
|
|
|
def set_response(self, name, body, status=200, delay=0):
|
|
return self.call('/__admin/' + name, 'PUT', body, {'Authorization': 'Bearer synthetic-test-token',
|
|
'X-Mock-Status': str(status), 'X-Mock-Delay-Ms': str(delay)})
|
|
|
|
def test_auth_limits_and_unknown_paths(self):
|
|
self.assertEqual(401, self.call('/__admin/clientes_consulta', 'PUT')[0])
|
|
self.assertEqual(404, self.call('/missing')[0])
|
|
self.assertEqual(405, self.call('/clientes_consulta', 'GET')[0])
|
|
self.assertEqual(400, self.set_response('clientes_consulta', b'{}', delay=-1)[0])
|
|
self.assertEqual(400, self.set_response('clientes_consulta', b'{}', status=204)[0])
|
|
self.assertEqual(413, self.set_response('clientes_consulta', b'x' * 262145)[0])
|
|
|
|
def test_java_healthcheck_and_custom_metadata(self):
|
|
subprocess.run([self.java, '-cp', str(self.work),
|
|
'MockServer', '--health'], env=self.env, check=True, timeout=15, capture_output=True)
|
|
body = b'{"documentId":"synthetic-document","requestId":"synthetic-request"}'
|
|
self.assertEqual(200, self.set_response('documents_metadata', body)[0])
|
|
self.assertEqual((200, body), self.call('/documents_metadata')[:2])
|
|
self.set_response('documents_metadata', b'{"error":"not configured"}', status=404)
|
|
|
|
def test_configurable_results_errors_delay_and_restart(self):
|
|
for body in (b'{"found":true,"customerId":"synthetic"}', b'{"found":false}'):
|
|
self.assertEqual(200, self.set_response('clientes_consulta', body)[0])
|
|
self.assertEqual((200, body), self.call('/clientes_consulta')[:2])
|
|
self.set_response('clientes_consulta', b'{"error":"unavailable"}', status=503, delay=200)
|
|
before = time.monotonic()
|
|
self.assertEqual(503, self.call('/clientes_consulta')[0])
|
|
self.assertGreaterEqual(time.monotonic() - before, .18)
|
|
cls = type(self)
|
|
cls.proc.terminate(); cls.proc.wait(timeout=10); cls.start()
|
|
status, body, headers = self.call('/__admin/clientes_consulta', 'GET', headers={'Authorization':'Bearer synthetic-test-token'})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual('503', headers['X-Mock-Status'])
|
|
self.assertEqual('200', headers['X-Mock-Delay-Ms'])
|
|
self.assertEqual('0', headers['X-Mock-Calls'])
|
|
|
|
def test_scoring_documents_and_malformed_output(self):
|
|
for score in (71, 70, 61, 60):
|
|
body = json.dumps({'score': score, 'consultationId': 'synthetic'}).encode()
|
|
self.set_response('buro_consulta', body)
|
|
self.assertEqual((200, body), self.call('/buro_consulta')[:2])
|
|
self.set_response('documento_valida', b'{"status":"INVALID","reasonCodes":["EXPIRED"]}')
|
|
self.assertEqual('INVALID', json.loads(self.call('/documento_valida')[1])['status'])
|
|
self.assertEqual(404, self.call('/documents_metadata')[0])
|
|
self.set_response('buro_consulta', b'not-json')
|
|
self.assertEqual(b'not-json', self.call('/buro_consulta')[1])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|