78 lines
5.0 KiB
Python
78 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Frozen local Workbench telemetry fixture; no live server or installed asset writes."""
|
|
from pathlib import Path
|
|
import hashlib,json,shutil,subprocess,sys,threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
ROOT=Path(__file__).resolve().parents[3]
|
|
sys.path.insert(0,str(ROOT/'tools'))
|
|
from source_manifest import addon_manifest
|
|
from headless_game import assert_prefix_idle
|
|
name=sys.argv[1]
|
|
if not name.isalnum(): raise ValueError('Use an alphanumeric run suffix')
|
|
source=Path('/tmp/erri-telemetry-source-'+name)
|
|
profile=Path('/tmp/erri-telemetry-run-'+name)
|
|
compat=ROOT/'.local/playtest-compat-20260920'
|
|
assert_prefix_idle(compat)
|
|
source.mkdir();profile.mkdir()
|
|
for rel in addon_manifest(ROOT):
|
|
target=source/rel;target.parent.mkdir(parents=True,exist_ok=True);shutil.copy2(ROOT/rel,target)
|
|
shutil.copytree(ROOT/'tools',source/'tools',ignore=shutil.ignore_patterns('__pycache__'))
|
|
(source/'docs').mkdir();shutil.copy2(ROOT/'docs/external-addons.json',source/'docs/external-addons.json')
|
|
fixture=Path(__file__).with_name('native-fixture.c')
|
|
shutil.copy2(fixture,source/'addons/core/Scripts/Game/ERR/00_Core/ERR_ZTelemetryFixture.c')
|
|
# Opt-in fixture is independent of the historical generic Core smoke.
|
|
p=source/'addons/core/Scripts/Game/ERR/00_Core/ERR_NativeBaseSmoke.c'
|
|
p.write_text(p.read_text().replace('array<string> independentFixtures = {','array<string> independentFixtures = {"telemetry", '))
|
|
p=source/'tools/headless_workbench.py'
|
|
p.write_text(p.read_text().replace("choices=['coordinated-ai',", "choices=['telemetry', 'coordinated-ai',").replace("a.smoke not in {'coordinated-ai',", "a.smoke not in {'telemetry', 'coordinated-ai',"))
|
|
# This fixture deliberately fails preparation before exercising its exit UI.
|
|
p.write_text(p.read_text().replace("a.smoke not in {'startup-failure',", "a.smoke not in {'telemetry', 'startup-failure',"))
|
|
p.write_text(p.read_text().replace("'-noThrow',", "'-errTelemetryTest', '1', '-noThrow',"))
|
|
# Loopback-only receiver: reject the first delivery, then accept retry and result.
|
|
received=[]
|
|
class Receiver(BaseHTTPRequestHandler):
|
|
def log_message(self,*args): pass
|
|
def do_POST(self):
|
|
(profile/'http-headers.json').write_text(json.dumps(dict(self.headers),indent=2)+'\n')
|
|
if self.headers.get('Transfer-Encoding','').lower() == 'chunked':
|
|
chunks=[]
|
|
while True:
|
|
size=int(self.rfile.readline().split(b';')[0],16)
|
|
if not size:
|
|
self.rfile.readline();break
|
|
chunks.append(self.rfile.read(size));self.rfile.read(2)
|
|
raw=b''.join(chunks)
|
|
else:
|
|
count=int(self.headers.get('Content-Length','0'));raw=self.rfile.read(count)
|
|
(profile/'http-body.txt').write_bytes(raw)
|
|
payload=json.loads(raw)
|
|
received.append({'path':self.path,'content_type':self.headers.get('Content-Type'),'payload':payload})
|
|
(profile/'http-events.json').write_text(json.dumps(received,indent=2)+'\n')
|
|
self.send_response(503 if len(received)==1 else 204); self.end_headers()
|
|
receiver=ThreadingHTTPServer(('127.0.0.1',0),Receiver)
|
|
threading.Thread(target=receiver.serve_forever,daemon=True).start()
|
|
endpoint='http://127.0.0.1:'+str(receiver.server_port)+'/ingest'
|
|
config=source/'addons/core/Configs/ERR/Telemetry.conf'
|
|
config.write_text(config.read_text().replace('Endpoint ""', 'Endpoint "'+endpoint+'"'))
|
|
# Only the frozen test clone permits this exact loopback URL; shipping policy is HTTPS.
|
|
telemetry=source/'addons/core/Scripts/Game/ERR/00_Core/ERR_Telemetry.c'
|
|
telemetry.write_text(telemetry.read_text().replace('!config.Endpoint.StartsWith("https://")', '!(config.Endpoint.StartsWith("https://") || config.Endpoint == "'+endpoint+'")'))
|
|
installed=Path.home()/'.local/share/Steam/steamapps/compatdata/1874880/pfx/drive_c/users/steamuser/Documents/My Games/ArmaReforger/addons'
|
|
(profile/'profile').mkdir()
|
|
(profile/'profile/EscapistsFactions.json').write_text(json.dumps({'escapists':'ussr','enemy':'us','independent':'fia'})+'\n')
|
|
(profiles:=profile/'fixture-inputs.json').write_text(json.dumps({'fixture_sha256':hashlib.sha256(fixture.read_bytes()).hexdigest(),'production_sources':addon_manifest(ROOT),'source':str(source),'profile':str(profile)},indent=2)+'\n')
|
|
command=[sys.executable,str(source/'tools/headless_workbench.py'),'--compat',str(compat),'--profile',str(profile),'--external-addons',str(installed),'--project','core','--smoke','telemetry','--rounds','1','--timeout','180']
|
|
with (profile/'runner-output.log').open('w') as output:
|
|
result=subprocess.run(command,stdout=output,stderr=subprocess.STDOUT)
|
|
receiver.shutdown()
|
|
if received:
|
|
assert len(received) >= 3, received
|
|
assert received[0]['payload'] == received[1]['payload']
|
|
assert received[0]['payload']['event']['type'] == 'mission_started'
|
|
assert received[2]['payload']['event']['outcome'] == 'preparation_failed'
|
|
assert all(r['content_type']=='application/json' and r['path']=='/ingest' for r in received)
|
|
print('HTTP proof: failed request retried with same event ID; result accepted.')
|
|
print(json.dumps({'exit':result.returncode,'source':str(source),'profile':str(profile)}))
|
|
print('\n'.join((profile/'runner-output.log').read_text().splitlines()[-45:]))
|
|
raise SystemExit(result.returncode)
|