mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-18 17:01:20 +02:00
saphidandClaude Opus 5.5
443813a088
Installs from the Android store, Import from my Quest and My Meta library now go into one server-side queue and run one after another, instead of being refused while another runs. The page shows the queue with Remove for installs that haven't started; each queued install runs only on the headset it was queued for. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1267 lines
68 KiB
Python
1267 lines
68 KiB
Python
"""My Meta library, Experimental (ui/frame_meta.py), against a fake Meta (GraphQL answers and
|
|
download server built from the shapes the public tools document) and a fake Frame. Offline:
|
|
nothing here reaches Meta, a headset or the compatibility database."""
|
|
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import urllib.parse
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'ui'))
|
|
import frame_android as android # noqa: E402
|
|
import frame_meta as meta # noqa: E402
|
|
import frame_quest as quest # noqa: E402
|
|
import frame_telemetry # noqa: E402
|
|
import test_quest # noqa: E402 (its APK builder; importing a module adds no tests here)
|
|
|
|
PKG = test_quest.PKG
|
|
TOKEN = 'OCAtestToken1234567890abcdefGHIJ-_xyz'
|
|
OTHER_TOKEN = 'FRLAnotherToken0987654321zyxwvuts'
|
|
APK = test_quest.apk_bytes(files=test_quest.OPENXR, version_code=100, version_name='1.0')
|
|
OBB = b'obb data ' * 100
|
|
PUBLIC = {'157.240.22.35'}
|
|
REAL_SEND = meta._send # before any test replaces it with a fake Meta
|
|
REAL_SSH = android.ssh
|
|
PRINTABLE = ''.join(chr(c) for c in range(32, 127))
|
|
|
|
|
|
def entitlements(apk=APK, obb=OBB, platform_sdk=False):
|
|
binary = {'id': '9001', 'package_name': PKG, 'version': '1.0', 'version_code': 100, 'size': str(len(apk)),
|
|
'obb_binary': {'id': '9002', 'file_name': 'game-data.obb', 'size': len(obb)},
|
|
'asset_files': {'nodes': [{'id': '9003', 'file_name': 'level1.pak', 'size': 3, 'is_required': True},
|
|
{'id': '9004', 'file_name': 'dlc.pak', 'size': 3, 'is_required': False}]}}
|
|
return {'data': {'viewer': {'user': {'active_entitlements': {'nodes': [
|
|
{'active_state': 'ACTIVE', 'grant_reason': 'PURCHASE',
|
|
'item': {'id': '111', 'display_name': 'Quest Game', 'platform': 'ANDROID_6DOF', 'latest_supported_binary': binary}},
|
|
# Details come from a second query (only the binary id in the list).
|
|
# A real account's purchases come back PERMANENT.
|
|
{'active_state': 'PERMANENT', 'item': {'id': '222', 'display_name': 'Another Game', 'platform': 'ANDROID_6DOF',
|
|
'latest_supported_binary': {'id': '9100'}}},
|
|
{'active_state': 'ACTIVE', 'item': {'id': '333', 'display_name': 'A Rift Game', 'platform': 'PC'}},
|
|
{'active_state': 'ACTIVE', 'expiration_time': 1000, 'item': {'id': '444', 'display_name': 'Expired',
|
|
'platform': 'ANDROID_6DOF'}},
|
|
{'active_state': 'INACTIVE', 'item': {'id': '555', 'display_name': 'Refunded', 'platform': 'ANDROID_6DOF'}},
|
|
]}}}}}
|
|
|
|
|
|
class Body(io.BytesIO):
|
|
pass
|
|
|
|
|
|
class FakeMeta:
|
|
"""graph.oculus.com and securecdn.oculus.com (which redirects to a CDN host), in memory."""
|
|
|
|
def __init__(self, test, token=TOKEN, apk=APK, obb=OBB, redirect=True):
|
|
self.token, self.redirect, self.requests = token, redirect, []
|
|
self.files = {'9001': apk, '9002': obb, '9003': b'pak', '9004': b'dlc'}
|
|
self.entitlements = entitlements(apk, obb)
|
|
self.details = {'9100': {'data': {'node': {'id': '9100', 'package_name': 'org.example.another', 'version': '2',
|
|
'version_code': '7', 'size': 10, 'obb_binary': None,
|
|
'asset_files': {'edges': []}}}}}
|
|
self.addresses = {} # host -> addresses its (fake) DNS gives
|
|
self.hold = None # an Event the "who am I" answer waits for
|
|
self.cdn_headers = {}
|
|
self.location = 'https://scontent.oculuscdn.com/v/{id}?sig=abc'
|
|
self.lookups = []
|
|
for p in (patch.object(meta, '_send', side_effect=self.send),
|
|
patch.object(meta.socket, 'getaddrinfo', side_effect=self.getaddrinfo)):
|
|
p.start()
|
|
test.addCleanup(p.stop)
|
|
|
|
def getaddrinfo(self, host, port, *args, **kwargs):
|
|
self.lookups.append(host)
|
|
return [(10 if ':' in a else 2, 1, 6, '', (a, port)) for a in sorted(self.addresses.get(host, PUBLIC))]
|
|
|
|
def answer(self, status, body, headers=None):
|
|
raw = body if isinstance(body, bytes) else json.dumps(body).encode()
|
|
return status, dict(headers or {}), Body(raw)
|
|
|
|
def send(self, url, data=None, headers=None, timeout=30):
|
|
self.requests.append({'url': url, 'data': data, 'headers': dict(headers or {})})
|
|
p = urllib.parse.urlsplit(url)
|
|
if p.hostname == 'graph.oculus.com':
|
|
form = dict(urllib.parse.parse_qsl(data.decode()))
|
|
if form.get('access_token') != self.token:
|
|
return self.answer(400, {'error': {'type': 'OAuthException', 'code': 190, 'message':
|
|
f"Invalid OAuth access token {form.get('access_token')}"}})
|
|
doc = form['doc_id']
|
|
if doc == meta.DOC_VIEWER:
|
|
if self.hold:
|
|
self.hold.wait(5)
|
|
return self.answer(200, {'data': {'viewer': {'user': {'id': '1234', 'alias': 'player_one'}}}})
|
|
if doc == meta.DOC_ENTITLEMENTS:
|
|
return self.answer(200, self.entitlements)
|
|
if doc == meta.DOC_BINARY:
|
|
return self.answer(200, self.details[json.loads(form['variables'])['binaryID']])
|
|
return self.answer(200, {'errors': [{'message': 'unknown doc'}]})
|
|
q = dict(urllib.parse.parse_qsl(p.query))
|
|
if p.hostname == 'securecdn.oculus.com':
|
|
if q.get('access_token') != self.token:
|
|
return self.answer(401, b'')
|
|
if q['id'] not in self.files:
|
|
return self.answer(404, b'')
|
|
if self.redirect:
|
|
return self.answer(302, b'', {'Location': self.location.format(id=q['id'], token=self.token)})
|
|
return self.answer(200, self.files[q['id']], {'Content-Length': str(len(self.files[q['id']]))})
|
|
if p.hostname == 'scontent.oculuscdn.com':
|
|
body = self.files[p.path.rsplit('/', 1)[-1]]
|
|
headers = dict({'Content-Length': str(len(body))}, **self.cdn_headers)
|
|
return self.answer(200, body, {k: v for k, v in headers.items() if v is not None})
|
|
return self.answer(404, b'')
|
|
|
|
def downloads(self):
|
|
return [r for r in self.requests if 'oculus.com/binaries' in r['url'] or 'oculuscdn' in r['url']]
|
|
|
|
|
|
def connected(test, **kw):
|
|
fake = FakeMeta(test, **kw)
|
|
meta.connect(TOKEN)
|
|
test.addCleanup(meta.disconnect)
|
|
return fake
|
|
|
|
|
|
class Tokens(unittest.TestCase):
|
|
def tearDown(self):
|
|
meta.disconnect()
|
|
|
|
def test_token_format(self):
|
|
self.assertEqual(meta.check_token(f' {TOKEN}\n'), TOKEN)
|
|
self.assertEqual(meta.check_token('Bearer ' + OTHER_TOKEN), OTHER_TOKEN)
|
|
for bad, why in (('', 'Paste'), ('hello world', 'unusual'), ('ABCDEFGHIJKLMNOPQRSTUVWX', 'OC or FRL'),
|
|
('OC' + '%7C' * 10, 'URL-encoded'), ('OC|752908224809889|', 'app token'),
|
|
('OC123456789012345', 'incomplete')):
|
|
with self.assertRaisesRegex(meta.MetaError, why):
|
|
meta.check_token(bad)
|
|
|
|
def test_connect_checks_then_stores_owner_only(self):
|
|
fake = FakeMeta(self)
|
|
result = meta.connect(TOKEN)
|
|
self.assertTrue(result['connected'])
|
|
self.assertEqual(result['user'], {'name': 'player_one'})
|
|
self.assertNotIn(TOKEN, json.dumps(result))
|
|
self.assertNotIn(TOKEN, json.dumps(meta.status()))
|
|
path = meta.account_path()
|
|
self.assertEqual(json.loads(path.read_text())['token'], TOKEN)
|
|
if os.name != 'nt':
|
|
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
|
self.assertFalse([p for p in path.parent.iterdir() if p.name.startswith('.meta-account-')]) # no temp left
|
|
# The token went to Meta's API host only, over HTTPS, in the POST body (not the address).
|
|
request = fake.requests[0]
|
|
self.assertTrue(request['url'].startswith('https://graph.oculus.com/'))
|
|
self.assertNotIn(TOKEN, request['url'])
|
|
self.assertIn(urllib.parse.quote(TOKEN, safe=''), request['data'].decode())
|
|
self.assertEqual(dict(urllib.parse.parse_qsl(request['data'].decode()))['doc_id'], meta.DOC_VIEWER)
|
|
|
|
def test_rejected_token_is_not_stored_or_echoed(self):
|
|
FakeMeta(self, token=OTHER_TOKEN)
|
|
with self.assertRaises(meta.TokenRejected) as e:
|
|
meta.connect(TOKEN)
|
|
self.assertNotIn(TOKEN, str(e.exception))
|
|
self.assertFalse(meta.account_path().exists())
|
|
self.assertFalse(meta.status()['connected'])
|
|
|
|
def test_disconnect_deletes_the_file(self):
|
|
FakeMeta(self)
|
|
meta.connect(TOKEN)
|
|
meta.disconnect()
|
|
self.assertFalse(meta.account_path().exists())
|
|
with self.assertRaisesRegex(meta.MetaError, 'Connect your Meta account first'):
|
|
meta.library()
|
|
meta.disconnect() # twice is fine
|
|
|
|
def test_disconnect_wins_over_a_connect_in_flight(self):
|
|
fake = FakeMeta(self)
|
|
fake.hold = threading.Event()
|
|
errors = []
|
|
|
|
def connect():
|
|
try:
|
|
meta.connect(TOKEN)
|
|
except meta.MetaError as e:
|
|
errors.append(e)
|
|
thread = threading.Thread(target=connect)
|
|
thread.start()
|
|
for _ in range(500):
|
|
if fake.requests:
|
|
break
|
|
threading.Event().wait(0.01)
|
|
meta.disconnect() # while Meta is still answering "who am I"
|
|
fake.hold.set()
|
|
thread.join(10)
|
|
self.assertEqual(len(errors), 1)
|
|
self.assertIn('disconnected', str(errors[0]))
|
|
self.assertFalse(meta.account_path().exists())
|
|
self.assertFalse(meta.status()['connected'])
|
|
|
|
|
|
class SignInAttempts(unittest.TestCase):
|
|
"""Sign in with Meta (app/meta-signin.js): a token from an attempt the app gave up on is
|
|
never kept, whether the cancel comes while it is checked or just after it was saved."""
|
|
|
|
def tearDown(self):
|
|
meta.disconnect()
|
|
|
|
def held_connect(self, fake, attempt):
|
|
"""Start connect(TOKEN, attempt) and wait until Meta is answering "who am I"."""
|
|
fake.hold = threading.Event()
|
|
errors = []
|
|
|
|
def connect():
|
|
try:
|
|
meta.connect(TOKEN, attempt=attempt)
|
|
except meta.MetaError as e:
|
|
errors.append(e)
|
|
thread = threading.Thread(target=connect)
|
|
thread.start()
|
|
for _ in range(500):
|
|
if fake.requests:
|
|
break
|
|
threading.Event().wait(0.01)
|
|
return thread, errors
|
|
|
|
def test_cancelling_while_meta_answers_wins(self):
|
|
fake = FakeMeta(self)
|
|
thread, errors = self.held_connect(fake, 'attempt-during-0003')
|
|
self.assertIn('attempt-during-0003', meta._attempts)
|
|
meta.cancel_signin('attempt-during-0003') # the window closed while Meta was answering
|
|
fake.hold.set()
|
|
thread.join(10)
|
|
self.assertEqual(len(errors), 1)
|
|
self.assertIsInstance(errors[0], meta.SignInCancelled)
|
|
self.assertFalse(meta.account_path().exists())
|
|
self.assertFalse(meta.status()['connected'])
|
|
self.assertEqual(meta._attempts, {})
|
|
|
|
def test_other_cancels_cant_push_out_one_in_flight(self):
|
|
fake = FakeMeta(self)
|
|
thread, errors = self.held_connect(fake, 'attempt-held-0004')
|
|
meta.cancel_signin('attempt-held-0004')
|
|
for i in range(300):
|
|
meta.cancel_signin(f'attempt-noise-{i:04d}')
|
|
fake.hold.set()
|
|
thread.join(10)
|
|
self.assertIsInstance(errors[0], meta.SignInCancelled)
|
|
self.assertFalse(meta.account_path().exists())
|
|
self.assertEqual(meta._attempts, {}) # unknown ids left nothing behind
|
|
|
|
def test_a_cancel_just_after_the_save_removes_that_token(self):
|
|
FakeMeta(self)
|
|
self.assertTrue(meta.connect(TOKEN, attempt='attempt-saved-0005')['connected'])
|
|
self.assertFalse(meta.cancel_signin('attempt-saved-0005')['connected'])
|
|
self.assertFalse(meta.account_path().exists())
|
|
|
|
def test_a_cancel_between_the_save_and_the_cleanup_removes_that_token(self):
|
|
# The connect has saved and let go of the lock, but hasn't yet dropped its attempt
|
|
# from the in-flight list when the cancel arrives.
|
|
FakeMeta(self)
|
|
real_status = meta.status
|
|
seen = {}
|
|
|
|
def cancel_in_the_gap():
|
|
# status() runs after the save; put the attempt back as still listed, as it is
|
|
# in the moment before connect's cleanup, and cancel there.
|
|
if 'done' not in seen and meta._saved_by[0] == 'attempt-gap-0011':
|
|
seen['done'] = True
|
|
with meta._lock:
|
|
meta._attempts['attempt-gap-0011'] = {'cancelled': False, 'running': 1}
|
|
meta.cancel_signin('attempt-gap-0011')
|
|
seen['listed'] = 'attempt-gap-0011' in meta._attempts
|
|
with meta._lock:
|
|
del meta._attempts['attempt-gap-0011']
|
|
return real_status()
|
|
with patch.object(meta, 'status', side_effect=cancel_in_the_gap):
|
|
meta.connect(TOKEN, attempt='attempt-gap-0011')
|
|
self.assertTrue(seen.get('listed'))
|
|
self.assertFalse(meta.account_path().exists())
|
|
self.assertFalse(meta.status()['connected'])
|
|
|
|
def test_a_cancel_while_still_listed_marks_and_never_touches_another_account(self):
|
|
FakeMeta(self)
|
|
meta.connect(TOKEN) # a pasted token owns the account
|
|
with meta._lock:
|
|
meta._attempts['attempt-listed-0012'] = {'cancelled': False, 'running': 1}
|
|
try:
|
|
meta.cancel_signin('attempt-listed-0012')
|
|
self.assertTrue(meta._attempts['attempt-listed-0012']['cancelled'])
|
|
finally:
|
|
with meta._lock:
|
|
meta._attempts.pop('attempt-listed-0012', None)
|
|
self.assertTrue(meta.status()['connected'])
|
|
|
|
def test_a_cancel_never_removes_a_later_account(self):
|
|
FakeMeta(self)
|
|
meta.connect(TOKEN, attempt='attempt-first-0006')
|
|
meta.connect(TOKEN, attempt='attempt-second-0007') # a later sign-in owns the account now
|
|
meta.cancel_signin('attempt-first-0006')
|
|
self.assertTrue(meta.status()['connected'])
|
|
meta.connect(TOKEN) # a pasted token
|
|
meta.cancel_signin('attempt-second-0007')
|
|
self.assertTrue(meta.status()['connected'])
|
|
meta.disconnect()
|
|
meta.connect(TOKEN, attempt='attempt-third-0008')
|
|
meta.disconnect()
|
|
meta.connect(TOKEN)
|
|
meta.cancel_signin('attempt-third-0008') # its account was disconnected already
|
|
self.assertTrue(meta.status()['connected'])
|
|
|
|
def test_unknown_attempts_change_nothing(self):
|
|
FakeMeta(self)
|
|
meta.cancel_signin('attempt-unknown-0009')
|
|
self.assertTrue(meta.connect(TOKEN, attempt='attempt-unknown-0009')['connected'])
|
|
self.assertEqual(meta._attempts, {})
|
|
|
|
def test_a_rejected_token_leaves_no_attempt_behind(self):
|
|
FakeMeta(self, token=OTHER_TOKEN)
|
|
with self.assertRaises(meta.TokenRejected):
|
|
meta.connect(TOKEN, attempt='attempt-rejected-0010')
|
|
self.assertEqual(meta._attempts, {})
|
|
|
|
def test_attempt_ids_are_checked(self):
|
|
FakeMeta(self)
|
|
for bad in ('short', 'x' * 65, 'has space here', 'a/b/c/d/e', 123, ''):
|
|
with self.assertRaisesRegex(meta.MetaError, 'attempt'):
|
|
meta.connect(TOKEN, attempt=bad)
|
|
with self.assertRaisesRegex(meta.MetaError, 'attempt'):
|
|
meta.cancel_signin(bad)
|
|
with self.assertRaisesRegex(meta.MetaError, 'attempt'):
|
|
meta.cancel_signin(None)
|
|
self.assertFalse(meta.account_path().exists())
|
|
|
|
|
|
class Redaction(unittest.TestCase):
|
|
def setUp(self):
|
|
FakeMeta(self)
|
|
meta.connect(TOKEN)
|
|
self.addCleanup(meta.disconnect)
|
|
|
|
def test_every_form_of_the_token(self):
|
|
def encoded(text, fmt):
|
|
return ''.join(fmt % ord(c) for c in text)
|
|
forms = [TOKEN, TOKEN.upper(), TOKEN.lower(), TOKEN.swapcase(), urllib.parse.quote(TOKEN, safe=''),
|
|
urllib.parse.quote_plus(TOKEN)]
|
|
# Fully percent-encoded, of the token recased too, with either case of hex digits, and mixed.
|
|
for text in (TOKEN, TOKEN.lower(), TOKEN.upper(), TOKEN.swapcase()):
|
|
forms += [encoded(text, '%%%02X'), encoded(text, '%%%02x')]
|
|
forms.append(''.join(c if i % 2 else '%%%02x' % ord(c.lower()) for i, c in enumerate(TOKEN)))
|
|
# Percent-encoded twice (%25XX), as an error quoting an encoded URL shows it: recased too.
|
|
once = encoded(TOKEN, '%%%02X')
|
|
forms += [urllib.parse.quote(once, safe=''), urllib.parse.quote(urllib.parse.quote(TOKEN, safe=''), safe='')]
|
|
for text in (TOKEN, TOKEN.lower(), TOKEN.swapcase()):
|
|
forms += [encoded(text, '%%25%02X'), encoded(text, '%%25%02x')]
|
|
forms.append(''.join(('%%25%02x' if i % 3 == 0 else '%%%02X' if i % 3 == 1 else '%s') % (ord(c) if i % 3 < 2 else c)
|
|
for i, c in enumerate(TOKEN.swapcase())))
|
|
for form in forms:
|
|
out = meta.redact(f'GET https://example.invalid/x?id=1&tok={form} failed')
|
|
self.assertNotIn(form, out)
|
|
self.assertIn(meta.REDACTED, out)
|
|
plain = urllib.parse.unquote(urllib.parse.unquote(out)).lower()
|
|
self.assertNotIn(TOKEN[2:12].lower(), plain)
|
|
self.assertNotIn(TOKEN[-10:].lower(), plain) # the whole credential, not just its start
|
|
# Any start of it of 8+ characters (as a cut elsewhere leaves), however written.
|
|
for n in (8, 12, 20):
|
|
for prefix in (TOKEN[:n], TOKEN[:n].lower(), encoded(TOKEN[:n].lower(), '%%%02x'),
|
|
encoded(TOKEN[:n].swapcase(), '%%25%02X'), encoded(TOKEN[:n], '%%25%02x')):
|
|
self.assertNotIn(prefix, meta.redact(f'cut: {prefix}', token=TOKEN))
|
|
|
|
def test_token_shaped_values_that_are_not_stored(self):
|
|
for text in (f'access_token={OTHER_TOKEN}&x=1', f'"access_token": "{OTHER_TOKEN}"',
|
|
f'Authorization: Bearer {OTHER_TOKEN}', f'OAuth {OTHER_TOKEN}', f'error near {OTHER_TOKEN} here',
|
|
f'ACCESS%5FTOKEN%3D{OTHER_TOKEN}'):
|
|
self.assertNotIn(OTHER_TOKEN[3:15], meta.redact(text), text)
|
|
self.assertEqual(meta.redact('com.oculus.systemactivities is missing'), 'com.oculus.systemactivities is missing')
|
|
self.assertEqual(meta.redact(None), '')
|
|
|
|
def test_a_token_sharing_the_known_tokens_start_is_redacted_whole(self):
|
|
known, other = 'OCabcdefX123456789012', 'OCabcdefY123456789012'
|
|
for text in (other, f'see {other} here', f'{other}{known}', f'{known} {other}'):
|
|
out = meta.redact(text, token=known)
|
|
self.assertNotIn('123456789012', out, text)
|
|
self.assertNotIn('abcdef', out, text)
|
|
|
|
def test_a_known_token_with_oc_inside_its_start_is_redacted_whole(self):
|
|
known = 'OCaOCabcdef123456789012'
|
|
enc = ''.join('%%%02x' % ord(c) for c in known[:3]) + known[3:]
|
|
for text in (f'error near {known} here', f'error near %4fCaOCabcdef123456789012 here',
|
|
f'error near oCaOCabcdef123456789012 here', f'error near {enc} here',
|
|
f'error near {known.swapcase()} here', f'x {known}{known} y'):
|
|
out = meta.redact(text, token=known)
|
|
self.assertTrue(out.startswith('error near ') or out.startswith('x '), out)
|
|
for frag in ('Ca', '4fCa', 'abcdef', '123456789012'):
|
|
self.assertNotIn(frag, out.replace('error near', '').replace(meta.REDACTED, ''), (text, out))
|
|
|
|
def test_a_partial_match_before_the_token_cannot_hide_its_start(self):
|
|
known = 'OCabcdefOCabcdefg1234567890'
|
|
for text in (known[:8].lower() + known.lower(), known[:9] + known, 'xx' + known[:8] + known + 'yy'):
|
|
out = meta.redact(text, token=known)
|
|
self.assertNotIn('1234567890', out, (text, out))
|
|
self.assertNotIn('g123', out, (text, out))
|
|
import time
|
|
t0 = time.monotonic(); meta.redact(('OCabcdef' * 12000)[:100000], token=known)
|
|
self.assertLess(time.monotonic() - t0, 2.0)
|
|
|
|
def test_redacted_before_truncation(self):
|
|
for cut in range(1, len(TOKEN) + 1):
|
|
text = 'x' * 50 + TOKEN + ' tail'
|
|
out = meta.redact(text, limit=50 + cut)
|
|
self.assertLessEqual(len(out), 50 + cut)
|
|
self.assertNotIn(TOKEN[:6], out)
|
|
self.assertEqual(meta.redact('short', limit=100), 'short')
|
|
|
|
|
|
class Hosts(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fake = FakeMeta(self)
|
|
|
|
def test_only_plain_https(self):
|
|
for url in ('http://securecdn.oculus.com/x', 'https://user:pw@securecdn.oculus.com/x',
|
|
'https://securecdn.oculus.com:8443/x', 'ftp://securecdn.oculus.com/x'):
|
|
with self.assertRaisesRegex(meta.MetaError, 'plain HTTPS'):
|
|
meta.check_url(url)
|
|
self.assertEqual(meta.check_url('https://scontent.oculuscdn.com/v/1'), 'https://scontent.oculuscdn.com/v/1')
|
|
self.assertEqual(self.fake.lookups, []) # the URL check never looks anything up
|
|
|
|
def test_public_addresses_only(self):
|
|
for ip in ('10.0.0.5', '127.0.0.1', '169.254.169.254', '100.64.0.1', '192.0.0.8', '0.0.0.0',
|
|
'224.0.0.1', '::1', '::', 'fd00::1', 'fe80::1', 'ff02::1', 'fec0::1', '64:ff9b::a00:1',
|
|
'::ffff:10.0.0.1', '2002:a00:1::1', '2001::1', '2001:db8::1', 'not an ip'):
|
|
self.assertFalse(meta.public_address(ip), ip)
|
|
for ip in ('157.240.22.35', '8.8.8.8', '2a03:2880:f12f:83:face:b00c:0:25de'):
|
|
self.assertTrue(meta.public_address(ip), ip)
|
|
self.fake.addresses['cdn.example.com'] = {'157.240.22.35', 'ff02::1'}
|
|
with self.assertRaisesRegex(meta.MetaError, 'private or local'):
|
|
meta.resolve_public('cdn.example.com')
|
|
self.assertEqual(meta.resolve_public('scontent.oculuscdn.com'), [('157.240.22.35', 443)])
|
|
|
|
def test_connects_only_to_the_checked_address(self):
|
|
self.fake.addresses['scontent.oculuscdn.com'] = {'157.240.22.35', '157.240.22.36'}
|
|
tried = []
|
|
|
|
def connect(host, address, timeout, deadline):
|
|
tried.append((host, address, deadline))
|
|
if address[0].endswith('35'):
|
|
raise OSError('unreachable')
|
|
return 'connection'
|
|
with patch.object(meta, '_connect', side_effect=connect):
|
|
self.assertEqual(meta._public_connection('scontent.oculuscdn.com', 60), 'connection')
|
|
self.assertEqual([a for _, a, _ in tried], [('157.240.22.35', 443), ('157.240.22.36', 443)])
|
|
self.assertEqual(len({d for _, _, d in tried}), 1) # one shared deadline
|
|
self.assertEqual(self.fake.lookups, ['scontent.oculuscdn.com']) # resolved once
|
|
self.fake.addresses['evil.example'] = {'10.1.1.1'}
|
|
with patch.object(meta, '_connect') as never:
|
|
with self.assertRaises(meta.MetaError):
|
|
meta._public_connection('evil.example', 60)
|
|
never.assert_not_called()
|
|
|
|
def test_send_uses_the_checked_connection_and_closes_it(self):
|
|
calls = []
|
|
|
|
class Conn:
|
|
def request(self, method, target, body=None, headers=None):
|
|
calls.append((method, target, body, headers))
|
|
|
|
def getresponse(self):
|
|
r = Body(b'{}')
|
|
r.status, r.headers = 200, {}
|
|
return r
|
|
|
|
def close(self):
|
|
calls.append('closed')
|
|
with patch.object(meta, '_public_connection', return_value=Conn()) as connect:
|
|
status, _, body = REAL_SEND('https://graph.oculus.com/graphql?x=1', data=b'a=1')
|
|
with body:
|
|
self.assertEqual(body.read(), b'{}')
|
|
self.assertEqual(status, 200)
|
|
connect.assert_called_once_with('graph.oculus.com', meta.TIMEOUT)
|
|
self.assertEqual(calls[0][:3], ('POST', '/graphql?x=1', b'a=1'))
|
|
self.assertEqual(calls[-1], 'closed')
|
|
with self.assertRaisesRegex(meta.MetaError, 'plain HTTPS'):
|
|
REAL_SEND('http://graph.oculus.com/graphql')
|
|
|
|
def test_redirects_off_meta_never_carry_the_token(self):
|
|
start = f'https://securecdn.oculus.com/binaries/download/?id=1&access_token={TOKEN}'
|
|
self.assertEqual(meta.next_hop(start, 'https://scontent.oculuscdn.com/v/1?sig=x', TOKEN),
|
|
'https://scontent.oculuscdn.com/v/1?sig=x')
|
|
same = meta.next_hop(start, f'/binaries/download/?id=2&access_token={TOKEN}', TOKEN)
|
|
self.assertIn(TOKEN, same) # still Meta's download host
|
|
quoted = urllib.parse.quote(TOKEN, safe='')
|
|
lower_hex = ''.join('%%%02x' % ord(c) for c in TOKEN.lower())
|
|
for target in (f'https://cdn.example/v/1?sig={TOKEN}', f'https://cdn.example/{TOKEN}/file',
|
|
f'https://cdn.example/v/1#{TOKEN}', f'https://cdn.example/v/1?sig={quoted}',
|
|
f'https://cdn.example/{lower_hex}', f'https://cdn.example/v?s={TOKEN.swapcase()}',
|
|
f'https://{TOKEN[:12].lower()}.cdn.example/v', f'https://cdn.example/v?x={TOKEN[:9]}',
|
|
'https://cdn.example/v?ACCESS_TOKEN=other'):
|
|
with self.assertRaisesRegex(meta.MetaError, 'carrying your token'):
|
|
meta.next_hop(start, target, TOKEN)
|
|
self.assertEqual(self.fake.lookups, []) # refused before any lookup (a token in a host name)
|
|
with self.assertRaisesRegex(meta.MetaError, 'plain HTTPS'):
|
|
meta.next_hop(start, 'http://evil.example/steal', TOKEN)
|
|
|
|
def test_download_redirect_with_the_token_is_refused_before_connecting(self):
|
|
meta.connect(TOKEN)
|
|
self.addCleanup(meta.disconnect)
|
|
token, generation = meta._session()
|
|
work = meta.cache_root() / 'refuse'
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
self.addCleanup(lambda: __import__('shutil').rmtree(work, ignore_errors=True))
|
|
self.fake.location = 'https://cdn.example/{token}/v/{id}'
|
|
with self.assertRaisesRegex(meta.MetaError, 'carrying your token'):
|
|
meta.fetch('9002', work / 'x.obb', meta.MAX_DATA_FILE, token, generation)
|
|
self.assertFalse([r for r in self.fake.requests if 'cdn.example' in r['url']])
|
|
self.assertEqual(list(work.iterdir()), [])
|
|
|
|
def test_download_redirect_sends_no_token_to_the_cdn(self):
|
|
meta.connect(TOKEN)
|
|
self.addCleanup(meta.disconnect)
|
|
token, generation = meta._session()
|
|
dest = Path(meta.cache_root()) / 'redirect-test'
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
self.addCleanup(lambda: [p.unlink() for p in dest.iterdir()] and dest.rmdir())
|
|
got = meta.fetch('9002', dest / 'x.obb', meta.MAX_DATA_FILE, token, generation, expected=len(OBB))
|
|
self.assertEqual(Path(got['path']).read_bytes(), OBB)
|
|
cdn = [r for r in self.fake.requests if 'oculuscdn' in r['url']]
|
|
self.assertEqual(len(cdn), 1)
|
|
self.assertNotIn(TOKEN, json.dumps(cdn[0]))
|
|
self.assertNotIn('access_token', cdn[0]['url'].lower())
|
|
self.assertNotIn('Authorization', cdn[0]['headers'])
|
|
|
|
def test_graphql_redirect_not_followed(self):
|
|
with patch.object(meta, '_send', return_value=(302, {'Location': 'https://evil.example/'}, Body(b''))):
|
|
with self.assertRaisesRegex(meta.MetaError, 'redirected'):
|
|
meta.graphql(TOKEN, meta.DOC_VIEWER)
|
|
|
|
def test_network_errors_are_redacted(self):
|
|
error = OSError(f'connection to https://graph.oculus.com/?access_token={TOKEN} reset')
|
|
with patch.object(meta, '_send', side_effect=error):
|
|
with self.assertRaises(meta.MetaError) as e:
|
|
meta.graphql(TOKEN, meta.DOC_VIEWER)
|
|
self.assertNotIn(TOKEN, str(e.exception))
|
|
self.assertIsNone(e.exception.__cause__)
|
|
self.assertTrue(e.exception.__suppress_context__)
|
|
|
|
|
|
class Library(unittest.TestCase):
|
|
def test_owned_quest_apps_with_download_facts(self):
|
|
fake = connected(self)
|
|
apps = {a['id']: a for a in meta.library()}
|
|
self.assertEqual(set(apps), {'111', '222'}) # no Rift app, nothing expired or inactive
|
|
game = apps['111']
|
|
self.assertEqual((game['title'], game['package'], game['version'], game['version_code'], game['size']),
|
|
('Quest Game', PKG, '1.0', 100, len(APK)))
|
|
self.assertEqual(game['binary_id'], '9001')
|
|
self.assertEqual(game['obb'], {'id': '9002', 'name': 'game-data.obb', 'size': len(OBB), 'sha256': None,
|
|
'required': True})
|
|
self.assertEqual([(f['id'], f['required']) for f in game['assets']], [('9003', True), ('9004', False)])
|
|
other = apps['222'] # filled in from the binary query
|
|
self.assertEqual((other['package'], other['version_code'], other['size']), ('org.example.another', 7, 10))
|
|
queries = len(fake.requests)
|
|
meta.library()
|
|
self.assertEqual(len(fake.requests), queries) # kept in memory
|
|
public = meta._public(game)
|
|
self.assertEqual(public['data_files'], 1) # the OBB; level1.pak isn't a standard OBB name
|
|
self.assertEqual(public['skipped_files'], ['level1.pak'])
|
|
self.assertNotIn(TOKEN, json.dumps(public))
|
|
meta.disconnect()
|
|
with self.assertRaises(meta.MetaError):
|
|
meta.library()
|
|
|
|
def test_binary_query_when_the_list_doesnt_say_what_data_there_is(self):
|
|
fake = connected(self)
|
|
binary = fake.entitlements['data']['viewer']['user']['active_entitlements']['nodes'][0]['item'][
|
|
'latest_supported_binary']
|
|
fake.details['9001'] = {'data': {'node': dict(binary)}}
|
|
for key in ('obb_binary', 'asset_files'):
|
|
binary.pop(key)
|
|
game = next(a for a in meta.library(refresh=True) if a['id'] == '111')
|
|
self.assertEqual(game['obb']['id'], '9002') # from the binary query
|
|
asked = [json.loads(dict(urllib.parse.parse_qsl(r['data'].decode()))['variables']).get('binaryID')
|
|
for r in fake.requests if r['data'] and meta.DOC_BINARY in r['data'].decode()]
|
|
self.assertIn('9001', asked)
|
|
with patch.object(meta, 'MAX_DETAILS', 0): # beyond the cap: filled in when chosen
|
|
game = next(a for a in meta.library(refresh=True) if a['id'] == '111')
|
|
self.assertIsNone(game['obb'])
|
|
self.assertEqual(meta.owned('111')['obb']['id'], '9002')
|
|
|
|
def test_cli(self):
|
|
FakeMeta(self)
|
|
self.addCleanup(meta.disconnect)
|
|
out = io.StringIO()
|
|
with patch.object(sys, 'argv', ['frame_meta.py', 'connect', '--token-stdin']), \
|
|
patch.object(sys, 'stdin', io.StringIO(TOKEN + '\n')), patch('sys.stdout', out), \
|
|
patch('sys.stderr', io.StringIO()):
|
|
meta.main()
|
|
self.assertTrue(json.loads(out.getvalue())['connected'])
|
|
out = io.StringIO()
|
|
with patch.object(sys, 'argv', ['frame_meta.py', 'list']), patch('sys.stdout', out), \
|
|
patch.object(meta, 'with_verdicts', side_effect=lambda apps: apps):
|
|
meta.main()
|
|
listed = json.loads(out.getvalue())
|
|
self.assertEqual({a['id'] for a in listed}, {'111', '222'})
|
|
self.assertNotIn(TOKEN, out.getvalue())
|
|
with patch.object(sys, 'argv', ['frame_meta.py', 'connect', TOKEN]):
|
|
with self.assertRaises(SystemExit) as e:
|
|
meta.main()
|
|
self.assertNotIn(TOKEN, str(e.exception.code)) # a token on the command line is refused, not echoed
|
|
|
|
def test_only_owned_apps_download(self):
|
|
fake = connected(self)
|
|
with self.assertRaisesRegex(meta.MetaError, "isn't in your Meta library"):
|
|
meta.owned('999')
|
|
with self.assertRaisesRegex(meta.MetaError, 'invalid'):
|
|
meta.owned('../1')
|
|
with self.assertRaisesRegex(meta.MetaError, "isn't in your Meta library"):
|
|
meta.download('333') # a Rift app the account owns isn't a Quest app
|
|
self.assertEqual(fake.downloads(), [])
|
|
|
|
def test_expired_token_while_listing(self):
|
|
fake = connected(self)
|
|
fake.token = OTHER_TOKEN # Meta signed it out
|
|
with self.assertRaises(meta.TokenRejected) as e:
|
|
meta.library(refresh=True)
|
|
self.assertIn('get a new one', str(e.exception))
|
|
|
|
|
|
class Downloads(unittest.TestCase):
|
|
def test_apk_and_obb_into_the_cache(self):
|
|
fake = connected(self)
|
|
r = meta.download('111')
|
|
self.addCleanup(lambda: __import__('shutil').rmtree(r['folder'], ignore_errors=True))
|
|
self.assertEqual(Path(r['apk']).read_bytes(), APK)
|
|
self.assertEqual([Path(p).name for p in r['obb']], [f'main.100.{PKG}.obb'])
|
|
self.assertEqual(Path(r['obb'][0]).read_bytes(), OBB)
|
|
self.assertEqual(r['skipped'], ['level1.pak'])
|
|
self.assertTrue(r['folder'].startswith(str(meta.cache_root())))
|
|
ids = [dict(urllib.parse.parse_qsl(urllib.parse.urlsplit(d['url']).query)).get('id')
|
|
for d in fake.downloads() if 'securecdn' in d['url']]
|
|
self.assertEqual(ids, ['9001', '9002']) # never the optional DLC, nor the non-OBB asset
|
|
|
|
def test_declared_size_over_the_limit_refused_before_downloading(self):
|
|
fake = connected(self)
|
|
token, generation = meta._session()
|
|
work = meta.cache_root() / 'limits'
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
self.addCleanup(lambda: __import__('shutil').rmtree(work, ignore_errors=True))
|
|
before = len(fake.requests)
|
|
with self.assertRaisesRegex(meta.MetaError, 'more than Frame Control downloads'):
|
|
meta.fetch('9002', work / 'a.obb', 100, token, generation, expected=1000)
|
|
self.assertEqual(len(fake.requests), before)
|
|
fake.entitlements['data']['viewer']['user']['active_entitlements']['nodes'][0]['item'][
|
|
'latest_supported_binary']['obb_binary']['size'] = meta.MAX_TOTAL + 1
|
|
meta.library(refresh=True)
|
|
with patch.object(meta.shutil, 'disk_usage', return_value=type('U', (), {'free': 10**15})()):
|
|
with self.assertRaisesRegex(meta.MetaError, 'more than Frame Control downloads'):
|
|
meta.download('111')
|
|
with patch.object(meta.shutil, 'disk_usage', return_value=type('U', (), {'free': 10})()):
|
|
with self.assertRaisesRegex(meta.MetaError, 'GB free on this computer'):
|
|
meta.download('111')
|
|
|
|
def test_declared_and_actual_sizes(self):
|
|
fake = connected(self)
|
|
token, generation = meta._session()
|
|
work = meta.cache_root() / 'sizes'
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
self.addCleanup(lambda: __import__('shutil').rmtree(work, ignore_errors=True))
|
|
with self.assertRaisesRegex(meta.MetaError, 'not the size'):
|
|
meta.fetch('9002', work / 'a.obb', 50, token, generation) # Content-Length over the limit
|
|
fake.cdn_headers = {'Content-Length': None} # no length given: the limit stops the stream
|
|
with self.assertRaisesRegex(meta.MetaError, 'larger than'):
|
|
meta.fetch('9002', work / 'a.obb', 50, token, generation)
|
|
fake.cdn_headers = {'Content-Length': '5'}
|
|
with self.assertRaisesRegex(meta.MetaError, 'not the size Meta said'):
|
|
meta.fetch('9002', work / 'b.obb', 10**6, token, generation, expected=len(OBB))
|
|
fake.cdn_headers = {'Content-Length': None}
|
|
with self.assertRaisesRegex(meta.MetaError, 'incomplete'):
|
|
meta.fetch('9003', work / 'c.obb', 10**6, token, generation, expected=10)
|
|
with self.assertRaisesRegex(meta.MetaError, 'checksum'):
|
|
meta.fetch('9003', work / 'd.obb', 10**6, token, generation, expected=3, sha256='0' * 64)
|
|
self.assertEqual(list(work.iterdir()), []) # nothing left behind, not even .part files
|
|
|
|
def test_disconnect_stops_a_download(self):
|
|
connected(self)
|
|
token, generation = meta._session()
|
|
work = meta.cache_root() / 'stop'
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
self.addCleanup(lambda: __import__('shutil').rmtree(work, ignore_errors=True))
|
|
meta.disconnect()
|
|
with self.assertRaisesRegex(meta.MetaError, 'disconnected'):
|
|
meta.fetch('9002', work / 'a.obb', 10**6, token, generation)
|
|
self.assertEqual(list(work.iterdir()), [])
|
|
|
|
def test_one_budget_for_every_file_including_unknown_sizes(self):
|
|
# Sizes Meta doesn't give count as they arrive: three files of unknown size that each fit
|
|
# the per-file limit still can't add up past the install's budget.
|
|
fake = connected(self)
|
|
node = fake.entitlements['data']['viewer']['user']['active_entitlements']['nodes'][0]['item'][
|
|
'latest_supported_binary']
|
|
node.pop('size')
|
|
node['obb_binary'].pop('size')
|
|
node['obb_binary']['file_name'] = f'main.100.{PKG}.obb'
|
|
node['asset_files']['nodes'][0].update(file_name=f'patch.100.{PKG}.obb')
|
|
node['asset_files']['nodes'][0].pop('size')
|
|
fake.files['9003'] = OBB
|
|
fake.cdn_headers = {'Content-Length': None}
|
|
meta.library(refresh=True)
|
|
budget = len(APK) + len(OBB) + len(OBB) // 2 # the second OBB overflows half way
|
|
with patch.object(meta, 'MAX_TOTAL', budget):
|
|
with self.assertRaisesRegex(meta.MetaError, 'past the'):
|
|
meta.download('111')
|
|
self.assertEqual([p for p in meta.cache_root().iterdir() if p.name.startswith(PKG)], []) # all removed
|
|
with patch.object(meta, 'MAX_TOTAL', len(APK) + 2 * len(OBB)):
|
|
r = meta.download('111') # exactly enough
|
|
self.addCleanup(lambda: __import__('shutil').rmtree(r['folder'], ignore_errors=True))
|
|
self.assertEqual(len(r['obb']), 2)
|
|
|
|
def test_not_an_apk(self):
|
|
connected(self, apk=b'<html>sign in</html>')
|
|
with self.assertRaisesRegex(meta.MetaError, 'not an APK'):
|
|
meta.install('111')
|
|
self.assertEqual(list(meta.cache_root().iterdir()) if meta.cache_root().is_dir() else [], [])
|
|
|
|
|
|
class Install(unittest.TestCase):
|
|
"""The handoff to Import from my Quest's checks, install and two-step game data."""
|
|
META = dict(test_quest.Import.META)
|
|
|
|
def setUp(self):
|
|
self.order, self.running, self.installed = [], False, None
|
|
|
|
def install(path, source=None, **kw):
|
|
self.order.append(('install', Path(path).name, source))
|
|
self.assertTrue(android._install_lock._is_owned())
|
|
self.assertEqual(Path(path).read_bytes()[:2], b'PK')
|
|
return dict(self.META)
|
|
|
|
def add_data(pkg, checked, obb, folder, report, warnings=(), source=''):
|
|
self.order.append(('data', [Path(p).name for p in obb], [Path(p).read_bytes() for p in obb]))
|
|
self.running = False
|
|
return {'message': 'Added the game data.', 'obb': [{'name': Path(p).name} for p in obb], 'warnings': list(warnings),
|
|
'app': dict(self.META), 'needs_game_data': False, 'closed': True, 'split': False}
|
|
|
|
def launch(pkg):
|
|
self.order.append(('launch',))
|
|
self.running = True
|
|
|
|
for target, name, effect in ((android, 'install', install), (android, 'launch', launch),
|
|
(android, 'read_meta', lambda pkg: self.installed),
|
|
(quest, '_running', lambda pkg: self.running),
|
|
(quest, '_sleep', lambda s: None),
|
|
(quest, 'add_game_data', add_data)):
|
|
p = patch.object(target, name, side_effect=effect)
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
def cache_left(self):
|
|
root = meta.cache_root()
|
|
return [p for p in root.iterdir() if p.name.startswith(PKG)] if root.is_dir() else []
|
|
|
|
def test_install_then_add_game_data_on_the_second_install(self):
|
|
connected(self)
|
|
result = meta.install('111')
|
|
self.assertEqual(self.order, [('install', 'base.apk', 'Meta Horizon Store')])
|
|
self.assertTrue(result['needs_game_data'])
|
|
self.assertIn('choose Install again', result['message'])
|
|
self.assertIn('level1.pak', result['message']) # not copied, and said so
|
|
self.assertEqual(result['source'], 'meta-store')
|
|
self.assertEqual(self.cache_left(), []) # nothing kept after the install
|
|
# Second Install with the game open: the OBB, under its standard name, then closed.
|
|
self.running, self.installed = True, dict(self.META)
|
|
result = meta.install('111')
|
|
self.assertEqual(self.order[-1], ('data', [f'main.100.{PKG}.obb'], [OBB]))
|
|
self.assertEqual(self.cache_left(), [])
|
|
|
|
def test_install_and_test_adds_the_data_itself_then_tests(self):
|
|
connected(self)
|
|
with patch.object(meta, 'run_test', side_effect=lambda pkg, **kw: self.order.append(('test', pkg)) or
|
|
{'message': 'Test: Quest Game runs.', 'outcome': 'runs'}):
|
|
result = meta.install('111', test=True)
|
|
self.assertEqual([o[0] for o in self.order], ['install', 'launch', 'data', 'test'])
|
|
self.assertFalse(result['needs_game_data'])
|
|
self.assertEqual(result['test']['outcome'], 'runs')
|
|
self.assertIn('Test: Quest Game runs.', result['message'])
|
|
|
|
def test_licence_check_installs_with_the_warning(self):
|
|
apk = test_quest.apk_bytes(files=dict(test_quest.OPENXR, **{'lib/arm64-v8a/libovrplatformloader.so': b''}),
|
|
version_code=100, version_name='1.0')
|
|
connected(self, apk=apk)
|
|
result = meta.install('111')
|
|
self.assertEqual(self.order[0][0], 'install')
|
|
self.assertTrue(result['platform_sdk'])
|
|
self.assertIn("Uses Meta's licence check", result['message'])
|
|
self.assertIn('may quit, hang, or run without online features', result['message'])
|
|
|
|
def test_still_refused_32_bit_and_newer_android(self):
|
|
connected(self, apk=test_quest.apk_bytes(files={'lib/armeabi-v7a/libgame.so': b''}, version_code=100))
|
|
with self.assertRaisesRegex(quest.QuestError, '32-bit'):
|
|
meta.install('111')
|
|
meta.disconnect()
|
|
connected(self)
|
|
real = android.apk_info
|
|
with patch.object(android, 'apk_info', side_effect=lambda p: dict(real(p), min_sdk=31)):
|
|
with self.assertRaisesRegex(android.FrameError, 'API 31'):
|
|
meta.install('111')
|
|
self.assertEqual(self.order, [])
|
|
self.assertEqual(self.cache_left(), [])
|
|
|
|
def test_version_mismatch_and_one_at_a_time(self):
|
|
connected(self, apk=test_quest.apk_bytes(files=test_quest.OPENXR, version_code=99))
|
|
with self.assertRaisesRegex(meta.MetaError, 'version code 100, but the APK it sent is 99'):
|
|
meta.install('111')
|
|
with meta._install_lock:
|
|
with self.assertRaisesRegex(meta.MetaError, 'Another My Meta library install'):
|
|
meta.install('111')
|
|
|
|
|
|
LOGS = {
|
|
'runs': 'I FrameXrCompat: created instance\nI Unity: XR_SESSION_STATE_FOCUSED\n',
|
|
'issues': 'I Unity: XR_SESSION_STATE_FOCUSED\nE Adreno: GL_INVALID_OPERATION in glDrawArrays\n',
|
|
'licence': 'I Unity: ovr_Entitlement_GetIsViewerEntitled\nE Unity: Entitlement check failed, quitting\n',
|
|
'crash': ('E AndroidRuntime: FATAL EXCEPTION: main\n'
|
|
'E AndroidRuntime: java.lang.NullPointerException: ClipboardManager.addPrimaryClipChangedListener\n'),
|
|
'native': 'F libc: Fatal signal 11 (SIGSEGV), code 1 in tid 4242 (UnityMain)\n',
|
|
'anr': 'E ActivityManager: ANR in com.example.questgame\n',
|
|
'extension': 'E OpenXR: xrCreateInstance: XR_ERROR_EXTENSION_NOT_PRESENT XR_FB_passthrough\n',
|
|
'device': 'E Unity: Failed to launch SystemActivities\n',
|
|
'gpu-crash': 'E zink: DEVICE LOST\nF libc: Fatal signal 6 (SIGABRT) in libgallium_dri.so\n',
|
|
'java': ('E AndroidRuntime: FATAL EXCEPTION: main\n'
|
|
f'E AndroidRuntime: java.lang.IllegalStateException: bad session {TOKEN} for user 1234\n'),
|
|
'licence-ok': ('I Unity: ovr_Entitlement_GetIsViewerEntitled called\n'
|
|
'I Unity: IsUserEntitledToApplication: true\nI Unity: entitlement check passed\n'),
|
|
}
|
|
LOGS['extension-session'] = LOGS['extension'] + 'I Unity: XR_SESSION_STATE_FOCUSED\n'
|
|
LOGS['licence-ok-session'] = LOGS['licence-ok'] + LOGS['runs']
|
|
|
|
|
|
class Classify(unittest.TestCase):
|
|
def test_outcomes_from_logs(self):
|
|
cases = [
|
|
('runs', True, 'runs', 'runs', '', ''),
|
|
('issues', True, 'runs-with-issues', 'runs', 'Graphics errors', ''),
|
|
('licence', False, 'quits', 'crashes', "Meta's licence check failed", ''),
|
|
# Open with a failed check but no sign of its VR scene: not enough to call it stuck.
|
|
('licence', True, 'inconclusive', None, "isn't recorded", ''),
|
|
('crash', False, 'crashes', 'crashes', 'clipboard service', ''),
|
|
('native', False, 'crashes', 'crashes', 'native crash signal', 'SIGSEGV'),
|
|
('java', False, 'crashes', 'crashes', 'Java exception', 'java.lang.IllegalStateException'),
|
|
('anr', True, 'hangs', 'crashes', 'stopped responding', ''),
|
|
('extension', True, 'inconclusive', None, 'OpenXR feature', 'XR_FB_passthrough'),
|
|
('extension-session', True, 'runs-with-issues', 'runs', 'OpenXR feature', 'XR_FB_passthrough'),
|
|
('device', False, 'quits', 'crashes', 'Meta Quest', ''),
|
|
('gpu-crash', False, 'crashes', 'crashes', 'Graphics errors', ''),
|
|
('licence-ok', True, 'inconclusive', None, "can't tell whether it runs", ''),
|
|
('licence-ok-session', True, 'runs', 'runs', '', ''),
|
|
]
|
|
for log, alive, outcome, result, cause, detail in cases:
|
|
v = meta.classify(LOGS[log], alive)
|
|
self.assertEqual((v['outcome'], v['result']), (outcome, result), log)
|
|
self.assertIn(cause, v['cause'], log)
|
|
self.assertEqual(v['detail'], detail, log)
|
|
# Causes are fixed texts, never log text.
|
|
self.assertTrue(v['cause'] == '' or v['cause'].startswith(tuple(meta.CAUSES.values())), log)
|
|
self.assertNotIn('licence', meta.classify(LOGS['licence-ok-session'], True)['signals']['matched'])
|
|
quiet = meta.classify('I Unity: loading\n', True)
|
|
self.assertEqual((quiet['outcome'], quiet['result']), ('inconclusive', None))
|
|
gone = meta.classify('', False, instance_up=False)
|
|
self.assertEqual(gone['outcome'], 'quits')
|
|
self.assertIn('stopped before its log could be read', gone['cause'])
|
|
never = meta.classify('', False, instance_up=False, started=False)
|
|
self.assertEqual((never['outcome'], never['result']), ('instance-failed', 'instance_failed'))
|
|
self.assertTrue(meta.classify(LOGS['runs'], True)['signals']['xr_session'])
|
|
self.assertEqual(meta.classify(LOGS['licence'], False)['signals']['matched'], ['licence'])
|
|
|
|
|
|
class Speed(unittest.TestCase):
|
|
"""A hostile or broken log can't make classifying or redacting slow: 100 KB adversarial
|
|
lines, each step well under 200 ms."""
|
|
LINES = {
|
|
'no .so': 'a' * 100000,
|
|
'dots': 'a.' * 50000,
|
|
'dotted names': ('a.' * 40 + ' ') * 1235,
|
|
'OC-': 'OC-' * 33334,
|
|
'OC run': ' OC' + 'a-' * 50000,
|
|
'zink': 'zink' * 25000,
|
|
'entitle': 'entitle' * 14286,
|
|
'entitlement calls': 'ovr_Entitlement' * 6667,
|
|
'platform': 'libovrplatform.so' * 5883,
|
|
'crash marker': '>>> ' + 'a' * 100000,
|
|
'access_token': 'access_token=' * 7693,
|
|
'percent': '%4f%43' * 16667,
|
|
'words': ('libx.so XR_FB_y java.lang.ZException SIGSEGV ' * 2300)[:100000],
|
|
}
|
|
|
|
def timed(self, fn, *args):
|
|
import time
|
|
start = time.perf_counter()
|
|
fn(*args)
|
|
return time.perf_counter() - start
|
|
|
|
def test_adversarial_lines(self):
|
|
for name, line in self.LINES.items():
|
|
self.assertTrue(99000 <= len(line) <= 101000, name)
|
|
for alive in (True, False):
|
|
self.assertLess(self.timed(meta.classify, line + '\nFATAL EXCEPTION\n', alive), 0.2, name)
|
|
self.assertLess(self.timed(meta.redact, line, None, TOKEN), 0.2, name)
|
|
for key in ('extension', 'library', 'crash-java', 'crash-native'):
|
|
self.assertLess(self.timed(meta._excerpt, key, line), 0.2, (name, key))
|
|
|
|
def test_names_are_still_found(self):
|
|
self.assertEqual(meta._excerpt('library', 'E linker: dlopen failed: library "libfoo.so" not found'), 'libfoo.so')
|
|
self.assertEqual(meta._excerpt('extension', 'XR_ERROR_EXTENSION_NOT_PRESENT: XR_FB_passthrough.'),
|
|
'XR_FB_passthrough')
|
|
self.assertEqual(meta._excerpt('crash-java', 'AndroidRuntime: java.lang.IllegalStateException: x'),
|
|
'java.lang.IllegalStateException')
|
|
self.assertEqual(meta._excerpt('crash-native', 'Fatal signal 11 (SIGSEGV)'), 'SIGSEGV')
|
|
self.assertEqual(meta.redact('error near ' + OTHER_TOKEN + ' here'), 'error near <redacted> here')
|
|
|
|
def test_every_accepted_token_is_redacted(self):
|
|
# Property: whatever check_token accepts, the generic pattern redacts in full, wherever
|
|
# it appears, without knowing the stored token.
|
|
import random
|
|
rng = random.Random(20261013)
|
|
allowed = ''.join(chr(c) for c in range(33, 127)
|
|
if re.fullmatch('[%s]' % meta.TOKEN_CHARS, chr(c)))
|
|
self.assertEqual(set(allowed), set('ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=_-'))
|
|
accepted = 0
|
|
for _ in range(3000):
|
|
prefix = rng.choice(('OC', 'FRL'))
|
|
body = ''.join(rng.choice(allowed) for _ in range(rng.randint(meta.TOKEN_MIN - len(prefix), 60)))
|
|
try:
|
|
token = meta.check_token(prefix + body)
|
|
except meta.MetaError:
|
|
continue
|
|
accepted += 1
|
|
for left, right in (('', ''), ('error near ', ' here'), ('library "', '.so" not found'),
|
|
('url?tok=', '&x=1'), ('(', ')'), ('', '_FLAG'), ('access_token=', '')):
|
|
self.assertGone(token, left, right)
|
|
# Random neighbours from all of printable ASCII, on either side.
|
|
for _ in range(5):
|
|
self.assertGone(token, ''.join(rng.choice(PRINTABLE) for _ in range(rng.randint(1, 3))),
|
|
''.join(rng.choice(PRINTABLE) for _ in range(rng.randint(1, 3))))
|
|
self.assertEqual(meta._excerpt('library', f'dlopen failed: library "{token}.so" not found'), '')
|
|
self.assertGreater(accepted, 2000)
|
|
|
|
def assertGone(self, token, left, right):
|
|
"""redact(left + token + right) is some start of left, one <redacted>, and some end of
|
|
right: every character of the token is gone, whatever surrounds it."""
|
|
out = meta.redact(left + token + right)
|
|
at = out.find(meta.REDACTED)
|
|
self.assertGreaterEqual(at, 0, (left, token, right))
|
|
before, after = out[:at], out[at + len(meta.REDACTED):]
|
|
self.assertTrue(left.startswith(before) and right.endswith(after), (left, token, right, out))
|
|
|
|
def test_round6_gaps(self):
|
|
# 1. All upper case is refused, so an "OC..._FLAG" constant can't be an accepted token.
|
|
for bad in ('OCABCDEFGHIJKLMNOPQR', 'OCABCDEFGHIJ_KLMNOPQRS', 'FRL0123456789ABCDEFGHIJ'):
|
|
with self.assertRaisesRegex(meta.MetaError, "doesn't look like a Meta token"):
|
|
meta.check_token(bad)
|
|
self.assertEqual(meta.redact('OCABCDEFGHIJKLMNOPQR_FLAG'), 'OCABCDEFGHIJKLMNOPQR_FLAG') # a name, kept
|
|
self.assertEqual(meta.redact('OCAbcdefGhijkLmnopQr_FLAG'), meta.REDACTED)
|
|
# 2. Found after any character, word characters included.
|
|
for left in ('https://example.invalid/_', 'x-', '=', '/', 'lib', '9', 'Z'):
|
|
self.assertEqual(meta.redact(left + 'OCAbcdefGhijkLmnopQr'), left + meta.REDACTED, left)
|
|
# 3. Whole-token redaction comes first, so no prefix survives a parameter match.
|
|
meta.check_token('OCaccess_token=123456')
|
|
self.assertEqual(meta.redact('OCaccess_token=123456'), meta.REDACTED)
|
|
self.assertEqual(meta.redact('x access_token=abc&y=1'), f'x access_token={meta.REDACTED}&y=1')
|
|
# Anything with other characters is refused at connect, with a clear message.
|
|
for bad in ('OCAbcdef.GhijkLmnopQrstUvwx', 'OCAbcdefGhijk~LmnopQrstUvwx', 'FRLAbcdef!GhijkLmnopQrstU',
|
|
'OCAbcdefGhijk#LmnopQrstUvwx', 'OCABCDEFGHIJ_KLMNOPQRS_TUV'):
|
|
with self.assertRaisesRegex(meta.MetaError, "doesn't look like a Meta token"):
|
|
meta.check_token(bad)
|
|
|
|
def test_dotted_value_regression(self):
|
|
# Accepted before (dots allowed) but not redacted, so it showed as a library name.
|
|
line = 'E linker: dlopen failed: library "OCAbcdef.GhijkLmnopQrstUvwx.so" not found'
|
|
with self.assertRaises(meta.MetaError):
|
|
meta.check_token('OCAbcdef.GhijkLmnopQrstUvwx')
|
|
# It can't be a connected token now; as a stored token it would be redacted by _clean too.
|
|
self.assertNotIn('OCAbcdef', meta.redact(line, token='OCAbcdef.GhijkLmnopQrstUvwx'))
|
|
|
|
def test_digit_free_token_shapes_are_redacted(self):
|
|
# check_token accepts tokens without digits, so the generic pattern must not need one.
|
|
shape = 'OCAbcdefGhijkLmnopQrstUvwx'
|
|
meta.check_token(shape)
|
|
meta.check_token('FRLAbcdefGhijkLmnopQrs')
|
|
meta.check_token('OCAbcdefGhijkLmnopQr') # 20 characters: the shortest accepted
|
|
with self.assertRaisesRegex(meta.MetaError, 'incomplete'):
|
|
meta.check_token('OCAbcdefGhijkLmnopQ') # 19: refused, so never needs redacting
|
|
for text in (shape, 'FRLAbcdefGhijkLmnopQrs', 'OC-' + 'x' * 20, 'OCabcdefghijklmnopqr', 'OCAbcdefGhijkLmnopQr'):
|
|
out = meta.redact(f'error near {text} here')
|
|
self.assertEqual(out, 'error near <redacted> here', text)
|
|
line = f'E linker: dlopen failed: library "{shape}.so" not found'
|
|
self.assertEqual(meta._excerpt('library', line), '')
|
|
v = meta.classify(line + '\n', False)
|
|
self.assertEqual((v['cause_key'], v['detail']), ('library', ''))
|
|
self.assertNotIn(shape[:10], json.dumps(v))
|
|
# Ordinary identifiers stay: upper-case constants, short words, package names, too-short runs.
|
|
for text in ('OCULUS_PLATFORM_SDK_VERSION_NAME', 'OCULUS_EXT_HAND_TRACKING_2', 'OCR', 'OCULUS',
|
|
'com.oculus.systemactivities', 'OCAbcdefGhijkLmnopQ', 'libOculusPlatform.so'):
|
|
self.assertEqual(meta.redact(f'error near {text} here'), f'error near {text} here', text)
|
|
self.assertEqual(meta._excerpt('library', 'dlopen failed: library "libOculusPlatformLoader.so" not found'),
|
|
'libOculusPlatformLoader.so')
|
|
long_line = 'x' * (meta.LINE_SCAN + 10) + ' Entitlement check failed'
|
|
self.assertNotIn('licence', meta.classify(long_line, False)['signals']['matched']) # only the start is scanned
|
|
|
|
|
|
class RunTest(unittest.TestCase):
|
|
META = dict(test_quest.Import.META)
|
|
|
|
def setUp(self):
|
|
self.reports, self.alive, self.log, self.running = [], True, LOGS['runs'], False
|
|
ctr = f"lepton-steamlaunch-{self.META['instance']}"
|
|
|
|
def ssh(cmd, input=None, timeout=120):
|
|
if 'pidof' in cmd:
|
|
return '4242\n' if self.alive else ''
|
|
if 'logcat' in cmd:
|
|
return self.log
|
|
raise AssertionError(cmd)
|
|
|
|
def launch(pkg):
|
|
self.running = True
|
|
for target, name, effect in ((android, '_meta_or_fail', lambda pkg: dict(self.META)),
|
|
(android, 'launch', launch), (android, 'ssh', ssh),
|
|
(android, 'running_instances', lambda strict=False: {ctr: '5555'} if self.running else {}),
|
|
(quest, '_running', lambda pkg: self.running),
|
|
(quest, '_close', lambda pkg: (setattr(self, 'running', False), (True, None))[1]),
|
|
(quest, '_sleep', lambda s: None)):
|
|
p = patch.object(target, name, side_effect=effect)
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
import frame_catalog
|
|
p = patch.object(frame_catalog, 'add_report', side_effect=lambda *a, **kw: self.reports.append((a, kw)) or
|
|
{'id': 'r1'})
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
def test_runs_and_report_carries_results_only(self):
|
|
connected(self)
|
|
self.log = LOGS['runs'] + f'I Game: debug url https://graph.oculus.com/?access_token={TOKEN}\n'
|
|
v = meta.run_test(PKG, wait=20)
|
|
self.assertEqual(v['outcome'], 'runs')
|
|
self.assertEqual(v['signals']['seconds'], 20)
|
|
(args, kwargs), = self.reports
|
|
self.assertEqual(args, (PKG, '1.0'))
|
|
self.assertEqual(kwargs['source'], 'meta-store')
|
|
self.assertEqual(kwargs['result'], 'runs')
|
|
self.assertEqual(kwargs['via'], 'probe')
|
|
self.assertIn('version code 100', kwargs['notes'])
|
|
self.assertEqual(set(kwargs), {'result', 'notes', 'via', 'label', 'source'})
|
|
payload = json.dumps([args, kwargs])
|
|
self.assertNotIn(TOKEN, payload)
|
|
self.assertNotIn(str(meta.cache_root()), payload)
|
|
self.assertNotIn('.apk', payload)
|
|
self.assertNotIn('.obb', payload)
|
|
|
|
def test_quit_with_licence_check_is_reported_as_it_happened(self):
|
|
connected(self)
|
|
self.alive, self.log = False, LOGS['licence']
|
|
v = meta.run_test(PKG, wait=30)
|
|
self.assertEqual((v['outcome'], v['result']), ('quits', 'crashes'))
|
|
self.assertIn("Meta's licence check failed", v['message'])
|
|
self.assertEqual(self.reports[0][1]['result'], 'crashes')
|
|
self.assertIn('Meta library test: quits', self.reports[0][1]['notes'])
|
|
|
|
def test_instance_never_starts(self):
|
|
connected(self)
|
|
with patch.object(android, 'launch'): # nothing comes up
|
|
v = meta.run_test(PKG, wait=5)
|
|
self.assertEqual(v['outcome'], 'instance-failed')
|
|
self.assertEqual(self.reports[0][1]['result'], 'instance_failed')
|
|
|
|
def test_hang_is_closed_and_no_report_when_asked(self):
|
|
connected(self)
|
|
self.log = LOGS['anr']
|
|
v = meta.run_test(PKG, wait=5, report_result=False)
|
|
self.assertEqual(v['outcome'], 'hangs')
|
|
self.assertFalse(self.running)
|
|
self.assertEqual(self.reports, [])
|
|
|
|
def test_crash_text_never_reaches_reports_or_messages(self):
|
|
connected(self)
|
|
self.alive, self.log = False, LOGS['java']
|
|
v = meta.run_test(PKG, wait=10)
|
|
self.assertEqual((v['outcome'], v['cause_key']), ('crashes', 'crash-java'))
|
|
self.assertIn('(java.lang.IllegalStateException)', v['message'])
|
|
(args, kwargs), = self.reports
|
|
everything = json.dumps([v, args, kwargs])
|
|
for fragment in (TOKEN, TOKEN[:8], 'bad session', 'user 1234'):
|
|
self.assertNotIn(fragment, everything)
|
|
self.assertNotIn('IllegalStateException', kwargs['notes']) # reports: fixed texts only
|
|
self.assertIn(meta.CAUSES['crash-java'].rstrip('.'), kwargs['notes'])
|
|
|
|
def test_long_token_in_a_library_name_is_never_shown(self):
|
|
connected(self)
|
|
long_token = TOKEN + 'A1b2C3d4E5f6G7h8I9j0K1l2M3n4O5p6Q7r8S9t0U1v2W3x4Y5z6'
|
|
meta.disconnect()
|
|
fake = FakeMeta(self, token=long_token)
|
|
meta.connect(long_token)
|
|
self.alive = False
|
|
for line in (f'E linker: dlopen failed: library "{long_token}.so" not found',
|
|
f'E linker: dlopen failed: library "lib{long_token[-40:]}.so" not found',
|
|
'E linker: dlopen failed: library "libZq9Xw3Rt7Yu1Io5Pa8Sd2Fg6.so" not found'):
|
|
self.log = line + '\n'
|
|
self.reports.clear()
|
|
v = meta.run_test(PKG, wait=5)
|
|
self.assertEqual(v['cause_key'], 'library')
|
|
self.assertEqual(v['detail'], '', line)
|
|
everything = json.dumps([v, self.reports])
|
|
for fragment in (long_token[-30:], long_token[-60:-30], long_token[:8], 'Zq9Xw3Rt7Yu1Io5Pa8Sd2Fg6'):
|
|
self.assertNotIn(fragment, everything)
|
|
self.log = 'E linker: dlopen failed: library "libfoo_engine.so" not found\n'
|
|
self.assertEqual(meta.run_test(PKG, wait=5)['detail'], 'libfoo_engine.so') # real names still shown
|
|
self.assertEqual(fake.token, long_token)
|
|
|
|
def test_log_read_failure_never_shows_the_ssh_output(self):
|
|
# A failed ssh command with empty stderr: frame_android.ssh raises its stdout, which
|
|
# here is the game's log, carrying a token. Neither the test, the CLI nor the job says it.
|
|
import subprocess
|
|
import server
|
|
connected(self)
|
|
leaked = f'I Game: url https://graph.oculus.com/?x=1&tok={TOKEN[5:]} {OTHER_TOKEN[3:]}\n' * 3
|
|
|
|
def run_ssh(cmd, **kw):
|
|
remote = cmd[-1]
|
|
if 'logcat' in remote:
|
|
return subprocess.CompletedProcess(cmd, 1, stdout=leaked, stderr='')
|
|
return subprocess.CompletedProcess(cmd, 0, stdout='4242\n' if 'pidof' in remote else '', stderr='')
|
|
with patch.object(android, 'ssh', REAL_SSH), patch.object(android.frame_host, 'run_ssh', side_effect=run_ssh):
|
|
with self.assertRaises(meta.MetaError) as e:
|
|
meta.run_test(PKG, wait=5)
|
|
self.assertEqual(str(e.exception), meta.LOG_UNREADABLE)
|
|
self.assertIsNone(e.exception.__cause__)
|
|
self.assertTrue(e.exception.__suppress_context__)
|
|
# The standalone Test route (a background job).
|
|
with patch.object(server, 'ensure_master'), patch.object(server, 'start_job',
|
|
return_value={'job': 'j'}) as job:
|
|
server.meta({'action': 'test', 'package': PKG})
|
|
with self.assertRaises(meta.MetaError) as e:
|
|
job.call_args.args[1]()
|
|
self.assertEqual(str(e.exception), meta.LOG_UNREADABLE)
|
|
# The CLI.
|
|
with patch.object(sys, 'argv', ['frame_meta.py', 'test', PKG, '--wait', '5']):
|
|
with self.assertRaises(SystemExit) as cli:
|
|
meta.main()
|
|
self.assertEqual(cli.exception.code, 'error: ' + meta.LOG_UNREADABLE)
|
|
for fragment in (TOKEN[5:20], OTHER_TOKEN[3:15]):
|
|
self.assertNotIn(fragment, str(e.exception) + str(cli.exception.code))
|
|
self.assertEqual(self.reports, [])
|
|
# Anything else that goes wrong in a test job is a fixed text too.
|
|
with patch.object(server, 'ensure_master'), patch.object(server, 'start_job', return_value={'job': 'j'}) as job:
|
|
server.meta({'action': 'test', 'package': PKG})
|
|
with patch.object(meta, 'run_test', side_effect=KeyError(TOKEN)):
|
|
with self.assertRaises(meta.MetaError) as e:
|
|
job.call_args.args[1]()
|
|
self.assertEqual(str(e.exception), 'The test stopped unexpectedly (KeyError); try again.')
|
|
|
|
def test_inconclusive_is_not_reported_or_closed(self):
|
|
connected(self)
|
|
self.log = 'I Unity: ovr_Entitlement_GetIsViewerEntitled called\nI Unity: loading\n'
|
|
v = meta.run_test(PKG, wait=10)
|
|
self.assertEqual((v['outcome'], v['result']), ('inconclusive', None))
|
|
self.assertEqual(self.reports, [])
|
|
self.assertIn('skipped', v['report'])
|
|
self.assertTrue(self.running) # left open: nothing says it is stuck
|
|
|
|
def test_must_close_a_running_game_before_testing(self):
|
|
connected(self)
|
|
self.running = True
|
|
launched = []
|
|
with patch.object(quest, '_close', return_value=(False, 'it is still running')), \
|
|
patch.object(android, 'launch', side_effect=launched.append):
|
|
with self.assertRaisesRegex(meta.MetaError, "couldn't close it, so it wasn't tested.*Close it from Steam"):
|
|
meta.run_test(PKG, wait=5)
|
|
self.assertEqual(launched, [])
|
|
self.assertEqual(self.reports, [])
|
|
|
|
def test_failed_hang_cleanup_is_said(self):
|
|
connected(self)
|
|
self.log = LOGS['anr']
|
|
with patch.object(quest, '_close', side_effect=[(False, 'Steam is not answering')]):
|
|
v = meta.run_test(PKG, wait=5)
|
|
self.assertEqual(v['outcome'], 'hangs')
|
|
self.assertFalse(v['closed'])
|
|
self.assertIn('the game may still be running; close it from Steam', v['message'])
|
|
self.assertEqual(self.reports[0][1]['result'], 'crashes')
|
|
|
|
def test_sharing_follows_the_privacy_setting(self):
|
|
sent = []
|
|
with patch.object(frame_telemetry, 'enabled', return_value=True), \
|
|
patch.object(frame_telemetry, 'capture', side_effect=lambda e, p, level: sent.append(p) or True):
|
|
frame_telemetry.compat_report({'id': 'x', 'package': PKG, 'result': 'runs', 'source': 'meta-store',
|
|
'notes': f'access_token={TOKEN}'})
|
|
self.assertEqual(sent[0]['source'], 'meta-store')
|
|
with patch.object(frame_telemetry, 'enabled', return_value=False), \
|
|
patch.object(frame_telemetry, 'capture') as capture:
|
|
self.assertFalse(frame_telemetry.compat_report({'id': 'y', 'package': PKG, 'result': 'runs'}))
|
|
capture.assert_not_called()
|
|
|
|
|
|
class Server(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
import server
|
|
cls.server = server
|
|
|
|
def test_requests_and_answers_never_carry_the_token(self):
|
|
s = self.server
|
|
FakeMeta(self)
|
|
self.addCleanup(meta.disconnect)
|
|
for body in ({'action': 'connect', 'token': TOKEN}, {'action': 'connect', 'acknowledged': True},
|
|
{'action': 'install', 'app': 'abc'}, {'action': 'install', 'app': '111', 'test': 'yes'},
|
|
{'action': 'test', 'package': '../x'}, {'action': 'nope'}):
|
|
with self.assertRaises(s.Failure) as e:
|
|
s.meta(body)
|
|
self.assertEqual(e.exception.status, 400, body)
|
|
result = s.meta({'action': 'connect', 'token': TOKEN, 'acknowledged': True})
|
|
self.assertTrue(result['connected'])
|
|
status = s.meta_status('library=1')
|
|
self.assertEqual({a['id'] for a in status['apps']}, {'111', '222'})
|
|
self.assertIn('sharing', status)
|
|
self.assertNotIn(TOKEN, json.dumps([result, status]))
|
|
with patch.object(s, 'ensure_master'), patch.object(s, 'queue_job', return_value={'job': 'j'}) as job:
|
|
self.assertEqual(s.meta({'action': 'install', 'app': '111', 'test': True}), {'job': 'j'})
|
|
work = job.call_args.args[1]
|
|
with patch.object(meta, 'install', side_effect=KeyError(f'access_token={TOKEN}')):
|
|
with self.assertRaises(meta.MetaError) as e:
|
|
work(lambda *a: None)
|
|
self.assertNotIn(TOKEN, str(e.exception))
|
|
with self.assertRaises(s.Failure) as e:
|
|
s.meta({'action': 'connect', 'token': OTHER_TOKEN, 'acknowledged': True})
|
|
self.assertNotIn(OTHER_TOKEN, str(e.exception))
|
|
self.assertEqual(e.exception.status, 400)
|
|
self.assertFalse(s.meta({'action': 'disconnect'})['connected'])
|
|
|
|
def test_sign_in_attempts_through_the_server(self):
|
|
s = self.server
|
|
FakeMeta(self)
|
|
self.addCleanup(meta.disconnect)
|
|
for body in ({'action': 'cancel-signin'}, {'action': 'cancel-signin', 'attempt': 'no'},
|
|
{'action': 'connect', 'token': TOKEN, 'acknowledged': True, 'attempt': 5},
|
|
{'action': 'connect', 'token': TOKEN, 'acknowledged': True, 'attempt': '../../x'}):
|
|
with self.assertRaises(s.Failure) as e:
|
|
s.meta(body)
|
|
self.assertEqual(e.exception.status, 400, body)
|
|
self.assertTrue(s.meta({'action': 'connect', 'token': TOKEN, 'acknowledged': True,
|
|
'attempt': 'server-attempt-0001'})['connected'])
|
|
result = s.meta({'action': 'cancel-signin', 'attempt': 'server-attempt-0001'})
|
|
self.assertFalse(result['connected'])
|
|
self.assertNotIn(TOKEN, json.dumps(result))
|
|
# A cancel caught while Meta answers is a 409, without the token.
|
|
with patch.object(meta, 'whoami', side_effect=lambda t: meta.cancel_signin('server-attempt-0002') or
|
|
{'id': '1', 'name': 'player_one'}):
|
|
with self.assertRaises(s.Failure) as e:
|
|
s.meta({'action': 'connect', 'token': TOKEN, 'acknowledged': True, 'attempt': 'server-attempt-0002'})
|
|
self.assertEqual(e.exception.status, 409)
|
|
self.assertNotIn(TOKEN, str(e.exception))
|
|
self.assertFalse(meta.status()['connected'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|