FreeDMR/tests/test_api.py

import io
import json
import sys
import types
import unittest


def install_dmr_utils_stub():
    if "dmr_utils3.utils" in sys.modules:
        return None
    dmr_utils3 = types.ModuleType("dmr_utils3")
    utils = types.ModuleType("dmr_utils3.utils")

    def bytes_4(value):
        return int(value).to_bytes(4, "big")

    utils.bytes_4 = bytes_4
    sys.modules["dmr_utils3"] = dmr_utils3
    sys.modules["dmr_utils3.utils"] = utils
    return ("dmr_utils3", "dmr_utils3.utils")


class FakeRequest:
    def __init__(self, path, payload=None):
        self.postpath = [part.encode("utf-8") for part in path.strip("/").split("/") if part]
        self.content = io.BytesIO(
            b"" if payload is None else json.dumps(payload).encode("utf-8")
        )
        self.code = None
        self.headers = {}

    def setResponseCode(self, code):
        self.code = code

    def setHeader(self, name, value):
        self.headers[name] = value

    def getHeader(self, name):
        if name == "content-length":
            return str(len(self.content.getvalue()))
        return None


class APITest(unittest.TestCase):
    def setUp(self):
        try:
            import twisted.web.resource  # noqa: F401
        except ModuleNotFoundError as exc:
            self.skipTest(f"Twisted is not installed: {exc}")
        self.stubbed_modules = install_dmr_utils_stub()
        import API

        self.api = API
        self.peer_id = (1234567).to_bytes(4, "big")
        self.config = {
            "GLOBAL": {"SYSTEM_API_KEY": "system-secret", "_KILL_SERVER": False},
            "SYSTEMS": {
                "MASTER-A": {
                    "MODE": "MASTER",
                    "PEERS": {self.peer_id: {}},
                    "_opt_key": "peer-secret",
                },
                "OBP-A": {
                    "MODE": "OPENBRIDGE",
                    "PEERS": {},
                },
            },
        }
        self.bridges = {}
        self.controller = API.FD_APIController(self.config, self.bridges)

    def tearDown(self):
        if self.stubbed_modules:
            for module in self.stubbed_modules:
                sys.modules.pop(module, None)

    def test_getoptions_returns_clear_no_options_response(self):
        result = self.controller.getoptions("MASTER-A")

        self.assertEqual(
            result,
            {"connected": True, "has_options": False, "options": ""},
        )

    def test_getoptions_decodes_byte_options_for_json(self):
        self.config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = b"KEY=peer-secret;TS1=91"

        result = self.controller.getoptions("MASTER-A")

        self.assertEqual(result["options"], "KEY=peer-secret;TS1=91")
        self.assertTrue(result["has_options"])

    def test_setoptions_stores_full_options_string_unchanged(self):
        options = "KEY=peer-secret;TS1=91;DIAL=2350"

        self.controller.options("MASTER-A", options)

        self.assertEqual(self.config["SYSTEMS"]["MASTER-A"]["OPTIONS"], options)

    def test_user_reset_is_allowed_only_for_matching_peer_key(self):
        system = self.controller.validateKey(1234567, "peer-secret")

        self.assertEqual(system, "MASTER-A")
        self.controller.reset(system)
        self.assertTrue(self.config["SYSTEMS"]["MASTER-A"]["_reset"])
        self.assertFalse(self.controller.validateKey(1234567, "wrong"))

    def test_system_kill_sets_existing_control_flag(self):
        self.assertTrue(self.controller.validateSystemKey("system-secret"))

        self.controller.killserver()

        self.assertTrue(self.config["GLOBAL"]["_KILL_SERVER"])

    def test_options_get_endpoint_returns_json(self):
        resource = self.api.make_api_resource(self.config, self.bridges)
        request = FakeRequest(
            "/api/v1/options/get",
            {"dmrid": 1234567, "key": "peer-secret"},
        )

        body = resource.render_POST(request)

        self.assertEqual(request.code, 200)
        self.assertEqual(
            json.loads(body.decode("utf-8")),
            {"ok": True, "connected": True, "has_options": False, "options": ""},
        )

    def test_options_get_endpoint_rejects_bad_key(self):
        resource = self.api.make_api_resource(self.config, self.bridges)
        request = FakeRequest(
            "/api/v1/options/get",
            {"dmrid": 1234567, "key": "wrong"},
        )

        body = resource.render_POST(request)

        self.assertEqual(request.code, 401)
        self.assertEqual(
            json.loads(body.decode("utf-8")),
            {"ok": False, "error": "invalid_credentials"},
        )

    def test_endpoint_rejects_large_request_body(self):
        resource = self.api.make_api_resource(self.config, self.bridges)
        request = FakeRequest(
            "/api/v1/options/set",
            {"dmrid": 1234567, "key": "peer-secret", "options": "A" * 9000},
        )

        body = resource.render_POST(request)

        self.assertEqual(request.code, 413)
        self.assertEqual(
            json.loads(body.decode("utf-8")),
            {"ok": False, "error": "request_too_large"},
        )


if __name__ == "__main__":
    unittest.main()