test: add pytest and implement a test runner

* use pytest as test runner
* delete scripts folder and create a test folder
* implement the test_runner at tests/test_runner.py

.gitignore

@@ -2,3 +2,4 @@ target/
 __pycache__
 build/
 *.egg-info/
+artifacts

README.md

@@ -117,3 +117,22 @@ $> --> pica_create_anchor 00:01 # Create another one
 
 Pica also implements HTTP commands, the documentation is available at `http://0.0.0.0:3000/openapi`.
 The set of HTTP commands let the user interact with Pica amd modify its scene.
+
+# Tests
+
+setup your python env:
+
+```bash
+python3 -m venv venv
+source venv/bin/activate
+pip install pytest
+pip install -e py/pica/
+```
+
+Then run the Tests
+
+```bash
+pytest --log-cli-level=DEBUG -v
+```
+
+The tests are located at `./tests/`

py/pica/pica/__init__.py

@@ -22,6 +22,7 @@ async def connect(address: str, port: int, mac_address: bytes) -> 'Host':
 
     def disconnect(self):
         self.writer.close()
+        self.reader_task.cancel()
 
     async def _read_exact(self, expected_len: int) -> bytes:
         """ Read an exact number of bytes from the socket.

scripts/data_transfer_example.py → tests/data_transfer.py

@@ -18,12 +18,13 @@
 import argparse
 from pica import Host
 from pica.packets import uci
-from helper import init
+from .helper import init
+from pathlib import Path
 
 MAX_DATA_PACKET_PAYLOAD_SIZE = 1024
 
 
-async def data_message_send(host: Host, peer: Host, file: str):
+async def data_message_send(host: Host, peer: Host, file: Path):
     await init(host)
 
     host.send_control(
@@ -96,7 +97,9 @@ async def data_message_send(host: Host, peer: Host, file: str):
     await host.expect_control(uci.SessionDeinitRsp(status=uci.StatusCode.UCI_STATUS_OK))
 
 
-async def data_transfer(host: Host, dst_mac_address: bytes, file: str, session_id: int):
+async def data_transfer(
+    host: Host, dst_mac_address: bytes, file: Path, session_id: int
+):
     try:
         with open(file, "rb") as f:
             b = f.read()
@@ -158,7 +161,7 @@ async def data_transfer(host: Host, dst_mac_address: bytes, file: str, session_i
         print(e)
 
 
-async def run(address: str, uci_port: int, http_port: int, file: str):
+async def run(address: str, uci_port: int, file: Path):
     try:
         host0 = await Host.connect(address, uci_port, bytes([0, 1]))
         host1 = await Host.connect(address, uci_port, bytes([0, 2]))
@@ -194,7 +197,7 @@ def main():
         "--http-port", type=int, default=3000, help="Select the pica HTTP port"
     )
     parser.add_argument(
-        "--file", type=str, required=True, help="Select the file to transfer"
+        "--file", type=Path, required=True, help="Select the file to transfer"
     )
     asyncio.run(run(**vars(parser.parse_args())))
 

scripts/ranging_example.py → tests/ranging.py

@@ -16,10 +16,11 @@
 
 import asyncio
 import argparse
+import logging
 
 from pica import Host
 from pica.packets import uci
-from helper import init
+from .helper import init
 
 
 async def controller(host: Host, peer: Host):
@@ -234,12 +235,12 @@ async def controlee(host: Host, peer: Host):
     await host.expect_control(uci.SessionDeinitRsp(status=uci.StatusCode.UCI_STATUS_OK))
 
 
-async def run(address: str, uci_port: int, http_port: int):
+async def run(address: str, uci_port: int):
     try:
         host0 = await Host.connect(address, uci_port, bytes([0, 1]))
         host1 = await Host.connect(address, uci_port, bytes([0, 2]))
     except Exception:
-        print(
+        logging.debug(
             f"Failed to connect to Pica server at address {address}:{uci_port}\n"
             + "Make sure the server is running"
         )
@@ -249,10 +250,12 @@ async def run(address: str, uci_port: int, http_port: int):
         tg.create_task(controller(host0, host1))
         tg.create_task(controlee(host1, host0))
 
+    logging.debug("calling disconnect")
+
     host0.disconnect()
     host1.disconnect()
 
-    print("Ranging test completed")
+    logging.debug("Ranging test completed")
 
 
 def main():

tests/test_runner.py

@@ -0,0 +1,66 @@
+import asyncio
+from asyncio.subprocess import Process
+import pytest
+import pytest_asyncio
+import logging
+import os
+
+from datetime import datetime
+from pathlib import Path
+from typing import Tuple
+
+from . import ranging, data_transfer
+
+PICA_BIN = Path("./target/debug/pica-server")
+DATA_FILE = Path("README.md")
+PICA_LOCALHOST = "127.0.0.1"
+
+logging.basicConfig(level=os.environ.get("PICA_LOGLEVEL", "DEBUG").upper())  # type: ignore
+
+
+def setup_artifacts(test_name: str) -> Tuple[Path, Path]:
+    artifacts = Path("./artifacts")
+    artifacts.mkdir(parents=True, exist_ok=True)
+
+    current_dt = datetime.now()
+    formatted_date = current_dt.strftime("%Y-%m-%d_%H-%M-%S-%f")[:-3]
+
+    f1 = artifacts / f"{formatted_date}_pica_{test_name}_stdout.txt"
+    f1.touch(exist_ok=True)
+
+    f2 = artifacts / f"{formatted_date}_pica_{test_name}_stderr.txt"
+    f2.touch(exist_ok=True)
+
+    return (f1, f2)
+
+
+@pytest_asyncio.fixture
+async def pica_port(request, unused_tcp_port):
+    (stdout, stderr) = setup_artifacts(request.node.name)
+    if not PICA_BIN.exists():
+        raise FileNotFoundError(f"{PICA_BIN} not found")
+
+    with stdout.open('w') as fstdout, stderr.open('w') as fstderr:
+        process = await asyncio.create_subprocess_exec(
+            PICA_BIN,
+            "--uci-port",
+            str(unused_tcp_port),
+            stdout=fstdout,
+            stderr=fstderr,
+        )
+        await asyncio.sleep(100 / 1000)  # Let pica boot up
+
+        yield unused_tcp_port
+
+        process.terminate()
+        await process.wait()
+
+
+@pytest.mark.asyncio
+async def test_ranging(pica_port):
+    await ranging.run(PICA_LOCALHOST, pica_port)
+
+
+@pytest.mark.asyncio
+async def test_data_transfer(pica_port):
+    await data_transfer.run(PICA_LOCALHOST, pica_port, DATA_FILE)