1#!/usr/bin/env python3
2"""UI-driver regression against two disposable demo-image containers.
3
4Pass client and remote container names. The client must have a pinned SSH
5host named remote reaching the remote container. No host session is touched.
6"""
7
8import json
9import subprocess
10import sys
11import time
12import uuid
13
14
15class InputBusy(AssertionError):
16 pass
17
18
19class Driver:
20 def __init__(self, container, session):
21 self.process = subprocess.Popen(
22 ["docker", "exec", "-i", container, "vev", "--ui-driver",
23 "--session", session, "--cols", "100", "--rows", "30"],
24 stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
25 )
26 ready = json.loads(self.process.stdout.readline())["result"]
27 self.attachment = ready["attachment"]
28 self.request_id = 0
29 # Ready never promises an attachment: the generation is zero until a
30 # committed attachment publishes one. Wait for the attached
31 # publication explicitly instead of assuming ready.Generation == 1.
32 self.generation = ready.get("generation", 0)
33 if ready.get("status") != "attached":
34 self.generation = self._await_attached()
35
36 def _await_attached(self):
37 """Poll capture until the committed attached generation is published."""
38 while True:
39 self.request_id += 1
40 self.process.stdin.write(json.dumps(dict(
41 version=1, id=self.request_id, op="capture",
42 attachment=self.attachment)) + "\n")
43 self.process.stdin.flush()
44 response = json.loads(self.process.stdout.readline())
45 assert "error" not in response, response
46 context = response["result"].get("context", {})
47 if context.get("status") == "attached" and context.get("generation"):
48 return context["generation"]
49
50 def call(self, operation, **fields):
51 # input_busy is an unaccepted refusal while the input owner still
52 # delivers an earlier terminal read; resending it is safe and bounded.
53 # Accepted actions and every other error are never retried.
54 deadline = time.monotonic() + 3
55 while True:
56 try:
57 return self._call_once(operation, **fields)
58 except InputBusy:
59 if time.monotonic() >= deadline:
60 raise
61 time.sleep(0.05)
62
63 def _call_once(self, operation, **fields):
64 self.request_id += 1
65 request = dict(version=1, id=self.request_id, op=operation,
66 attachment=self.attachment, **fields)
67 if operation in ("keys", "text"):
68 request["generation"] = self.generation
69 self.process.stdin.write(json.dumps(request) + "\n")
70 self.process.stdin.flush()
71 response = json.loads(self.process.stdout.readline())
72 error = response.get("error")
73 if isinstance(error, dict) and error.get("code") == "input_busy" and not error.get("accepted"):
74 raise InputBusy(response)
75 assert "error" not in response, response
76 result = response["result"]
77 self.generation = result.get("context", {}).get("generation", self.generation)
78 return result
79
80 def palette(self, query, label):
81 self.call("keys", keys=["Alt+Space"])
82 self.call("text", text=query)
83 self.call("wait", expect={"text_contains": label}, timeout_ms=10000)
84 return self.call("capture")["text"]
85
86 def close(self):
87 self.process.stdin.close()
88 self.process.wait(timeout=10)
89
90
91def run(client_container, remote_container):
92 suffix = uuid.uuid4().hex[:8]
93 first, second, remote = (name + suffix for name in ("sample", "second", "other"))
94 fixture = Driver(remote_container, remote)
95 fixture.close()
96 driver = Driver(client_container, first)
97 try:
98 initial = driver.call("capture")["context"]["session"]
99 fixture = Driver(client_container, second)
100 fixture.close()
101 driver.palette(second, "Switch to session " + second + "@local")
102 driver.call("keys", keys=["Enter"])
103 driver.palette(remote, "Switch to session " + remote + "@remote")
104 entered = driver.call("keys", keys=["Enter"])
105 assert entered["context"]["session"]["session_name"] == remote
106 label = "Switch to session " + first + "@local"
107 screen = driver.palette(first, label)
108 assert screen.count(label) == 1, screen
109 assert "Switch to session " + first + " " not in screen, screen
110 selected = driver.call("keys", keys=["Enter"])
111 assert selected["context"]["session"] == initial, selected
112 driver.call("wait", expect={"session": initial}, timeout_ms=10000)
113 print("PASS: exact local return, one client-qualified row, committed UI action")
114 finally:
115 driver.close()
116
117
118if __name__ == "__main__":
119 if len(sys.argv) != 3:
120 raise SystemExit("usage: navigation_repro.py CLIENT_CONTAINER REMOTE_CONTAINER")
121 run(*sys.argv[1:])