72 lines
4.0 KiB
Python
72 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
import importlib.util
|
|
import json
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
spec = importlib.util.spec_from_file_location('spectator', Path(__file__).with_name('spectator-server.py'))
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
class SpectatorTests(unittest.TestCase):
|
|
def test_public_projection_excludes_secrets_and_whispers(self):
|
|
bots = [{'name': n, 'status': 'online', 'race': r, 'role': 'Death Knight', 'level': 55,
|
|
'autonomy': {'managed': True, 'accountId': 123}, 'memories': ['PRIVATE_MEMORY'],
|
|
'world': {'team': team, 'groupId': 'internal-guid', 'groupNames': ['A', 'B', 'HumanPlayer']}}
|
|
for n, r, team in [('A', 'Human', 0), ('B', 'Orc', 1)]]
|
|
feed = [{'id': i, 'kind': 'conversation', 'status': 'confirmed', 'names': names, 'text': t}
|
|
for i, names, t in [('whisper-1', ['HumanPlayer', 'A'], 'PRIVATE_WHISPER'),
|
|
('social-1', ['A', 'B'], 'SYNTHETIC_PROMPT'),
|
|
('social-1:reply', ['A', 'B'], 'Hello friend'),
|
|
('social-2:reply', ['HumanPlayer', 'A'], 'NOT_BOT_CHAT')]]
|
|
source={'bots': bots, 'settings': {'token': 'SECRET'}, 'community': {'feed': feed}}
|
|
projected = m.project(source)
|
|
public = m.public_state(source)
|
|
for forbidden in ['PRIVATE_MEMORY','PRIVATE_WHISPER','SYNTHETIC_PROMPT','NOT_BOT_CHAT','SECRET','accountId']:
|
|
self.assertNotIn(forbidden,json.dumps(public))
|
|
serialized = json.dumps(projected)
|
|
for private in ['PRIVATE_MEMORY', 'PRIVATE_WHISPER', 'SYNTHETIC_PROMPT', 'NOT_BOT_CHAT', 'HumanPlayer', 'SECRET', 'accountId', 'internal-guid']:
|
|
self.assertNotIn(private, serialized)
|
|
self.assertEqual([b['faction'] for b in projected['bots']], ['Alliance', 'Horde'])
|
|
self.assertEqual([e['text'] for e in projected['feed']], ['Hello friend'])
|
|
|
|
def test_http_allowlist_and_mutations(self):
|
|
server = m.Server(('127.0.0.1', 0), m.Handler)
|
|
worker = threading.Thread(target=server.serve_forever, daemon=True)
|
|
worker.start()
|
|
base = 'http://127.0.0.1:' + str(server.server_port)
|
|
try:
|
|
for route in ['/app.js', '/style.css', '/health']:
|
|
with urllib.request.urlopen(base + route) as response:
|
|
self.assertEqual(response.status, 200)
|
|
self.assertIn("default-src 'none'", response.headers['Content-Security-Policy'])
|
|
for path in ['/api/actions', '/api/ale/operator', '/.dev.vars', '/../.dev.vars', '/%2e%2e/.dev.vars', '/api/bridge/ale/poll', '/assets/../../.dev.vars', '/assets/private.map']:
|
|
with self.assertRaises(urllib.error.HTTPError) as error:
|
|
urllib.request.urlopen(base + path)
|
|
self.assertEqual(error.exception.code, 404)
|
|
error.exception.close()
|
|
with patch.object(m,'local_get',return_value=b'<html>Visitor</html>') as upstream:
|
|
with urllib.request.urlopen(base+'/') as response:
|
|
self.assertEqual(response.read(),b'<html>Visitor</html>')
|
|
upstream.assert_called_once_with('/visitor')
|
|
with patch.object(m.SNAPSHOT,'get',return_value={'state':{'bots':[]}}):
|
|
with urllib.request.urlopen(base+'/api/state') as response:
|
|
self.assertEqual(json.load(response),{'bots':[]})
|
|
for method in ['POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS', 'TRACE']:
|
|
with self.assertRaises(urllib.error.HTTPError) as error:
|
|
urllib.request.urlopen(urllib.request.Request(base + '/api/actions', method=method, data=b'{}'))
|
|
self.assertEqual(error.exception.code, 405)
|
|
error.exception.close()
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
worker.join()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|