Pytest tests and CI

.gitignore

@@ -2,6 +2,7 @@
 *~
 /*.o
 /microcom
+*.pyc
 
 # autotools stuff
 /.deps

can.c

@@ -184,8 +184,8 @@ struct ios_ops *can_init(char *interface_id)
 		return NULL;
 	}
 
-	printf("connected to %s (rx_id=%x, tx_id=%x)\n",
-	       interface, filter->can_id, data.can_id);
+	msg_printf("connected to %s (rx_id=%x, tx_id=%x)\n",
+		   interface, filter->can_id, data.can_id);
 
 	return ios;
 }

microcom.c

@@ -36,6 +36,7 @@ static struct termios sots;	/* old stdout/in termios settings to restore */
 
 struct ios_ops *ios;
 int debug;
+int quiet;
 
 void init_terminal(void)
 {
@@ -98,6 +99,7 @@ void main_usage(int exitcode, char *str, char *dev)
 		"                                         default: (%s:%x:%x)\n"
 		"    -f, --force                          ignore existing lock file\n"
 		"    -d, --debug                          output debugging info\n"
+		"    -q, --quiet                          do not print status information to stdout\n"
 		"    -l, --logfile=<logfile>              log output to <logfile>\n"
 		"    -o, --listenonly                     Do not modify local terminal, do not send input\n"
 		"                                         from stdin\n"
@@ -136,6 +138,7 @@ int main(int argc, char *argv[])
 		{ "telnet", required_argument, NULL, 't'},
 		{ "can", required_argument, NULL, 'c'},
 		{ "debug", no_argument, NULL, 'd' },
+		{ "quiet", no_argument, NULL, 'q' },
 		{ "force", no_argument, NULL, 'f' },
 		{ "logfile", required_argument, NULL, 'l'},
 		{ "listenonly", no_argument, NULL, 'o'},
@@ -144,54 +147,57 @@ int main(int argc, char *argv[])
 		{ },
 	};
 
-	while ((opt = getopt_long(argc, argv, "hp:s:t:c:dfl:oi:a:e:v", long_options, NULL)) != -1) {
+	while ((opt = getopt_long(argc, argv, "hp:s:t:c:dqfl:oi:a:e:v", long_options, NULL)) != -1) {
 		switch (opt) {
-			case '?':
-				main_usage(1, "", "");
-				break;
-			case 'h':
-				main_usage(0, "", "");
-				break;
-			case 'v':
-				printf("%s\n", PACKAGE_VERSION);
-				exit(EXIT_SUCCESS);
-				break;
-			case 'p':
-				device = optarg;
-				break;
-			case 's':
-				current_speed = strtoul(optarg, NULL, 0);
-				break;
-			case 't':
-				telnet = 1;
-				hostport = optarg;
-				break;
-			case 'c':
-				can = 1;
-				interfaceid = optarg;
-				break;
-			case 'f':
-				opt_force = 1;
-				break;
-			case 'd':
-				debug = 1;
-				break;
-			case 'l':
-				logfile = optarg;
-				break;
-			case 'o':
-				listenonly = 1;
-				break;
-			case 'a':
-				answerback = optarg;
-				break;
-			case 'e':
-				if (strlen(optarg) != 1) {
-					fprintf(stderr, "Option -e requires a single character argument.\n");
-					exit(EXIT_FAILURE);
-				}
-				escape_char = *optarg;
-				break;
+		case '?':
+			main_usage(1, "", "");
+			break;
+		case 'h':
+			main_usage(0, "", "");
+			break;
+		case 'v':
+			printf("%s\n", PACKAGE_VERSION);
+			exit(EXIT_SUCCESS);
+			break;
+		case 'p':
+			device = optarg;
+			break;
+		case 's':
+			current_speed = strtoul(optarg, NULL, 0);
+			break;
+		case 't':
+			telnet = 1;
+			hostport = optarg;
+			break;
+		case 'c':
+			can = 1;
+			interfaceid = optarg;
+			break;
+		case 'f':
+			opt_force = 1;
+			break;
+		case 'd':
+			debug = 1;
+			break;
+		case 'q':
+			quiet = 1;
+			break;
+		case 'l':
+			logfile = optarg;
+			break;
+		case 'o':
+			listenonly = 1;
+			break;
+		case 'a':
+			answerback = optarg;
+			break;
+		case 'e':
+			if (strlen(optarg) != 1) {
+				fprintf(stderr, "Option -e requires a single character argument.\n");
+				exit(EXIT_FAILURE);
+			}
+			escape_char = *optarg;
+			break;
 		}
 	}
 
@@ -239,8 +245,8 @@ int main(int argc, char *argv[])
 	ios->set_flow(ios, current_flow);
 
 	if (!listenonly) {
-		printf("Escape character: Ctrl-%c\n", escape_char);
-		printf("Type the escape character to get to the prompt.\n");
+		msg_printf("Escape character: Ctrl-%c\n", escape_char);
+		msg_printf("Type the escape character to get to the prompt.\n");
 
 		/* Now deal with the local terminal side */
 		tcgetattr(STDIN_FILENO, &sots);

microcom.h

@@ -72,6 +72,7 @@ void main_usage(int exitcode, char *str, char *dev);
 
 extern struct ios_ops *ios;
 extern int debug;
+extern int quiet;
 extern int opt_force;
 extern int listenonly;
 extern char *answerback;
@@ -121,7 +122,8 @@ extern int current_flow;
 int do_commandline(void);
 int do_script(char *script);
 
-#define dbg_printf(fmt,args...)  ({ if (debug) printf(fmt ,##args); })
+#define dbg_printf(...) do { if (debug) printf(__VA_ARGS__); } while (0)
+#define msg_printf(...) do { if (!quiet) printf(__VA_ARGS__); } while (0)
 
 /*
  * Some telnet options according to

serial.c

@@ -267,7 +267,7 @@ struct ios_ops * serial_init(char *device)
 	memcpy(&pots, &pts, sizeof (pots));
 	init_comm(&pts);
 	tcsetattr(fd, TCSANOW, &pts);
-	printf("connected to %s\n", device);
+	msg_printf("connected to %s\n", device);
 
 	return ops;
 }

telnet.c

@@ -603,7 +603,7 @@ struct ios_ops *telnet_init(char *hostport)
 			fprintf(stderr, "getnameinfo: %s\n", gai_strerror(ret));
 			goto out;
 		}
-		printf("connected to %s (port %s)\n", connected_host, connected_port);
+		msg_printf("connected to %s (port %s)\n", connected_host, connected_port);
 
 		/* send intent we WILL do COM_PORT stuff */
 		dbg_printf("-> WILL COM_PORT_CONTROL\n");

test/conftest.py

@@ -0,0 +1,24 @@
+import os
+from pathlib import Path
+import shlex
+import pytest
+
+
+def pytest_sessionstart(session):
+    # run tests in toplevel directory always, not in test/
+    os.chdir(Path(__file__).resolve().parent.parent)
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--cmd",
+        action="store",
+        default="./microcom",
+        help="Command used to invoke microcom"
+    )
+
+
+@pytest.fixture(scope="session")
+def cmd(pytestconfig):
+    cmd = pytestconfig.getoption("--cmd")
+    return shlex.split(cmd)

test/test_telnet.py

@@ -0,0 +1,101 @@
+import socket
+import threading
+import time
+import pytest
+import subprocess
+import os
+import pty
+
+RFC2217_CMD = bytes([255, 254, 44])  # RFC2217 IAC DONT COM-PORT-OPTION
+
+
+def make_pattern(length):
+    ascii_range = list(range(32, 127))
+    return bytes(ascii_range[i % len(ascii_range)] for i in range(length))
+
+
+@pytest.fixture
+def telnet_recv(cmd):
+    def _recv(buf, timeout=1):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.bind(("127.0.0.1", 0))
+        sock.listen(1)
+        host, port = sock.getsockname()
+
+        def run():
+            conn, _ = sock.accept()
+            with conn:
+                conn.sendall(buf)
+                time.sleep(0.01)  # sadly without this, microcom may miss the transmission
+                sock.close()
+
+        thread = threading.Thread(target=run, daemon=True)
+        thread.start()
+
+        telnet_cmd = cmd + [f"--telnet={host}:{port}", "--quiet"]
+        master_fd, slave_fd = pty.openpty()
+        proc = subprocess.Popen(
+            telnet_cmd,
+            stdin=slave_fd,
+            stdout=slave_fd,
+            stderr=os.dup(2),
+            close_fds=True,
+        )
+        os.close(slave_fd)
+        output = bytearray()
+        start_time = time.time()
+        while (time.time() - start_time) < timeout:
+            try:
+                chunk = os.read(master_fd, 1024)
+                if not chunk:
+                    break
+                output.extend(chunk)
+            except OSError:
+                break
+        os.close(master_fd)
+        proc.wait()
+        assert proc.returncode in (0, 1), f"Exit code must be 0 or 1, got {proc.returncode}"
+
+        return bytes(output)
+    return _recv
+
+
+@pytest.mark.parametrize("buf", [10, 1023, 1024, 1025, 4000])
+def test_no_cmd(telnet_recv, buf):
+    payload = make_pattern(buf)
+
+    assert telnet_recv(payload) == payload
+
+
+def test_cmd_across_buffers(telnet_recv):
+    before_pattern = make_pattern(1023)
+    after_pattern = make_pattern(20)
+    payload = before_pattern + RFC2217_CMD + after_pattern
+    expected_output = before_pattern + after_pattern
+
+    assert telnet_recv(payload) == expected_output
+
+
+def test_cmd_buffer_end(telnet_recv):
+    pattern = make_pattern(1023)
+    payload = pattern + RFC2217_CMD
+
+    assert telnet_recv(payload) == pattern
+
+
+def test_cmd_within_buffer(telnet_recv):
+    before_pattern = make_pattern(345)
+    after_pattern = make_pattern(890)
+    payload = before_pattern + RFC2217_CMD + after_pattern
+    expected_output = before_pattern + after_pattern
+
+    assert telnet_recv(payload) == expected_output
+
+
+def test_iac_escape(telnet_recv):
+    before_pattern = make_pattern(42)
+    after_pattern = make_pattern(42)
+    payload = before_pattern + bytes([255, 255]) + after_pattern
+    expected_output = before_pattern + bytes([255]) + after_pattern
+
+    assert telnet_recv(payload) == expected_output