61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
"""Exercise the external adapter against a real local HTTP/SOAP fixture."""
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import xml.etree.ElementTree as ET
|
|
|
|
root = Path(__file__).resolve().parents[2]
|
|
spec = importlib.util.spec_from_file_location('testbot_client', root / 'tools/testbots/client.py')
|
|
client = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(client)
|
|
requests = []
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def do_POST(self):
|
|
assert self.headers['Authorization'] == 'Basic ' + base64.b64encode(b'operator:secret').decode()
|
|
tree = ET.fromstring(self.rfile.read(int(self.headers['Content-Length'])))
|
|
command = next(n.text for n in tree.iter() if n.tag == 'command')
|
|
requests.append(command)
|
|
if 'submit' in command:
|
|
assert base64.b64decode(command.split()[-1]) == b'testbot.check(true, "hello <&>")\n'
|
|
text = 'Testbot run adapter_case: queued'
|
|
else:
|
|
text = json.dumps({'run_id': 'adapter_case', 'state': 'passed', 'checks': 1})
|
|
result = ET.Element('{http://schemas.xmlsoap.org/soap/envelope/}Envelope')
|
|
body = ET.SubElement(result, '{http://schemas.xmlsoap.org/soap/envelope/}Body')
|
|
ET.SubElement(body, 'result').text = text
|
|
data = ET.tostring(result)
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
|
|
server = HTTPServer(('127.0.0.1', 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
script = Path(directory) / 'case.lua'
|
|
script.write_text('testbot.check(true, "hello <&>")\n')
|
|
result = client.run(f'http://127.0.0.1:{server.server_port}/', 'operator', 'secret', 'adapter_case', script)
|
|
assert result['state'] == 'passed'
|
|
assert requests[-1] == 'clientless test result adapter_case'
|
|
try:
|
|
client.run('', '', '', '../bad', script)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError('invalid run id accepted')
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
print('External TestBot SOAP submission, base64 script integrity, result retrieval and run-ID validation passed.')
|