92 lines
4.5 KiB
Python
92 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""TmuxPlexerController.ask with a real tmux client attached through a pty.
|
|
|
|
For every question kind both implementations (tai.py and the C++ controller)
|
|
run against a private tmux server whose client is driven like a user: wait for
|
|
the menu or popup title to be drawn, type the answer, read the reply. The
|
|
replies must equal the expected value and each other.
|
|
|
|
Usage: python3 tests/tmux_ask.py [--probe PATH]
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import foundation_fixtures as fixtures # noqa: E402
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.dirname(os.path.dirname(HERE))
|
|
PY_RUNNER = os.path.join(HERE, "foundation_parity.py")
|
|
|
|
CASES = [
|
|
# label, screen, timeout, keys typed once the title is drawn, expected reply
|
|
("menu option 2", {"title": "askmenu", "text": "pick", "kind": "choice", "options": ["first", "second", "third"]}, 10, "2", {"status": "ok", "value": 1}),
|
|
("menu option 1", {"title": "askmenu", "text": "pick", "kind": "choice", "options": ["first", "second"]}, 10, "1", {"status": "ok", "value": 0}),
|
|
("menu escape", {"title": "askmenu", "text": "pick", "kind": "choice", "options": ["first", "second"]}, 10, "\x1b", {"status": "cancelled"}),
|
|
("text answer", {"title": "asktext", "text": "your name?", "kind": "text", "initial": ""}, 10, "hello popup\r", {"status": "ok", "value": "hello popup"}),
|
|
("text unicode", {"title": "asktext", "text": "name?", "kind": "text", "initial": ""}, 10, "héllo über\r", {"status": "ok", "value": "héllo über"}),
|
|
("message enter", {"title": "askmsg", "text": "read this", "kind": "message"}, 10, "\r", {"status": "ok", "value": True}),
|
|
("popup ctrl-c", {"title": "asktext", "text": "name?", "kind": "text", "initial": ""}, 10, "\x03", {"status": "cancelled"}),
|
|
("popup escape is typed into the popup, so it times out", {"title": "asktext", "text": "name?", "kind": "text", "initial": ""}, 2, "\x1b", {"status": "timeout"}),
|
|
("popup timeout", {"title": "asktext", "text": "name?", "kind": "text", "initial": ""}, 2, None, {"status": "timeout"}),
|
|
]
|
|
|
|
|
|
def run_case(impl, probe, screen, timeout, keys, work):
|
|
server = fixtures.TmuxServer(os.path.join(work, "root-%s-%d" % (impl, int(time.time() * 1000))))
|
|
client = fixtures.AttachedClient(server)
|
|
try:
|
|
time.sleep(0.5)
|
|
scenario = os.path.join(server.tmpdir, "scenario.json")
|
|
with open(scenario, "w", encoding="utf-8") as handle:
|
|
json.dump([["tmux_ctl", ["%0", "ask", [screen, timeout]]]], handle)
|
|
home = os.path.join(server.tmpdir, "home")
|
|
os.makedirs(home)
|
|
env = {key: value for key, value in os.environ.items() if not key.startswith(("TAI_", "TMUX"))}
|
|
env.update({"TAI_HOME": home, "TAI_PASSPHRASE": "", "HOME": home, "TMUX_TMPDIR": server.tmpdir})
|
|
cmd = [sys.executable, "-I", PY_RUNNER, "--py-runner", home, scenario] if impl == "python" else [probe, scenario]
|
|
mark = client.mark()
|
|
proc = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
drawn = client.wait_for(screen["title"], since=mark)
|
|
if keys is not None and drawn:
|
|
time.sleep(0.3)
|
|
client.type(keys)
|
|
out, err = proc.communicate(timeout=60)
|
|
if proc.returncode != 0:
|
|
return {"error": err[-500:]}
|
|
if keys is not None and not drawn:
|
|
return {"error": "title never drawn"}
|
|
return json.loads(out)[0]
|
|
finally:
|
|
client.close()
|
|
server.close()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--probe", default=os.path.join(os.path.dirname(HERE), "build", "foundation_probe"))
|
|
opts = parser.parse_args()
|
|
if shutil.which("tmux") is None:
|
|
print("SKIP: tmux is not installed")
|
|
return 0
|
|
work = tempfile.mkdtemp(prefix="detz-tmuxask-")
|
|
failures = 0
|
|
for label, screen, timeout, keys, expected in CASES:
|
|
got = {impl: run_case(impl, opts.probe, screen, timeout, keys, work) for impl in ("python", "cpp")}
|
|
for impl, reply in got.items():
|
|
if reply != expected:
|
|
failures += 1
|
|
print("FAIL %s [%s]: expected %r, got %r" % (label, impl, expected, reply))
|
|
print("%d cases x 2 implementations, %d failures" % (len(CASES), failures))
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|