data-collector-debugtool/intergration/bacnet/common.py

import atexit
import json
import os
import socket
import threading
import time
import urllib.error
import urllib.request
from contextlib import closing

import uvicorn

from app.main import app


CONFIGURED_BASE_URL = os.getenv("DATA_COLLECTOR_BASE_URL", "").rstrip("/")
REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "30"))

BACNET_DEVICE_IP = os.getenv("BACNET_DEVICE_IP", "192.168.75.240")
BACNET_DEVICE_ID = int(os.getenv("BACNET_DEVICE_ID", "12345"))
BACNET_PORT = int(os.getenv("BACNET_PORT", "47808"))

_LOCAL_API_SERVER = None


def find_free_port() -> int:
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


class LocalApiServer:
    def __init__(self) -> None:
        self.port = find_free_port()
        self.base_url = f"http://127.0.0.1:{self.port}"
        config = uvicorn.Config(app, host="127.0.0.1", port=self.port, log_level="critical", access_log=False)
        self.server = uvicorn.Server(config)
        self.thread = threading.Thread(target=self.server.run, daemon=True)

    def start(self) -> None:
        self.thread.start()
        deadline = time.time() + 10
        while time.time() < deadline:
            try:
                with urllib.request.urlopen(f"{self.base_url}/api/dc-debugtool/health", timeout=1) as response:
                    if response.status == 200:
                        return
            except OSError:
                time.sleep(0.05)
        raise RuntimeError("local API test server did not start")

    def stop(self) -> None:
        self.server.should_exit = True
        self.thread.join(timeout=5)


def base_url() -> str:
    global _LOCAL_API_SERVER
    if CONFIGURED_BASE_URL:
        return CONFIGURED_BASE_URL
    if _LOCAL_API_SERVER is None:
        _LOCAL_API_SERVER = LocalApiServer()
        _LOCAL_API_SERVER.start()
    return _LOCAL_API_SERVER.base_url


def stop_local_api_server() -> None:
    if _LOCAL_API_SERVER is not None:
        _LOCAL_API_SERVER.stop()


atexit.register(stop_local_api_server)


def post_json(path: str, payload: dict) -> tuple[int, dict]:
    data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        f"{base_url()}{path}",
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response:
            body = response.read().decode("utf-8")
            return response.status, json.loads(body)
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8")
        return exc.code, json.loads(body)


def device_payload() -> dict:
    return {
        "ip": BACNET_DEVICE_IP,
        "bacnet_device_id": BACNET_DEVICE_ID,
        "port": BACNET_PORT,
    }


def assert_response_contract(testcase, status: int, data: dict) -> None:
    testcase.assertEqual(200, status, data)
    testcase.assertIn("code", data)
    testcase.assertIn("msg", data)
    testcase.assertIn("data", data)
    testcase.assertIsInstance(data["data"], dict)


def assert_bacnet_device(testcase, payload: dict, data: dict) -> None:
    testcase.assertEqual(
        {
            "device_type": "BACnet/IP",
            "ip": payload["ip"],
            "port": payload["port"],
            "bacnet_device_id": payload["bacnet_device_id"],
        },
        data["data"]["device"],
    )