Merge origin/main into apk-store-fixes, preserving store and headset features

Keep the union of server routes, desktop resources and responsive controls. Preserve OpenXR install defaults and telemetry hooks alongside library artwork. Adapt the resource test to single-file entries and avoid a completed-refresh race in the F-Droid test.

Co-Authored-By: GPT-6 Astra <noreply@openai.com>
This commit is contained in:
saphidandGPT-6 Astra committed 2026-09-29 12:07:09 +10:00
commit 8fca0fe0a2
169 files changed
+47085 -292

No files matched your search

+43
View File
@@ -0,0 +1,43 @@
// Run the actual page script with a tiny DOM/fetch fixture; no browser dependency.
const fs = require('node:fs');
const vm = require('node:vm');
const assert = require('node:assert/strict');
const elements = new Map();
const events = new Map();
const requests = [];
const element = id => {
if (!elements.has(id)) elements.set(id, {value:'', checked:false, disabled:false, textContent:'',
addEventListener(){}, reset(){}});
return elements.get(id);
};
const context = {
document:{getElementById:element}, location:{hash:''}, URLSearchParams,
window:{addEventListener:(name, fn) => events.set(name, fn)},
fetch:(path, options) => new Promise(resolve => requests.push({path, options, resolve})),
};
const html = fs.readFileSync(process.argv[2], 'utf8');
vm.runInNewContext(html.match(/<script>([\s\S]*?)<\/script>/)[1].replace('__FRAME_KEY__', '"test"'), context);
const answer = (index, data) => requests[index].resolve({ok:true,json:async () => data});
(async () => {
context.location.hash = '#confirm=first';
const first = events.get('hashchange')();
context.location.hash = '#confirm=second';
const second = events.get('hashchange')();
answer(1, {action:{name:'second'},approved:false});
await second;
answer(0, {action:{name:'first'},approved:false});
await first;
assert.match(element('action').textContent, /second/);
assert.doesNotMatch(element('action').textContent, /first/);
const approved = element('approve').onclick();
assert.equal(JSON.parse(requests[2].options.body).confirmation, 'second');
context.location.hash = '#confirm=third';
const third = events.get('hashchange')();
answer(3, {action:{name:'third'},approved:false});
await third;
answer(2, {message:'Approved for one use'});
await approved;
assert.equal(element('approval-status').textContent, '');
assert.match(element('action').textContent, /third/);
console.log('Approval navigation races: pass');
})().catch(error => { console.error(error); process.exitCode=1; });
+46
View File
@@ -0,0 +1,46 @@
"""Real HTTP/MCP adapter against fake-Frame SSH; no model service needed."""
import json
from pathlib import Path
import sys
import harness
from harness import api, ok, finished, ssh
sys.path.insert(0, str(harness.ROOT / 'ui'))
import frame_mcp
class Agents(harness.FrameTestCase):
def client(self):
return frame_mcp.Client('http://127.0.0.1:%d' % harness.Server.port)
def call(self, name, args):
return json.loads(frame_mcp.call(self.client(), name, args)['content'][0]['text'])
def approve(self, proposal):
ok('POST', '/api/agent/approval', {'confirmation': proposal['confirmation'], 'accept': True})
return proposal['confirmation']
def test_status_and_approved_install_job(self):
self.assertIn('battery', self.call('status', {}))
proposal = self.call('install', {'id': 'org.example.AgentTest'})
before = api('POST', '/api/agent/call', {'name': 'install', 'arguments': {'id': 'org.example.AgentTest'}, 'confirmation': proposal['confirmation']})
self.assertEqual(before[0], 400)
token = self.approve(proposal)
job = self.call('install', {'id': 'org.example.AgentTest', 'confirmation': token})
self.assertFalse(finished(job).get('error'))
self.assertIn('org.example.AgentTest', ssh('flatpak list --app --columns=application'))
denied = api('POST', '/api/agent/call', {'name': 'install', 'arguments': {'id': 'org.example.AgentTest'}, 'confirmation': token})
self.assertEqual(denied[0], 400)
def test_approved_file_and_text(self):
path = Path(self.path('agent-note.txt'))
path.write_text('MCP file content\n')
args = {'path': str(path)}
token = self.approve(self.call('send_file', args))
self.call('send_file', {**args, 'confirmation': token})
self.assertEqual(ssh('cat ~/Downloads/agent-note.txt'), path.read_text())
args = {'text': 'MCP clipboard text'}
token = self.approve(self.call('send_text', args))
self.call('send_text', {**args, 'confirmation': token})
self.assertEqual(harness.state()['clipboard'], ['MCP clipboard text'])
+21
View File
@@ -0,0 +1,21 @@
"""Media transfer through the real HTTP/SSH path; the fake has no VR renderer."""
import harness
from harness import ok, ssh
harness.require()
class Media(harness.FrameTestCase):
def test_upload_and_list_without_launching_a_viewer(self):
# The transfer endpoint doesn't decode. Rendering belongs to the real
# headset smoke checks documented in docs/vr-video.md.
self.assertEqual(ssh('stat -c "%U:%G %a" ~/.local/share').strip(), 'steamos:steamos 755')
result = ok('POST', '/api/upload', raw=b'fake-media', headers={
'X-Mode': 'media', 'X-Filename': 'test_SBS.png'})
identity = result['id']
try:
files = ok('POST', '/api/media', {'action': 'list'})['files']
self.assertIn(identity, [f['id'] for f in files])
self.assertEqual(ssh('cat ~/Videos/FrameControl/'+identity), 'fake-media')
finally:
ssh('rm -rf ~/Videos/FrameControl/'+identity.split('/')[0])
@@ -231,6 +231,13 @@ def sysfs(state):
def runtimes(state):
"""Installed compat tools as Steam app manifests, and the Lepton launcher itself."""
# Verified 2026-09-28, BUILD_ID 20260925.6191901: both parents are
# steamos:steamos 0755. The root supervisor must not leave them root-owned
# when creating Steam's fake manifests; user-account app installs need them.
for directory in (HOME + '/.local', HOME + '/.local/share'):
os.makedirs(directory, mode=0o755, exist_ok=True)
chown(directory)
os.chmod(directory, 0o755)
apps = fs.STEAM_ROOT + '/steamapps'
for alias, installed in state['runtimes'].items():
acf = f'{apps}/appmanifest_{fs.RUNTIME_APPIDS[alias]}.acf'
+244
View File
@@ -0,0 +1,244 @@
// Control on the live view (tap, drag, hold, scroll, type), run in node against the real
// functions from ui/index.html with a fake canvas and a fake server.
import { readFileSync } from "fs";
const src = readFileSync(new URL("../../ui/index.html", import.meta.url), "utf8");
const grab = name => {
const one = src.match(new RegExp(`\\n((?:async )?function ${name}\\(.*\\}\\n)`)); // one-line function
if (one) return one[1];
const m = src.match(new RegExp(`(?:\\nconst ${name} = [^\\n]*\\n)|((?:async )?function ${name}\\([\\s\\S]*?\\n}\\n)`));
if (!m) throw new Error("not found: " + name);
return m[0];
};
const NAMES = ["panelKey", "ctrlAimedAt", "ctrlSameTarget", "isMoveEvent", "isRelease", "ctrlKeepable", "TAP_MOVE", "ctrlAim", "ctrlKeyEvent", "ctrlTouchCancel", "ctrlSend", "ctrlFlush", "ctrlMoveTo", "ctrlMoveBy", "ctrlSchedule",
"ctrlFlushMoves", "ctrlButton", "ctrlClick", "ctrlRelease", "ctrlFraction", "ctrlTouchDown", "centroid",
"ctrlTouchMove", "ctrlTouchUp", "ctrlTap", "ctrlText"];
const code = NAMES.map(grab).join("");
const fail = msg => { console.log("FAIL " + msg); process.exit(1); };
const tick = (ms = 0) => new Promise(r => setTimeout(r, ms));
function page({ mode = "abs", rect = { left: 0, top: 0, width: 640, height: 360 }, api } = {}) {
const target = { panel: { window: 42, display: ":1" } };
const ctrl = { on: true, queue: [], sending: false, state: "ready", message: "", move: null, rel: [0, 0], raf: 0,
held: new Set(), keys: new Set(), pointers: new Map(), g: null, retry: null };
const sent = [];
const canvas = { width: 1280, height: 720, getBoundingClientRect: () => rect };
const env = {
ctrl, $: () => canvas, ctrlMode: () => mode, ctrlShow: () => {}, toast: () => {},
ctrlTarget: () => (mode !== "abs" ? { ok: true } : target.panel ? { ok: true, panel: target.panel } : { why: "gone" }),
api: api || (async (path, body) => { sent.push(...body.events); return { state: "ready", sent: true }; }),
requestAnimationFrame: cb => { setTimeout(cb, 0); return 1; },
navigator: {},
};
const fns = new Function(...Object.keys(env), `let ctrlWarned = false;\n${code}
return { ctrlTouchDown, ctrlTouchMove, ctrlTouchUp, ctrlTouchCancel, ctrlSend, ctrlText, ctrlButton, ctrlKeyEvent, ctrlRelease, ctrlFlushMoves };`)(...Object.values(env));
const at = (id, x, y) => ({ pointerId: id, clientX: x, clientY: y });
return { ctrl, sent, target, ...fns, at };
}
// A tap lands where it was tapped: pointer there first, then the click.
{
const p = page();
p.ctrlTouchDown(p.at(1, 320, 90)); p.ctrlTouchUp(p.at(1, 320, 90));
await tick(10);
const [move, down, up] = p.sent;
if (!(move.fx === 0.5 && Math.abs(move.fy - 0.25) < 1e-9 && move.window === 42 && move.display === ":1")) fail("tap position " + JSON.stringify(move));
if (!(down.window === 42 && down.display === ":1")) fail("a press names its panel " + JSON.stringify(down));
if (!(down.button === "left" && down.down && up.button === "left" && up.down === false && p.sent.length === 3))
fail("tap click " + JSON.stringify(p.sent));
}
// The panel is letterboxed in a taller view: taps map inside it, taps on the bars do nothing.
{
const p = page({ rect: { left: 0, top: 0, width: 640, height: 480 } }); // 640x360 picture, 60px bars
p.ctrlTouchDown(p.at(1, 320, 150)); p.ctrlTouchUp(p.at(1, 320, 150));
p.ctrlTouchDown(p.at(2, 320, 10)); p.ctrlTouchUp(p.at(2, 320, 10));
await tick(10);
if (!(p.sent.length === 3 && Math.abs(p.sent[0].fy - 0.25) < 1e-9)) fail("letterbox " + JSON.stringify(p.sent));
}
// A drag is a mouse drag: down where it started, moves, up at the end.
{
const p = page();
p.ctrlTouchDown(p.at(1, 100, 100));
p.ctrlTouchMove(p.at(1, 140, 100)); await tick(5);
p.ctrlTouchMove(p.at(1, 200, 120)); await tick(5);
p.ctrlTouchUp(p.at(1, 200, 120)); await tick(10);
const kinds = p.sent.map(e => "fx" in e ? "move" : `${e.button}-${e.down ? "down" : "up"}`);
if (!(kinds[0] === "move" && kinds[1] === "left-down" && kinds.at(-1) === "left-up" && kinds.includes("move", 2)))
fail("drag " + kinds.join(","));
if (Math.abs(p.sent[0].fx - 100 / 640) > 1e-9) fail("drag starts where the finger went down");
if (p.ctrl.held.size) fail("drag left the button held");
}
// Press and hold is a right-click where the finger is.
{
const p = page();
p.ctrlTouchDown(p.at(1, 64, 36));
await tick(620);
p.ctrlTouchUp(p.at(1, 64, 36)); await tick(10);
const buttons = p.sent.filter(e => "button" in e).map(e => `${e.button}-${e.down}`);
if (buttons.join() !== "right-true,right-false") fail("hold " + buttons.join());
if (!(p.sent[0].fx === 0.1)) fail("hold position " + JSON.stringify(p.sent[0]));
}
// Two fingers scroll, and the content follows them (fingers up scrolls down).
{
const p = page();
p.ctrlTouchDown(p.at(1, 100, 200)); p.ctrlTouchDown(p.at(2, 200, 200));
p.ctrlTouchMove(p.at(1, 100, 180)); p.ctrlTouchMove(p.at(2, 200, 180));
p.ctrlTouchUp(p.at(1, 100, 180)); p.ctrlTouchUp(p.at(2, 200, 180));
await tick(10);
const dy = p.sent.filter(e => e.scroll).reduce((a, e) => a + e.scroll[1], 0);
if (!(dy === 40 && !p.sent.some(e => "button" in e))) fail("scroll " + JSON.stringify(p.sent));
}
// On the headset view it's a trackpad: drags move the pointer, taps click where it is.
{
const p = page({ mode: "rel" });
p.ctrlTouchDown(p.at(1, 100, 100));
p.ctrlTouchMove(p.at(1, 110, 100)); p.ctrlTouchMove(p.at(1, 120, 105));
p.ctrlTouchUp(p.at(1, 120, 105)); await tick(10);
p.ctrlTouchDown(p.at(1, 50, 50)); p.ctrlTouchUp(p.at(1, 50, 50)); await tick(10);
const moved = p.sent.filter(e => "dx" in e).reduce((a, e) => [a[0] + e.dx, a[1] + e.dy], [0, 0]);
if (!(moved[0] === 32 && moved[1] === 8)) fail("trackpad move " + JSON.stringify(moved));
if (p.sent.some(e => "fx" in e)) fail("trackpad sent an absolute position");
if (p.sent.filter(e => e.button === "left").length !== 2) fail("trackpad tap " + JSON.stringify(p.sent));
}
// Text: plain ASCII only, in chunks the server takes.
{
const p = page();
p.ctrlText("héllo " + "x".repeat(600));
await tick(10);
const text = p.sent.map(e => e.text).join("");
if (!(text === "hllo " + "x".repeat(600) && p.sent.every(e => e.text.length <= 500))) fail("text " + text.length);
}
// Not connected yet: clicks and keys wait with their position. The request fails: only releases wait.
{
let calls = 0;
const p = page({ api: async () => { calls++; if (calls === 1) return { state: "starting", sent: false }; throw new Error("offline"); } });
p.ctrlSend([{ fx: 0.1, fy: 0.1, window: 42 }, { button: "left", down: true }, { key: 30, down: true }]);
await tick(5);
const q = p.ctrl.queue;
if (!(q.length === 3 && "fx" in q[0] && q[1].button === "left" && q[2].key === 30)) fail("kept while starting " + JSON.stringify(q));
clearTimeout(p.ctrl.retry);
p.ctrl.queue = []; p.ctrl.retryAt = 0;
p.ctrlSend([{ button: "left", down: false }, { key: 31, down: true }, { dx: 3, dy: 1 }]);
await tick(5);
if (!(p.ctrl.queue.length === 1 && p.ctrl.queue[0].button === "left" && p.ctrl.queue[0].down === false))
fail("release kept on failure " + JSON.stringify(p.ctrl.queue));
if (calls !== 2) fail("retried in a tight loop: " + calls + " requests");
clearTimeout(p.ctrl.retry);
}
// Turning Control off (or the view losing focus) lets go of anything held.
{
const p = page();
p.ctrlButton("left", true);
p.ctrlRelease(); await tick(10);
if (!(p.sent.at(-1).button === "left" && p.sent.at(-1).down === false && !p.ctrl.held.size)) fail("release " + JSON.stringify(p.sent));
}
// Keys held on the Frame are let go too.
{
const p = page();
p.ctrlKeyEvent(42, true);
p.ctrlRelease(); await tick(10);
if (!(p.sent.at(-1).key === 42 && p.sent.at(-1).down === false && !p.ctrl.keys.size)) fail("key release " + JSON.stringify(p.sent));
}
// A long queue sheds moves and old scrolls, never a release.
{
const p = page({ api: () => new Promise(() => {}) }); // stuck request
p.ctrlSend([{ dx: 1, dy: 1 }]);
p.ctrlSend([{ button: "left", down: false }]);
for (let i = 0; i < 320; i++) p.ctrlSend([{ scroll: [0, 1] }]);
if (!p.ctrl.queue.some(e => e.button === "left" && e.down === false)) fail("trim dropped a release");
if (p.ctrl.queue.length > 300) fail("trim kept " + p.ctrl.queue.length);
}
// Broken on the Frame side: no hammering, and only releases wait.
{
let calls = 0;
const p = page({ api: async () => { calls++; return { state: "error", sent: false }; } });
p.ctrlSend([{ button: "left", down: true }, { button: "left", down: false }]);
await tick(50);
if (calls !== 1) fail("error response reposted " + calls + " times");
if (!(p.ctrl.queue.length === 1 && p.ctrl.queue[0].down === false)) fail("error kept " + JSON.stringify(p.ctrl.queue));
clearTimeout(p.ctrl.retry);
}
// Connecting: a tap keeps its position, so it lands where it was made.
{
const p = page({ api: async () => ({ state: "starting", sent: false }) });
p.ctrlTouchDown(p.at(1, 320, 90)); p.ctrlTouchUp(p.at(1, 320, 90));
await tick(10);
const q = p.ctrl.queue;
if (!("fx" in q[0] && q[1].button === "left")) fail("connecting tap " + JSON.stringify(q));
clearTimeout(p.ctrl.retry);
}
// Lifting one of two scrolling fingers doesn't jump the scroll or start a drag.
{
const p = page();
p.ctrlTouchDown(p.at(1, 100, 200)); p.ctrlTouchDown(p.at(2, 300, 200));
p.ctrlTouchUp(p.at(1, 100, 200));
p.ctrlTouchMove(p.at(2, 300, 199));
p.ctrlTouchUp(p.at(2, 300, 199)); await tick(10);
if (p.sent.length) fail("one finger left after scrolling " + JSON.stringify(p.sent));
}
// A cancelled touch isn't a tap.
{
const p = page();
p.ctrlTouchDown(p.at(1, 100, 100)); p.ctrlTouchCancel(p.at(1, 100, 100)); await tick(10);
if (p.sent.length) fail("cancel clicked " + JSON.stringify(p.sent));
}
// Press and hold on the bars around the picture does nothing.
{
const p = page({ rect: { left: 0, top: 0, width: 640, height: 480 } });
p.ctrlTouchDown(p.at(1, 320, 10)); await tick(620); p.ctrlTouchUp(p.at(1, 320, 10)); await tick(10);
if (p.sent.length) fail("hold on the bars " + JSON.stringify(p.sent));
}
// Taps only reach the panel in use, and only once its picture is the one on screen.
{
const fns = ["panelKey", "deskPanel", "ctrlTarget"].map(grab).join("");
const check = (desk, live = true) => new Function("desk", "live", "ctrlMode", fns + "; return ctrlTarget();")(desk, live, () => "abs");
const a = { display: ":1", window: 5 }, b = { display: ":0", window: 5 };
const base = { panels: [a, b], loaded: true, pick: "", focus: ":1/5", shown: ":1/5" };
if (!check(base).ok) fail("target: shown and in use");
if (check({ ...base, shown: ":0/5" }).ok) fail("target: the picture is another panel with the same id");
if (!/Capture or Live/.test(check({ ...base, shown: null }, false).why)) fail("target: stale picture message");
if (check({ ...base, pick: ":0/5", shown: ":0/5" }).ok) fail("target: a watched panel that isn't in use");
if (check({ ...base, focus: null }).ok) fail("target: nothing in use");
}
// Focus moves to another panel mid-gesture: the tap or hold does nothing.
{
const p = page();
p.ctrlTouchDown(p.at(1, 100, 100));
p.target.panel = { window: 43, display: ":1" };
p.ctrlTouchUp(p.at(1, 100, 100)); await tick(10);
p.target.panel = { window: 42, display: ":1" };
p.ctrlTouchDown(p.at(1, 100, 100));
p.target.panel = null;
await tick(620); p.ctrlTouchUp(p.at(1, 100, 100)); await tick(10);
if (p.sent.some(e => "button" in e)) fail("gesture outlived its panel " + JSON.stringify(p.sent));
}
// Trimming keeps a click together with its position.
{
const p = page({ api: () => new Promise(() => {}) });
p.ctrlSend([{ dx: 1, dy: 0 }]); // in flight forever
for (let i = 0; i < 100; i++) p.ctrlSend([{ scroll: [0, 1] }]);
p.ctrlSend([{ fx: 0.5, fy: 0.5, window: 42, display: ":1" }, { button: "left", down: true }, { button: "left", down: false }]);
for (let i = 0; i < 198; i++) p.ctrlSend([{ scroll: [0, 1] }]);
const q = p.ctrl.queue, i = q.findIndex(e => e.button === "left" && e.down);
if (i < 1 || !("fx" in q[i - 1])) fail("trim split a click from its position");
}
// Backing off holds for new input too.
{
let calls = 0;
const p = page({ api: async () => { calls++; return { state: "error", sent: false }; } });
p.ctrlSend([{ button: "left", down: false }]);
await tick(5);
for (let i = 0; i < 10; i++) { p.ctrlSend([{ key: 30, down: false }]); await tick(2); }
if (calls !== 1) fail("new input bypassed the backoff: " + calls + " requests");
clearTimeout(p.ctrl.retry);
}
// A long paste goes in several requests, so a release never waits behind all of it.
{
const batches = [];
const p = page({ api: async (path, body) => { batches.push(body.events); return { state: "ready", sent: true }; } });
p.ctrlText("y".repeat(450));
p.ctrlButton("left", false);
await tick(30);
if (!batches.every(b => b.reduce((a, e) => a + (e.text?.length || 0), 0) <= 100)) fail("a batch carried too much text");
if (batches.length < 5) fail("paste went in " + batches.length + " requests");
}
console.log("control gestures ok");
+48
View File
@@ -0,0 +1,48 @@
// The page's input queue, run in node against the real functions from ui/index.html.
import { readFileSync } from "fs";
const src = readFileSync(new URL("../../ui/index.html", import.meta.url), "utf8");
const grab = name => src.match(new RegExp(`(?:const ${name} = [^\\n]*\\n)|((?:async )?function ${name}\\([\\s\\S]*?\\n}\\n)`))[0];
const code = ["isMove", "padSend", "padKeep", "padFlush"].map(grab).join("");
const fail = msg => { console.log("FAIL " + msg); process.exit(1); };
function page(api) {
const pad = { state: "ready", queue: [], sending: false };
const shown = [];
const fns = new Function("pad", "api", "padShow", code + "; return { padSend, padFlush };")(pad, api, s => { shown.push(s); pad.state = s.state; });
return { pad, shown, ...fns };
}
// While a request is in flight, events queue by these rules.
{
const { pad, padSend } = page(() => new Promise(() => {})); // a request that never returns
const send = (state, e) => { pad.state = state; padSend(e); };
send("ready", { singlehold: true }); // goes out, and stays in flight
send("error", { dx: 3 }); // dropped: a stale move while broken
send("error", { singlerelease: true }); // kept
send("off", { singlerelease: true }); // off: nothing was held
send("pairing", { key: "a" }); // kept while reconnecting
send("pairing", { dx: 1 }); // dropped
for (let i = 0; i < 250; i++) send("pairing", { key: "x" });
send("pairing", { singlerelease: true }); // kept past the 200 cap
const q = pad.queue;
if (!(q[0].singlerelease && q[1].key === "a" && q.at(-1).singlerelease && !q.some(e => "dx" in e)
&& q.filter(e => e.singlerelease).length === 2)) fail("queueing " + JSON.stringify(q.slice(0, 3)) + " len " + q.length);
}
// A request that fails, or reaches a link that had dropped, keeps its keys and releases.
for (const [label, api] of [["request failed", async () => { throw new Error("offline"); }],
["not sent", async () => ({ state: "starting", sent: false })]]) {
const { pad, padSend, shown } = page(api);
padSend({ dx: 5 }); // sent and lost: a move, fine to drop
await new Promise(r => setTimeout(r, 0));
pad.state = "ready";
padSend({ singlerelease: true });
padSend({ dx: 7 }); // queued while that request is in flight: stale once it fails
padSend({ key: "b" });
await new Promise(r => setTimeout(r, 0));
const q = pad.queue;
if (!(q.some(e => e.singlerelease) && q.some(e => e.key === "b") && !q.some(e => "dx" in e)))
fail(label + ": " + JSON.stringify(q));
if (!shown.length || shown.at(-1).state === "ready") fail(label + ": state not shown");
}
console.log("queue rules ok");
+16
View File
@@ -0,0 +1,16 @@
"""Imported first by every test module: nothing a test does reaches this person's
app data, their telemetry, or the shared compatibility database.
Must run before any ui module is imported, since those read these at import time.
"""
import atexit
import os
import shutil
import tempfile
_dir = tempfile.mkdtemp(prefix="frame-control-tests-")
atexit.register(shutil.rmtree, _dir, ignore_errors=True)
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"
+253
View File
@@ -0,0 +1,253 @@
"""MCP protocol, exact-action approvals and explicit assistant data sharing."""
import io
import json
import os
import shutil
from pathlib import Path
import subprocess
import sys
import tempfile
import threading
import unittest
from unittest import mock
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
import frame_agent as agent
import frame_assistant as assistant
import frame_mcp as mcp
import server
class Approvals(unittest.TestCase):
def test_requires_human_decision_exact_action_and_single_use(self):
gate = agent.Approvals()
action = {'name': 'power', 'arguments': {'action': 'reboot'}}
token = gate.request(action)['confirmation']
with self.assertRaises(ValueError):
gate.consume(token, action)
gate.decide(token, True)
with self.assertRaises(ValueError):
gate.consume(token, {'name': 'power', 'arguments': {'action': 'poweroff'}})
gate.consume(token, action)
with self.assertRaises(ValueError):
gate.consume(token, action)
def test_expiry_rejection_and_non_boolean_approval(self):
gate = agent.Approvals()
token = gate.request({})['confirmation']
gate.decide(token, 'true')
with self.assertRaises(ValueError):
gate.inspect(token)
token = gate.request({})['confirmation']
with mock.patch.object(agent.time, 'monotonic', return_value=float('inf')):
with self.assertRaises(ValueError):
gate.decide(token, True)
def test_concurrent_consumption_executes_once(self):
gate = agent.Approvals()
token = gate.request({})['confirmation']
gate.decide(token, True)
results = []
def consume():
try:
gate.consume(token, {})
results.append(True)
except ValueError:
results.append(False)
threads = [threading.Thread(target=consume) for _ in range(8)]
for thread in threads: thread.start()
for thread in threads: thread.join()
self.assertEqual(results.count(True), 1)
def test_action_never_runs_before_approval(self):
with mock.patch.object(agent, 'approvals', agent.Approvals()), mock.patch.object(server, 'flatpak') as install:
body = {'name': 'install', 'arguments': {'id': 'org.example.App'}}
result = agent.call(server, body)
install.assert_not_called()
body['confirmation'] = result['confirmation']
with self.assertRaises(ValueError): agent.call(server, body)
agent.approvals.decide(body['confirmation'], True)
agent.call(server, body)
install.assert_called_once_with({'id': 'org.example.App', 'action': 'install'})
with self.assertRaises(ValueError): agent.call(server, body)
def test_file_content_change_invalidates_approval(self):
with tempfile.TemporaryDirectory() as tmp, mock.patch.object(agent, 'approvals', agent.Approvals()), mock.patch.object(server, 'push_file') as push:
path = Path(tmp) / 'note.txt'
path.write_text('first')
body = {'name': 'send_file', 'arguments': {'path': str(path)}}
result = agent.call(server, body)
agent.approvals.decide(result['confirmation'], True)
body['confirmation'] = result['confirmation']
path.write_text('second')
with self.assertRaises(ValueError): agent.call(server, body)
push.assert_not_called()
def test_no_arbitrary_commands_or_arguments(self):
for name, args in [('shell', {'command': 'true'}), ('panel', {'id': 'org.example.App', 'args': '--evil'}),
('power', {'action': 'factory-reset'}), ('send_text', {'text': ''})]:
with self.assertRaises(ValueError): agent.call(server, {'name': name, 'arguments': args})
class Assistant(unittest.TestCase):
def setUp(self):
self.received = []
owner = self
class Endpoint(BaseHTTPRequestHandler):
def log_message(self, *args): pass
def do_POST(self):
owner.received.append((dict(self.headers), json.loads(self.rfile.read(int(self.headers['Content-Length'])))))
if self.path == '/redirect':
self.send_response(302)
self.send_header('Location', '/other')
self.end_headers()
return
data = json.dumps({'choices': [{'message': {'content': '<script>not executed</script>'}}]}).encode()
self.send_response(200)
self.send_header('Content-Length', str(len(data)))
self.end_headers()
self.wfile.write(data)
self.httpd = ThreadingHTTPServer(('127.0.0.1', 0), Endpoint)
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
self.thread.start()
self.body = {'endpoint': 'http://127.0.0.1:%d/chat' % self.httpd.server_port, 'model': 'local', 'prompt': 'Hello', 'consent': True}
def tearDown(self):
self.httpd.shutdown()
self.httpd.server_close()
self.thread.join()
def test_no_opt_in_no_request_or_capture(self):
capture = mock.Mock()
for consent in (False, None, 'true', 1):
with self.assertRaises(ValueError): assistant.chat({**self.body, 'consent': consent, 'screenshot': True}, capture)
capture.assert_not_called()
self.assertEqual(self.received, [])
def test_text_only_keyless_and_optional_screenshot(self):
capture = mock.Mock(return_value=b'png')
self.assertIn('script', assistant.chat(self.body, capture)['reply'])
capture.assert_not_called()
headers, body = self.received[-1]
self.assertNotIn('Authorization', headers)
self.assertEqual(body['messages'], [{'role': 'user', 'content': 'Hello'}])
assistant.chat({**self.body, 'screenshot': True, 'key': 'test-key'}, capture)
capture.assert_called_once()
headers, body = self.received[-1]
self.assertEqual(headers['Authorization'], 'Bearer test-key')
self.assertEqual(body['messages'][0]['content'][1]['image_url']['url'], 'data:image/png;base64,cG5n')
def test_redirects_do_not_forward_context_or_credentials(self):
with self.assertRaises(ValueError):
assistant.chat({**self.body, 'endpoint': self.body['endpoint'].replace('/chat', '/redirect'), 'key': 'secret'}, mock.Mock())
self.assertEqual(len(self.received), 1)
def test_bad_urls_fail_before_capture(self):
for url in ('file:///etc/passwd', 'http://example.com/chat', 'https://user:pass@example.com', 'https://example.com?key=secret'):
capture = mock.Mock()
with self.assertRaises(ValueError): assistant.chat({**self.body, 'endpoint': url, 'screenshot': True}, capture)
capture.assert_not_called()
class AssistantPage(unittest.TestCase):
@unittest.skipUnless(shutil.which('node'), 'Node is required for the page script regression')
def test_approval_navigation_races(self):
root = Path(__file__).resolve().parents[1]
result = subprocess.run(['node', str(root / 'tests/assistant_ui.cjs'), str(root / 'ui/assistant.html')],
capture_output=True, text=True, timeout=10)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
class Protocol(unittest.TestCase):
def test_stdio_initialize_list_call_errors_and_eof(self):
messages = [
{'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {'protocolVersion': '2025-06-18'}},
{'jsonrpc': '2.0', 'method': 'notifications/initialized'},
{'jsonrpc': '2.0', 'id': 2, 'method': 'tools/list'},
{'jsonrpc': '2.0', 'id': 3, 'method': 'tools/call', 'params': {'name': 'shell'}},
{'jsonrpc': '2.0', 'id': 4, 'method': 'ping'},
]
result = subprocess.run([sys.executable, str(Path(mcp.__file__))], input='\n'.join(map(json.dumps, messages)) + '\n', text=True, capture_output=True, timeout=10)
self.assertEqual(result.returncode, 0, result.stderr)
replies = list(map(json.loads, result.stdout.splitlines()))
self.assertEqual([r['id'] for r in replies], [1, 2, 3, 4])
self.assertEqual(replies[0]['result']['protocolVersion'], '2025-06-18')
self.assertIn('screenshot', [t['name'] for t in replies[1]['result']['tools']])
self.assertTrue(replies[2]['result']['isError'])
def test_mcp_cannot_approve_and_returns_review_url(self):
client = mock.Mock(url='http://127.0.0.1:47810')
client.request.return_value = {'approvalPath': '/assistant#confirm=token'}
result = mcp.call(client, 'power', {'action': 'reboot'})
self.assertIn('http://127.0.0.1:47810/assistant', result['content'][0]['text'])
with self.assertRaises(ValueError): mcp.call(client, 'approve', {'confirmation': 'token'})
with self.assertRaises(ValueError): mcp.call(client, 'status', {'path': '/api/open'})
def test_loopback_only_backend(self):
for url in ('https://example.com', 'http://127.0.0.1/api', 'http://secret@localhost:1234', 'file:///tmp/x'):
with self.assertRaises(ValueError): mcp.Client(url)
class ManagedBackend(unittest.TestCase):
def test_private_backend_auth_and_cleanup(self):
from urllib.error import HTTPError, URLError
from urllib.request import urlopen
with mock.patch.dict(os.environ, {'FRAME_ALIAS': 'frame-control-test.invalid'}):
with mcp.backend() as client:
url = client.url
self.assertIn('os', client.request('/api/host'))
with self.assertRaises(HTTPError) as error:
urlopen(url + '/api/host', timeout=2)
self.assertEqual(error.exception.code, 403)
error.exception.close()
# A second client has its own backend and key.
with mcp.backend() as other:
self.assertNotEqual(client.url, other.url)
self.assertNotEqual(client.key, other.key)
self.assertIn('os', client.request('/api/host'))
with self.assertRaises(URLError):
urlopen(url + '/', timeout=2)
def test_private_ssh_socket_is_not_the_desktop_socket(self):
with mock.patch.object(server.frame_host, 'MUX', True), \
mock.patch.object(server.frame_host.os, 'getuid', return_value=501, create=True), \
mock.patch.object(server.frame_host.os, 'getpid', return_value=123):
self.assertEqual(server.frame_host.control_path(), '/tmp/frame-ui-501-%C')
self.assertEqual(server.frame_host.control_path(private=True), '/tmp/frame-ui-501-123-%C')
class ComputerState(unittest.TestCase):
def test_gamescope_triplets_and_empty_focus(self):
import frame_computer
parsed = frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 16, 42, 123, 32, 55, 999\nGAMESCOPE_FOCUSED_APP(CARDINAL) = \n')
self.assertEqual(parsed['windows'], [{'windowId': '0x10', 'appid': 42, 'pid': 123}, {'windowId': '0x20', 'appid': 55, 'pid': 999}])
self.assertIsNone(parsed['focusedApp'])
with self.assertRaises(ValueError):
frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 1, 2')
with self.assertRaises(ValueError):
frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = untrusted')
with self.assertRaises(ValueError):
frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS: no such atom on any window.')
def test_partial_snapshot_reports_failure_not_empty_success(self):
import frame_computer
with mock.patch.object(frame_computer.subprocess, 'run', side_effect=OSError('no display')), \
mock.patch.object(frame_computer, 'accessibility', side_effect=OSError('no AT-SPI')):
result = frame_computer.snapshot()
self.assertIn('windowError', result)
self.assertIn('accessibilityError', result)
self.assertFalse(result['inputEnabled'])
self.assertNotIn('windows', result)
def test_mcp_computer_state_is_read_only(self):
client = mock.Mock()
client.request.return_value = {'windows': []}
mcp.call(client, 'computer_state', {})
client.request.assert_called_once_with('/api/computer/state')
spec = next(t for t in mcp.TOOLS if t['name'] == 'computer_state')
self.assertTrue(spec['annotations']['readOnlyHint'])
if __name__ == '__main__':
unittest.main()
+257
View File
@@ -0,0 +1,257 @@
"""Fake-Frame session clock/actions plus real helper serialization and sensor probes."""
import os
import shutil
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
import frame_comfort as comfort
import frame_status as status
OPTIONS = {'action': 'start', 'minutes': 3, 'breakMinutes': 1, 'stillMinutes': 1,
'batteryAlert': True, 'heatAlert': True}
class SessionTests(unittest.TestCase):
def setUp(self):
self.s = comfort.new_session(OPTIONS, 0, 'boot-one')
self.warn, self.home = Mock(), Mock()
def step(self, now, **sample):
comfort.tick(self.s, now, sample, self.warn, self.home, read_clock=lambda: now)
def test_warning_then_home_never_closes_a_game(self):
self.step(119)
self.warn.assert_not_called()
self.step(120)
self.warn.assert_called_once()
self.step(179)
self.home.assert_not_called()
self.step(180)
self.home.assert_called_once()
self.assertFalse(self.s['active'])
self.step(181)
self.home.assert_called_once()
def test_late_wakeup_always_gets_a_full_warning_minute(self):
self.step(400)
self.home.assert_not_called()
self.step(459)
self.home.assert_not_called()
self.step(460)
self.home.assert_called_once()
def test_slow_warning_still_leaves_a_full_minute(self):
comfort.tick(self.s, 120, {}, self.warn, self.home, read_clock=lambda: 140)
self.assertEqual(self.s['warned'], 140)
self.step(180)
self.home.assert_not_called()
self.step(199)
self.home.assert_not_called()
self.step(200)
self.home.assert_called_once()
def test_failed_warning_never_stops_session(self):
self.warn.side_effect = RuntimeError('offline')
with self.assertRaises(RuntimeError):
self.step(200)
self.assertIsNone(self.s['warned'])
self.home.assert_not_called()
self.warn.side_effect = None
self.step(300)
self.step(359)
self.home.assert_not_called()
self.step(360)
self.home.assert_called_once()
def test_failed_home_stays_active_and_retries(self):
self.step(120)
self.home.side_effect = RuntimeError('Steam offline')
with self.assertRaises(RuntimeError):
self.step(180)
self.assertTrue(self.s['active'])
self.home.side_effect = None
self.step(185)
self.assertFalse(self.s['active'])
def test_cancel_prevents_all_actions(self):
self.s['active'] = False
self.step(999, battery={'percent': 1, 'status': 'Discharging'})
self.warn.assert_not_called()
self.home.assert_not_called()
self.assertEqual(self.s['events'], [])
def test_breaks_and_checkin_require_measured_activity(self):
self.step(20, activity=1)
self.step(40, activity=2)
self.step(60, activity=1)
self.assertEqual([e['kind'] for e in self.s['events']], ['break', 'still'])
self.step(70, activity=1)
self.assertEqual(len(self.s['events']), 2)
self.step(75, activity=3)
self.assertEqual(self.s['used'], 0)
self.assertFalse(self.s['stillSent'])
def test_unknown_activity_and_gaps_do_not_count_as_wear(self):
self.step(25)
self.assertEqual(self.s['used'], 0)
self.assertEqual(self.s['unavailable'], ['battery', 'temperature', 'activity'])
self.step(100, activity=1)
self.assertEqual(self.s['used'], 30)
def test_alerts_latch_and_rearm_without_battery_chatter(self):
low = {'percent': 10, 'status': 'Discharging'}
self.step(1, battery=low, thermal=['cpu'])
self.step(2, battery=low, thermal=['cpu'])
self.step(3)
self.assertEqual(len(self.s['events']), 2)
self.step(4, battery={'percent': 16, 'status': 'Discharging'}, thermal=[])
self.step(5, battery=low, thermal=[])
self.assertEqual(len(self.s['events']), 2)
self.step(6, battery={'percent': 22, 'status': 'Discharging'}, thermal=[])
self.step(7, battery=low, thermal=['cpu'])
self.assertEqual([e['kind'] for e in self.s['events']], ['battery', 'heat', 'battery', 'heat'])
def test_disabled_alerts_and_charging(self):
self.s['options']['heatAlert'] = False
self.step(1, battery={'percent': 2, 'status': 'Charging'}, thermal=['cpu'])
self.assertEqual(self.s['events'], [])
def test_invalid_options(self):
for key, value in [('minutes', 0), ('minutes', 241), ('minutes', True), ('minutes', 2.5),
('breakMinutes', -1), ('stillMinutes', '1'), ('heatAlert', 1)]:
with self.subTest(key=key, value=value), self.assertRaises(ValueError):
comfort.validate({**OPTIONS, key: value})
for value in (None, [], {'action': 'shutdown'}):
with self.assertRaises(ValueError):
comfort.validate(value)
def test_restart_invalidates_session_and_stale_worker_is_explicit(self):
with patch.object(comfort, 'boot', return_value='boot-one'), patch.object(comfort.time, 'time', return_value=999):
current = comfort.current(self.s, 100)
self.assertIn('not responding', current['error'])
self.assertEqual(current['time'], 999)
with patch.object(comfort, 'boot', return_value='boot-two'):
result = comfort.current(self.s, 100)
self.assertFalse(result['active'])
self.assertIn('restarted', result['error'])
@unittest.skipUnless(os.name == "posix", "on-headset state uses POSIX flock")
def test_real_state_commands_share_one_session_and_cancel(self):
with tempfile.TemporaryDirectory() as tmp, patch.object(comfort, 'ROOT', Path(tmp)), \
patch.object(comfort, 'boot', return_value='boot-one'), \
patch.object(comfort, 'clock', return_value=0), patch.object(comfort.subprocess, 'Popen') as spawn:
started = comfort.command(OPTIONS)
self.assertEqual(comfort.command({'action': 'status'})['id'], started['id'])
with self.assertRaises(ValueError):
comfort.command(OPTIONS)
self.assertFalse(comfort.command({'action': 'cancel'})['active'])
spawn.assert_called_once()
self.assertEqual((Path(tmp) / 'session.json').stat().st_mode & 0o777, 0o600)
@unittest.skipUnless(os.name == "posix", "on-headset state uses POSIX flock")
def test_cancel_clears_stale_worker_error(self):
with tempfile.TemporaryDirectory() as tmp, patch.object(comfort, 'ROOT', Path(tmp)), \
patch.object(comfort, 'boot', return_value='boot-one'), \
patch.object(comfort, 'clock', return_value=200):
with comfort.locked():
comfort.save(self.s)
self.assertIn('not responding', comfort.command({'action': 'status'})['error'])
cancelled = comfort.command({'action': 'cancel'})
self.assertFalse(cancelled['active'])
self.assertIsNone(cancelled['error'])
self.assertIsNone(comfort.command({'action': 'status'})['error'])
@unittest.skipUnless(os.name == "posix", "on-headset state uses POSIX flock")
def test_failed_spawn_leaves_session_inactive_and_retryable(self):
with tempfile.TemporaryDirectory() as tmp, patch.object(comfort, 'ROOT', Path(tmp)), \
patch.object(comfort, 'boot', return_value='boot-one'), \
patch.object(comfort, 'clock', return_value=0), patch.object(comfort.subprocess, 'Popen') as spawn:
spawn.side_effect = OSError('process limit')
with self.assertRaises(OSError):
comfort.command(OPTIONS)
failed = comfort.command({'action': 'status'})
self.assertFalse(failed['active'])
self.assertIn('Could not start', failed['error'])
spawn.side_effect = None
self.assertTrue(comfort.command(OPTIONS)['active'])
@unittest.skipUnless(os.name == "posix", "on-headset state uses POSIX flock")
def test_unreadable_state_is_preserved_and_can_be_replaced(self):
for contents in (b'{broken', b'\xff', b'null', b'[]', b'42', b'"x"'):
with self.subTest(contents=contents):
with tempfile.TemporaryDirectory() as tmp, patch.object(comfort, 'ROOT', Path(tmp)), \
patch.object(comfort, 'boot', return_value='boot-one'), \
patch.object(comfort, 'clock', return_value=0), patch.object(comfort.subprocess, 'Popen'):
(Path(tmp) / 'session.json').write_bytes(contents)
failed = comfort.command({'action': 'status'})
self.assertFalse(failed['active'])
self.assertIn('unreadable', failed['error'])
backups = list(Path(tmp).glob('session-unreadable-*.json'))
self.assertEqual(len(backups), 1)
self.assertEqual(backups[0].read_bytes(), contents)
self.assertTrue(comfort.command(OPTIONS)['active'])
def test_home_has_total_process_deadline_and_propagates_timeout(self):
with patch.object(comfort.subprocess, 'run', side_effect=subprocess.TimeoutExpired('home', 15)) as run:
with self.assertRaises(subprocess.TimeoutExpired):
comfort.home()
self.assertEqual(run.call_args.kwargs['timeout'], 15)
self.assertEqual(run.call_args.args[0][-1], '--home')
def test_native_warning_reports_failures_and_quotes_as_one_argument(self):
with patch.object(comfort.subprocess, 'run') as run:
run.return_value = subprocess.CompletedProcess([], 0, 'Notification succeeded', '')
comfort.notify('Save "now"; $(nothing)')
args = run.call_args.args[0]
self.assertEqual(args, [comfort.VRCMD, '--notify', 'Frame Control: Save "now"; $(nothing)'])
run.return_value.stdout = 'Notification failed with error 1'
with self.assertRaises(RuntimeError):
comfort.notify('test')
@unittest.skipUnless(shutil.which("node"), "Node exercises the fake Steam JS context")
def test_home_javascript_against_fake_steam_preserves_game(self):
# Same JS runs in Steam CDP. This fake records navigation and refuses any
# unexpected API call; it offers no shutdown or terminate-game primitive.
js = '''let running = [123], path = '/routes/library/app/123', visible = false;
const location = {get pathname() {return path;}};
const SteamUIStore = {Navigate(p) {path = '/routes' + p;}};
const SteamClient = {OpenVR: {VROverlay: {
async ShowDashboard(key) {if (key !== 'valve.steam.gamepadui.main') throw Error(key); visible = true;},
async IsDashboardVisible() {return visible;}
}}};
'''
js += comfort.HOME_JS + '.then(result => console.log(JSON.stringify({result, running, visible})));'
r = subprocess.run(['node', '-e', js], capture_output=True, text=True, check=True)
result = json.loads(r.stdout)
self.assertEqual(result['running'], [123])
self.assertTrue(result['visible'])
self.assertEqual(result['result']['path'], '/routes/library/home')
class SensorTests(unittest.TestCase):
def test_hot_trip_uses_its_own_zone_not_hottest_unrelated_chip(self):
values = {'/z/a/temp': '90000', '/z/a/trip_point_0_type': 'hot', '/z/a/trip_point_0_temp': '110000',
'/z/b/temp': '45000', '/z/b/trip_point_0_type': 'hot', '/z/b/trip_point_0_temp': '44000', '/z/b/type': 'battery'}
def glob(pattern):
if pattern.endswith('thermal_zone*'):
return ['/z/a', '/z/b']
return [pattern.replace('*', '0')]
with patch.object(status.glob, 'glob', side_effect=glob), patch.object(status, 'read', side_effect=values.get):
self.assertEqual(status.thermal_alerts(), [{'zone': 'battery', 'tempC': 45, 'limitC': 44}])
def test_missing_thermal_and_activity_are_unknown(self):
with patch.object(status.glob, 'glob', return_value=[]):
self.assertIsNone(status.thermal_alerts())
with patch.object(status, 'run', return_value='unavailable'):
self.assertIsNone(status.activity_level())
for malformed in ('{}', '[null, 42, "bad"]'):
with patch.object(status, 'run', return_value=malformed):
self.assertIsNone(status.activity_level())
with patch.object(status, 'run', return_value='[{"operation":"status","activity_level":3}]'):
self.assertEqual(status.activity_level(), 3)
+60
View File
@@ -0,0 +1,60 @@
"""Run the actual shared page's comfort renderer against a minimal DOM/bridge."""
import pathlib
import shutil
import subprocess
import unittest
ROOT = pathlib.Path(__file__).resolve().parents[1]
@unittest.skipUnless(shutil.which('node'), 'Node exercises the shared page JS')
class ComfortUI(unittest.TestCase):
def test_notification_failure_survives_poll_until_success(self):
page = (ROOT / 'ui/index.html').read_text(encoding='utf-8')
code = page[page.index('let comfortBusy ='):page.index('async function pollComfort()')]
setup = r'''
const assert = require('node:assert/strict');
const elements = new Map();
const $ = id => {
if (!elements.has(id)) elements.set(id, {textContent:'', hidden:true, disabled:false, type: 'number'});
return elements.get(id);
};
let denied = 0;
const window = {frameApp:{notify:async()=>{denied++;throw Error('permission denied');}}};
const log = ()=>{}, toast = ()=>{};
'''
checks = r'''
(async()=>{
const active = {id:'session-one',active:true,time:100,remaining:120,
options:{minutes:2,breakMinutes:1,stillMinutes:1,batteryAlert:true,heatAlert:true},
events:[{id:'event-one',kind:'battery',time:99,message:'Low battery'}]};
renderComfort(active); // initial history must not replay even a fresh event
assert.equal(denied,0);
assert.equal($('comfortAnnouncement').textContent,'');
active.events.push({id:'event-two',kind:'break',time:100,message:'Take a break'});
renderComfort(active);
await new Promise(resolve=>setImmediate(resolve));
assert.equal(denied,1);
assert.equal($('comfortAnnouncement').textContent,'Take a break');
assert.equal($('comfortNotificationStatus').hidden,false);
assert.match($('comfortNotificationStatus').textContent,/notification settings/);
renderComfort({...active,time:105}); // the next normal poll must not erase failure
assert.equal($('comfortNotificationStatus').hidden,false);
assert.equal($('sessionStart').disabled,true);
assert.equal($('sessionMinutes').disabled,true);
assert.equal($('sessionCancel').disabled,false);
let requests=0;
window.frameApp.notify=async()=>{requests++;};
renderComfort({...active,time:106});
assert.equal($('comfortAnnouncement').textContent,'Take a break');
assert.equal(requests,0); // polling does not replay an already-seen event
await localNotification('test',true);
assert.equal($('comfortNotificationStatus').hidden,true);
renderComfort({...active,active:false});
assert.equal($('sessionStart').disabled,false);
assert.equal($('sessionMinutes').disabled,false);
assert.equal($('sessionCancel').disabled,true);
})().catch(e=>{console.error(e);process.exitCode=1;});
'''
result = subprocess.run(['node', '-e', setup + code + checks], capture_output=True, text=True)
self.assertEqual(result.returncode, 0, result.stderr)
+1
View File
@@ -2,6 +2,7 @@
Run: python3 -m unittest discover -s tests
"""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import os
import sys
import tempfile
+1
View File
@@ -3,6 +3,7 @@ steamos-devkit-service, the ~/.ssh/config block, and the mDNS output parsers.
Run: python3 -m unittest discover -s tests
"""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import json
import socket
import sys
+4 -1
View File
@@ -516,7 +516,10 @@ class Repositories(unittest.TestCase):
self.expire(source)
self.fetch_mock.side_effect = urllib.error.HTTPError(URL, 503, 'unavailable', None, None)
self.assertEqual(len(fdroid.search(source, 'example')), 1)
fdroid._refreshing[source['id']].join(5)
# A fast failed refresh may already have removed itself from the registry.
refresh = fdroid._refreshing.get(source['id'])
if refresh is not None:
refresh.join(5)
calls = self.fetch_mock.call_count
self.assertEqual(fdroid.details(source, 'org.example.app')['version_code'], 2)
self.assertTrue(fdroid.stale(source))
+1
View File
@@ -1,4 +1,5 @@
"""frame_apk against a small APK built here: binary manifest plus resource table."""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import io
import os
import struct
+25
View File
@@ -1,4 +1,5 @@
"""Offline version lookup with small index-v2 fixtures."""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import io
import json
import os
@@ -242,6 +243,7 @@ class UploadVersionsTest(unittest.TestCase):
handler.rfile = io.BytesIO(b'x')
with patch.object(frame_android, 'apk_info', return_value=dict(info)), \
patch.object(versions, 'alternatives', side_effect=AssertionError('lookup during upload')) as lookup, \
patch.object(frame_android, 'install_hooks', []), \
patch.object(server, 'ensure_master') as ssh:
if mode == 'apkinfo':
reply = handler.upload()
@@ -256,5 +258,28 @@ class UploadVersionsTest(unittest.TestCase):
ssh.assert_not_called()
def test_blocked_uploads_are_reported_like_failed_installs(self):
import server
info = {'package': 'org.example.app', 'label': 'Example', 'version': '5',
'version_code': 5, 'min_sdk': 33, 'abis': [], 'icon_png': None}
for apk_info, expected_info in ((dict(info), 'org.example.app'),
(frame_android.FrameError('not an APK'), None)):
handler = object.__new__(server.Handler)
handler.headers = {'X-Filename': 'app.apk', 'X-Mode': 'apk', 'Content-Length': '1'}
handler.rfile = io.BytesIO(b'x')
calls = []
patch_info = (patch.object(frame_android, 'apk_info', side_effect=apk_info)
if isinstance(apk_info, Exception) else
patch.object(frame_android, 'apk_info', return_value=apk_info))
with patch_info, patch.object(frame_android, 'install_hooks', [lambda *a: calls.append(a)]), \
patch.object(server, 'ensure_master'):
with self.assertRaises(server.Failure):
handler.upload()
self.assertEqual(len(calls), 1)
got_info, meta, error, _ = calls[0]
self.assertEqual((got_info or {}).get('package'), expected_info)
self.assertIsNone(meta)
self.assertIsInstance(error, frame_android.FrameError)
if __name__ == '__main__':
unittest.main()
+1
View File
@@ -1,4 +1,5 @@
"""frame_titles without a headset: executable headers, launch targets, zips, runtimes."""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import json
import os
import shutil
+632
View File
@@ -0,0 +1,632 @@
"""Keyboard and pointer: event checks, and the agent's KDE Connect protocol against a fake kdeconnectd.
The fake follows what KDE Connect 24.02 does (read from its source,
core/backends/lan/lanlinkprovider.cpp): it accepts the TCP connection, reads the
identity line, then starts TLS as the *client*.
Run: python3 -m unittest discover -s tests
"""
import json
import os
import shutil
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "ui"))
class InputEvents(unittest.TestCase):
@classmethod
def setUpClass(cls):
import server
cls.server = server
def check(self, event):
return self.server.input_event(event)
def test_keeps_known_fields(self):
self.assertEqual(self.check({"dx": 3, "dy": -1.234}), {"dx": 3.0, "dy": -1.23})
self.assertEqual(self.check({"singleclick": True, "other": 1}), {"singleclick": True})
self.assertEqual(self.check({"key": "héllo", "shift": True}), {"key": "héllo", "shift": True})
self.assertEqual(self.check({"specialKey": 12, "ctrl": True}), {"specialKey": 12, "ctrl": True})
self.assertEqual(self.check({"scroll": True, "dy": 1}), {"scroll": True, "dy": 1.0})
def test_clamps_movement(self):
self.assertEqual(self.check({"dx": 1e9})["dx"], self.server.INPUT_MOVE_LIMIT)
self.assertEqual(self.check({"dy": -1e9})["dy"], -self.server.INPUT_MOVE_LIMIT)
def test_rejects_bad_events(self):
bad = [None, [], "a", {}, {"shift": True}, {"dx": "1"}, {"dx": True}, {"dx": float("nan")},
{"key": ""}, {"key": 5}, {"key": "x" * 501}, {"specialKey": 0}, {"specialKey": 33},
{"specialKey": True}, {"specialKey": 1.5}, {"singleclick": "yes"}]
for event in bad:
with self.assertRaises(self.server.Failure, msg=repr(event)):
self.check(event)
def test_send_while_link_is_down_says_not_sent(self):
# The page keeps unsent keys and clicks and sends them once the agent is ready again.
from types import SimpleNamespace
class Probe(self.server.InputAgent):
def start(self):
self.proc, self.status = SimpleNamespace(poll=lambda: None, stdin=None), {"state": "starting"}
agent = Probe()
agent.status, agent.proc = {"state": "ready"}, SimpleNamespace(poll=lambda: 255) # ssh has exited
self.assertEqual(agent.send([{"key": "x"}]), {"state": "starting", "sent": False})
def test_command_passes_client_quoted(self):
os.environ["FRAME_CLIENT"] = "test-client-1"
try:
cmd = self.server.InputAgent().command()
finally:
del os.environ["FRAME_CLIENT"]
self.assertTrue(cmd.startswith("python3 -u -c '"))
self.assertIn(" test-client-1 ", cmd)
cmd = self.server.InputAgent(packages=[("a.pkg.tar.zst", "ab")]).command("~/in/x")
self.assertTrue(cmd.endswith(""" '~/in/x' '[["a.pkg.tar.zst","ab"]]'"""), cmd)
def test_batch_limits(self):
with self.assertRaises(self.server.Failure):
self.server.remote_input({"events": "dx"})
with self.assertRaises(self.server.Failure):
self.server.remote_input({"events": [{"dx": 1}] * (self.server.INPUT_BATCH_LIMIT + 1)})
class Bundled(unittest.TestCase):
"""KDE Connect ships with Frame Control: the manifest, the notice, the copy to the Frame."""
@classmethod
def setUpClass(cls):
import server
cls.server = server
cls.manifest = json.loads((ROOT / "frame/kdeconnect/packages.json").read_text())
def test_manifest_pins_every_package(self):
packages = self.manifest["packages"]
self.assertEqual({p["name"] for p in packages},
{"kdeconnect", "kcontacts", "kpeople", "libfakekey", "modemmanager-qt", "pulseaudio-qt"})
for p in packages:
self.assertRegex(p["sha256"], r"^[0-9a-f]{64}$")
self.assertTrue(p["file"].startswith(f"{p['name']}-{p['version']}-") and p["file"].endswith("-aarch64.pkg.tar.zst"), p)
self.assertTrue(p["source"].startswith(self.manifest["release"]), p)
self.assertRegex(p["source_sha256"], r"^[0-9a-f]{64}$")
self.assertTrue(self.manifest["release"].startswith("https://github.com/") and self.manifest["release"].endswith("/"))
self.assertNotIn("DO_NOT_SHARE", json.dumps(self.manifest))
def test_notice_names_each_version_and_its_licence_texts_ship(self):
notice = (ROOT / "frame/kdeconnect/NOTICE.md").read_text()
for p in self.manifest["packages"]:
self.assertIn(f"{p['name']} {p['version']}", notice)
self.assertIn(p["source"], notice)
self.assertIn(p["upstream"], notice)
for spdx in p["licenses"]:
self.assertTrue((ROOT / "frame/kdeconnect/LICENSES" / p["name"] / f"{spdx}.txt").is_file(), spdx)
self.assertIn("frame/kdeconnect/NOTICE.md", (ROOT / "THIRD_PARTY_NOTICES.md").read_text())
def test_about_dialog_has_the_licences(self):
titles = [n["title"] for n in self.server.licenses()]
self.assertIn("Frame Control (MIT)", titles)
self.assertIn("KDE Connect for the Frame", titles)
self.assertIn("kdeconnect: GPL-2.0-only", titles)
def test_both_apps_bundle_the_packages(self):
pkg = json.loads((ROOT / "app/package.json").read_text())["build"]["extraResources"]
kde = next(r for r in pkg if r["from"] == "../frame/kdeconnect")
self.assertIn("packages/*.pkg.tar.zst", kde["filter"])
self.assertIn("LICENSES/**/*", kde["filter"])
bundle = (ROOT / "ios/scripts/make_frame_bundle.py").read_text()
for pattern in ("frame/kdeconnect/packages/*.pkg.tar.zst", "frame/kdeconnect/LICENSES/*/*", "THIRD_PARTY_NOTICES.md"):
self.assertIn(pattern, bundle)
def fake_bundle(self, damaged=False):
folder = Path(tempfile.mkdtemp())
self.addCleanup(shutil.rmtree, folder)
(folder / "packages").mkdir()
data = {"a-1-1-aarch64.pkg.tar.zst": b"first", "b-2-1-aarch64.pkg.tar.zst": b"second"}
packages = []
for name, body in data.items():
(folder / "packages" / name).write_bytes(b"x" + body if damaged else body)
packages.append((name, __import__("hashlib").sha256(body).hexdigest()))
return folder, packages
def deliver(self, frame_has, damaged=False):
folder, packages = self.fake_bundle(damaged)
calls = []
def ssh(remote, stdin=None, timeout=30, text=True):
calls.append((remote, stdin))
answer = "yes\n" if remote.startswith("{ test -x") and frame_has else ""
return answer if text else answer.encode()
old = self.server.ssh, self.server.KDECONNECT, self.server.LOCAL
self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = ssh, folder, False
try:
agent = self.server.InputAgent(packages=packages)
return agent.deliver(lambda m: calls.append(("report", m))), calls, packages
finally:
self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = old
def test_copies_nothing_when_the_frame_has_them(self):
folder, calls, packages = self.deliver(frame_has=True)
self.assertEqual(folder, "")
self.assertEqual(len(calls), 1)
# Bytes: on Windows a text pipe would send CRLF and never match.
self.assertEqual(calls[0][1], "".join(f"{sha} {name}\n" for name, sha in packages).encode())
def test_copies_each_package_over_ssh(self):
folder, calls, packages = self.deliver(frame_has=False)
self.assertTrue(folder.startswith("~/.local/share/frame-control/kdeconnect/incoming/"))
copies = [c for c in calls if c[0] != "report" and "cat >" in c[0]]
self.assertEqual([c[1] for c in copies], [b"first", b"second"])
self.assertIn("a-1-1-aarch64.pkg.tar.zst.part", copies[0][0])
def test_refuses_a_damaged_bundle(self):
with self.assertRaises(self.server.Failure):
self.deliver(frame_has=False, damaged=True)
def test_a_failed_copy_is_removed(self):
calls = []
def ssh(remote, stdin=None, timeout=30, text=True):
calls.append(remote)
if "cat >" in remote and len([c for c in calls if "cat >" in c]) == 2:
raise self.server.Failure("Broken pipe")
return b"" if not text else ""
folder, packages = self.fake_bundle()
old = self.server.ssh, self.server.KDECONNECT, self.server.LOCAL
self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = ssh, folder, False
try:
with self.assertRaises(self.server.Failure):
self.server.InputAgent(packages=packages).deliver(lambda m: None)
finally:
self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = old
self.assertTrue(calls[-1].startswith("rm -rf .local/share/frame-control/kdeconnect/incoming/"), calls[-1])
def lifecycle(self, agent_lines, deliver=None):
"""An InputAgent whose ssh and agent are fakes. Returns it, the folders each launch
used, and the fake agents. A fake agent reports its lines, then keeps running
(unless its lines end with need-packages) until the test ends."""
launches, procs, done = [], [], threading.Event()
self.addCleanup(done.set)
class Agent(self.server.InputAgent):
def deliver(self, report, force=False):
if deliver:
deliver()
return "~/copied" if force else ""
def command(self, folder=""):
launches.append(folder)
return "agent"
class Proc:
stdin = None
def __init__(self, lines):
self.lines, self.ended = lines, threading.Event()
procs.append(self)
@property
def stdout(self):
yield from self.lines
if self.lines and b"need-packages" not in self.lines[-1]:
while not (done.is_set() or self.ended.is_set()):
self.ended.wait(0.02)
self.ended.set()
def wait(self):
self.ended.wait(5)
return 0
def poll(self):
return 0 if self.ended.is_set() else None
def terminate(self):
self.ended.set()
def popen(*a, **k):
return Proc(agent_lines.pop(0) if agent_lines else [b'{"state": "ready"}\n'])
old = self.server.ensure_master, self.server.subprocess.Popen
self.server.ensure_master, self.server.subprocess.Popen = lambda: None, popen
self.addCleanup(lambda: (setattr(self.server, "ensure_master", old[0]),
setattr(self.server.subprocess, "Popen", old[1])))
agent = Agent(packages=[("a.pkg.tar.zst", "0" * 64)])
def settle(): # nothing of this test may still be launching when the next one starts
agent.stop()
for _ in range(250):
if agent.launching is None and all(p.ended.is_set() for p in procs):
return
time.sleep(0.02)
self.addCleanup(settle)
return agent, launches, procs
def wait_for(self, check):
for _ in range(250):
if check():
return
time.sleep(0.02)
self.fail("timed out")
def test_agent_asking_for_packages_gets_them_and_starts_again(self):
agent, launches, procs = self.lifecycle([[b'{"state": "need-packages"}\n'], [b'{"state": "ready"}\n']])
agent.start()
self.wait_for(lambda: agent.status == {"state": "ready"})
self.assertEqual(launches, ["", "~/copied"])
self.assertIs(agent.proc, procs[1])
def test_asking_twice_is_an_error_not_a_loop(self):
need = [b'{"state": "need-packages"}\n']
agent, launches, _ = self.lifecycle([need, list(need)])
agent.start()
self.wait_for(lambda: agent.status.get("state") == "error")
self.assertEqual(launches, ["", "~/copied"])
self.assertIn("didn't reach", agent.status["message"])
def test_start_after_stop_during_the_copy_still_starts(self):
gate, entered = threading.Event(), threading.Event()
agent, launches, procs = self.lifecycle([], deliver=lambda: (entered.set(), gate.wait(5)))
agent.start()
self.assertTrue(entered.wait(5))
agent.stop()
entered.clear()
agent.start() # while the first launch is still copying
self.assertTrue(entered.wait(5), "the second start didn't launch")
gate.set()
self.wait_for(lambda: agent.status == {"state": "ready"})
time.sleep(0.1)
# The stopped launch started no agent; the new one is the one in use.
self.assertEqual(len(procs), 1)
self.assertIs(agent.proc, procs[0])
agent.stop()
self.wait_for(lambda: all(p.ended.is_set() for p in procs))
def test_retry_and_a_new_start_never_run_two_agents(self):
# A start() landing just as an agent that asked for the packages exits: exactly one
# of it and the retry launches, and stop() ends everything. The new start is held
# in its copy until the retry has decided, so neither can see the other's agent.
first, decided = [], threading.Event()
agent, launches, procs = self.lifecycle([[b'{"state": "need-packages"}\n'], [b'{"state": "ready"}\n']])
watch, launch, deliver = agent._watch, agent._launch, agent.deliver
def racing_watch(proc, errors, retry=False):
wanted, heard = watch(proc, errors, retry)
if wanted and not first:
first.append(threading.current_thread())
agent.start() # lands between the agent exiting and the retry
return wanted, heard
def held_deliver(report, force=False):
if force:
decided.set() # the retry went ahead
elif first and threading.current_thread() is not first[0]:
decided.wait(5) # the new start waits until the retry has decided
return deliver(report, force)
def first_launch(generation, force=False):
try:
launch(generation, force)
finally:
if first and threading.current_thread() is first[0]:
decided.set() # the retry declined
agent._watch, agent._launch, agent.deliver = racing_watch, first_launch, held_deliver
agent.start()
# Both have decided once the new start has launched its agent (it always does).
self.wait_for(lambda: decided.is_set() and launches.count("") == 2 and len(procs) == len(launches))
self.wait_for(lambda: agent.status == {"state": "ready"})
self.assertEqual(sum(not p.ended.is_set() for p in procs), 1, launches)
agent.stop()
self.wait_for(lambda: all(p.ended.is_set() for p in procs))
def test_stopped_during_the_copy_removes_it(self):
gate, entered = threading.Event(), threading.Event()
agent, launches, procs = self.lifecycle([], deliver=lambda: (entered.set(), gate.wait(5)))
removed = []
agent.discard = removed.append
agent.deliver = lambda report, force=False: (entered.set(), gate.wait(5), "~/incoming/x")[2]
self.addCleanup(lambda: self.assertEqual(removed, ["~/incoming/x"]))
agent.start()
self.assertTrue(entered.wait(5))
agent.stop()
gate.set()
self.wait_for(lambda: removed == ["~/incoming/x"])
self.assertEqual(procs, []) # no agent started for it
def test_a_copy_whose_agent_never_starts_is_removed(self):
agent, launches, procs = self.lifecycle([[]]) # the agent dies before saying anything
removed = []
agent.discard = removed.append
agent.deliver = lambda report, force=False: "~/incoming/y"
agent.start()
self.wait_for(lambda: removed == ["~/incoming/y"])
def test_discard_only_touches_copies(self):
calls = []
old = self.server.ssh, self.server.LOCAL
self.server.ssh, self.server.LOCAL = (lambda remote, **k: calls.append(remote)), False
try:
agent = self.server.InputAgent(packages=[])
agent.discard("")
agent.discard("~/.local/share/frame-control/kdeconnect")
agent.discard("~/.local/share/frame-control/kdeconnect/incoming/ab12")
self.server.LOCAL = True
agent.discard("~/.local/share/frame-control/kdeconnect/incoming/ab12")
finally:
self.server.ssh, self.server.LOCAL = old
self.assertEqual(calls, ["rm -rf .local/share/frame-control/kdeconnect/incoming/ab12"])
def test_a_launch_error_leaves_it_ready_to_try_again(self):
# Popen fails (say, out of file handles) and so does removing the copy: the
# page gets the error and the next start launches.
agent, launches, procs = self.lifecycle([])
agent.deliver = lambda report, force=False: "~/.local/share/frame-control/kdeconnect/incoming/z"
old_popen, old_ssh = self.server.subprocess.Popen, self.server.ssh
self.server.subprocess.Popen = self.server.ssh = lambda *a, **k: (_ for _ in ()).throw(OSError(24, "Too many open files"))
try:
agent.start()
self.wait_for(lambda: agent.status.get("state") == "error")
finally:
self.server.subprocess.Popen, self.server.ssh = old_popen, old_ssh
self.assertIsNone(agent.launching)
agent.start()
self.wait_for(lambda: agent.status == {"state": "ready"})
def test_any_unexpected_error_still_reports_and_allows_a_retry(self):
agent, launches, procs = self.lifecycle([])
old = self.server.tempfile.TemporaryFile
self.server.tempfile.TemporaryFile = lambda: (_ for _ in ()).throw(OSError(24, "Too many open files"))
try:
agent.start()
self.wait_for(lambda: agent.status.get("state") == "error")
finally:
self.server.tempfile.TemporaryFile = old
self.assertIsNone(agent.launching)
agent.deliver = lambda report, force=False: (_ for _ in ()).throw(ValueError("something odd"))
agent.start()
self.wait_for(lambda: "something odd" in agent.status.get("message", ""))
self.assertIsNone(agent.launching)
def test_copies_go_to_a_folder_of_their_own(self):
first, _, _ = self.deliver(frame_has=False)
second, _, _ = self.deliver(frame_has=False)
self.assertNotEqual(first, second)
@unittest.skipIf(sys.platform == "win32", "the agent runs on the Frame (Linux)")
class AgentInstall(unittest.TestCase):
"""The agent unpacks what the server copied, after checking each SHA-256."""
@classmethod
def setUpClass(cls):
import frame_input_agent
cls.agent = frame_input_agent
def setUp(self):
self.dir = Path(tempfile.mkdtemp())
self.addCleanup(shutil.rmtree, self.dir)
saved = self.agent.BASE, self.agent.ROOT, self.agent.stop_daemon, self.agent.say
self.addCleanup(lambda: setattr_all(self.agent, saved))
self.agent.BASE, self.agent.ROOT = self.dir / "base", self.dir / "base/root"
self.agent.stop_daemon, self.agent.say = lambda: None, lambda *a, **k: None
self.agent.BASE.mkdir()
def package(self):
src = self.dir / "src"
(src / "usr/lib").mkdir(parents=True)
(src / "usr/lib/kdeconnectd").write_text("#!/bin/sh\n")
out = self.dir / "incoming/kdeconnect-24.02.2-1-aarch64.pkg.tar.zst"
out.parent.mkdir()
if subprocess.run(["tar", "--zstd", "-cf", str(out), "-C", str(src), "usr"], capture_output=True).returncode:
self.skipTest("this tar can't write zstd")
return out, [(out.name, self.agent.sha256(out))]
def test_unpacks_and_stamps(self):
path, packages = self.package()
self.assertFalse(self.agent.installed(packages))
self.agent.install(path.parent, packages)
self.assertTrue((self.agent.ROOT / "usr/lib/kdeconnectd").is_file())
self.assertTrue(self.agent.installed(packages))
self.assertFalse(self.agent.installed([(path.name, "0" * 64)]))
def test_refuses_a_damaged_package(self):
path, packages = self.package()
with open(path, "ab") as f:
f.write(b"!")
with self.assertRaisesRegex(RuntimeError, "damaged"):
self.agent.install(path.parent, packages)
self.assertFalse(self.agent.ROOT.exists())
def test_missing_package_and_empty_manifest(self):
with self.assertRaisesRegex(RuntimeError, "didn't reach"):
self.agent.install(self.dir, [("nope.pkg.tar.zst", "0" * 64)])
with self.assertRaisesRegex(RuntimeError, "doesn't include"):
self.agent.install(self.dir, [])
def test_leaves_another_devices_running_copy_alone(self):
# A different build is running for another device: use it, don't stop it to reinstall.
self.agent.listening, old = (lambda: True), self.agent.listening
self.agent.our_daemons, old_ours = (lambda: [123]), self.agent.our_daemons
try:
self.agent.ensure_daemon("", [("new.pkg.tar.zst", "1" * 64)])
finally:
self.agent.listening, self.agent.our_daemons = old, old_ours
self.assertFalse(self.agent.ROOT.exists())
def test_asks_for_packages_it_was_not_sent(self):
self.agent.listening, old = (lambda: False), self.agent.listening
try:
for folder in ("", str(self.dir / "gone")): # none sent, or already tidied away
with self.assertRaises(self.agent.NeedPackages):
self.agent.ensure_daemon(folder, [("new.pkg.tar.zst", "1" * 64)])
finally:
self.agent.listening = old
def test_restart_can_still_unpack_its_copy_then_tidies_it(self):
folder = self.agent.BASE / "incoming/abc"
folder.mkdir(parents=True)
seen = []
stubs = {"ensure_daemon": lambda f, p: seen.append(Path(f).is_dir()),
"identity": lambda c: ("id", "cert", "key"), "our_daemons": lambda: [123],
"connect": lambda *a: (_ for _ in ()).throw(OSError("no answer"))}
saved = {k: getattr(self.agent, k) for k in stubs}
self.addCleanup(lambda: [setattr(self.agent, k, v) for k, v in saved.items()])
for k, v in stubs.items():
setattr(self.agent, k, v)
self.assertEqual(self.agent.run("c", "n", str(folder), []), 1)
self.assertEqual(seen, [True, True]) # there for the first start and the restart
self.assertFalse(folder.exists())
def test_tidy_keeps_copies_in_use(self):
incoming = self.agent.BASE / "incoming"
held_dir = incoming / "held"
held_dir.mkdir(parents=True)
held = self.agent.hold_incoming(str(held_dir))
self.addCleanup(held.close)
os.utime(held_dir, (time.time() - 7200,) * 2)
self.agent.tidy_incoming("")
self.assertTrue(held_dir.is_dir())
def test_tidy_keeps_other_starts_copies(self):
incoming = self.agent.BASE / "incoming"
mine, theirs, stale = incoming / "mine", incoming / "theirs", incoming / "stale"
for d in (mine, theirs, stale):
d.mkdir(parents=True)
os.utime(stale, (time.time() - 7200,) * 2)
self.agent.tidy_incoming(str(mine))
self.assertEqual(sorted(p.name for p in incoming.iterdir()), ["theirs"])
def test_server_and_agent_agree_on_the_stamp(self):
import server
packages = [("a.pkg.tar.zst", "1" * 64), ("b.pkg.tar.zst", "2" * 64)]
self.assertEqual(server.kdeconnect_stamp(packages), self.agent.stamp(packages))
def setattr_all(module, saved):
module.BASE, module.ROOT, module.stop_daemon, module.say = saved
class FakeKdeConnect:
"""Just enough of kdeconnectd: pairs when asked and records remote-input packets."""
def __init__(self, paired=False):
self.listener = socket.socket()
self.listener.bind(("127.0.0.1", 0))
self.listener.listen(1)
self.port = self.listener.getsockname()[1]
self.paired, self.identity, self.received, self.pair_requests = paired, None, [], 0
threading.Thread(target=self.serve, daemon=True).start()
def serve(self):
conn, _ = self.listener.accept()
line = b""
while not line.endswith(b"\n"):
line += conn.recv(1)
self.identity = json.loads(line)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname, ctx.verify_mode = False, ssl.CERT_NONE
tls = ctx.wrap_socket(conn) # "Starting client ssl (but I'm the server TCP socket)"
tls.sendall(b'{"id": 1, "type": "kdeconnect.mousepad.keyboardstate", "body": {"state": true}}\n')
buf = b""
while True:
try:
chunk = tls.recv(65536)
except OSError:
return
if not chunk:
return
buf += chunk
while b"\n" in buf:
raw, buf = buf.split(b"\n", 1)
p = json.loads(raw)
if p["type"] == "kdeconnect.pair":
self.pair_requests += 1
elif p["type"] == "kdeconnect.mousepad.request" and self.paired:
self.received.append(p["body"])
@unittest.skipIf(sys.platform == "win32", "the agent runs on the Frame (Linux)")
@unittest.skipUnless(shutil.which("openssl"), "needs openssl to make a certificate")
class AgentProtocol(unittest.TestCase):
@classmethod
def setUpClass(cls):
import frame_input_agent
cls.agent = frame_input_agent
cls.dir = tempfile.TemporaryDirectory()
cls.cert, cls.key = Path(cls.dir.name, "cert.pem"), Path(cls.dir.name, "key.pem")
subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1",
"-subj", "/CN=framecontrol_test", "-keyout", str(cls.key), "-out", str(cls.cert)],
check=True, capture_output=True)
@classmethod
def tearDownClass(cls):
cls.dir.cleanup()
def wait_for(self, check):
for _ in range(100):
if check():
return
time.sleep(0.02)
self.fail("timed out")
def test_pairs_then_forwards_events(self):
fake = FakeKdeConnect()
link = self.agent.Link("framecontrol_0123456789abcdef01234567", self.cert, self.key, port=fake.port)
link.pair(lambda: fake.paired, lambda: setattr(fake, "paired", fake.pair_requests > 0), timeout=5)
self.assertEqual(fake.pair_requests, 1)
ident = fake.identity["body"]
self.assertEqual(fake.identity["type"], "kdeconnect.identity")
self.assertEqual(ident["protocolVersion"], 7)
self.assertIn("kdeconnect.mousepad.request", ident["outgoingCapabilities"])
link.read(0.5)
self.assertTrue(link.keyboard)
for body in self.agent.events(b'[{"dx": 4, "dy": -2}, {"key": "hi"}]'):
link.send(body)
self.wait_for(lambda: len(fake.received) == 2)
self.assertEqual(fake.received, [{"dx": 4, "dy": -2}, {"key": "hi"}])
def test_already_paired_sends_no_pair_request(self):
# A pair request to a device that's already paired makes KDE Connect unpair it.
fake = FakeKdeConnect(paired=True)
link = self.agent.Link("framecontrol_0123456789abcdef01234567", self.cert, self.key, port=fake.port)
link.pair(lambda: True, lambda: self.fail("accepted a pairing that wasn't needed"))
link.send({"singleclick": True})
self.wait_for(lambda: fake.received == [{"singleclick": True}])
self.assertEqual(fake.pair_requests, 0)
def test_client_args_are_folder_safe(self):
old = sys.argv
try:
sys.argv = ["-c", "../../etc x", "Alex's Mac"]
self.assertEqual(self.agent.client_args(), ("etcx", "Frame Control (Alex's Mac)", "", []))
sys.argv = ["-c"]
self.assertEqual(self.agent.client_args(), ("default", "Frame Control", "", []))
sys.argv = ["-c", "mac", "Mac", "~/x", '[["a.pkg.tar.zst", "ab"]]']
self.assertEqual(self.agent.client_args()[2:], (os.path.expanduser("~/x"), [("a.pkg.tar.zst", "ab")]))
finally:
sys.argv = old
def test_events_parsing(self):
self.assertEqual(self.agent.events(b'{"dx": 1}'), [{"dx": 1}])
self.assertEqual(self.agent.events(b'[{"dx": 1}, 5, {}]'), [{"dx": 1}])
self.assertEqual(self.agent.events(b"not json"), [])
@unittest.skipUnless(shutil.which("node"), "needs node")
class PageQueue(unittest.TestCase):
"""The page's input queue (tests/page/pad_queue.mjs runs the real functions from index.html)."""
def test_queue_rules(self):
r = subprocess.run(["node", str(ROOT / "tests" / "page" / "pad_queue.mjs")], capture_output=True, text=True)
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
if __name__ == "__main__":
unittest.main()
+2 -1
View File
@@ -304,7 +304,8 @@ class Renderer(unittest.TestCase):
def test_desktop_packages_include_renderer_and_settings(self):
config = json.loads((ROOT / 'app/package.json').read_text())
resources = {r['from']: r['filter'] for r in config['build']['extraResources']}
# Single-file resources (licenses/notices) do not need a filter.
resources = {r['from']: r.get('filter', ['**/*']) for r in config['build']['extraResources']}
self.assertIn('*.js', resources['../ui'])
self.assertIn('*.js', resources['../frame/android'])
+421
View File
@@ -0,0 +1,421 @@
"""Mac in the headset (ui/frame_macview.py and the frame-mac-view agent).
The Python checks run anywhere. On macOS the agent is built and driven over
HTTP and WebSocket with its test pattern, which needs no Screen Recording
permission: status, the token, the viewer page, H.264 keyframes, pointer
input reaching the source, and closing.
Run: python3 -m unittest discover -s tests
"""
import base64
import http.client
import json
import os
import shutil
import socket
import struct
import subprocess
import sys
import tempfile
import time
import threading
import unittest
from unittest import mock
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "ui"))
import frame_macview # noqa: E402
class Helpers(unittest.TestCase):
def test_panel_id_is_stable_and_in_range(self):
a = frame_macview.panel_id("window:123")
self.assertEqual(a, frame_macview.panel_id("window:123"))
self.assertNotEqual(a, frame_macview.panel_id("window:124"))
self.assertTrue(2_001_000_000 <= a < 2_002_000_000)
def test_fit_keeps_aspect_inside_the_panel(self):
self.assertEqual(frame_macview.fit(2560, 1440), (1920, 1080))
w, h = frame_macview.fit(800, 1600)
self.assertEqual(h, 1080)
self.assertAlmostEqual(w / h, 0.5, places=2)
@unittest.skipUnless(shutil.which("bash"), "needs bash")
@unittest.skipIf(os.name == "nt", "Windows' bash.exe is WSL's launcher, and runners have no distribution")
def test_launch_script_parses(self):
r = subprocess.run(["bash", "-n"], input=frame_macview.LAUNCH, text=True, capture_output=True)
self.assertEqual(r.returncode, 0, r.stderr)
def test_show_checks_the_source_before_anything_else(self):
mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
with self.assertRaises(frame_macview.MacViewError):
mv.show("rm -rf /")
def test_missing_browser_is_explained(self):
calls = []
def run(remote, stdin=None, timeout=30):
calls.append(remote)
e = RuntimeError("exit 3")
e.stdout = "NO_BROWSER\n"
raise e
mv = frame_macview.MacView(["ssh"], run, "frame")
mv.call = lambda path, **kw: {"screen": True, "ticket": "tk"}
mv.ensure_tunnel = lambda **kw: None
mv.remote_port = 47900
with self.assertRaises(frame_macview.MacViewError) as cm:
mv.show("window:5")
self.assertIn("Chromium", str(cm.exception))
self.assertTrue(calls[0].startswith("bash -s -- "))
def test_separate_windows_need_accessibility(self):
mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
mv.call = lambda path, **kw: {"screen": True, "accessibility": False}
with self.assertRaises(frame_macview.MacViewError) as cm:
mv.show("separate:42")
self.assertIn("Accessibility", str(cm.exception))
def test_screen_permission_is_checked_before_the_headset(self):
mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
mv.call = lambda path, **kw: {"screen": False}
with self.assertRaises(frame_macview.MacViewError) as cm:
mv.show("display:1")
self.assertIn("Screen Recording", str(cm.exception))
def test_stop_all_ends_the_viewer_browser_unless_shown_again(self):
calls = []
mv = frame_macview.MacView(["ssh"], lambda remote, **kw: calls.append(remote) or "", "frame")
mv.agent = mock.Mock(poll=lambda: None)
mv.call = lambda path, **kw: {"closed": 1}
mv.shown = {"window:5"}
with mock.patch.object(frame_macview.time, "sleep"):
gen = mv.shows
mv.shown.clear()
mv._end_viewer_browser(gen)
self.assertEqual(len(calls), 1)
self.assertIn("pkill -f '[f]rame-control/mac-view", calls[0])
mv.shows += 1 # Show pressed during the wait: leave the new viewer alone
mv._end_viewer_browser(gen)
self.assertEqual(len(calls), 1)
mv.launching = 1 # a Show replacing its own stream is still launching
mv._end_viewer_browser(mv.shows)
self.assertEqual(len(calls), 1)
# A Show waits while the cleanup checks and runs pkill.
mv.launching = 0
in_pkill, release, entered = threading.Event(), threading.Event(), threading.Event()
def blocking_pkill(remote, **kw):
in_pkill.set()
release.wait(5)
mv.run = blocking_pkill
mv._show = lambda *a: entered.set() or "shown"
with mock.patch.object(frame_macview.time, "sleep"):
cleanup = threading.Thread(target=mv._end_viewer_browser, args=(mv.shows,))
cleanup.start()
self.assertTrue(in_pkill.wait(2))
shower = threading.Thread(target=mv.show, args=("window:5",))
shower.start()
self.assertFalse(entered.wait(0.3), "Show started while the cleanup held the lock")
release.set()
self.assertTrue(entered.wait(2))
cleanup.join()
shower.join()
def test_tunnel_prefers_the_usb_c_network_when_plugged_in(self):
mv = frame_macview.MacView(["ssh"], lambda remote, **kw: "13: usb0 inet 10.86.200.233/29 scope global", "frame")
ssh_g = mock.Mock(stdout="user steamos\nhostname frame.example.ts.net\n")
with mock.patch.object(frame_macview.socket, "create_connection") as conn, \
mock.patch.object(frame_macview.subprocess, "run", return_value=ssh_g):
self.assertEqual(mv._usb_route(), ["-o", "HostName=10.86.200.233", "-o", "HostKeyAlias=frame.example.ts.net"])
conn.assert_called_once_with(("10.86.200.233", 22), timeout=1)
ssh_g.stdout = "hostname frame.example.ts.net\nhostkeyalias paired-frame\n" # a configured alias wins
conn.side_effect = None
self.assertIn("HostKeyAlias=paired-frame", mv._usb_route())
conn.side_effect = OSError("unplugged")
self.assertEqual(mv._usb_route(), [])
mv.run = lambda remote, **kw: "" # no usb0
self.assertEqual(mv._usb_route(), [])
mv.prefer_usb = False
mv.run = lambda remote, **kw: self.fail("no ssh when USB is off")
self.assertEqual(mv._usb_route(), [])
def test_a_failed_usb_tunnel_falls_back_to_the_network(self):
mv = frame_macview.MacView(["ssh"], lambda *a, **k: "", "frame")
mv._usb_route = lambda: ["-o", "HostName=10.86.200.233"]
tried = []
def attempt(via, ports):
tried.append(list(via))
mv._last_tunnel_error = "Connection refused"
return not via # USB fails, the normal path works
mv._open_tunnel = attempt
mv.tunnel_up = lambda: False
mv.ensure_tunnel()
self.assertEqual(tried, [["-o", "HostName=10.86.200.233"], []])
self.assertEqual(mv.route, "network")
class WS:
"""A minimal WebSocket client (masked frames out, plain frames in)."""
def __init__(self, port, path):
self.s = socket.create_connection(("127.0.0.1", port), timeout=10)
key = base64.b64encode(os.urandom(16)).decode()
self.s.sendall(f"GET {path} HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n".encode())
head = b""
while b"\r\n\r\n" not in head:
head += self.s.recv(1)
self.status = int(head.split()[1])
self.buf = b""
def _read(self, n):
while len(self.buf) < n:
chunk = self.s.recv(65536)
if not chunk:
raise EOFError
self.buf += chunk
out, self.buf = self.buf[:n], self.buf[n:]
return out
def recv(self):
b0, b1 = self._read(2)
n = b1 & 0x7F
if n == 126:
n = struct.unpack(">H", self._read(2))[0]
elif n == 127:
n = struct.unpack(">Q", self._read(8))[0]
return b0 & 0x0F, self._read(n)
def send_text(self, text):
data, mask = text.encode(), os.urandom(4)
head = bytes([0x81, 0x80 | len(data)]) if len(data) < 126 else bytes([0x81, 0xFE]) + struct.pack(">H", len(data))
self.s.sendall(head + mask + bytes(c ^ mask[i % 4] for i, c in enumerate(data)))
def close(self):
self.s.close()
@unittest.skipUnless(sys.platform == "darwin" and shutil.which("xcrun"), "the agent is macOS-only")
class Agent(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.tmp = tempfile.mkdtemp()
cls.bin = Path(cls.tmp) / "frame-mac-view"
r = subprocess.run(["/bin/sh", str(ROOT / "mac" / "frame-mac-view" / "build.sh"), str(cls.bin)],
capture_output=True, text=True, timeout=600)
if r.returncode: # a real failure on a Mac with Xcode: don't hide it as a skip
raise AssertionError("agent didn't build:\n" + (r.stderr or r.stdout)[-2000:])
cls.token = "t0ken-" + os.urandom(6).hex()
cls.proc = subprocess.Popen([str(cls.bin), "serve", "--port", "0", "--page", str(ROOT / "ui" / "mac-view.html"),
"--exit-on-eof"], env={**os.environ, "FRAME_MAC_VIEW_TOKEN": cls.token},
stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
line = cls.proc.stdout.readline()
cls.port = int(line.rsplit(":", 1)[1])
@classmethod
def tearDownClass(cls):
cls.proc.stdin.close() # --exit-on-eof
cls.proc.stdout.close()
try:
cls.proc.wait(5)
except subprocess.TimeoutExpired:
cls.proc.kill()
shutil.rmtree(cls.tmp, ignore_errors=True)
def get(self, path, method="GET"):
c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
c.request(method, path)
r = c.getresponse()
return r.status, r.read()
def test_token_is_required(self):
self.assertEqual(self.get("/status")[0], 403)
self.assertEqual(self.get("/status?k=wrong")[0], 403)
status, body = self.get(f"/status?k={self.token}")
self.assertEqual(status, 200)
self.assertIn("screen", json.loads(body))
def test_serves_the_viewer_page(self):
status, body = self.get(f"/view?k={self.token}&src=test")
self.assertEqual(status, 200)
self.assertIn(b"VideoDecoder", body)
def test_snapshot_needs_the_key(self):
self.assertEqual(self.get("/snapshot?display=1")[0], 403)
def test_separate_without_accessibility_explains(self):
if json.loads(self.get(f"/status?k={self.token}")[1])["accessibility"]:
self.skipTest("this Mac allows Accessibility here")
ws = WS(self.port, f"/stream?k={self.token}&src=separate:1")
self.assertEqual(ws.status, 101)
for _ in range(5):
op, data = ws.recv()
msg = json.loads(data) if op == 1 else {}
if msg.get("t") in ("error", "closed"):
break
self.assertIn("Accessibility", msg.get("message", msg.get("reason", "")))
ws.close()
def test_lists_displays(self):
status, body = self.get(f"/displays?k={self.token}")
self.assertEqual(status, 200)
self.assertIsInstance(json.loads(body)["displays"], list)
def ticket(self, src):
status, body = self.get(f"/ticket?k={self.token}&src={src}", method="POST")
self.assertEqual(status, 200)
return json.loads(body)["ticket"]
def test_ping_and_page_are_open_but_streams_are_not(self):
self.assertEqual(self.get("/ping"), (200, b"frame-mac-view"))
self.assertEqual(WS(self.port, "/stream?src=test").status, 403)
self.assertEqual(self.get("/ticket?src=test", method="POST")[0], 403)
def test_tickets_are_single_use_and_tied_to_one_source(self):
t = self.ticket("test")
self.assertEqual(WS(self.port, f"/stream?src=display:1&t={t}").status, 403) # wrong source
ws = WS(self.port, f"/stream?src=test&t={t}")
self.assertEqual(ws.status, 101)
hello = json.loads(ws.recv()[1])
self.assertEqual(hello["t"], "hello")
# Until the viewer acknowledges, a retry (the hello got lost) gets the
# same key, and replaces the first viewer rather than adding one.
retry = WS(self.port, f"/stream?src=test&t={t}")
self.assertEqual(retry.status, 101)
self.assertEqual(json.loads(retry.recv()[1])["r"], hello["r"])
time.sleep(0.3)
_, body = self.get(f"/status?k={self.token}")
self.assertEqual(len(json.loads(body)["streams"]), 1)
ws.close()
ws = retry
ws.send_text(json.dumps({"t": "ack"}))
time.sleep(0.3)
self.assertEqual(WS(self.port, f"/stream?src=test&t={t}").status, 403) # spent
again = WS(self.port, f"/stream?src=test&r={hello['r']}") # the viewer reconnecting
self.assertEqual(again.status, 101)
again.close()
ws.close()
# Stop revokes the reconnect key.
self.get(f"/close?k={self.token}&src=test", method="POST")
self.assertEqual(WS(self.port, f"/stream?src=test&r={hello['r']}").status, 403)
def test_stop_revokes_tickets_not_yet_used(self):
t = self.ticket("test")
self.get(f"/close?k={self.token}&src=test", method="POST")
self.assertEqual(WS(self.port, f"/stream?src=test&t={t}").status, 403)
def test_bad_frame_lengths_close_the_socket_not_the_agent(self):
ws = WS(self.port, f"/stream?src=test&t={self.ticket('test')}")
self.assertEqual(ws.status, 101)
# A masked frame claiming 2^63 bytes.
ws.s.sendall(bytes([0x81, 0xFF]) + struct.pack(">Q", 1 << 63) + os.urandom(4))
with self.assertRaises((EOFError, OSError)):
for _ in range(1000):
ws.recv()
ws.close()
self.assertEqual(self.get(f"/status?k={self.token}")[0], 200)
def test_stream_input_and_close(self):
ws = WS(self.port, f"/stream?k={self.token}&src=test&codec=h264&fps=30&max=640")
self.assertEqual(ws.status, 101)
info = None
key = None
for _ in range(200):
op, data = ws.recv()
if op == 1:
msg = json.loads(data)
if msg["t"] == "hello":
continue
if msg["t"] == "error":
self.skipTest("no H.264 encoder here: " + msg["message"])
if msg["t"] == "info":
info = msg
elif op == 2 and data[0] == 1:
key = data
if info and key:
break
self.assertEqual(info["src"], "test")
self.assertTrue(info["input"])
# A keyframe: flags, 8-byte timestamp, sequence number, input echoed,
# then Annex B with the SPS first.
self.assertEqual(key[17:21], b"\x00\x00\x00\x01")
self.assertEqual(key[21] & 0x1F, 7) # NAL type 7, SPS
ws.send_text(json.dumps({"t": "m", "e": "down", "b": 0, "x": 0.5, "y": 0.5}))
ws.send_text(json.dumps({"t": "key-frame"}))
got_key = False
for _ in range(200):
op, data = ws.recv()
if op == 2 and data[0] == 1:
got_key = True
break
self.assertTrue(got_key, "no keyframe after asking for one")
_, body = self.get(f"/status?k={self.token}")
self.assertEqual([s["src"] for s in json.loads(body)["streams"]], ["test"])
status, body = self.get(f"/close?k={self.token}", method="POST")
self.assertEqual(json.loads(body)["closed"], 1)
for _ in range(400):
op, data = ws.recv()
if op == 1 and json.loads(data)["t"] == "close":
break
self.assertEqual(json.loads(data)["t"], "close")
# Without the viewer doing anything, the agent ends the stream itself.
time.sleep(1)
_, body = self.get(f"/status?k={self.token}")
self.assertEqual(json.loads(body)["streams"], [])
ws.close()
def test_timing_reports_and_input_echo(self):
# What a viewer does: sync clocks, report each frame, stamp input.
ws = WS(self.port, f"/stream?k={self.token}&src=test&codec=h264&fps=30&max=640")
self.assertEqual(ws.status, 101)
ws.send_text(json.dumps({"t": "w"})) # keep-warm filler: ignored
ws.send_text(json.dumps({"t": "ping", "c": 1.5}))
pong, echoed, seqs, info = None, None, [], None
clicked = False
deadline = time.time() + 10
while time.time() < deadline and not (pong and echoed):
op, data = ws.recv()
if op == 1:
msg = json.loads(data)
if msg["t"] == "error":
self.skipTest("no H.264 encoder here: " + msg["message"])
if msg["t"] == "pong":
pong = msg
if msg["t"] == "info":
info = msg
elif op == 2:
seq, echo = struct.unpack(">II", data[9:17])
seqs.append(seq)
now = pong["a"] if pong else 0
ws.send_text(json.dumps({"t": "rx", "s": seq, "r": now}))
ws.send_text(json.dumps({"t": "fd", "f": [[seq, now + 1, now + 2, now + 3]], "drop": 0}))
if echo == 7:
echoed = seq
if len(seqs) == 3 and not clicked:
ws.send_text(json.dumps({"t": "m", "e": "down", "b": 0, "x": 0.5, "y": 0.5, "i": 7, "tv": now}))
clicked = True
self.assertEqual(pong["c"], 1.5)
self.assertGreater(pong["a"], 0)
self.assertEqual(seqs[:3], sorted(seqs[:3]))
self.assertIsNotNone(echoed, "no frame was tagged as the first reply to the click")
time.sleep(1.7) # frames are reported once the viewer has had time
stream = json.loads(self.get(f"/stats?k={self.token}")[1])["streams"][0]
first = stream["frames"][0]
self.assertGreater(first["e1"], first["e0"])
self.assertGreaterEqual(first["e0"], first["cap"])
self.assertEqual(first["vs"] - first["rx"], 3)
self.assertEqual([i["frame"] for i in stream["inputs"] if i["id"] == 7], [echoed])
self.assertIn("total", stream["summary"])
self.assertEqual(info["warm"], 0) # off unless FRAME_MAC_VIEW_WARM is set
live = json.loads(self.get(f"/status?k={self.token}")[1])["streams"][0]
self.assertEqual(live["controller"]["tier"], 0)
self.assertIn("fps", live["stats"])
ws.close()
if __name__ == "__main__":
unittest.main()
+141
View File
@@ -0,0 +1,141 @@
"""Owned media planning, eye isolation, decoder choice and fake-Frame ownership."""
import json
import os
from pathlib import Path
import struct
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
import frame_media as media
import frame_media_player as player
import frame_media_remote as remote
import frame_splat as splat
import server
class Media(unittest.TestCase):
def test_layout_evidence_and_override(self):
for name, layout in [('film_SBS.mp4', 'sbs'), ('film.OU.mkv', 'ou'),
('film_FSBS.mp4', 'full-sbs'), ('photo_TB.png', 'ou')]:
self.assertEqual(media.plan(name)['layout'], layout)
self.assertEqual(media.plan('film_SBS_OU.mp4', 'mono')['source'], 'explicit')
self.assertEqual(media.plan('film.mkv', metadata={'stereo_mode': 'top_bottom'})['layout'], 'full-ou')
for name in ('film.mp4', 'film_SBS_OU.mp4', 'businessbs.mp4'):
with self.assertRaises(ValueError):
media.plan(name)
with self.assertRaises(ValueError):
media.plan('film.mkv', metadata={'stereo_mode': 'right_left'})
def test_unsupported_containers_do_not_flatten_spatial_photos(self):
for name in ('spatial.HEIC', 'stereo.mpo', 'cloud.ply', 'cloud.spz', 'app.exe'):
with self.assertRaises(ValueError):
media.plan(name, 'sbs')
def test_ou_pixels_keep_each_eye_and_row(self):
a, b, c, d = [bytes([n])*8 for n in (1, 2, 3, 4)]
data, width, height = media.stereo_pixels(a+b+c+d, 2, 4, 'ou')
self.assertEqual((data, width, height), (a+c+b+d, 4, 2))
with self.assertRaises(ValueError):
media.stereo_pixels(b'bad', 2, 4, 'sbs')
self.assertEqual(media.geometry(3840, 2160, 'sbs'), (1920, 1080, 2))
self.assertEqual(media.geometry(1920, 1080, 'ou'), (1920, 1080, .5))
def test_decode_is_hardware_and_one_clock_for_audio(self):
for codec, decoder in [('h264', 'h264_v4l2m2m'), ('hevc', 'hevc_v4l2m2m')]:
cmd = player.decoder_command(Path('/tmp/a file.mp4'), {'codec_name': codec}, 1280, 720, True)
self.assertIn(decoder, cmd)
self.assertIn('-re', cmd)
self.assertIn('pulse', cmd)
self.assertIn(str(Path('/tmp/a file.mp4')), cmd)
with self.assertRaises(ValueError):
player.decoder_command(Path('x.webm'), {'codec_name': 'vp9'}, 640, 480, False)
cmd = player.decoder_command(Path('x.png'), {'codec_name': 'png'}, 640, 480, False, True)
self.assertNotIn('-re', cmd)
self.assertIn('-frames:v', cmd)
def test_fake_frame_library_and_traversal(self):
with tempfile.TemporaryDirectory() as d, patch.object(remote, 'ROOT', Path(d)):
identity = 'a'*32+'/space and quote\'.png'
path = Path(d)/identity
path.parent.mkdir()
path.write_bytes(b'test')
self.assertEqual(remote.media_path(identity), path.resolve())
for bad in ('../../etc/passwd', '/etc/passwd', 'a'*32+'/..', 'a'*32+'/x/y', None):
with self.assertRaises((ValueError, FileNotFoundError)):
remote.media_path(bad)
link = path.parent/'link.png'
link.symlink_to(path)
with self.assertRaises(ValueError):
remote.media_path('a'*32+'/link.png')
with patch.object(remote, 'status', return_value={'state': 'idle'}):
files = remote.run({'action': 'list'})['files']
self.assertEqual([f['id'] for f in files], [identity])
def test_server_rejects_bad_actions_before_ssh(self):
with patch.object(server, 'ssh') as ssh:
for body in ({'action': 'delete'}, {'action': 'play', 'id': '../x'},
{'action': 'play', 'id': 'a'*32+'/x', 'layout': 'invalid'},
{'action': 'play', 'id': 'a'*32+'/x', 'theatre': 'false'}):
with self.assertRaises(server.Failure):
server.media(body)
ssh.assert_not_called()
def test_fake_frame_stop_only_owns_our_unit(self):
for rc in (0, 5): # 5: already collected ("not loaded"), a no-op
with patch.object(remote.subprocess, 'run') as run, patch.object(remote, 'status', return_value={'state': 'ended'}):
run.return_value.returncode = rc
self.assertEqual(remote.run({'action': 'stop'})['state'], 'ended')
self.assertEqual(run.call_args.args[0], ['systemctl', '--user', 'stop', 'frame-control-media.service'])
with patch.object(remote.subprocess, 'run') as run, patch.object(remote, 'status', return_value={}):
run.return_value.returncode, run.return_value.stderr = 1, 'Access denied'
with self.assertRaisesRegex(RuntimeError, 'Access denied'):
remote.run({'action': 'stop'})
def test_start_failure_reports_systemd_error(self):
with tempfile.TemporaryDirectory() as d, patch.object(remote, 'ROOT', Path(d)), \
patch.object(remote, 'STATUS', Path(d)/'status.json'), \
patch.object(remote, 'active', return_value=False), \
patch.object(remote.subprocess, 'run') as run:
identity = 'b'*32+'/still_SBS.png'
(Path(d)/identity).parent.mkdir()
(Path(d)/identity).write_bytes(b'x')
run.return_value.returncode, run.return_value.stderr = 1, 'Unit already exists'
with patch.object(remote, 'probe', return_value=({}, False)), \
self.assertRaisesRegex(RuntimeError, 'Unit already exists'):
remote.run({'action': 'play', 'id': identity})
self.assertEqual(run.call_args.args[0][0], 'systemd-run') # reset-failed's result is ignored
def test_upload_rejects_unplayable_names_and_keeps_copy_error(self):
with patch.object(server, 'ssh') as ssh, patch.object(server, 'push_file') as push:
# On Windows a backslash is a separator, so such a name can't reach here.
for name in ('.hidden.mp4',) + (('a\\b_SBS.mp4',) if os.sep == '/' else ()):
with self.assertRaises(server.Failure):
server.push_media(Path('/tmp')/name)
ssh.assert_not_called()
push.side_effect = server.Failure('copy failed')
ssh.side_effect = [None, server.Failure('link down')]
with self.assertRaisesRegex(server.Failure, 'copy failed'):
server.push_media(Path('/tmp/film_SBS.mp4'))
def test_splat_invalid_records_and_stereo_parallax(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d)/'small.splat'
path.write_bytes(struct.pack('<6f8B', 0, 0, 0, .001, .001, .001,
255, 0, 0, 255, 255, 128, 128, 128))
data, w, h = splat.render(path, 64, 48)
self.assertEqual((len(data), w, h), (128*48*4, 128, 48))
def centroid(eye):
weights = [(x, data[(y*w+x+eye*64)*4]) for y in range(h) for x in range(64)]
return sum(x*v for x,v in weights)/sum(v for x,v in weights)
self.assertGreater(centroid(0), centroid(1))
for bad in (b'', b'bad', struct.pack('<6f8B', float('nan'), 0, 0, 1, 1, 1, *([128]*8))):
path.write_bytes(bad)
with self.assertRaises(ValueError):
splat.read(path)
if __name__ == '__main__':
unittest.main()
+102
View File
@@ -0,0 +1,102 @@
"""Fake-Frame panel responses and the headset page's real HTTP guards."""
import http.client
import json
from pathlib import Path
import sys
import threading
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'ui'))
import frame_panels as panels
# Shape verified with vrcmd on SteamVR 2.18.1 / BUILD_ID 20260925.6191901.
OVERLAYS = """---- OVERLAYS ----
'valve.steam.desktopgame.12' -- 'Alex's <notes>', 1920x1080 visible VROverlayType_Dashboard_Main
'valve.steam.desktopgame.12.thumb' -- 'Thumb', not_visible VROverlayType_Dashboard_Thumbnail
'valve.steam.desktopgame.12.layer1' -- 'Layer', visible VROverlayType_Subview
'system.pointer' -- 'Pointer', visible VROverlayType_Basic
'valve.steam.desktopgame.13' -- 'Other', not_visible VROverlayType_Dashboard_Main
"""
class Panels(unittest.TestCase):
def test_enumerates_only_main_panels_including_hidden(self):
rows = panels.parse_overlays(OVERLAYS)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[0]['title'], "Alex's <notes>")
self.assertTrue(rows[0]['visible'])
self.assertFalse(rows[1]['visible'])
def test_unavailable_runtime_is_not_an_empty_workspace(self):
with self.assertRaises(panels.PanelError):
panels.parse_overlays('SteamVR not running')
self.assertEqual(panels.parse_overlays('---- OVERLAYS ----'), [])
@patch.object(panels, 'run', return_value=OVERLAYS)
def test_focus_revalidates_and_dispatches_without_shell(self, run):
key = 'valve.steam.desktopgame.12'
self.assertEqual(panels.focus(key)['requested'], key)
self.assertEqual(run.call_args.args[0], [panels.VRCMD, '--showdashboard', key])
@patch.object(panels, 'run', return_value=OVERLAYS)
def test_closed_or_injected_panel_never_dispatches(self, run):
for key in (None, 12, 'x;touch /tmp/bad', '../other', 'missing'):
with self.assertRaises(panels.PanelError):
panels.focus(key)
self.assertEqual(run.call_count, 1) # only valid-looking 'missing' enumerates
class HeadsetHTTP(unittest.TestCase):
def setUp(self):
self.server = panels.HTTPServer(('127.0.0.1', 0), panels.Handler)
self.server.key = 'test-key'
self.server.closing = False
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
def tearDown(self):
self.server.shutdown()
self.thread.join()
self.server.server_close()
def request(self, path, body=None, headers=None):
c = http.client.HTTPConnection('127.0.0.1', self.server.server_port, timeout=5)
try:
c.request('POST' if body is not None else 'GET', path, body=body, headers=headers or {})
r = c.getresponse()
return r.status, r.read().decode()
finally:
c.close()
def test_page_is_public_but_contains_no_key_or_private_titles(self):
status, page = self.request('/')
self.assertEqual(status, 200)
self.assertNotIn('test-key', page)
self.assertIn('textContent=p.title', page) # titles never become HTML
self.assertEqual(self.request('/panels')[0], 403)
@patch.object(panels, 'state', return_value={'panels': []})
def test_auth_host_and_origin_checks(self, state):
auth = {'X-Panel-Key': 'test-key'}
self.assertEqual(self.request('/panels', headers=auth)[0], 200)
for extra in ({'Host': 'evil.test'}, {'Origin': 'https://evil.test'}, {'X-Panel-Key': 'wrong'}):
self.assertEqual(self.request('/panels', headers={**auth, **extra})[0], 403)
self.assertEqual(state.call_count, 1)
@patch.object(panels, 'focus', return_value={'requested': 'panel'})
def test_post_validation_and_close(self, focus):
auth = {'X-Panel-Key': 'test-key'}
for body in ('[]', '{broken', '0', '"text"', 'x' * 1025):
self.assertEqual(self.request('/focus', body, auth)[0], 400)
self.assertEqual(focus.call_count, 0)
self.assertEqual(self.request('/focus', '{"key":"panel"}', auth)[0], 200)
self.assertEqual(focus.call_args.args, ('panel',))
self.assertEqual(self.request('/close', '{}', auth)[0], 200)
self.assertTrue(self.server.closing)
@patch.object(panels, 'state', side_effect=panels.PanelError('offline'))
def test_offline_is_an_error_not_a_successful_empty_list(self, state):
status, body = self.request('/panels', headers={'X-Panel-Key': 'test-key'})
self.assertEqual(status, 502)
self.assertEqual(json.loads(body), {'error': 'offline'})
+4
View File
@@ -5,6 +5,7 @@ request guards and input validation, which all run before any SSH call.
Run: python3 -m unittest discover -s tests
"""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import http.client
import io
import json
@@ -83,6 +84,7 @@ class ServerGuards(unittest.TestCase):
def test_api_needs_custom_header(self):
# <img src> and plain form posts from other sites can't set it.
self.assertEqual(self.request("POST", "/api/comfort", {"action": "start"})[0], 403)
self.assertEqual(self.request("GET", "/api/status")[0], 403)
self.assertEqual(self.request("GET", "/api/screenshot?view=headset")[0], 403)
self.assertEqual(self.request("GET", "/api/shots")[0], 403)
@@ -98,6 +100,8 @@ class ServerGuards(unittest.TestCase):
def test_input_validation(self):
cases = [
("/api/comfort", {"action": "poweroff"}),
("/api/comfort", {"action": "start", "minutes": 0}),
("/api/launch", {"appid": "620; rm -rf ~"}),
("/api/launch", {"appid": ""}),
("/api/flatpak", {"id": "org.example.App;id", "action": "install"}),
+35
View File
@@ -2,6 +2,7 @@
Run: python3 -m unittest discover -s tests
"""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import json
import subprocess
import sys
@@ -13,6 +14,7 @@ ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "ui"))
import frame_store # noqa: E402
import frame_steam # noqa: E402
import test_server # noqa: E402 (not `from … import`, or unittest runs ServerGuards twice)
@@ -49,6 +51,39 @@ class FrameSteamHelper(unittest.TestCase):
self.assertEqual(out.returncode, 1, args)
self.assertIn("error", json.loads(out.stdout), args)
def test_installed_game_is_not_reinstalled(self):
with mock.patch.object(frame_steam, "Page") as page, \
mock.patch.object(frame_steam, "steam_url") as launch:
page.return_value.eval.return_value = {"name": "Gravitas", "installed": True}
self.assertEqual(frame_steam.install(1067310)["state"], "installed")
launch.assert_not_called()
def test_license_and_eula_wait_for_headset(self):
# Seen during #26's free Gravitas install: state 3 is not permission
# to click through the license. The same guard applies to an EULA.
for state in (3, 8):
with self.subTest(state=state), \
mock.patch.object(frame_steam, "Page") as page, \
mock.patch.object(frame_steam, "steam_url") as launch, \
mock.patch.object(frame_steam.time, "sleep"):
page.return_value.eval.side_effect = [
None, {"app": 1067310, "state": state, "need": 1, "free": 100}]
self.assertEqual(frame_steam.install(1067310)["state"], "headset")
launch.assert_called_once_with("steam://install/1067310")
self.assertNotIn(mock.call("SteamClient.Installs.ContinueInstall()"),
page.return_value.eval.call_args_list)
def test_insufficient_space_leaves_options_open(self):
with mock.patch.object(frame_steam, "Page") as page, \
mock.patch.object(frame_steam, "steam_url"), \
mock.patch.object(frame_steam.time, "sleep"):
page.return_value.eval.side_effect = [
{"name": "Gravitas", "installed": False},
{"app": 1067310, "state": 7, "need": 100, "free": 10}]
self.assertEqual(frame_steam.install(1067310)["state"], "headset")
self.assertNotIn(mock.call("SteamClient.Installs.ContinueInstall()"),
page.return_value.eval.call_args_list)
class HelperErrors(unittest.TestCase):
def test_json_error_found_despite_ssh_stderr(self):
+448
View File
@@ -0,0 +1,448 @@
"""Anonymous analytics (ui/frame_telemetry.py): what's collected at each level,
what's scrubbed, and that nothing is sent without a key, the notice, or consent.
Run: python3 -m unittest discover -s tests
"""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import json
import os
import sys
import tempfile
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "ui"))
import frame_compat_db as db # noqa: E402
import frame_report as fr # noqa: E402
import frame_telemetry as tm # noqa: E402
class Base(unittest.TestCase):
"""A packaged build with a key, its state in a temp folder."""
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
state = Path(tmp.name)
for name, value in (("STATE", state), ("SETTINGS", state / "settings.json"),
("OUTBOX", state / "outbox.jsonl"), ("SENT", state / "sent.jsonl")):
p = mock.patch.object(tm, name, value)
p.start()
self.addCleanup(p.stop)
env = mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_KEY": "phc_test", "FRAME_CONTROL_PACKAGED": "1",
"FRAME_CONTROL_POSTHOG_HOST": "http://127.0.0.1:9",
"FRAME_CONTROL_VERSION": "9.9.9"})
env.start()
self.addCleanup(env.stop)
for k in ("DO_NOT_TRACK", "FRAME_CONTROL_TELEMETRY"):
os.environ.pop(k, None)
tm._seen_errors.clear()
def queued(self):
return tm._read_lines(tm.OUTBOX)
class Gates(Base):
def test_blocked_without_key_or_in_a_checkout_or_by_do_not_track(self):
self.assertIsNone(tm.blocked())
with mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_KEY": ""}), \
mock.patch.object(tm, "HERE", Path(tempfile.gettempdir()) / "no-config-here"):
self.assertIn("key", tm.blocked())
with mock.patch.dict(os.environ, {"FRAME_CONTROL_PACKAGED": ""}):
self.assertIn("source checkout", tm.blocked())
with mock.patch.dict(os.environ, {"DO_NOT_TRACK": "1"}):
self.assertIn("DO_NOT_TRACK", tm.blocked())
self.assertFalse(tm.capture("app_opened"))
self.assertFalse(tm.OUTBOX.exists())
def test_usage_is_on_by_default_the_others_are_opt_in(self):
self.assertTrue(tm.capture("app_opened"))
self.assertFalse(tm.capture("compat_report", {}, level="compat"))
self.assertFalse(tm.capture("$exception", {}, level="diagnostics"))
self.assertEqual([e["event"] for e in self.queued()], ["app_opened"])
def test_events_are_anonymous(self):
tm.capture("app_opened")
e = self.queued()[0]
self.assertEqual(e["distinct_id"], tm.settings()["id"])
self.assertIs(e["properties"]["$process_person_profile"], False)
self.assertIs(e["properties"]["$geoip_disable"], True)
self.assertEqual(e["properties"]["app_version"], "9.9.9")
def test_turning_a_level_off_drops_its_unsent_events(self):
tm.update_settings({"diagnostics": True})
tm.capture("app_opened")
tm.diagnostic("somewhere", RuntimeError("boom"))
self.assertEqual(len(self.queued()), 2)
tm.update_settings({"diagnostics": False})
self.assertEqual([e["event"] for e in self.queued()], ["app_opened"])
tm.update_settings({"usage": False})
self.assertEqual(self.queued(), [])
self.assertFalse(tm.capture("app_opened"))
def test_the_same_error_is_sent_once_in_a_while(self):
tm.update_settings({"diagnostics": True})
for _ in range(3):
tm.diagnostic("POST /api/android install", RuntimeError("boom"))
self.assertEqual(len(self.queued()), 1)
def test_page_events_are_checked(self):
self.assertTrue(tm.page_event({"event": "tab_viewed", "properties": {"tab": "android", "extra": "x"}})["queued"])
self.assertEqual(self.queued()[0]["properties"].get("extra"), None)
with self.assertRaises(ValueError):
tm.page_event({"event": "anything_else"})
with self.assertRaises(ValueError):
tm.page_event({"event": "tab_viewed", "properties": {"tab": "/Users/me/secret"}})
class Lifecycle(Base):
def test_install_update_and_one_open_a_day(self):
tm.app_started()
tm.app_started()
self.assertEqual([e["event"] for e in self.queued()], ["app_installed", "app_opened"])
with mock.patch.dict(os.environ, {"FRAME_CONTROL_VERSION": "10.0.0"}):
tm.app_started()
e = self.queued()[-1]
self.assertEqual((e["event"], e["properties"]["from_version"]), ("app_updated", "9.9.9"))
def test_frame_build_once(self):
tm.frame_seen("20260922.1", "3.8")
tm.frame_seen("20260922.1", "3.8")
self.assertEqual(len(self.queued()), 1)
class Sending(Base):
def serve(self, status=200):
got = []
class H(BaseHTTPRequestHandler):
def do_POST(self):
got.append((self.path, json.loads(self.rfile.read(int(self.headers["Content-Length"])))))
self.send_response(status)
self.end_headers()
self.wfile.write(b'{"status": 1}')
def log_message(self, *a):
pass
httpd = HTTPServer(("127.0.0.1", 0), H)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
self.addCleanup(httpd.server_close)
self.addCleanup(httpd.shutdown)
os.environ["FRAME_CONTROL_POSTHOG_HOST"] = f"http://127.0.0.1:{httpd.server_port}"
return got
def test_nothing_is_sent_before_the_notice_was_shown(self):
got = self.serve()
tm.capture("app_opened")
self.assertEqual(tm.flush(), 0)
self.assertEqual(got, [])
tm.update_settings({"noticeShown": True})
self.assertEqual(tm.flush(), 1)
path, body = got[0]
self.assertEqual((path, body["api_key"], body["batch"][0]["event"]), ("/batch/", "phc_test", "app_opened"))
self.assertEqual(self.queued(), [])
self.assertEqual([e["event"] for e in tm.state()["sent"]], ["app_opened"])
def test_a_failed_send_keeps_the_events(self):
self.serve(status=500)
tm.update_settings({"noticeShown": True})
tm.capture("app_opened")
self.assertEqual(tm.flush(), 0)
self.assertEqual(len(self.queued()), 1)
class Scrub(unittest.TestCase):
def test_personal_details_are_removed(self):
home = str(Path.home())
text = (f"open {home}/Downloads/My Game.apk failed; ssh alex@192.168.1.20 (frame.local) "
"key ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIM steam 76561198000000000 mac 3c:22:fb:12:34:56 "
"url https://example.com/private/path?token=abc phc_abcdefghijklmnopqrstu C:\\Users\\Bob\\x "
"/home/carol/y")
out = tm.scrub(text)
for leaked in (home, "192.168.1.20", "frame.local", "AAAAC3Nza", "76561198000000000", "3c:22:fb",
"private/path", "phc_abcdefghijklmnopqrstu", "Bob", "carol", "alex@"):
self.assertNotIn(leaked, out)
self.assertIn("https://example.com/…", out)
self.assertIn("~/Downloads", out)
def test_categories(self):
self.assertEqual(tm.categorize("adb: failed to install: INSTALL_FAILED_NO_MATCHING_ABIS: x"),
("android_installer", "INSTALL_FAILED_NO_MATCHING_ABIS"))
self.assertEqual(tm.categorize("X has no arm64-v8a build (armeabi-v7a)")[0], "apk_wrong_abi")
self.assertEqual(tm.categorize("ssh: connect to host 10.0.0.2 port 22: Connection refused")[0],
"frame_unreachable")
self.assertEqual(tm.categorize("something new")[0], "other")
class Compat(Base):
def test_reports_are_shared_only_after_opting_in_without_file_names(self):
r = {"id": "r1", "package": "org.example", "version": "1.0", "rating": "works", "via": "user",
"notes": f"from {Path.home()}/x", "source": "MyPrivateBuild.apk", "date": "2026-09-28T10:00:00"}
self.assertFalse(tm.compat_report(r))
tm.update_settings({"compat": True})
self.assertTrue(tm.compat_report(r))
p = self.queued()[-1]["properties"]
self.assertEqual((p["package"], p["rating"], p["id"]), ("org.example", "works", "r1"))
self.assertNotIn("source", p)
self.assertNotIn(str(Path.home()), p["notes"])
r2 = dict(r, id="r2", source="https://f-droid.org/repo/org.example_1.apk")
tm.compat_report(r2)
self.assertEqual(self.queued()[-1]["properties"]["source"], "https://f-droid.org/…")
def test_opting_in_shares_earlier_local_reports(self):
with mock.patch.object(db, "shared", return_value=False), \
mock.patch.object(db, "_outbox", return_value=[{"id": "old1", "package": "org.a", "rating": "works",
"date": "2026-09-01T00:00:00"}]):
tm.update_settings({"compat": True})
tm.update_settings({"compat": True}) # already sent: not again
self.assertEqual([e["properties"]["id"] for e in self.queued() if e["event"] == "compat_report"], ["old1"])
class ApkInstallReports(unittest.TestCase):
"""server.apk_installed: an APK that won't install is reported; connection trouble isn't."""
def setUp(self):
import server
self.server = server
for target, name in ((server.frame_catalog, "add_report"), (server.frame_telemetry, "install_finished")):
p = mock.patch.object(target, name)
setattr(self, name, p.start())
self.addCleanup(p.stop)
def test_wrong_abi_is_an_install_failed_report(self):
info = {"package": "org.x", "version": "2.0", "label": "X"}
self.server.apk_installed(info, None, self.server.frame_android.FrameError(
"X has no arm64-v8a build (armeabi-v7a); Lepton is 64-bit ARM only"), 3.0)
args, kw = self.add_report.call_args
self.assertEqual((args[0], args[1], kw["result"], kw["via"]), ("org.x", "2.0", "install_failed", "install"))
self.assertIs(self.install_finished.call_args[0][1], False)
def test_connection_trouble_is_not_reported(self):
self.server.apk_installed({"package": "org.x", "version": "2.0"}, None,
self.server.frame_android.FrameError("timed out talking to frame"), 3.0)
self.add_report.assert_not_called()
def test_private_package_names_stay_here(self):
with mock.patch.dict(self.server.frame_catalog._cache, {"by_pkg": {"org.public": {}}}):
self.server.apk_installed({"package": "com.private.thing", "version": "1"}, {"package": "com.private.thing"},
None, 2.0)
self.assertIsNone(self.install_finished.call_args[1]["package"])
self.server.apk_installed({"package": "org.public", "version": "1"}, {"package": "org.public"}, None, 2.0)
self.assertEqual(self.install_finished.call_args[1]["package"], "org.public")
class CommunitySync(unittest.TestCase):
def ev(self, i, who="a", day="2026-09-28", **kw):
return ({"id": f"id{i}", "package": "org.x", "rating": "works", "date": f"{day}T00:00:00",
"via": "probe", **kw}, who, f"{day} 10:00:00")
def test_rows_are_validated_marked_and_capped_per_reporter(self):
events = [self.ev(i) for i in range(5)] + [self.ev(9, who="b", rating="nonsense"), self.ev(10, who="b")]
rows, skipped = db.community_rows(events, {}, cap=3)
self.assertEqual([r["id"] for r in rows], ["id0", "id1", "id2", "id10"])
self.assertTrue(all(r["via"] == "community-probe" for r in rows))
self.assertEqual(len(skipped), 3)
def test_the_cap_and_duplicates_hold_across_syncs(self):
state = {}
rows, _ = db.community_rows([self.ev(i) for i in range(3)], state, cap=3)
self.assertEqual(len(rows), 3)
rows, skipped = db.community_rows([self.ev(i) for i in range(6)], state, cap=3) # overlapping re-read
self.assertEqual(rows, [])
self.assertEqual([why for _, why in skipped], ["over the daily limit for one reporter"] * 3)
def test_a_malformed_event_is_skipped_not_fatal(self):
rows, skipped = db.community_rows([self.ev(1, via=["probe"]), ("not json", "a", "2026-09-28"), self.ev(2)], {})
self.assertEqual([r["id"] for r in rows], ["id2"])
self.assertEqual(len(skipped), 2)
class Regressions(Base):
"""Findings from the cross-provider review."""
def test_urls_lose_credentials_paths_and_private_hosts(self):
for text, leaked in (("https://alice:secret@example.com/private.apk?token=credential", ("alice", "secret", "private", "credential")),
("https://alice:secret@192.168.1.4/private.apk", ("alice", "192.168", "private")),
("fe80::1234 and 2001:db8::5", ("fe80", "2001:db8")),
("sk-proj-abcdefghijklmnopqrstuv", ("abcdefghijk",)),
("http://frame.local:8080/x", ("frame.local", "8080"))):
out = tm.scrub(text)
for s in leaked:
self.assertNotIn(s, out, (text, out))
def test_compat_labels_versions_and_sources_are_scrubbed(self):
tm.update_settings({"compat": True})
tm.compat_report({"id": "r9", "package": "org.x", "rating": "works", "date": "2026-09-28T00:00:00",
"label": "alice@example.com build", "version": "1.0-alice@example.com",
"source": "https://alice:secret@192.168.1.4/private.apk"})
p = self.queued()[-1]["properties"]
self.assertNotIn("alice", json.dumps(p))
self.assertNotIn("source", p)
def test_an_unsent_report_is_shared_again_after_opting_out_and_in(self):
with mock.patch.object(db, "shared", return_value=False), \
mock.patch.object(db, "_outbox", return_value=[{"id": "q1", "package": "org.a", "rating": "works",
"date": "2026-09-01T00:00:00"}]):
tm.update_settings({"compat": True})
tm.update_settings({"compat": False})
self.assertEqual(self.queued(), [])
tm.update_settings({"compat": True})
self.assertEqual([e["properties"]["id"] for e in self.queued() if e["event"] == "compat_report"], ["q1"])
def test_opting_out_waits_for_a_send_in_progress(self):
tm.update_settings({"noticeShown": True})
tm.capture("app_opened")
order = []
started = threading.Event()
def slow_open(req, timeout):
started.set()
time.sleep(0.3)
order.append("sent")
return mock.MagicMock(__enter__=lambda s: s, __exit__=lambda *a: False, read=lambda: b"{}")
with mock.patch.object(tm.urllib.request, "urlopen", side_effect=slow_open):
th = threading.Thread(target=tm.flush)
th.start()
started.wait(2)
tm.update_settings({"usage": False})
order.append("opted out")
th.join()
self.assertEqual(order, ["sent", "opted out"])
def test_project_id_comes_from_the_config(self):
with mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_PROJECT": "12345"}):
self.assertEqual(tm.config()["project"], "12345")
class ReportProblem(Base):
"""Report a problem: diagnostics are scrubbed and bounded; the report goes privately to PostHog."""
def serve(self, status=200):
got = []
class H(BaseHTTPRequestHandler):
def do_POST(self):
got.append((self.path, json.loads(self.rfile.read(int(self.headers["Content-Length"])))))
self.send_response(status)
self.end_headers()
self.wfile.write(b'{"status":"Ok"}')
def log_message(self, *a):
pass
httpd = HTTPServer(("127.0.0.1", 0), H)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
self.addCleanup(httpd.server_close)
self.addCleanup(httpd.shutdown)
p = mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_HOST": f"http://127.0.0.1:{httpd.server_port}"})
p.start()
self.addCleanup(p.stop)
return got
def test_diagnostics_are_scrubbed_and_include_the_log(self):
log = tm.STATE / "server.log"
log.write_text("GET /api/status 200\nTraceback: ssh alice@192.168.1.9 failed in %s/x\n" % Path.home())
with mock.patch.dict(os.environ, {"FRAME_CONTROL_LOG": str(log)}):
text = fr.diagnostics(["16:00 Install failed: https://bob:pw@example.com/a.apk"], include_logs=True, limit=5000)
self.assertIn("Frame Control 9.9.9", text)
self.assertIn("Traceback", text)
self.assertNotIn("GET /api/status", text)
for leaked in ("alice", "192.168.1.9", str(Path.home()), "bob", "pw@"):
self.assertNotIn(leaked, text)
def test_a_report_is_bounded_in_utf16_units(self):
body = {"title": "Live view stops", "message": "It stops 😀 " * 800, "diagnostics": "log 😀 line\n" * 2000}
title, text, diag = fr.compose(body)
self.assertLessEqual(fr.u16(text), fr.TEXT_MAX)
self.assertLessEqual(fr.u16(diag), fr.DIAG_MAX)
self.assertTrue(text.startswith("It stops"))
with self.assertRaises(ValueError):
fr.compose({"title": "hi", "message": "It stops after a minute."})
def test_logs_only_when_asked_and_environment_is_kept_first(self):
log = tm.STATE / "server.log"
log.write_text("".join(f"old line {i}\n" for i in range(200)) + "newest line\n")
with mock.patch.dict(os.environ, {"FRAME_CONTROL_LOG": str(log)}):
plain = fr.diagnostics(["Copy Jane Doe tax return.pdf to ~/Downloads"])
full = fr.diagnostics(["Install failed"], include_logs=True, limit=400)
self.assertNotIn("Jane Doe", plain)
self.assertNotIn("line", plain)
self.assertTrue(full.startswith("Frame Control 9.9.9"))
self.assertIn("Install failed", full)
self.assertIn("newest line", full)
self.assertLessEqual(fr.u16(full), 400)
def test_the_previewed_diagnostics_are_what_is_sent(self):
got = self.serve()
fr.send({"title": "Live view stops", "message": "It stops after a minute.",
"diagnostics": "Frame Control 9.9.9\nssh janes-mac.tail12345.ts.net failed"})
diag = got[0][1]["batch"][0]["properties"]["diagnostics"]
self.assertIn("Frame Control 9.9.9", diag)
self.assertNotIn("janes-mac", diag)
def test_send_is_a_private_posthog_event_whatever_the_settings(self):
got = self.serve()
tm.update_settings({"usage": False}) # analytics off: a deliberate report still goes
res = fr.send({"kind": "idea", "title": "Live view stops", "message": "It stops after a minute.",
"contact": "me@example.com"})
path, body = got[0]
event = body["batch"][0]
self.assertEqual((path, body["api_key"], event["event"]), ("/batch/", "phc_test", "problem_report"))
props = event["properties"]
self.assertEqual((props["kind"], props["title"], props["message"], props["contact"], props["report_id"]),
("idea", "Live view stops", "It stops after a minute.", "me@example.com", res["id"]))
self.assertEqual((props["$process_person_profile"], props["$geoip_disable"]), (False, True))
self.assertNotEqual(event["distinct_id"], tm.settings()["id"]) # not linked to the analytics
self.assertIn(res["id"], res["message"])
self.assertEqual([e["event"] for e in tm._read_lines(tm.SENT)], ["problem_report"])
def test_a_sent_report_is_not_an_error_if_the_local_log_fails(self):
self.serve()
with mock.patch.object(tm, "record_sent", side_effect=OSError("disk full")):
res = fr.send({"title": "Live view stops", "message": "It stops after a minute."})
self.assertTrue(res["id"])
def test_events_queued_by_older_versions_get_the_placeholder_address(self):
got = self.serve()
tm.update_settings({"noticeShown": True})
tm._write_lines(tm.OUTBOX, [{"event": "app_opened", "distinct_id": "x", "uuid": "u1",
"properties": {"level": "usage"}}])
self.assertEqual(tm.flush(), 1)
self.assertEqual(got[0][1]["batch"][0]["properties"]["$ip"], "0.0.0.0")
self.assertEqual(tm._read_lines(tm.SENT)[0]["properties"]["$ip"], "0.0.0.0")
def test_the_inbox_skips_malformed_reports(self):
good = ["2026-09-28T09:50:00Z", "AB12CD34", "bug", "Live view stops", "It stops.", None,
"0.4.0", "macOS", "", ""]
rows = [["2026-09-28T10:00:00Z", "X", "bug", "Hand-made", None, None, None, None, None, None], ["short"], good]
with mock.patch.object(db, "_posthog_query", return_value={"results": rows}), \
mock.patch.object(sys, "argv", ["frame_report.py", "inbox"]), \
mock.patch("builtins.print") as out:
fr.main()
printed = " ".join(str(c.args[0]) for c in out.call_args_list if c.args)
self.assertIn("AB12CD34", printed)
self.assertIn("Hand-made", printed)
def test_a_refused_report_is_an_error(self):
self.serve(status=401)
with self.assertRaisesRegex(fr.ReportError, "HTTP 401"):
fr.send({"title": "Live view stops", "message": "It stops after a minute."})
self.assertEqual(tm._read_lines(tm.SENT), [])
def test_no_key_means_no_report(self):
with mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_KEY": ""}), \
mock.patch.object(tm, "HERE", tm.STATE):
with self.assertRaisesRegex(fr.ReportError, "no PostHog project key"):
fr.send({"title": "Live view stops", "message": "It stops after a minute."})
if __name__ == "__main__":
unittest.main()
+283
View File
@@ -0,0 +1,283 @@
"""Touch: the live view's Control (ui/frame_touch.py on the Frame, and the server's checks).
Run: python3 -m unittest discover -s tests
"""
import shutil
import subprocess
import sys
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "ui"))
class Mapping(unittest.TestCase):
@classmethod
def setUpClass(cls):
import frame_touch
cls.t = frame_touch
def test_panel_bigger_than_its_display_is_scaled(self):
# Verified 2026-09-29: a 1920x1080 window on :1 (1280x720) took Accept at
# window (1828, 1020) as pointer (1219, 680).
panel = {"root": [1280, 720], "width": 1920, "height": 1080}
x, y = self.t.to_root(panel, 1828 / 1920, 1020 / 1080)
self.assertAlmostEqual(x, 1218.7, delta=1)
self.assertAlmostEqual(y, 679.4, delta=1)
def test_same_size_is_one_to_one_and_clamped(self):
panel = {"root": [1280, 720], "width": 1280, "height": 720}
self.assertEqual(self.t.to_root(panel, 0, 0), (0, 0))
self.assertEqual(self.t.to_root(panel, 1, 1), (1279, 719))
self.assertEqual(self.t.to_root(panel, -3, 7), (0, 719))
def test_other_shapes_are_letterboxed(self):
panel = {"root": [1280, 720], "width": 800, "height": 800} # square: bars left and right
x, y = self.t.to_root(panel, 0, 0.5)
self.assertAlmostEqual(x, 280, delta=0.5)
self.assertAlmostEqual(y, 359.5, delta=0.5)
def test_same_id_on_both_displays_is_told_apart_by_pid(self):
t = self.t
saved = t.displays, t.window_info, t.window_pid
self.addCleanup(lambda: (setattr(t, "displays", saved[0]), setattr(t, "window_info", saved[1]),
setattr(t, "window_pid", saved[2])))
t.displays = lambda: [":0", ":1"]
t.window_info = lambda d, w: {"name": f"on {d}", "width": 1280 if w != "root" else 1920, "height": 720}
t.window_pid = lambda d, w: {":0": 111, ":1": 222}[d]
self.assertEqual(t.locate(5, 222)["display"], ":1")
self.assertEqual(t.locate(5, 111)["display"], ":0")
self.assertEqual(t.locate(5, None)["display"], ":0")
def test_focus_display_is_decoded(self):
t = self.t
saved = t.xprop_root
self.addCleanup(lambda: setattr(t, "xprop_root", saved))
for values, want in (([12602, 0, 58], ":1"), ([12346], ":0"), ([], None), ([0x41], None)):
t.xprop_root = lambda d, n, v=values: v
self.assertEqual(t.focus_display(), want)
def test_ascii_table_covers_printable_characters(self):
for code in range(0x20, 0x7F):
self.assertIn(chr(code), self.t.ASCII, chr(code))
self.assertEqual(self.t.ASCII["a"], (30, False))
self.assertEqual(self.t.ASCII["A"], (30, True))
self.assertEqual(self.t.ASCII["?"], (53, True))
self.assertEqual(self.t.ASCII["1"], (2, False))
self.assertEqual(self.t.ASCII["0"], (11, False))
def test_events_parsing(self):
self.assertEqual(self.t.events(b'{"dx": 1}'), [{"dx": 1}])
self.assertEqual(self.t.events(b'[{"dx": 1}, 5]'), [{"dx": 1}])
self.assertEqual(self.t.events(b"nope"), [])
class FakeGamescope:
def __init__(self):
self.calls = []
def __getattr__(self, name):
return lambda *a: self.calls.append((name, *a))
class Apply(unittest.TestCase):
@classmethod
def setUpClass(cls):
import frame_touch
cls.t = frame_touch
def setUp(self):
self.focused = {"window": 7, "display": ":1", "root": [1280, 720], "width": 1280, "height": 720, "name": "x"}
saved = self.t.focus, self.t.say, self.t.focus_now
self.t.STALE[0] = False
self.said = []
self.t.focus = lambda: dict(self.focused)
self.t.focus_now = lambda: (self.focused["window"], self.focused["display"])
self.t.say = lambda state, **more: self.said.append((state, more))
self.addCleanup(lambda: (setattr(self.t, "focus", saved[0]), setattr(self.t, "say", saved[1]),
setattr(self.t, "focus_now", saved[2])))
def test_tap_moves_then_clicks_in_order(self):
gs, panel = FakeGamescope(), None
for e in ({"fx": 0.5, "fy": 0.5, "window": 7, "display": ":1"}, {"button": "left", "down": True, "window": 7, "display": ":1"},
{"button": "left", "down": False}):
panel = self.t.apply(gs, e, panel)
self.assertEqual([c[0] for c in gs.calls], ["move_to", "button", "button"])
self.assertEqual(gs.calls[1][1:], ("left", True))
def test_a_tap_meant_for_another_panel_goes_nowhere(self):
# Focus moved on: the position and the press are dropped; the release still goes.
gs, panel = FakeGamescope(), None
for e in ({"fx": 0.5, "fy": 0.5, "window": 99, "display": ":1"}, {"button": "left", "down": True, "window": 99, "display": ":1"},
{"key": 30, "down": True, "window": 7, "display": ":0"}, {"button": "left", "down": False, "window": 99, "display": ":1"}):
panel = self.t.apply(gs, e, panel)
self.assertEqual(gs.calls, [("button", "left", False)])
self.assertEqual(self.said[0], ("ready", {"focus": 7, "display": ":1", "stale": True}))
def test_presses_read_focus_afresh(self):
# A move may use a recent reading; a press checks again, so a panel that just
# took focus doesn't get a click meant for another.
gs, calls = FakeGamescope(), []
self.t.focus = lambda: calls.append(1) or dict(self.focused)
panel = self.t.apply(gs, {"fx": 0.5, "fy": 0.5, "window": 7, "display": ":1"}, None)
panel = self.t.apply(gs, {"fx": 0.6, "fy": 0.5, "window": 7, "display": ":1"}, panel)
self.assertEqual(len(calls), 1)
self.t.apply(gs, {"button": "left", "down": True, "window": 7, "display": ":1"}, panel)
self.assertEqual(len(calls), 1) # same panel still: the quick check was enough
self.focused["window"] = 8
self.t.apply(gs, {"button": "left", "down": True, "window": 7, "display": ":1"}, panel)
self.assertEqual(len(calls), 2)
self.assertEqual([c[0] for c in gs.calls], ["move_to", "move_to", "button"])
def test_stale_is_said_once_and_cleared(self):
gs, panel = FakeGamescope(), None
for e in ({"fx": 0.5, "fy": 0.5, "window": 99, "display": ":1"}, {"fx": 0.5, "fy": 0.5, "window": 7, "display": ":1"},
{"fx": 0.6, "fy": 0.5, "window": 7, "display": ":1"}):
panel = self.t.apply(gs, e, panel)
self.assertEqual(self.said, [("ready", {"focus": 7, "display": ":1", "stale": True}),
("ready", {"focus": 7, "display": ":1"})])
def test_stale_clears_on_a_trackpad_move_but_not_on_a_stale_release(self):
gs = FakeGamescope()
panel = self.t.apply(gs, {"fx": 0.5, "fy": 0.5, "window": 99, "display": ":1"}, None)
panel = self.t.apply(gs, {"button": "left", "down": False, "window": 99, "display": ":1"}, panel)
self.assertTrue(self.t.STALE[0]) # that release was still aimed at the old panel
self.t.apply(gs, {"dx": 3, "dy": 0}, panel)
self.assertFalse(self.t.STALE[0])
self.assertEqual(self.said[-1][1].get("stale"), None)
def test_same_window_id_on_the_other_display_is_another_panel(self):
gs = FakeGamescope()
self.t.apply(gs, {"fx": 0.5, "fy": 0.5, "window": 7, "display": ":0"}, None)
self.assertEqual(gs.calls, [])
def test_relative_scroll_keys_text(self):
gs = FakeGamescope()
for e in ({"dx": 5, "dy": -3}, {"scroll": [0, 120]}, {"key": 30, "down": True}, {"text": "hi"},
{"key": True}, {"button": "sideways"}):
self.t.apply(gs, e, None)
self.assertEqual([c[0] for c in gs.calls], ["move_by", "scroll", "key", "text"])
def test_everything_held_is_let_go(self):
class Held(self.t.Gamescope):
def __init__(self):
self.held, self.keys, self.log = set(), set(), []
def frame(self):
pass
g = Held()
g.L = type("L", (), {"ei_device_button_button": lambda *a: g.log.append(("button",) + a[2:]),
"ei_device_keyboard_key": lambda *a: g.log.append(("key",) + a[2:])})()
g.device = object()
g.button("left", True)
g.key(42, True)
g.release_all()
self.assertEqual(g.log[-2:], [("button", 0x110, False), ("key", 42, False)])
self.assertEqual((g.held, g.keys), (set(), set()))
def test_resuming_after_a_pause_lets_go_of_everything(self):
t = self.t
log, queue = [], [1]
class L:
def __getattr__(self, name):
if name == "ei_get_event":
return lambda ei: queue.pop(0) if queue else None
return {"ei_event_get_type": lambda ev: t.EV_DEVICE_RESUMED, "ei_event_get_device": lambda ev: "dev",
"ei_device_has_capability": lambda d, c: True,
"ei_device_button_button": lambda d, c, down: log.append(("button", c, down)),
"ei_device_keyboard_key": lambda d, c, down: log.append(("key", c, down))}.get(name, lambda *a: 0)
g = t.Gamescope.__new__(t.Gamescope)
g.L, g.ei, g.fd, g.device, g.sequence, g.alive = L(), None, None, None, 0, True
g.held, g.keys = {0x110}, {42}
g.frame = lambda: None
old = t.select.select
t.select.select = lambda *a: ([], [], [])
try:
g.pump()
finally:
t.select.select = old
self.assertEqual(sorted(log), [("button", 0x110, False), ("key", 42, False)])
self.assertEqual(g.device, "dev")
def test_text_uses_shift_for_capitals(self):
class Keys(self.t.Gamescope):
def __init__(self):
self.pressed = []
def key(self, code, down):
self.pressed.append((code, down))
k = Keys()
k.text("Hié") # the é has no key on a US layout and is left out
self.assertEqual(k.pressed, [(42, True), (35, True), (35, False), (42, False), (23, True), (23, False)])
class ServerChecks(unittest.TestCase):
@classmethod
def setUpClass(cls):
import server
cls.s = server
def test_touch_event_keeps_known_fields(self):
ev = self.s.touch_event
self.assertEqual(ev({"fx": 0.5, "fy": 2, "window": 5, "display": ":1"}), {"fx": 0.5, "fy": 1.0, "window": 5, "display": ":1"})
self.assertEqual(ev({"button": "left", "window": 5, "display": ":0"}), {"button": "left", "down": True, "window": 5, "display": ":0"})
self.assertEqual(ev({"button": "right"}), {"button": "right", "down": True})
self.assertEqual(ev({"key": 30, "down": False}), {"key": 30, "down": False})
self.assertEqual(ev({"scroll": [0, 1e9]}), {"scroll": [0.0, 5000.0]})
self.assertEqual(ev({"dx": 3, "other": 1}), {"dx": 3.0})
def test_touch_event_rejects_bad_ones(self):
for bad in (None, {}, {"fx": 0.5, "fy": 0.5}, {"fx": "1", "fy": 0, "window": 1, "display": ":1"}, {"button": "side"},
{"fx": 0.5, "fy": 0.5, "window": 1}, {"fx": 0.5, "fy": 0.5, "window": 1, "display": ":1;x"},
{"window": 1, "display": ":1"},
{"key": 0}, {"key": 999}, {"key": True}, {"scroll": [1]}, {"text": ""}, {"text": "x" * 501},
{"fx": 0.1, "fy": 0.1, "window": True, "display": ":1"}):
with self.assertRaises(self.s.Failure, msg=repr(bad)):
self.s.touch_event(bad)
def test_panel_stream_is_the_window_and_checked(self):
cmd = self.s.stream_command("src=panel&window=10485777&display=:1&h=720&fps=30")
self.assertIn("DISPLAY=:1 ffmpeg", cmd)
self.assertIn("-window_id 10485777 -i :1", cmd)
for bad in ("src=panel&window=1;rm&display=:1", "src=panel&window=1&display=:1;x", "src=panel&display=:1"):
with self.assertRaises(self.s.Failure):
self.s.stream_command(bad + "&h=720&fps=30")
def test_touch_agent_runs_the_helper_with_nothing_to_copy(self):
agent = self.s.TouchAgent()
self.assertEqual(agent.deliver(lambda m: None), "")
cmd = agent.command()
self.assertTrue(cmd.startswith("python3 -u -c '"))
self.assertIn("frame_touch", cmd)
def test_batch_limit(self):
with self.assertRaises(self.s.Failure):
self.s.remote_touch({"events": [{"dx": 1}] * (self.s.INPUT_BATCH_LIMIT + 1)})
@unittest.skipUnless(shutil.which("node"), "needs node")
class PageGestures(unittest.TestCase):
"""Control's gestures and queue (tests/page/ctrl_gestures.mjs runs the real functions from index.html)."""
def test_gestures(self):
r = subprocess.run(["node", str(ROOT / "tests" / "page" / "ctrl_gestures.mjs")], capture_output=True, text=True,
timeout=60)
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
class Script(unittest.TestCase):
def test_helper_compiles_on_the_frames_python(self):
# The Frame runs it with its own python3 (3.13 on SteamOS 0.4.1); stdlib and ctypes only.
src = (ROOT / "ui/frame_touch.py").read_text()
compile(src, "frame_touch.py", "exec")
for mod in ("import ctypes", "import json", "import select"):
self.assertIn(mod, src)
self.assertNotIn("import requests", src)
if __name__ == "__main__":
unittest.main()
+1
View File
@@ -4,6 +4,7 @@ the localhost-testing rule allows.
Run: python3 -m unittest discover -s tests
"""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import hashlib
import json
import os