1import asyncio
2import pytest
3import pytest_asyncio
4import logging
5import os
6
7from datetime import datetime
8from pathlib import Path
9from typing import Tuple
10
11from . import ranging, data_transfer
12
13PICA_BIN = Path("./target/debug/pica")
14DATA_FILE = Path("README.md")
15PICA_LOCALHOST = "127.0.0.1"
16
17logging.basicConfig(level=os.environ.get("PICA_LOGLEVEL", "DEBUG").upper())
18
19
20def setup_artifacts(test_name: str) -> Tuple[Path, Path]:
21    artifacts = Path("./artifacts")
22    artifacts.mkdir(parents=True, exist_ok=True)
23
24    current_dt = datetime.now()
25    formatted_date = current_dt.strftime("%Y-%m-%d_%H-%M-%S-%f")[:-3]
26
27    f1 = artifacts / f"{formatted_date}_pica_{test_name}_stdout.txt"
28    f1.touch(exist_ok=True)
29
30    f2 = artifacts / f"{formatted_date}_pica_{test_name}_stderr.txt"
31    f2.touch(exist_ok=True)
32
33    return (f1, f2)
34
35
36@pytest_asyncio.fixture
37async def pica_port(request, unused_tcp_port):
38    (stdout, stderr) = setup_artifacts(request.node.name)
39    if not PICA_BIN.exists():
40        raise FileNotFoundError(f"{PICA_BIN} not found")
41
42    with stdout.open("w") as fstdout, stderr.open("w") as fstderr:
43        process = await asyncio.create_subprocess_exec(
44            PICA_BIN,
45            "--uci-port",
46            str(unused_tcp_port),
47            stdout=fstdout,
48            stderr=fstderr,
49        )
50        await asyncio.sleep(100 / 1000)  # Let pica boot up
51
52        yield unused_tcp_port
53
54        process.terminate()
55        await process.wait()
56
57
58@pytest.mark.asyncio
59async def test_ranging(pica_port):
60    await ranging.run(PICA_LOCALHOST, pica_port)
61
62
63@pytest.mark.asyncio
64async def test_data_transfer(pica_port):
65    await data_transfer.run(PICA_LOCALHOST, pica_port, DATA_FILE)
66