jakub
6a667d8deb
Include the current addon, terrain integrations, tools, regression tests and recorded development evidence.
273 lines
13 KiB
Python
273 lines
13 KiB
Python
"""Binary transport regressions plus Google request/cache boundary tests."""
|
|
import base64
|
|
import importlib.util
|
|
import http.client
|
|
import io
|
|
import json
|
|
import random
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import wave
|
|
from pathlib import Path
|
|
from policy_harness import ROOT, body, run
|
|
|
|
path = ROOT / 'addons/core/Scripts/Game/ERR/00_Core/ERR_WelcomeVoice.c'
|
|
decode = body(path, 'static bool Decode(').replace('array<int> bytes', 'Array& bytes').replace('Math.Min', 'std::min')
|
|
ready = body(path, 'static bool Ready(')
|
|
weather = body(path, 'static string Weather(')
|
|
briefing = body(path, 'static string Briefing(')
|
|
key = body(path, 'static bool ClipKey(')
|
|
connection = 'class Connection { bool m_Attempted = false; public: ' + '\n'.join(body(path, signature) for signature in ['void MainMenu()', 'bool Claim()', 'bool Attempted()']) + '};'
|
|
cases = []
|
|
for size in [44, 45, 46, 47, 255, 256, 257, 8192]:
|
|
data = random.Random(size).randbytes(size)
|
|
text = base64.b64encode(data).decode()
|
|
cases.append('assert(Decode("' + text + '",v)); assert(v==Array({' + ','.join(map(str, data)) + '}));')
|
|
valid = base64.b64encode(bytes(44)).decode()
|
|
for invalid in ['', 'RIFF', valid[:-1], valid[:5] + '?' + valid[6:],
|
|
valid[:8] + '=' + valid[9:], valid[:-2] + 'B=', valid[:-2] + '=A']:
|
|
cases.append('assert(!Decode(' + json.dumps(invalid) + ',v));')
|
|
run(r'''
|
|
#include <cassert>
|
|
#include <string>
|
|
#include <vector>
|
|
struct string:std::string {
|
|
using std::string::string;string(std::string s):std::string(s){}
|
|
bool IsEmpty(){return empty();} void ToLower(){for(auto &c:*this)c=std::tolower(c);}
|
|
int Length(){return size();} string Substring(int i,int n){return substr(i,n);}
|
|
int IndexOf(string s){auto p=find(s);return p==npos?-1:int(p);}
|
|
};
|
|
struct Array:std::vector<int>{using std::vector<int>::vector;void Clear(){clear();}void Insert(int v){push_back(v);}int Count(){return size();}};
|
|
''' + decode + '\n' + ready + '\n' + weather + '\n' + briefing + '\n' + key + '\n' + connection + '\nint main(){Array v;\n' + '\n'.join(cases) + r'''
|
|
Connection c; assert(!c.Attempted()); assert(c.Claim());
|
|
for(int i=0;i<100;i++) assert(!c.Claim()); // repeated ticks/respawns share this object
|
|
assert(c.Attempted()); c.MainMenu(); assert(c.Claim()); // actual new connection
|
|
assert(Briefing("Nogova","Clear","Rainy").find("This is Nogova.")!=string::npos);
|
|
assert(Briefing("Chernarus","Clear","Rainy").find("Expected next: rain.")!=string::npos);
|
|
assert(Briefing("Everon","Clear","Clear").find("Expected next:")==string::npos);
|
|
assert(Briefing("","alien","unknown").find("Current weather:")==string::npos);
|
|
assert(Briefing("","alien","unknown").find("Expected next:")==string::npos);
|
|
assert(ClipKey(string(64,'a'))); assert(!ClipKey(string(64,'z'))); assert(!ClipKey("../welcome"));
|
|
assert(Ready(false,true,true,false));
|
|
assert(!Ready(true,true,true,false));
|
|
assert(!Ready(false,false,true,false));
|
|
assert(!Ready(false,true,false,false));
|
|
assert(!Ready(false,true,true,true));
|
|
}''')
|
|
|
|
spec = importlib.util.spec_from_file_location('welcome_voice', ROOT/'tools/welcome_voice.py')
|
|
tool = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(tool)
|
|
|
|
|
|
def clip():
|
|
output = io.BytesIO()
|
|
with wave.open(output, 'wb') as wav:
|
|
wav.setnchannels(1)
|
|
wav.setsampwidth(2)
|
|
wav.setframerate(24000)
|
|
wav.writeframes(bytes(4800))
|
|
return output.getvalue()
|
|
|
|
|
|
class Response:
|
|
status_code = 200
|
|
def __enter__(self): return self
|
|
def __exit__(self, *_): pass
|
|
def iter_content(self, _):
|
|
yield json.dumps({'audioContent': base64.b64encode(clip()).decode()}).encode()
|
|
|
|
|
|
class Session:
|
|
def __init__(self): self.calls = []
|
|
def post(self, endpoint, **kwargs):
|
|
self.calls.append((endpoint, kwargs))
|
|
return Response()
|
|
|
|
|
|
class WelcomeTests(unittest.TestCase):
|
|
def test_bounded_http_request_framing(self):
|
|
read = tool.read_request_body
|
|
self.assertEqual(read({'Content-Length': '2'}, io.BytesIO(b'{}')), b'{}')
|
|
chunked = {'Transfer-Encoding': 'chunked'}
|
|
self.assertEqual(read(chunked, io.BytesIO(b'1\r\n{\r\n1;ext=yes\r\n}\r\n0\r\nX-Test: ok\r\n\r\n')), b'{}')
|
|
for headers, data in [
|
|
({}, b'{}'), ({'Content-Length': '0'}, b''),
|
|
({'Content-Length': '8193'}, b'x'),
|
|
({'Content-Length': '4'}, b'{}'),
|
|
({'Transfer-Encoding': 'gzip'}, b'{}'),
|
|
({'Transfer-Encoding': 'chunked', 'Content-Length': '2'}, b'{}'),
|
|
(chunked, b'2001\r\n'), (chunked, b'3\r\n{}'),
|
|
(chunked, b'2\r\n{}XX0\r\n\r\n'),
|
|
(chunked, b'Q\r\n'), (chunked, b'0\r\n\r\n'),
|
|
(chunked, b'2\r\n{}\r\n0\r\n'),
|
|
(chunked, b'2\r\n{}\r\n0\r\n' + b'a'*1025 + b'\r\n'),
|
|
(chunked, b'1\r\na\r\n'*2049 + b'0\r\n\r\n'),
|
|
]:
|
|
with self.subTest(headers=headers, prefix=data[:12]), self.assertRaises(ValueError):
|
|
read(headers, io.BytesIO(data))
|
|
with self.assertRaises(ValueError):
|
|
read(chunked, io.BytesIO(b'2\r\nab\r\n2\r\ncd\r\n0\r\n\r\n'), limit=3)
|
|
|
|
def test_real_chunked_http_generation_and_playback(self):
|
|
# An iterable body makes http.client send Transfer-Encoding: chunked
|
|
# without Content-Length, as allowed by HTTP/1.1. No Google call.
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
calls = []
|
|
def generate(text):
|
|
calls.append(text)
|
|
return clip()
|
|
token = 'test-only-secret-' + 'x'*32
|
|
cache = tool.BriefingCache(directory, token, generate, 'chunked-test')
|
|
server = tool.ThreadingHTTPServer(('127.0.0.1', 0), tool.handler_for(clip(), cache))
|
|
worker = threading.Thread(target=server.serve_forever, daemon=True)
|
|
worker.start()
|
|
connection = http.client.HTTPConnection(*server.server_address, timeout=3)
|
|
try:
|
|
payload = json.dumps({'token': token, 'text': 'Welcome to Chernarus.'}).encode()
|
|
chunks = [payload[:15], payload[15:50], payload[50:]]
|
|
connection.request('POST', '/', iter(chunks), {'Content-Type': 'application/json'}, encode_chunked=True)
|
|
response = connection.getresponse()
|
|
self.assertEqual(response.status, 200)
|
|
key = json.loads(response.read())['clip']
|
|
connection.request('GET', '/clips/' + key + '.b64')
|
|
response = connection.getresponse()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(base64.b64decode(response.read()), clip())
|
|
connection.request('POST', '/', payload, {'Content-Type': 'application/json'})
|
|
response = connection.getresponse()
|
|
self.assertEqual(response.status, 200)
|
|
self.assertEqual(json.loads(response.read())['clip'], key)
|
|
self.assertEqual(calls, ['Welcome to Chernarus.'])
|
|
finally:
|
|
connection.close()
|
|
server.shutdown()
|
|
server.server_close()
|
|
worker.join(timeout=3)
|
|
|
|
def test_http_generation_requires_token_and_get_only_reads_cache(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
calls = []
|
|
def generate(text):
|
|
calls.append(text)
|
|
return clip()
|
|
cache = tool.BriefingCache(directory, 'x'*40, generate, 'veteran')
|
|
handler = tool.handler_for(clip(), cache)
|
|
def request(method, path, payload=None):
|
|
instance = handler.__new__(handler)
|
|
data = json.dumps(payload).encode() if payload is not None else b''
|
|
instance.path, instance.headers = path, {'Content-Length': str(len(data))}
|
|
instance.rfile, instance.wfile = io.BytesIO(data), io.BytesIO()
|
|
statuses = []
|
|
instance.send_response = statuses.append
|
|
instance.send_error = statuses.append
|
|
instance.send_header = lambda *args: None
|
|
instance.end_headers = lambda: None
|
|
getattr(instance, 'do_' + method)()
|
|
return statuses[-1], instance.wfile.getvalue()
|
|
self.assertEqual(request('POST', '/', {'text': 'Nogova'})[0], 403)
|
|
self.assertEqual(request('GET', '/?text=Nogova')[0], 404)
|
|
self.assertEqual(calls, [])
|
|
status, payload = request('POST', '/', {'token': 'x'*40, 'text': 'Nogova'})
|
|
self.assertEqual(status, 200)
|
|
key = json.loads(payload)['clip']
|
|
status, audio = request('GET', '/clips/' + key + '.b64')
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(base64.b64decode(audio), clip())
|
|
self.assertNotIn(b'x'*40, payload + audio)
|
|
self.assertEqual(calls, ['Nogova'])
|
|
|
|
def test_veteran_style_and_dynamic_text(self):
|
|
session = Session()
|
|
tool.synthesize(session, tool.ENDPOINT, 'Algenib', 'en-US',
|
|
'This is Nogova. Current weather: rain.', 'gemini-2.5-flash-tts', tool.VETERAN_PROMPT)
|
|
request = session.calls[0][1]['json']
|
|
self.assertIn('older war veteran', request['input']['prompt'])
|
|
self.assertIn('Nogova', request['input']['text'])
|
|
self.assertEqual(request['voice']['modelName'], 'gemini-2.5-flash-tts')
|
|
|
|
def test_authenticated_cache_and_billing_bounds(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
calls, now = [], [100.0]
|
|
def generate(text):
|
|
calls.append(text)
|
|
return clip()
|
|
token = 'x' * 40
|
|
cache = tool.BriefingCache(directory, token, generate, 'veteran', lambda: now[0])
|
|
for request in [{}, {'token': 'wrong', 'text': 'hello'}]:
|
|
with self.assertRaises(PermissionError): cache.prepare(request)
|
|
self.assertEqual(calls, [])
|
|
request = {'token': token, 'text': 'Nogova, clear skies.'}
|
|
key = cache.prepare(request)
|
|
self.assertEqual(cache.prepare(request), key)
|
|
self.assertEqual(cache.read(key), clip())
|
|
self.assertEqual(len(calls), 1)
|
|
with self.assertRaises(BlockingIOError):
|
|
cache.prepare({'token': token, 'text': 'Chernarus, rain.'})
|
|
now[0] += 61
|
|
changed = cache.prepare({'token': token, 'text': 'Chernarus, rain.'})
|
|
self.assertNotEqual(changed, key)
|
|
self.assertEqual(len(calls), 2)
|
|
for invalid in ['../welcome', 'z' * 64, key + '?token=secret']:
|
|
with self.assertRaises(FileNotFoundError): cache.read(invalid)
|
|
# Service restart still reuses completed clips, with no Google call.
|
|
restarted = tool.BriefingCache(directory, token, generate, 'veteran')
|
|
self.assertEqual(restarted.prepare(request), key)
|
|
self.assertEqual(len(calls), 2)
|
|
for text in ['', 'a' * 801, 'line\nline', 42]:
|
|
with self.assertRaises(ValueError): cache.prepare({'token': token, 'text': text})
|
|
|
|
def test_cache_failure_is_not_retried_immediately(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
calls = []
|
|
def failed(text):
|
|
calls.append(text)
|
|
raise RuntimeError('provider failed')
|
|
cache = tool.BriefingCache(directory, 'x'*40, failed, 'veteran')
|
|
request = {'token': 'x'*40, 'text': 'hello'}
|
|
with self.assertRaises(RuntimeError): cache.prepare(request)
|
|
with self.assertRaises(BlockingIOError): cache.prepare(request)
|
|
self.assertEqual(calls, ['hello'])
|
|
self.assertEqual(list(Path(directory).iterdir()), [])
|
|
|
|
def test_provider_failure_is_not_retried(self):
|
|
class FailedSession(Session):
|
|
def post(self, endpoint, **kwargs):
|
|
response = super().post(endpoint, **kwargs)
|
|
response.status_code = 503
|
|
return response
|
|
session = FailedSession()
|
|
with self.assertRaises(RuntimeError):
|
|
tool.synthesize(session, tool.ENDPOINT, 'en-GB-Neural2-B', 'en-GB')
|
|
self.assertEqual(len(session.calls), 1)
|
|
|
|
def test_google_request_and_response(self):
|
|
session = Session()
|
|
self.assertEqual(tool.synthesize(session, tool.ENDPOINT, 'en-GB-Neural2-B', 'en-GB'), clip())
|
|
endpoint, args = session.calls[0]
|
|
self.assertEqual(endpoint, tool.ENDPOINT)
|
|
self.assertEqual(args['json']['input']['text'], tool.WELCOME)
|
|
self.assertEqual(args['json']['audioConfig']['audioEncoding'], 'LINEAR16')
|
|
self.assertFalse(args['allow_redirects'])
|
|
self.assertEqual(len(session.calls), 1)
|
|
|
|
def test_no_insecure_or_secret_url(self):
|
|
for endpoint in ['http://example.com', 'https://user:secret@example.com/', 'https://example.com/?key=secret']:
|
|
session = Session()
|
|
with self.assertRaises(ValueError):
|
|
tool.synthesize(session, endpoint, '', '')
|
|
self.assertEqual(session.calls, [])
|
|
|
|
def test_bad_clip_does_not_replace_cached_clip(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory)/'welcome.wav'
|
|
tool.save_clip(path, clip())
|
|
for bad in [b'<html>error</html>', clip()[:-10], bytes(tool.MAX_AUDIO+1)]:
|
|
with self.assertRaises(ValueError): tool.save_clip(path, bad)
|
|
self.assertEqual(path.read_bytes(), clip())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|