tests: Add test coverage for console and terminal handling

Add unit tests for the terminal handling in term.py, covering the
external console launch (microcom and telnet fallback), the internal
console read/write loop, and terminal setup and teardown. Also add
tests for the new is_allowed() helper and the new parser argument for
the internal console.

Use real pipes for stdin rather than mocking os.read(), giving more
realistic coverage of the keystroke and txdelay paths. The exit-char
deadline test uses a threading.Event to synchronise with the run
loop, avoiding brittle fixed sleeps.

Fix an UnboundLocalError in term.internal() where log_fd could be
referenced in the finally block before being assigned.

Also fix a duplicate --logfile append in term.external() when using
microcom.

Series-changes: 5
- Feed keystrokes through a prompt_toolkit Input, so that raw mode
  and the event-loop hookup are covered too
- Add a test that escape sequences and control characters reach the
  board untouched

Signed-off-by: Simon Glass <sjg@chromium.org>
Co-developed-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: sjg20

labgrid/util/term.py

@@ -54,8 +54,6 @@ async def external(check_allowed, host, port, resource, logfile, listen_only):
         if logfile:
             logging.warning("--logfile option not supported by telnet, ignoring")
 
-    if logfile:
-        call.append(f"--logfile={logfile}")
     logging.info("connecting to %s calling %s", resource, " ".join(call))
     p = await asyncio.create_subprocess_exec(*call)
     while p.returncode is None:
@@ -166,6 +164,7 @@ def read_keys():
 
                 to_cons += data
 
+            # Drain one byte at a time, honouring txdelay between bytes
             if to_cons and time.monotonic() > next_cons:
                 cons._write(to_cons[:1])
                 to_cons = to_cons[1:]

tests/test_client_unit.py

@@ -0,0 +1,94 @@
+"""Unit tests for labgrid.remote.client"""
+
+import argparse
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from labgrid.remote.client import ClientSession, UserError, get_parser
+
+
+# --- is_allowed() tests ---
+
+@pytest.fixture
+def session():
+    """Create a minimal ClientSession-like object for testing"""
+    s = object.__new__(ClientSession)
+    s.args = argparse.Namespace()
+    return s
+
+
+@pytest.fixture
+def mock_place():
+    place = MagicMock()
+    place.name = "testplace"
+    place.acquired = "myhost/myuser"
+    place.allowed = {"myhost/myuser"}
+    return place
+
+
+class TestIsAllowed:
+    def test_place_not_acquired(self, session, mock_place):
+        mock_place.acquired = None
+        with patch.object(session, "gethostname", return_value="myhost"), \
+             patch.object(session, "getuser", return_value="myuser"):
+            result = session.is_allowed(mock_place)
+        assert "not acquired" in result
+
+    def test_place_acquired_by_us(self, session, mock_place):
+        with patch.object(session, "gethostname", return_value="myhost"), \
+             patch.object(session, "getuser", return_value="myuser"):
+            result = session.is_allowed(mock_place)
+        assert result is None
+
+    def test_place_acquired_by_different_user(self, session, mock_place):
+        mock_place.acquired = "myhost/otheruser"
+        mock_place.allowed = {"myhost/otheruser"}
+        with patch.object(session, "gethostname", return_value="myhost"), \
+             patch.object(session, "getuser", return_value="myuser"):
+            result = session.is_allowed(mock_place)
+        assert "not acquired by your user" in result
+        assert "otheruser" in result
+
+    def test_place_acquired_on_different_host(self, session, mock_place):
+        mock_place.acquired = "otherhost/myuser"
+        mock_place.allowed = {"otherhost/myuser"}
+        with patch.object(session, "gethostname", return_value="myhost"), \
+             patch.object(session, "getuser", return_value="myuser"):
+            result = session.is_allowed(mock_place)
+        assert "not acquired on this computer" in result
+        assert "otherhost" in result
+
+    def test_place_acquired_elsewhere_but_allowed(self, session, mock_place):
+        """User is in the allowed set even though place was acquired elsewhere"""
+        mock_place.acquired = "otherhost/otheruser"
+        mock_place.allowed = {"otherhost/otheruser", "myhost/myuser"}
+        with patch.object(session, "gethostname", return_value="myhost"), \
+             patch.object(session, "getuser", return_value="myuser"):
+            result = session.is_allowed(mock_place)
+        assert result is None
+
+
+# --- _check_allowed() tests ---
+
+class TestCheckAllowed:
+    def test_raises_on_not_allowed(self, session, mock_place):
+        mock_place.acquired = None
+        with patch.object(session, "gethostname", return_value="myhost"), \
+             patch.object(session, "getuser", return_value="myuser"):
+            with pytest.raises(UserError, match="not acquired"):
+                session._check_allowed(mock_place)
+
+    def test_no_raise_when_allowed(self, session, mock_place):
+        with patch.object(session, "gethostname", return_value="myhost"), \
+             patch.object(session, "getuser", return_value="myuser"):
+            session._check_allowed(mock_place)  # should not raise
+
+
+# --- get_parser() tests ---
+
+class TestGetParser:
+    def test_console_internal_argument(self):
+        parser = get_parser()
+        args = parser.parse_args(["console", "--internal"])
+        assert args.internal is True