You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
140 lines
4.4 KiB
140 lines
4.4 KiB
import io
|
|
import json
|
|
import unittest
|
|
|
|
|
|
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}")
|
|
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 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()
|