Merge pull request 'fix(client): flush the device code, or nobody ever sees it' (#8) from fix/link-output-unbuffered into main

This commit is contained in:
Nirav Patel
2026-08-23 15:58:36 -04:00
2 changed files with 33 additions and 2 deletions
+7 -2
View File
@@ -599,8 +599,13 @@ def device_flow():
form={"client_id": DEVICE_CLIENT_ID, "scope": DEVICE_SCOPE}) form={"client_id": DEVICE_CLIENT_ID, "scope": DEVICE_SCOPE})
if status != 200: if status != 200:
raise SystemExit(f"device authorization failed (HTTP {status}): {resp}") raise SystemExit(f"device authorization failed (HTTP {status}): {resp}")
print(f"\nTo link this device, open:\n\n {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n") # flush=True is not cosmetic. Python buffers stdout when it is not a
print(f"and enter code: {resp['user_code']}\n") # terminal, so `granthi-sync link | tee setup.log`, a wrapper script, or
# anything capturing output shows NOTHING while the code silently expires
# five minutes later. Hit for real on 2026-08-23 driving a first sign-in.
print(f"\nTo link this device, open:\n\n {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n",
flush=True)
print(f"and enter code: {resp['user_code']}\n", flush=True)
interval = int(resp.get("interval", 5)) interval = int(resp.get("interval", 5))
deadline = time.time() + int(resp.get("expires_in", 300)) deadline = time.time() + int(resp.get("expires_in", 300))
while time.time() < deadline: while time.time() < deadline:
+26
View File
@@ -1284,5 +1284,31 @@ class TestResolveGranted(unittest.TestCase):
client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant"), client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant"),
"Nirlabinc/Ai-Assistant") "Nirlabinc/Ai-Assistant")
class TestDeviceFlowOutputIsVisible(unittest.TestCase):
"""The code has a five-minute life. If it is sitting in a buffer, the user
never sees it and it expires -- which is what happened on 2026-08-23 while
driving a first real sign-in through a wrapper."""
def test_the_url_and_code_are_flushed_immediately(self):
seen = []
real_print = print
def spy(*a, **kw):
seen.append(kw.get("flush", False))
return real_print(*a, **{k: v for k, v in kw.items() if k != "flush"})
resp = {"verification_uri_complete": "https://id.example/device?user_code=AB-CD",
"user_code": "AB-CD", "device_code": "dc", "interval": 0,
"expires_in": 0}
with mock.patch.object(client, "http_json", lambda *a, **k: (200, resp)), \
mock.patch("builtins.print", spy), \
mock.patch.object(client.time, "sleep", lambda *_: None):
with self.assertRaises(SystemExit): # expires_in 0 -> times out
client.device_flow()
self.assertTrue(seen, "device_flow printed nothing")
self.assertTrue(all(seen[:2]),
"the URL and code must be printed with flush=True")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()