152 lines
7.0 KiB
Python
152 lines
7.0 KiB
Python
"""Real headless Bash/tmux integration tests; only test-owned profiles are closed."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
WORK = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(WORK / "expanded"))
|
|
os.environ["PYTHONPATH"] = str(WORK / "expanded")
|
|
os.environ["FRAMEBRIDGE_EXPANDED_PREFIX"] = str(WORK)
|
|
from framebridge_expanded import tty
|
|
|
|
|
|
class RealTtyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.state = Path(tempfile.mkdtemp(prefix="tty-test-", dir=WORK / "qa"))
|
|
self.sessions = []
|
|
self.previous_display = os.environ.pop("DISPLAY", None)
|
|
self.previous_x11 = os.environ.pop("XDG_SESSION_TYPE", None)
|
|
self.assertTrue(tty.start(self.state).get("ok"))
|
|
|
|
def tearDown(self):
|
|
for sid in self.sessions:
|
|
tty.request(self.state, {"op": "close", "authorized": True, "session_id": sid})
|
|
self.assertTrue(tty.stop(self.state).get("ok"))
|
|
self.assertFalse((tty.runtime_path(self.state) / "agent.sock").exists())
|
|
if self.previous_display is not None:
|
|
os.environ["DISPLAY"] = self.previous_display
|
|
if self.previous_x11 is not None:
|
|
os.environ["XDG_SESSION_TYPE"] = self.previous_x11
|
|
|
|
def call(self, op, **args):
|
|
return tty.request(self.state, {"op": op, "authorized": True, **args})
|
|
|
|
def open(self, **args):
|
|
value = self.call("open", **args)
|
|
self.assertTrue(value.get("ok"), value)
|
|
self.sessions.append(value["result"]["session_id"])
|
|
return value["result"]
|
|
|
|
def enter(self, sid, text):
|
|
self.assertTrue(self.call("text.type", session_id=sid, text=text).get("ok"))
|
|
self.assertTrue(self.call("key.send", session_id=sid, key="ENTER").get("ok"))
|
|
|
|
def await_file(self, path):
|
|
deadline = time.monotonic() + 3
|
|
while not path.exists() and time.monotonic() < deadline:
|
|
time.sleep(0.05)
|
|
self.assertTrue(path.is_file(), str(path))
|
|
return path.read_text()
|
|
|
|
def test_real_persistent_bash_and_unicode_without_x11(self):
|
|
self.assertNotIn("DISPLAY", os.environ)
|
|
opened = self.open(rows=30, cols=100)
|
|
self.assertTrue(opened["real_terminal"])
|
|
self.assertEqual(opened["backend"], "tmux")
|
|
self.assertTrue(Path(opened["tty"]).exists())
|
|
self.assertGreater(opened["pid"], 1)
|
|
sid = opened["session_id"]
|
|
directory = self.state / "working-directory"
|
|
directory.mkdir()
|
|
self.enter(sid, "export FRAMEBRIDGE_TTY_VALUE=" + shlex.quote("persistent ñ 漢字")
|
|
+ "; cd " + shlex.quote(str(directory)))
|
|
marker = self.state / "unicode-and-state.txt"
|
|
self.enter(sid, 'printf %s "$FRAMEBRIDGE_TTY_VALUE" > ' + shlex.quote(str(marker)))
|
|
self.assertEqual(self.await_file(marker), "persistent ñ 漢字")
|
|
observed = self.call("observe", session_id=sid)
|
|
self.assertTrue(observed["ok"])
|
|
self.assertEqual(observed["result"]["pid"], opened["pid"])
|
|
self.assertEqual(observed["result"]["cwd"], str(directory))
|
|
self.assertGreaterEqual(observed["result"]["sequence"], 4)
|
|
|
|
def test_control_c_interrupts_a_real_foreground_process(self):
|
|
sid = self.open()["session_id"]
|
|
self.enter(sid, "sleep 30")
|
|
deadline = time.monotonic() + 2
|
|
sleeping = False
|
|
while time.monotonic() < deadline:
|
|
value = self.call("observe", session_id=sid)
|
|
if value["result"]["current_command"] == "sleep":
|
|
sleeping = True
|
|
break
|
|
time.sleep(0.05)
|
|
self.assertTrue(sleeping)
|
|
self.assertTrue(self.call("key.send", session_id=sid, key="CTRL_C")["ok"])
|
|
marker = self.state / "interrupted.txt"
|
|
self.enter(sid, "printf interrupted > " + shlex.quote(str(marker)))
|
|
self.assertEqual(self.await_file(marker), "interrupted")
|
|
|
|
def test_real_resize_can_exceed_the_old_machine_specific_bounds(self):
|
|
sid = self.open()["session_id"]
|
|
self.assertTrue(self.call("resize", session_id=sid, rows=110, cols=240)["ok"])
|
|
observed = self.call("observe", session_id=sid)["result"]
|
|
self.assertEqual((observed["screen"]["rows"], observed["screen"]["cols"]), (110, 240))
|
|
|
|
def test_authorization_and_socket_identity_are_explicit(self):
|
|
before = self.call("status")["result"]["active_sessions"]
|
|
denied = tty.request(self.state, {"op": "open", "authorized": False})
|
|
self.assertEqual(denied.get("error"), "AUTHORIZED_TRUE_REQUIRED")
|
|
denied = tty.request(self.state, {"op": "open", "authorized": 1})
|
|
self.assertEqual(denied.get("error"), "AUTHORIZED_TRUE_REQUIRED")
|
|
self.assertEqual(self.call("status")["result"]["active_sessions"], before)
|
|
path = tty.runtime_path(self.state) / "agent.sock"
|
|
st = path.stat()
|
|
self.assertTrue(stat.S_ISSOCK(st.st_mode))
|
|
self.assertEqual(stat.S_IMODE(st.st_mode), 0o600)
|
|
self.assertEqual(st.st_uid, os.geteuid())
|
|
status = self.call("status")["result"]
|
|
self.assertEqual(status["input_policy"], "same-uid-authorized")
|
|
self.assertNotIn("owner_device_connected", status)
|
|
self.assertNotIn("owner_signature_required", status)
|
|
|
|
def test_errors_do_not_expose_arbitrary_payloads_or_paths(self):
|
|
self.assertEqual(self.call("open", program="demo-readonly").get("error"),
|
|
"PROGRAM_NOT_SUPPORTED")
|
|
self.assertEqual(self.call("observe", session_id="../../outside").get("error"),
|
|
"SESSION_ID_INVALID")
|
|
sid = self.open()["session_id"]
|
|
self.assertEqual(self.call("key.send", session_id=sid, key="INVALID_PRIVATE_TEXT").get("error"),
|
|
"KEY_NOT_ALLOWLISTED")
|
|
closed = self.call("close", session_id=sid)
|
|
self.assertTrue(closed["result"]["cleanup_verified"])
|
|
self.sessions.remove(sid)
|
|
self.assertEqual(self.call("observe", session_id=sid).get("error"), "SESSION_NOT_FOUND")
|
|
|
|
def test_configurable_session_limit_replaces_the_old_constructor_cap(self):
|
|
self.assertTrue(tty.stop(self.state)["ok"])
|
|
(self.state / "tty.json").write_text(json.dumps({"limits": {"max_sessions": 5}}))
|
|
self.assertTrue(tty.start(self.state)["ok"])
|
|
for _ in range(5):
|
|
self.open(rows=10, cols=40)
|
|
self.assertEqual(self.call("status")["result"]["active_sessions"], 5)
|
|
self.assertEqual(self.call("open").get("error"), "SESSION_LIMIT")
|
|
|
|
def test_configured_input_limit_is_reported_and_enforced(self):
|
|
self.assertTrue(tty.stop(self.state)["ok"])
|
|
(self.state / "tty.json").write_text(json.dumps({"limits": {"max_text_bytes": 100}}))
|
|
self.assertTrue(tty.start(self.state)["ok"])
|
|
sid = self.open()["session_id"]
|
|
self.assertEqual(self.call("status")["result"]["limits"]["max_text_bytes"], 100)
|
|
self.assertEqual(self.call("text.type", session_id=sid, text="x" * 101).get("error"),
|
|
"TEXT_INVALID_OR_TOO_LARGE")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|