From 58b6f3feda3662c2004bdfaa1e7300f14dabd931 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Mon, 9 Nov 2026 18:12:33 +0100 Subject: feat: HTTP server with trust, auth, module routing --- tests/test_server.py | 91 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) create mode 100644 tests/test_server.py (limited to 'tests/test_server.py') diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..1583ad2 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,91 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import json +import os +import socket +import tempfile +import threading +import unittest + +import http.client + +import deckit.module as mod +import deckit.server as srv + +TOKEN = "test-token" +HOST = {"Host": "127.0.0.1:8765", "Cookie": f"deckit={TOKEN}"} +ORIGIN = {"Origin": "http://127.0.0.1:8765"} + + +class UnixConnection(http.client.HTTPConnection): + def __init__(self, path): + super().__init__("localhost") + self.path = path + + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.connect(self.path) + + +class Server(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.dir = tempfile.TemporaryDirectory() + cls.sock = os.path.join(cls.dir.name, "http.sock") + srv.TOKEN = TOKEN + srv.REGISTRY = mod.Registry([ + mod.Module("demo", "Demo", lambda: {"ok": True}, + {"ping": lambda body: {"got": body.get("n")}}), + ]) + cls.server = srv.UnixHTTPServer(cls.sock, srv.Handler) + threading.Thread(target=cls.server.serve_forever, daemon=True).start() + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + cls.dir.cleanup() + + def req(self, method, path, headers=None, body=None): + conn = UnixConnection(self.sock) + conn.request(method, path, body=body, headers=headers or {}) + r = conn.getresponse() + data = r.read() + conn.close() + return r.status, data + + def test_assets_and_api(self): + self.assertEqual(self.req("GET", "/", HOST)[0], 200) + self.assertEqual(self.req("GET", "/theme.css", HOST)[0], 200) + self.assertEqual(json.loads(self.req("GET", "/api/modules", HOST)[1]), + [{"name": "demo", "title": "Demo"}]) + self.assertEqual(json.loads(self.req("GET", "/api/demo", HOST)[1]), {"ok": True}) + + def test_trust_and_auth(self): + self.assertEqual(self.req("GET", "/api/demo", {"Host": "evil.com"})[0], 403) + self.assertEqual(self.req("POST", "/api/demo/x", {**HOST, "Origin": "https://evil.com"})[0], 403) + self.assertEqual(self.req("GET", "/api/demo", {"Host": "127.0.0.1:8765"})[0], 401) + + def test_unlock_and_bad_action(self): + conn = UnixConnection(self.sock) + conn.request("GET", f"/?t={TOKEN}", headers={"Host": "127.0.0.1:8765"}) + r = conn.getresponse() + r.read() + conn.close() + self.assertEqual(r.status, 303) + self.assertIn("deckit=", r.getheader("Set-Cookie")) + self.assertIn("SameSite=Strict", r.getheader("Set-Cookie")) + self.assertEqual(self.req("POST", "/api/demo/nope", {**HOST, **ORIGIN})[0], 404) + + def test_rejects_oversized_body(self): + big = "x" * 5000 + status, _ = self.req("POST", "/api/demo/ping", + {**HOST, **ORIGIN, "Content-Type": "application/json"}, big) + self.assertEqual(status, 400) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3