Add e2e test for file ops

.github/workflows/ci.yml

@@ -21,7 +21,7 @@ jobs:
     name: test
     runs-on: ubuntu-latest
     env:
-      MICROPYPATH: ".frozen:~/.micropython/lib:/usr/lib/micropython:$(pwd)"
+      MICROPYPATH: "$(pwd)/tests/mocks:$(pwd):.frozen:~/.micropython/lib:/usr/lib/micropython"
     steps:
       - uses: actions/checkout@v2
       - name: Build micropython
@@ -37,7 +37,11 @@ jobs:
           rm -rf micropython
       - name: Install test dependencies
         run: ./bin/setup
-      - name: Install the package
+      - name: Install mosquitto
+        run: sudo apt-get install -y mosquitto
+      - name: Install the package dependencies
         run: "micropython -m mip install github:${{ github.repository }}@${{ github.ref_name }}"
       - name: Run tests
         run: micropython -m unittest
+      - name: Run integration tests
+        run: ./bin/test_e2e

README.md

@@ -31,30 +31,64 @@ TODO
 
 You need python and a build of micropython with `asyncio` support. Follow the steps in the CI workflow to get a `micropython` binary and add it to your `PATH`.
 
-Before making changes, install the development dependencies:
+Before making changes, install the development (CPython) dependencies:
 
 ```bash
 pip install -r dev-requirements.txt
 ```
 
-After making changes, you can run the linter:
+### Linting
+
+This project uses [ruff](https://github.com/astral-sh/ruff) for linting. After making changes, you can run the linter:
 
 ```bash
 ruff check
 ```
 
-Before running tests, install the test dependencies:
+### Testing
+
+Before running tests, install the test (micropython) dependencies:
 
 ```bash
 ./bin/setup
 ```
 
-Then, you can run the tests using the micropython version of `unittest`:
+Note that you need to set up your `MICROPYPATH` environment variable so that the local copy of the package is loaded before any installed packages.
+
+```bash
+export MICROPYPATH="$(pwd)/tests/mocks:$(pwd):.frozen:~/.micropython/lib:/usr/lib/micropython"
+```
+
+#### Unit tests
+
+You can run the unit tests using the micropython version of `unittest`:
 
 ```bash
 micropython -m unittest
 ```
 
+#### Integration tests
+
+Integration tests use a running MQTT broker ([mosquitto](https://mosquitto.org/)), which you need to have installed (e.g. with `brew`).
+
+There is a script that will set up the test environment, run the tests, and tear down the broker afterward:
+
+```bash
+./bin/test_e2e
+```
+
+Sometimes it's useful to debug an individual integration test. To do this, you need to run the broker yourself, then set up the environment and invoke the test directly:
+
+```bash
+mosquitto -v  # keep open to check the broker logs
+```
+
+Then in another terminal:
+
+```bash
+LOG_LEVEL=DEBUG MICROPYPATH="$(pwd)/tests/mocks:$(pwd):.frozen:~/.micropython/lib:/usr/lib/micropython" micropython ./tests/e2e/e2e_file_ops.py
+```
+
 ## Releasing
 
 To release a new version, update the version in `package.json`. Commit your changes and make a pull request. After merging, create a new tag and push to GitHub:

bin/test_e2e

@@ -0,0 +1,35 @@
+#!/usr/bin/env bash
+
+# ensure we can import the modules from the package
+# import any mocked modules from tests/mocks/ first so they take precedence
+export MICROPYPATH="$(pwd)/tests/mocks:$(pwd):.frozen:~/.micropython/lib:/usr/lib/micropython"
+
+# bail out if mosquitto executable not found with `which mosquitto`
+if ! which mosquitto > /dev/null; then
+    echo "mosquitto not found"
+    exit 1
+fi
+
+# start mosquitto broker
+echo "starting mosquitto broker"
+mosquitto -d
+
+# loop over all test files in tests/mpy/e2e and run them with ./bin/micropython
+for test in tests/e2e/e2e_*.py; do
+    # print test file name
+    echo -en "$test "
+
+    # run test and if any test fails, set failed=1
+    if ! micropython $test; then
+        failed=1
+    fi
+done
+
+# stop mosquitto broker
+echo "shutting down mosquitto broker"
+pkill mosquitto
+
+# if failed is set, exit with status 1 (so CI will fail)
+if [ -n "$failed" ]; then
+    exit 1
+fi

mqterm/terminal.py

@@ -104,7 +104,7 @@ async def stream_job_output(self, job):
                 self.logger.debug(f"Streaming {bytes_read} bytes")
                 await self.mqtt_client.publish(
                     self.out_topic,
-                    self.out_view[:bytes_read].tobytes(),
+                    self.out_view[:bytes_read],
                     qos=1,
                     properties={
                         self.PROP_CORR: job.client_id.encode("utf-8"),

package.json

@@ -5,7 +5,7 @@
     ["mqterm/terminal.py", "github:solanus-systems/mqterm/mqterm/terminal.py"]
   ],
   "deps": [
-    ["github:solanus-systems/amqc", "v0.1.0"],
+    ["github:solanus-systems/amqc", "v0.3.0"],
     ["logging", "latest"]
   ],
   "version": "0.2.0"

tests/e2e/e2e_file_ops.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env micropython
+
+"""End-to-end tests for copying files to and from the device."""
+
+import asyncio
+import logging
+import os
+import sys
+from io import BytesIO
+
+from amqc.client import MQTTClient, config
+
+from mqterm.terminal import MqttTerminal
+
+# Set up logging; pass LOG_LEVEL=DEBUG if needed for local testing
+logger = logging.getLogger()
+logger.setLevel(getattr(logging, os.getenv("LOG_LEVEL", "WARNING").upper()))
+formatter = logging.Formatter(
+    "%(asctime)s.%(msecs)d - %(levelname)s - %(name)s - %(message)s"
+)
+handler = logging.StreamHandler(sys.stdout)
+handler.setFormatter(formatter)
+logger.handlers = []
+logger.addHandler(handler)
+device_logger = logging.getLogger("device")
+control_logger = logging.getLogger("control")
+
+
+# MQTT Client config
+config["server"] = "localhost"
+config["queue_len"] = 1  # use event queue
+device_config = config.copy()
+control_config = config.copy()
+device_config["client_id"] = "device"
+control_config["client_id"] = "server"
+
+# Set up MQTT clients
+device_client = MQTTClient(device_config, logger=device_logger)
+control_client = MQTTClient(control_config, logger=control_logger)
+
+# Set up the terminal
+term = MqttTerminal(device_client, topic_prefix="/test")
+
+
+def create_props(seq: int, client_id: str) -> dict:
+    """Create MQTTv5 properties with a seq number and client ID."""
+    return {
+        MqttTerminal.PROP_CORR: client_id.encode("utf-8"),
+        MqttTerminal.PROP_USER: {"seq": str(seq)},
+    }
+
+
+async def send_file(buffer: BytesIO):
+    """Send a file to the terminal."""
+    # Send the first message that will create the job
+    seq = 0
+    props = create_props(seq, "tty0")
+    await control_client.publish(
+        "/test/tty/in", "cp test.txt test.txt".encode("utf-8"), properties=props
+    )
+
+    # Send the file in 4-byte chunks; close when done
+    fp = BytesIO(b"Hello world!")
+    while True:
+        await asyncio.sleep(0.5)
+        chunk = fp.read(4)
+        if chunk:
+            seq += 1
+        else:
+            seq = -1
+        props = create_props(seq, "tty0")
+        logger.debug(f"Sending chunk {seq} of size {len(chunk)}: {chunk!r}")
+        await control_client.publish("/test/tty/in", chunk, properties=props)
+        if seq == -1:
+            break
+
+    # Wait until the received buffer gets populated with the response
+    await asyncio.sleep(1)
+
+    # Return the bytes received and empty the output buffer
+    buffer.seek(0)
+    output = buffer.read()
+    logger.debug(f"Buffer contents: {output!r}")
+    buffer.flush()
+    buffer.seek(0)
+    return output
+
+
+async def get_file(buffer: BytesIO):
+    """Send a file to the terminal and read it back."""
+    # Send the request for the file
+    seq = 0
+    props = create_props(seq, "tty0")
+    await control_client.publish(
+        "/test/tty/in", "cat test.txt".encode(), properties=props
+    )
+
+    # Wait until the received buffer gets populated with the response
+    await asyncio.sleep(1)
+
+    # Return the bytes received and empty the output buffer
+    buffer.seek(0)
+    output = buffer.read()
+    buffer.flush()
+    return output
+
+
+# Handler for device messages that passes them to the terminal
+async def device_handler():
+    async for topic, payload, _retained, properties in device_client.queue:
+        device_logger.debug(f"Device received {len(payload)} bytes on topic '{topic}'")
+        await term.handle_msg(topic, payload, properties)
+
+
+# Handler for control messages that logs and stores them
+async def control_handler(buffer):
+    async for topic, payload, _retained, properties in control_client.queue:
+        if topic == "/test/tty/err":
+            logger.error(f"Control received error: {payload.decode('utf-8')}")
+        else:
+            buffer.write(payload)  # Don't decode yet
+            logger.debug(f"Control received {len(payload)} bytes on topic '{topic}'")
+
+
+# Main test function
+async def main():
+    # Connect all clients and the terminal
+    await control_client.connect(True)
+    await control_client.subscribe("/test/tty/out")
+    await control_client.subscribe("/test/tty/err")
+    await device_client.connect(True)
+    await term.connect()
+
+    # Run handlers in the background and test task in the foreground
+    buffer = BytesIO()  # buffer for received bytes
+    asyncio.create_task(control_handler(buffer))
+    asyncio.create_task(device_handler())
+
+    bytes_sent = await send_file(buffer)
+    logger.debug(f"Sent {bytes_sent.decode()} bytes to the device")
+    bytes_received = await get_file(buffer)
+    logger.debug(f"Received {len(bytes_received)} bytes from the device")
+
+    # Disconnect and clean up
+    await term.disconnect()
+    await device_client.disconnect()
+    await control_client.disconnect()
+
+    # Read out the received file
+    received_str = bytes_received.decode("utf-8")
+    assert (
+        received_str == "Hello world!"
+    ), f"Expected 'Hello world!', got '{received_str}'"
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+    print("\033[1m\tOK\033[0m")