"""HTTP-Roundtrip-Tests ohne laufende Java-App oder externe Python-Pakete. Ausführen: python3 -B -m unittest discover -s tools -p 'test_*.py' -v """ import contextlib import http.client import io import json import threading import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from unittest.mock import patch from editor_test_server import EditorTestServer, MAX_BODY_BYTES, RESULT_TYPE class FakeEditorHandler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_POST(self): self.server.imports.append(json.loads(self.rfile.read(int(self.headers["Content-Length"])))) token = "test-token-" + str(len(self.server.imports)) response = {"token": token, "deepLink": self.server.origin + "/?import=" + token} if self.server.external_link: response["deepLink"] = "https://unexpected.example/?import=" + token body = json.dumps(response).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) class TestEditorTestServer(unittest.TestCase): @classmethod def setUpClass(cls): cls.editor = ThreadingHTTPServer(("127.0.0.1", 0), FakeEditorHandler) cls.editor.origin = "http://127.0.0.1:" + str(cls.editor.server_port) cls.server = EditorTestServer(0, cls.editor.origin, {"tasks": []}) cls.threads = [] for server in (cls.editor, cls.server): thread = threading.Thread(target=lambda s=server: s.serve_forever(poll_interval=0.05), daemon=True) thread.start() cls.threads.append(thread) @classmethod def tearDownClass(cls): for server in (cls.server, cls.editor): server.shutdown() server.server_close() for thread in cls.threads: thread.join(timeout=2) def setUp(self): self.editor.imports = [] self.editor.external_link = False self.server.sessions.clear() def request(self, method, path, value=None, origin="local", raw=None, extra_headers=None): connection = http.client.HTTPConnection("127.0.0.1", self.server.server_port, timeout=2) headers = {"Content-Type": "application/json"} if origin: headers["Origin"] = self.server.local_origin if origin == "local" else origin if extra_headers: headers.update(extra_headers) body = raw if raw is not None else json.dumps(value).encode("utf-8") if value is not None else None try: connection.request(method, path, body=body, headers=headers) response = connection.getresponse() body = response.read() if response.getheader("Content-Type", "").startswith("application/json"): body = json.loads(body) return response.status, body finally: connection.close() def open_editor(self, payload=None): status, result = self.request("POST", "/api/open", {"tasks": []} if payload is None else payload) self.assertEqual(status, 200, result) return result["token"] def test_page_and_config(self): status, body = self.request("GET", "/") self.assertEqual(status, 200) self.assertIn(b"TaskList Editor testen", body) status, config = self.request("GET", "/api/config") self.assertEqual(status, 200) self.assertEqual(config["editorOrigin"], self.editor.origin) self.assertEqual(config["initialPayload"], {"tasks": [], "customTypes": []}) def test_import_sets_local_return_origin_and_preserves_metadata(self): payload = {"name": "Grüße", "tasks": [{"type": 3}], "metadata": {"id": 42, "nested": [True, None]}, "returnOrigin": "https://untrusted.example"} self.open_editor(payload) imported = self.editor.imports[0] self.assertEqual(imported["returnOrigin"], self.server.local_origin) self.assertEqual(imported["tasks"], payload["tasks"]) self.assertEqual(imported["metadata"], payload["metadata"]) self.assertEqual(imported["name"], payload["name"]) def test_bare_array_and_missing_tasks_create_new_lists(self): self.open_editor([]) self.open_editor({"name": "Neu"}) self.open_editor({"tasks": None}) self.assertEqual([value["tasks"] for value in self.editor.imports], [[], [], []]) self.assertEqual([value["customTypes"] for value in self.editor.imports], [[], [], []]) def test_custom_types_roundtrip_preserves_full_catalog_and_deletions(self): payload = json.loads(Path(__file__).with_name("example_tasks.json").read_text(encoding="utf-8")) original_types = payload["customTypes"] self.assertEqual(payload["tasks"][1]["type"], original_types[0]["taskType"]) token = self.open_editor(payload) self.assertEqual(self.editor.imports[0]["customTypes"], original_types) result = {"type": RESULT_TYPE, "token": token, "status": "completed", **payload} with contextlib.redirect_stdout(io.StringIO()) as output: self.assertEqual(self.request("POST", "/api/result", result), (200, {"ok": True})) printed = json.loads(output.getvalue().split("--- Editor-Rückgabe ---", 1)[1]) self.assertEqual(printed["customTypes"], original_types) # Der Aufrufer ersetzt den Katalog durch die gesamte Rückgabe, auch [] nach einer Löschung. token = self.open_editor({"tasks": [], "customTypes": original_types}) result = {"type": RESULT_TYPE, "token": token, "status": "completed", "tasks": [], "customTypes": []} with contextlib.redirect_stdout(io.StringIO()): self.assertEqual(self.request("POST", "/api/result", result), (200, {"ok": True})) self.open_editor({"tasks": result["tasks"], "customTypes": result["customTypes"]}) self.assertEqual(self.editor.imports[-1]["customTypes"], []) def test_new_list_can_receive_custom_types_without_tasks(self): custom_types = [{"typeId": "custom-1000", "taskType": 1000}] self.open_editor({"customTypes": custom_types}) self.assertEqual(self.editor.imports[0]["tasks"], []) self.assertEqual(self.editor.imports[0]["customTypes"], custom_types) def test_completed_and_retry_print_once(self): token = self.open_editor() result = {"type": RESULT_TYPE, "token": token, "status": "completed", "name": "Grüße", "tasks": [{"type": 3}], "customTypes": [], "metadata": {"id": 42}} output = io.StringIO() with contextlib.redirect_stdout(output): self.assertEqual(self.request("POST", "/api/result", result), (200, {"ok": True})) self.assertEqual(self.request("POST", "/api/result", result), (200, {"ok": True, "duplicate": True})) self.assertEqual(output.getvalue().count("--- Editor-Rückgabe ---"), 1) self.assertIn('"name": "Grüße"', output.getvalue()) printed = output.getvalue().split("--- Editor-Rückgabe ---", 1)[1] self.assertEqual(json.loads(printed), result) result["tasks"] = [] self.assertEqual(self.request("POST", "/api/result", result)[0], 409) def test_cancelled_has_no_tasks_or_custom_types(self): result = {"type": RESULT_TYPE, "token": self.open_editor(), "status": "cancelled", "metadata": {"id": 42}} with contextlib.redirect_stdout(io.StringIO()): self.assertEqual(self.request("POST", "/api/result", result)[0], 200) for key in ("tasks", "customTypes"): with self.subTest(key=key): self.assertEqual(self.request("POST", "/api/result", {**result, key: []})[0], 400) def test_invalid_result_and_unknown_token(self): result = {"type": RESULT_TYPE, "token": "unknown", "status": "completed", "tasks": [], "customTypes": []} self.assertEqual(self.request("POST", "/api/result", result)[0], 404) result["token"] = self.open_editor() result["tasks"] = "bad" self.assertEqual(self.request("POST", "/api/result", result)[0], 400) result["status"] = "invalid" self.assertEqual(self.request("POST", "/api/result", result)[0], 400) def test_completed_requires_full_custom_type_catalog(self): result = {"type": RESULT_TYPE, "token": self.open_editor(), "status": "completed", "tasks": []} self.assertEqual(self.request("POST", "/api/result", result)[0], 400) for invalid_types in (None, False, {}, [1]): with self.subTest(custom_types=invalid_types): result["customTypes"] = invalid_types self.assertEqual(self.request("POST", "/api/result", result)[0], 400) def test_expired_token(self): result = {"type": RESULT_TYPE, "token": self.open_editor(), "status": "cancelled"} self.server.sessions[result["token"]]["created"] -= 3601 self.assertEqual(self.request("POST", "/api/result", result)[0], 404) def test_untrusted_origin_and_host(self): for origin in (None, "https://untrusted.example"): with self.subTest(origin=origin): self.assertEqual(self.request("POST", "/api/open", [], origin=origin)[0], 403) self.assertEqual(self.request("GET", "/api/config", extra_headers={"Host": "untrusted.example"})[0], 403) self.assertFalse(self.editor.imports) def test_invalid_payloads(self): for payload in (42, "bad", {"tasks": False}, {"tasks": [1]}, {"name": 42}, {"customTypes": {}}, {"customTypes": [1]}): with self.subTest(payload=payload): self.assertEqual(self.request("POST", "/api/open", payload)[0], 400) for raw in (b'{"tasks": [}', b'{"tasks": [], "metadata": NaN}', b'\xff'): with self.subTest(raw=raw): self.assertEqual(self.request("POST", "/api/open", raw=raw)[0], 400) self.assertFalse(self.editor.imports) def test_body_limit_and_content_type(self): self.assertEqual(self.request("POST", "/api/open", [], extra_headers={"Content-Length": str(MAX_BODY_BYTES + 1)})[0], 413) self.assertEqual(self.request("POST", "/api/open", [], extra_headers={"Content-Type": "text/plain"})[0], 415) def test_external_deep_link_rejected(self): self.editor.external_link = True self.assertEqual(self.request("POST", "/api/open", [])[0], 502) self.assertFalse(self.server.sessions) def test_unreachable_editor_has_useful_error(self): with patch("editor_test_server.urllib.request.urlopen", side_effect=OSError("Connection refused")): status, body = self.request("POST", "/api/open", []) self.assertEqual(status, 502) self.assertIn("nicht erreichbar", body["error"]) if __name__ == "__main__": unittest.main()