mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-12 13:01:10 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5f3798b79b | ||
|
|
6e0b140efc | ||
|
|
226ec40720 | ||
|
|
f1aaed6279 |
No files matched your search
@@ -0,0 +1,87 @@
|
||||
"""Small APKs built in memory (binary manifest plus resource table), for test_frame_apk
|
||||
and the end-to-end tests. Deliberately doesn't import sandbox: tests/e2e runs in a
|
||||
throwaway container and keeps its real home, which sandbox would move."""
|
||||
import io
|
||||
import os
|
||||
import struct
|
||||
import sys
|
||||
import zipfile
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'ui'))
|
||||
import frame_apk # noqa: E402
|
||||
|
||||
|
||||
def pool(strings, utf8=False):
|
||||
"""A ResStringPool chunk."""
|
||||
data, offsets = b'', []
|
||||
for s in strings:
|
||||
offsets.append(len(data))
|
||||
if utf8:
|
||||
b = s.encode()
|
||||
data += bytes([len(s), len(b)]) + b + b'\0'
|
||||
else:
|
||||
data += struct.pack('<H', len(s)) + s.encode('utf-16-le') + b'\0\0'
|
||||
data += b'\0' * (-len(data) % 4)
|
||||
start = 28 + 4 * len(strings)
|
||||
body = struct.pack(f'<{len(strings)}I', *offsets) + data
|
||||
return struct.pack('<HHIIIIII', 1, 28, 28 + len(body), len(strings), 0, 0x100 if utf8 else 0, start, 0) + body
|
||||
|
||||
|
||||
def manifest(package, label_ref, version_ref, min_sdk, package_raw=True, foreign_label=False):
|
||||
"""<manifest package versionName><uses-sdk minSdkVersion/><application label icon/></manifest>.
|
||||
|
||||
package_raw=False drops the package's raw string (as some repackers do);
|
||||
foreign_label adds a non-android `label` attribute after android:label.
|
||||
"""
|
||||
strings = ['label', 'icon', 'versionName', 'minSdkVersion', 'package', 'manifest', 'uses-sdk',
|
||||
'application', package, 'junk', 'label', 'versionCode'] # the second 'label' has no android id
|
||||
resmap = struct.pack('<4I', 0x01010001, 0x01010002, 0x0101021c, 0x0101020c)
|
||||
resmap = struct.pack('<HHI', 0x0180, 8, 8 + len(resmap)) + resmap
|
||||
|
||||
def element(name, attrs):
|
||||
body = struct.pack('<IIHHHHHH', 0xffffffff, name, 20, 20, len(attrs), 0, 0, 0)
|
||||
for aname, raw, dtype, value in attrs:
|
||||
body += struct.pack('<IIIHBBI', 0xffffffff, aname, raw, 8, 0, dtype, value)
|
||||
return struct.pack('<HHIII', 0x0102, 16, 16 + len(body), 1, 0xffffffff) + body
|
||||
|
||||
none = 0xffffffff
|
||||
chunks = (pool(strings) + resmap
|
||||
+ element(5, [(4, 8 if package_raw else none, frame_apk.T_STRING, 8),
|
||||
(2, none, frame_apk.T_REF, version_ref),
|
||||
(11, none, frame_apk.T_INT_DEC, 210)])
|
||||
+ element(6, [(3, none, frame_apk.T_INT_DEC, min_sdk)])
|
||||
+ element(7, [(0, none, frame_apk.T_REF, label_ref), (1, none, frame_apk.T_REF, 0x7f020000)]
|
||||
+ ([(10, 9, frame_apk.T_STRING, 9)] if foreign_label else [])))
|
||||
return struct.pack('<HHI', 3, 8, 8 + len(chunks)) + chunks
|
||||
|
||||
|
||||
def resources(values):
|
||||
"""resources.arsc with package 0x7f; values: {(type id, entry, language, density): global string index}."""
|
||||
strings = ['French label', 'App label', '2.1', 'res/icon_lo.png', 'res/icon_hi.png', 'res/icon.xml']
|
||||
pkg_body = b''
|
||||
for (tid, lang, density), entries in values.items():
|
||||
cfg = struct.pack('<I4x2s4xH', 64, lang.encode().ljust(2, b'\0'), density).ljust(64, b'\0')
|
||||
count = max(entries) + 1
|
||||
offsets, data = [], b''
|
||||
for i in range(count):
|
||||
if i in entries:
|
||||
offsets.append(len(data))
|
||||
data += struct.pack('<HHI', 8, 0, 0) + struct.pack('<HBBI', 8, 0, frame_apk.T_STRING, entries[i])
|
||||
else:
|
||||
offsets.append(0xffffffff)
|
||||
header = 20 + 64
|
||||
estart = header + 4 * count
|
||||
body = struct.pack(f'<{count}I', *offsets) + data
|
||||
pkg_body += struct.pack('<HHIBBHII', 0x0201, header, header + len(body), tid, 0, 0, count, estart) + cfg + body
|
||||
pkg_header = struct.pack('<HHII', 0x0200, 288, 288 + len(pkg_body), 0x7f).ljust(288, b'\0')
|
||||
pkg = pkg_header + pkg_body
|
||||
table = pool(strings, utf8=True) + pkg
|
||||
return struct.pack('<HHII', 2, 12, 12 + len(table), 1) + table
|
||||
|
||||
|
||||
def apk(files):
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w') as z:
|
||||
for name, data in files.items():
|
||||
z.writestr(name, data)
|
||||
return buf.getvalue()
|
||||
+27
-4
@@ -10,6 +10,7 @@ import atexit
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -22,7 +23,22 @@ from pathlib import Path
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path[:0] = [str(ROOT / 'ui'), str(ROOT / 'tests'), str(ROOT / 'tests' / 'smoke')]
|
||||
|
||||
ENABLED = os.environ.get('FRAME_E2E') == '1'
|
||||
# Only inside the compose `host` container (FRAME_E2E_CONTAINER from compose.yaml, plus
|
||||
# the container runtime's marker file): these tests drive ui/server.py with the real
|
||||
# home, ~/.ssh and cache, which is only safe in a throwaway container.
|
||||
REQUESTED = os.environ.get('FRAME_E2E') == '1'
|
||||
IN_CONTAINER = os.environ.get('FRAME_E2E_CONTAINER') == '1' and (
|
||||
os.path.exists('/.dockerenv') or os.path.exists('/run/.containerenv'))
|
||||
ENABLED = REQUESTED and IN_CONTAINER
|
||||
if ENABLED:
|
||||
# What tests/sandbox.py does, minus moving HOME: the container's real ~/.ssh is what
|
||||
# points `ssh frame` at the fake Frame, and OpenSSH reads ~/.ssh from the password
|
||||
# database, not $HOME. (These tests used to get this through test_frame_apk.)
|
||||
_data = tempfile.mkdtemp(prefix='frame-control-e2e-')
|
||||
atexit.register(shutil.rmtree, _data, ignore_errors=True)
|
||||
os.environ['FRAME_CONTROL_DATA_DIR'] = _data
|
||||
os.environ['FRAME_CONTROL_TELEMETRY'] = '0'
|
||||
os.environ['FRAME_COMPAT_DB_URL'] = 'http://127.0.0.1:9'
|
||||
CTL = os.environ.get('FAKEFRAME_CTL', 'http://fakeframe:9999').rstrip('/')
|
||||
FAKE_HOST = os.environ.get('FAKEFRAME_HOST', 'fakeframe')
|
||||
HOME = '/home/steamos'
|
||||
@@ -30,9 +46,14 @@ SERVER_LOG = os.path.join(tempfile.gettempdir(), 'fakeframe-e2e-server.log')
|
||||
|
||||
|
||||
def require():
|
||||
"""Call at module level: skips the module unless the fake Frame is up."""
|
||||
if not ENABLED:
|
||||
raise unittest.SkipTest('needs the fake Frame: run scripts/e2e.sh (sets FRAME_E2E=1)')
|
||||
"""Skips (the module, at module level; the class, in setUpClass) unless this is
|
||||
scripts/e2e.sh's container with the fake Frame up."""
|
||||
if ENABLED:
|
||||
return
|
||||
if REQUESTED:
|
||||
raise unittest.SkipTest('FRAME_E2E=1 outside the e2e container: these tests use the real home, ~/.ssh '
|
||||
'and app cache, so they only run in scripts/e2e.sh\'s throwaway container')
|
||||
raise unittest.SkipTest('needs the fake Frame: run scripts/e2e.sh (sets FRAME_E2E=1)')
|
||||
|
||||
|
||||
# ---- the fake Frame's control port --------------------------------------------
|
||||
@@ -97,6 +118,7 @@ class Server:
|
||||
|
||||
@classmethod
|
||||
def start(cls):
|
||||
require() # never a server on this computer's real data
|
||||
if cls.proc and cls.proc.poll() is None:
|
||||
return
|
||||
with socket.socket() as s:
|
||||
@@ -193,6 +215,7 @@ class FrameTestCase(unittest.TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
require()
|
||||
Server.start()
|
||||
|
||||
def setUp(self):
|
||||
|
||||
@@ -6,6 +6,8 @@ import sys
|
||||
import harness
|
||||
from harness import api, ok, finished, ssh
|
||||
|
||||
harness.require()
|
||||
|
||||
sys.path.insert(0, str(harness.ROOT / 'ui'))
|
||||
import frame_mcp
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ import unittest
|
||||
|
||||
import harness
|
||||
from harness import HOME, exists, ok, state, upload, wait_for
|
||||
from test_frame_apk import apk, manifest, resources
|
||||
from apk_fixtures import apk, manifest, resources
|
||||
|
||||
harness.require()
|
||||
|
||||
|
||||
@@ -43,6 +43,7 @@ services:
|
||||
FAKEFRAME_CTL: http://fakeframe:9999
|
||||
FAKEFRAME_HOST: fakeframe
|
||||
FRAME_E2E: "1"
|
||||
FRAME_E2E_CONTAINER: "1" # tests/e2e/harness.py refuses to run anywhere else
|
||||
healthcheck:
|
||||
test: ["CMD", "test", "-f", "/home/tester/.ready"]
|
||||
interval: 1s
|
||||
|
||||
+31
-1
@@ -1,12 +1,20 @@
|
||||
"""Imported first by every test module: nothing a test does reaches this person's
|
||||
app data, their telemetry, or the shared compatibility database.
|
||||
app data, their ~/.ssh, their telemetry, or the shared compatibility database.
|
||||
|
||||
Must run before any ui module is imported, since those read these at import time.
|
||||
The variables are set in os.environ, so servers and scripts the tests start
|
||||
inherit them too (they pass {**os.environ, ...}).
|
||||
"""
|
||||
import atexit
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# The environment and real ~/.ssh from before the changes below: test_sandbox
|
||||
# reruns tests from that environment and checks ~/.ssh is left alone.
|
||||
ORIGINAL_ENV = dict(os.environ)
|
||||
REAL_SSH_DIR = Path.home() / ".ssh"
|
||||
|
||||
_dir = tempfile.mkdtemp(prefix="frame-control-tests-")
|
||||
atexit.register(shutil.rmtree, _dir, ignore_errors=True)
|
||||
@@ -14,3 +22,25 @@ os.environ["FRAME_CONTROL_DATA_DIR"] = _dir
|
||||
os.environ["FRAME_CONTROL_TELEMETRY"] = "0"
|
||||
# A maintainer's machine holds the database key; send anything that slips through nowhere.
|
||||
os.environ["FRAME_COMPAT_DB_URL"] = "http://127.0.0.1:9"
|
||||
|
||||
# A throwaway home, so Path.home() / "~" land here: ~/.ssh (config, host-key pins,
|
||||
# its lock file) and the other per-user folders some modules use. A server started
|
||||
# without FRAME_CONTROL_SSH_DIR used to import the real ~/.ssh/config and seed a pin
|
||||
# into the real ~/.ssh/frame-control-hosts on every run.
|
||||
#
|
||||
# tests/e2e must not import this (directly or through a test_* module): they run in
|
||||
# a throwaway container whose real ~/.ssh points `ssh frame` at the fake Frame, and
|
||||
# OpenSSH finds ~/.ssh from the password database, not $HOME. Shared helpers they
|
||||
# need live in sandbox-free modules such as apk_fixtures.py.
|
||||
HOME = os.path.join(_dir, "home")
|
||||
SSH_DIR = os.path.join(HOME, ".ssh")
|
||||
os.makedirs(SSH_DIR, mode=0o700)
|
||||
os.environ["HOME"] = HOME
|
||||
os.environ["USERPROFILE"] = HOME # Path.home() on Windows
|
||||
os.environ["FRAME_CONTROL_SSH_DIR"] = SSH_DIR # frame_devices.ssh_dir(), frame_connect.SSH_DIR
|
||||
# The per-user folders that don't follow HOME when these are set: frame_host.cache_dir()
|
||||
# (servers warm indexes there and sweep its staging folders) and data_dir()'s fallbacks.
|
||||
for _name, _sub in (("LOCALAPPDATA", "AppData/Local"), ("APPDATA", "AppData/Roaming"),
|
||||
("XDG_CACHE_HOME", ".cache"), ("XDG_DATA_HOME", ".local/share"),
|
||||
("XDG_STATE_HOME", ".local/state"), ("XDG_CONFIG_HOME", ".config")):
|
||||
os.environ[_name] = os.path.join(HOME, *_sub.split("/"))
|
||||
@@ -1,4 +1,5 @@
|
||||
"""MCP protocol, exact-action approvals and explicit assistant data sharing."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import http.client
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import hashlib, io, json, os, sys, tempfile, time, unittest, urllib.error, urllib.request, zipfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import http.client
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Fake-Frame session clock/actions plus real helper serialization and sensor probes."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import os
|
||||
import shutil
|
||||
import json
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Run the actual shared page's comfort renderer against a minimal DOM/bridge."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import pathlib
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Offline authenticated repository fixtures; no tests contact a server."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Offline artwork, Steam API and launcher supervision regressions."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
|
||||
+1
-76
@@ -11,82 +11,7 @@ import zipfile
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'ui'))
|
||||
import frame_apk # noqa: E402
|
||||
|
||||
|
||||
def pool(strings, utf8=False):
|
||||
"""A ResStringPool chunk."""
|
||||
data, offsets = b'', []
|
||||
for s in strings:
|
||||
offsets.append(len(data))
|
||||
if utf8:
|
||||
b = s.encode()
|
||||
data += bytes([len(s), len(b)]) + b + b'\0'
|
||||
else:
|
||||
data += struct.pack('<H', len(s)) + s.encode('utf-16-le') + b'\0\0'
|
||||
data += b'\0' * (-len(data) % 4)
|
||||
start = 28 + 4 * len(strings)
|
||||
body = struct.pack(f'<{len(strings)}I', *offsets) + data
|
||||
return struct.pack('<HHIIIIII', 1, 28, 28 + len(body), len(strings), 0, 0x100 if utf8 else 0, start, 0) + body
|
||||
|
||||
|
||||
def manifest(package, label_ref, version_ref, min_sdk, package_raw=True, foreign_label=False):
|
||||
"""<manifest package versionName><uses-sdk minSdkVersion/><application label icon/></manifest>.
|
||||
|
||||
package_raw=False drops the package's raw string (as some repackers do);
|
||||
foreign_label adds a non-android `label` attribute after android:label.
|
||||
"""
|
||||
strings = ['label', 'icon', 'versionName', 'minSdkVersion', 'package', 'manifest', 'uses-sdk',
|
||||
'application', package, 'junk', 'label', 'versionCode'] # the second 'label' has no android id
|
||||
resmap = struct.pack('<4I', 0x01010001, 0x01010002, 0x0101021c, 0x0101020c)
|
||||
resmap = struct.pack('<HHI', 0x0180, 8, 8 + len(resmap)) + resmap
|
||||
|
||||
def element(name, attrs):
|
||||
body = struct.pack('<IIHHHHHH', 0xffffffff, name, 20, 20, len(attrs), 0, 0, 0)
|
||||
for aname, raw, dtype, value in attrs:
|
||||
body += struct.pack('<IIIHBBI', 0xffffffff, aname, raw, 8, 0, dtype, value)
|
||||
return struct.pack('<HHIII', 0x0102, 16, 16 + len(body), 1, 0xffffffff) + body
|
||||
|
||||
none = 0xffffffff
|
||||
chunks = (pool(strings) + resmap
|
||||
+ element(5, [(4, 8 if package_raw else none, frame_apk.T_STRING, 8),
|
||||
(2, none, frame_apk.T_REF, version_ref),
|
||||
(11, none, frame_apk.T_INT_DEC, 210)])
|
||||
+ element(6, [(3, none, frame_apk.T_INT_DEC, min_sdk)])
|
||||
+ element(7, [(0, none, frame_apk.T_REF, label_ref), (1, none, frame_apk.T_REF, 0x7f020000)]
|
||||
+ ([(10, 9, frame_apk.T_STRING, 9)] if foreign_label else [])))
|
||||
return struct.pack('<HHI', 3, 8, 8 + len(chunks)) + chunks
|
||||
|
||||
|
||||
def resources(values):
|
||||
"""resources.arsc with package 0x7f; values: {(type id, entry, language, density): global string index}."""
|
||||
strings = ['French label', 'App label', '2.1', 'res/icon_lo.png', 'res/icon_hi.png', 'res/icon.xml']
|
||||
pkg_body = b''
|
||||
for (tid, lang, density), entries in values.items():
|
||||
cfg = struct.pack('<I4x2s4xH', 64, lang.encode().ljust(2, b'\0'), density).ljust(64, b'\0')
|
||||
count = max(entries) + 1
|
||||
offsets, data = [], b''
|
||||
for i in range(count):
|
||||
if i in entries:
|
||||
offsets.append(len(data))
|
||||
data += struct.pack('<HHI', 8, 0, 0) + struct.pack('<HBBI', 8, 0, frame_apk.T_STRING, entries[i])
|
||||
else:
|
||||
offsets.append(0xffffffff)
|
||||
header = 20 + 64
|
||||
estart = header + 4 * count
|
||||
body = struct.pack(f'<{count}I', *offsets) + data
|
||||
pkg_body += struct.pack('<HHIBBHII', 0x0201, header, header + len(body), tid, 0, 0, count, estart) + cfg + body
|
||||
pkg_header = struct.pack('<HHII', 0x0200, 288, 288 + len(pkg_body), 0x7f).ljust(288, b'\0')
|
||||
pkg = pkg_header + pkg_body
|
||||
table = pool(strings, utf8=True) + pkg
|
||||
return struct.pack('<HHII', 2, 12, 12 + len(table), 1) + table
|
||||
|
||||
|
||||
def apk(files):
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w') as z:
|
||||
for name, data in files.items():
|
||||
z.writestr(name, data)
|
||||
return buf.getvalue()
|
||||
from apk_fixtures import apk, manifest, pool, resources # noqa: E402,F401
|
||||
|
||||
|
||||
class ApkInfo(unittest.TestCase):
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Local-only VR manifest, raw ZIP and independent signature checks."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
@@ -15,7 +16,7 @@ import frame_apk
|
||||
import frame_apk_vr as vr
|
||||
import frame_apk_sign as signing
|
||||
import frame_android
|
||||
from test_frame_apk import pool
|
||||
from apk_fixtures import pool
|
||||
|
||||
|
||||
DEFAULT = 'android.intent.category.DEFAULT'
|
||||
|
||||
@@ -6,6 +6,7 @@ identity line, then starts TLS as the *client*.
|
||||
|
||||
Run: python3 -m unittest discover -s tests
|
||||
"""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Every public sideload entry point reaches the mandatory five-slot writer."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
|
||||
@@ -7,6 +7,7 @@ input reaching the source, and closing.
|
||||
|
||||
Run: python3 -m unittest discover -s tests
|
||||
"""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import base64
|
||||
import http.client
|
||||
import json
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Owned media planning, eye isolation, decoder choice and fake-Frame ownership."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import argparse
|
||||
import io
|
||||
import json
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Fake-Frame panel responses and the headset page's real HTTP guards."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import http.client
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT / "ui"))
|
||||
import frame_connect # noqa: E402
|
||||
import frame_devices # noqa: E402
|
||||
import frame_host # noqa: E402
|
||||
|
||||
|
||||
def inside(path, folder):
|
||||
path, folder = os.path.realpath(path), os.path.realpath(folder)
|
||||
return path == folder or path.startswith(folder + os.sep)
|
||||
|
||||
|
||||
def snapshot(folder):
|
||||
"""{relative path: (size, mtime)} for everything under FOLDER."""
|
||||
out = {}
|
||||
for base, dirs, files in os.walk(folder):
|
||||
for name in dirs + files:
|
||||
p = os.path.join(base, name)
|
||||
try:
|
||||
st = os.lstat(p)
|
||||
except OSError:
|
||||
continue
|
||||
out[os.path.relpath(p, folder)] = (st.st_size, st.st_mtime_ns)
|
||||
return out
|
||||
|
||||
|
||||
class Sandbox(unittest.TestCase):
|
||||
"""Tests once wrote a host-key pin into the real ~/.ssh/frame-control-hosts on every
|
||||
run (a server started without FRAME_CONTROL_SSH_DIR imported the real ~/.ssh/config)."""
|
||||
|
||||
def test_home_and_ssh_dir_are_throwaway(self):
|
||||
tmp = tempfile.gettempdir()
|
||||
self.assertTrue(inside(Path.home(), tmp), Path.home())
|
||||
self.assertTrue(inside(os.path.expanduser("~"), tmp))
|
||||
for d in (frame_devices.ssh_dir(), frame_devices.pin_dir(), frame_connect.SSH_DIR,
|
||||
os.environ["FRAME_CONTROL_SSH_DIR"], frame_host.cache_dir(), frame_host.data_dir()):
|
||||
self.assertTrue(inside(d, tmp), d)
|
||||
self.assertFalse(inside(d, sandbox.REAL_SSH_DIR), d)
|
||||
|
||||
def test_cache_folders_on_every_platform_are_throwaway(self):
|
||||
"""cache_dir() reads LOCALAPPDATA on Windows and XDG_CACHE_HOME on Linux, which a
|
||||
moved HOME alone doesn't cover: servers sweep staging folders there."""
|
||||
tmp = tempfile.gettempdir()
|
||||
for platform in ("MAC", "WINDOWS", None):
|
||||
with mock.patch.object(frame_host, "MAC", platform == "MAC"), \
|
||||
mock.patch.object(frame_host, "WINDOWS", platform == "WINDOWS"):
|
||||
self.assertTrue(inside(frame_host.cache_dir("apk-sources"), tmp), platform)
|
||||
for name in ("LOCALAPPDATA", "APPDATA", "XDG_CACHE_HOME", "XDG_DATA_HOME", "XDG_STATE_HOME"):
|
||||
self.assertTrue(inside(os.environ[name], tmp), name)
|
||||
|
||||
def test_end_to_end_tests_stay_out_of_it(self):
|
||||
"""tests/e2e keeps the container's real home (OpenSSH reads ~/.ssh from the password
|
||||
database, not $HOME), so nothing there may import sandbox, even through a test_* module."""
|
||||
here = Path(__file__).resolve().parent
|
||||
pattern = re.compile(r"^\s*(import sandbox|from sandbox |import test_|from test_)", re.M)
|
||||
for f in [*(here / "e2e").glob("*.py"), *(here / "smoke").glob("*.py"), here / "apk_fixtures.py"]:
|
||||
self.assertIsNone(pattern.search(f.read_text(encoding="utf-8")), f)
|
||||
|
||||
def test_end_to_end_tests_start_no_server_outside_their_container(self):
|
||||
"""Plain discovery over tests/e2e, with or without FRAME_E2E=1, must skip everything
|
||||
and never start ui/server.py: outside scripts/e2e.sh's container it would run on
|
||||
this computer's real app data, ~/.ssh and cache."""
|
||||
e2e = str(Path(__file__).resolve().parent / "e2e")
|
||||
code = ("import io, sys, unittest; sys.path.insert(0, sys.argv[1]); import harness; calls = []\n"
|
||||
"harness.Server.start = classmethod(lambda cls: calls.append(cls))\n"
|
||||
"suite = unittest.defaultTestLoader.discover(sys.argv[1], top_level_dir=sys.argv[1])\n"
|
||||
"r = unittest.TextTestRunner(stream=io.StringIO()).run(suite)\n"
|
||||
"print(len(calls), r.testsRun, len(r.skipped), len(r.errors) + len(r.failures))")
|
||||
for extra in ({}, {"FRAME_E2E": "1"}):
|
||||
env = {k: v for k, v in os.environ.items() if not k.startswith("FRAME_E2E")}
|
||||
env.update(extra, PYTHONDONTWRITEBYTECODE="1")
|
||||
out = subprocess.run([sys.executable, "-c", code, e2e], env=env, capture_output=True, text=True,
|
||||
timeout=120)
|
||||
self.assertEqual(out.returncode, 0, out.stderr[-2000:])
|
||||
started, ran, skipped, broken = map(int, out.stdout.split())
|
||||
self.assertEqual((started, broken), (0, 0), (extra, out.stdout))
|
||||
self.assertEqual(ran, skipped, (extra, out.stdout))
|
||||
self.assertGreater(skipped, 0, extra)
|
||||
|
||||
def test_child_processes_inherit_it(self):
|
||||
out = subprocess.run(
|
||||
[sys.executable, "-c", "import sys; sys.path.insert(0, sys.argv[1]); import frame_devices, frame_connect;"
|
||||
" from pathlib import Path; print(Path.home()); print(frame_devices.ssh_dir()); print(frame_connect.SSH_DIR)",
|
||||
str(ROOT / "ui")], env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
|
||||
capture_output=True, text=True, check=True).stdout.split("\n")
|
||||
for d in filter(None, out):
|
||||
self.assertFalse(inside(d, sandbox.REAL_SSH_DIR), d)
|
||||
self.assertTrue(inside(d, tempfile.gettempdir()), d)
|
||||
|
||||
def test_real_ssh_folder_untouched_by_tests_that_start_servers(self):
|
||||
"""Reruns, from the environment the suite was started in, the tests that used to
|
||||
leak (servers started without FRAME_CONTROL_SSH_DIR) plus the device and connect
|
||||
tests, and checks nothing under the real ~/.ssh changed."""
|
||||
real = sandbox.REAL_SSH_DIR
|
||||
if inside(real, tempfile.gettempdir()):
|
||||
self.skipTest("this run's HOME is already a temporary folder")
|
||||
before = snapshot(real)
|
||||
env = {k: v for k, v in sandbox.ORIGINAL_ENV.items() if k != "FRAME_CONTROL_SSH_DIR"}
|
||||
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
||||
r = subprocess.run([sys.executable, "-m", "unittest", "-q", "test_server.OneServer", "test_server.LocalMode",
|
||||
"test_devices", "test_connect"],
|
||||
cwd=str(Path(__file__).resolve().parent), env=env, capture_output=True, text=True,
|
||||
timeout=300)
|
||||
self.assertEqual(r.returncode, 0, r.stderr[-3000:])
|
||||
after = snapshot(real)
|
||||
changed = sorted(k for k in set(before) | set(after) if before.get(k) != after.get(k)
|
||||
and not k.startswith("agent")) # ssh-agent sockets come and go on their own
|
||||
self.assertEqual(changed, [], f"tests changed the real {real}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,3 +1,4 @@
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
Run: python3 -m unittest discover -s tests
|
||||
"""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Own VR telemetry, optional catalogue and entitlement guards; no headset needed."""
|
||||
import sandbox # noqa: F401 (first: keeps tests off real data and services)
|
||||
import ctypes
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
+2
-1
@@ -31,7 +31,8 @@ import frame_host # noqa: E402
|
||||
FRAME_USER = os.environ.get("FRAME_USER", "steamos")
|
||||
USER_FROM_ENV = "FRAME_USER" in os.environ
|
||||
FRAME_ALIAS = os.environ.get("FRAME_ALIAS", "frame")
|
||||
SSH_DIR = Path.home() / ".ssh"
|
||||
# $FRAME_CONTROL_SSH_DIR in tests, as frame_devices.ssh_dir(), so they never touch the real one.
|
||||
SSH_DIR = Path(os.environ.get("FRAME_CONTROL_SSH_DIR") or Path.home() / ".ssh")
|
||||
KEY = SSH_DIR / "id_ed25519_frame"
|
||||
# The devkit service only accepts ssh-rsa keys (write_key in Valve's
|
||||
# steamos-devkit-service), so pairing uses a second key next to the ed25519 one.
|
||||
|
||||
Reference in new issue
Block a user