Add simple tests for gambaterm and gambaterm-ssh

.github/workflows/build.yml

@@ -1,12 +1,17 @@
 name: Build
 
-on: [push, pull_request]
+on:
+  pull_request:
+  push:
+    branches:
+      - main
 
 jobs:
   build_wheels:
     name: Build wheels on ${{ matrix.os }}
     runs-on: ${{ matrix.os }}
     strategy:
+      fail-fast: false
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
 

.gitignore

@@ -75,6 +75,8 @@ parsec/core/gui/rc/generated_misc/
 *#
 *_ext/*.cpp
 *_ext/*.c
+tests/*.rtc
+tests/*.sav
 
 # Build artefacts
 *.o

gambaterm/file_input.py

@@ -35,9 +35,11 @@ def open_input_log_file(path: str) -> Iterator[TextIOWrapper]:
         with ZipFile(path) as myzip:
             with myzip.open("Input Log.txt") as myfile:
                 yield TextIOWrapper(myfile, "utf-8")
+        return
     except BadZipFile:
-        with open(path) as myfile:
-            yield myfile
+        pass
+    with open(path) as myfile:
+        yield myfile
 
 
 @contextmanager

gambaterm/ssh.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import time
 import asyncio
 import argparse
@@ -38,7 +39,7 @@ async def detect_true_color_support(
             header = await asyncio.wait_for(process.stdin.readuntil("\033\\"), timeout)
         except asyncssh.TerminalSizeChanged:
             pass
-        except asyncio.TimeoutError:
+        except (asyncio.TimeoutError, asyncio.IncompleteReadError):
             return False
         else:
             break
@@ -230,8 +231,19 @@ def validate_password(self, username: str, password: str) -> bool:
 async def run_server(
     app_config: argparse.Namespace, executor: ThreadPoolExecutor
 ) -> None:
-    user_private_key = str(Path("~/.ssh/id_rsa").expanduser())
-    user_public_key = str(Path("~/.ssh/id_rsa.pub").expanduser())
+    ssh_key_dir = Path(os.environ.get("GAMBATERM_SSH_KEY_DIR", "~/.ssh"))
+    user_private_key = (ssh_key_dir / "id_rsa").expanduser()
+    user_public_key = (ssh_key_dir / "id_rsa.pub").expanduser()
+    if not user_private_key.exists():
+        raise SystemExit(
+            f"The server requires a private RSA key to use as a host hey.\n"
+            f"You may generate one by running the following command:\n\n"
+            f"    ssh-keygen -f {ssh_key_dir / 'id_rsa'} -P ''\n"
+        )
+    server_host_keys = [str(user_private_key)]
+    authorized_client_keys = []
+    if user_public_key.exists():
+        authorized_client_keys = [str(user_public_key)]
 
     # Remove chacha20 from encryption_algs because it's a bit too expensive
     encryption_algs = [
@@ -247,14 +259,14 @@ async def run_server(
         lambda: SSHServer(app_config, executor),
         app_config.bind,
         app_config.port,
-        server_host_keys=[user_private_key],
-        authorized_client_keys=user_public_key,
+        server_host_keys=server_host_keys,
+        authorized_client_keys=authorized_client_keys,
         x11_forwarding=True,
         encryption_algs=encryption_algs,
         line_editor=False,
     )
     bind, port = server.sockets[0].getsockname()
-    print(f"Running ssh server on {bind}:{port}...")
+    print(f"Running ssh server on {bind}:{port}...", flush=True)
 
     await server.wait_closed()
 
@@ -271,7 +283,7 @@ def main(
         "--bind",
         "-b",
         type=str,
-        default="localhost",
+        default="127.0.0.1",
         help="Bind adress of the SSH server, "
         "use `0.0.0.0` for all interfaces (default is localhost)",
     )

gambaterm/ssh_app_session.py

@@ -28,6 +28,8 @@ async def vt100_output_from_process(
 ) -> AsyncIterator[Vt100_Output]:
     def get_size() -> Size:
         width, height, _, _ = process.get_terminal_size()
+        if width == height == 0:
+            width, height = 80, 24
         return Size(rows=height, columns=width)
 
     term = process.get_terminal_type()

pyproject.toml

@@ -19,7 +19,15 @@ disallow_untyped_decorators = true
 [tool.cibuildwheel]
 build = "cp*-win_amd64 cp*-manylinux_x86_64 cp*-macosx_x86_64"
 skip = "cp36*"
-test-command = "python -m gambaterm --help"
+
+[tool.cibuildwheel.windows]
+test-command = "gambaterm --help"
+
+[tool.cibuildwheel.macos]
+test-requires = "pytest"
+test-command = "pytest {project}/tests -v -k 'not test_gambaterm[interactive]'"
 
 [tool.cibuildwheel.linux]
-before-all = "yum install -y libsamplerate portaudio"
+before-all = "yum install -y libsamplerate portaudio openssh-clients"
+test-requires = "pytest"
+test-command = "pytest {project}/tests -v"

tests/test_gambaterm.py

@@ -0,0 +1,71 @@
+import os
+import sys
+import pytest
+import asyncssh
+from pathlib import Path
+from typing import Iterator
+from subprocess import Popen, PIPE, run
+
+TEST_ROM = Path(__file__).parent / "test_rom.gb"
+
+
+@pytest.fixture  # type: ignore[misc]
+def ssh_config(tmp_path: Path) -> Iterator[Path]:
+    rsa_key = asyncssh.generate_private_key("ssh-rsa")
+    (tmp_path / "id_rsa").write_bytes(rsa_key.export_private_key())
+    (tmp_path / "id_rsa.pub").write_bytes(rsa_key.export_public_key())
+    os.chmod(tmp_path / "id_rsa", 0o600)
+    os.chmod(tmp_path / "id_rsa.pub", 0o600)
+    os.environ["GAMBATERM_SSH_KEY_DIR"] = str(tmp_path)
+    yield tmp_path
+    del os.environ["GAMBATERM_SSH_KEY_DIR"]
+
+
+@pytest.mark.parametrize(  # type: ignore[misc]
+    "interactive", (False, True), ids=("non-interactive", "interactive")
+)
+def test_gambaterm(interactive: bool) -> None:
+    assert TEST_ROM.exists()
+    command = f"gambaterm {TEST_ROM} --break-after 10 --input-file /dev/null --disable-audio --color-mode 4"
+    result = run(
+        f"script -e -q -c '{command}' /dev/null" if interactive else command,
+        shell=True,
+        check=True,
+        text=True,
+        capture_output=True,
+    )
+    if interactive:
+        assert result.stderr == ""
+        assert "| test_rom.gb |" in result.stdout
+    else:
+        assert result.stderr == "Warning: Input is not a terminal (fd=0).\n"
+    if sys.platform == "linux":
+        assert "▀ ▄▄ ▀" in result.stdout
+
+
+def test_gambaterm_ssh(ssh_config: Path) -> None:
+    assert TEST_ROM.exists()
+    command = f"gambaterm-ssh {TEST_ROM} --break-after 10 --input-file /dev/null --color-mode 4"
+    server = Popen(command.split(), stdout=PIPE, stderr=PIPE, bufsize=0, text=True)
+    assert server.stdout is not None
+    assert server.stderr is not None
+    try:
+        assert server.stdout.readline() == "Running ssh server on 127.0.0.1:8022...\n"
+        client = run(
+            f"ssh -tt -q localhost -p 8022 -X -i {ssh_config / 'id_rsa'} -o StrictHostKeyChecking=no",
+            shell=True,
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        assert client.stderr == ""
+        assert "| test_rom.gb |" in client.stdout
+        if sys.platform == "linux":
+            assert "▀ ▄▄ ▀" in client.stdout
+    finally:
+        server.terminate()
+        server.wait()
+        print(server.stdout.read(), end="", file=sys.stdout)
+        print(server.stderr.read(), end="", file=sys.stderr)
+        server.stdout.close()
+        server.stderr.close()