""" Tests for security/decorators.py — require_json_body decorator. Covers all three error formats and common edge cases: empty body, non-dict JSON payloads, malformed JSON, and valid dicts. """ import json import pytest from flask import Flask from local_deep_research.security.decorators import require_json_body @pytest.fixture def app(): """Minimal Flask app with routes using each error format.""" app = Flask(__name__) @app.route("/simple", methods=["POST"]) @require_json_body() def simple_route(): return {"ok": True} @app.route("/simple-custom", methods=["POST"]) @require_json_body(error_message="Query parameter is required") def simple_custom_route(): return {"ok": True} @app.route("/status", methods=["POST"]) @require_json_body(error_format="status") def status_route(): return {"ok": True} @app.route("/status-custom", methods=["POST"]) @require_json_body( error_format="status", error_message="No settings data provided" ) def status_custom_route(): return {"ok": True} @app.route("/success", methods=["POST"]) @require_json_body(error_format="success") def success_route(): return {"ok": True} @app.route("/success-custom", methods=["POST"]) @require_json_body( error_format="success", error_message="Missing required fields" ) def success_custom_route(): return {"ok": True} return app @pytest.fixture def client(app): return app.test_client() # --------------------------------------------------------------------------- # Valid dict payloads — should always pass through # --------------------------------------------------------------------------- class TestValidPayloads: def test_valid_dict_passes_simple(self, client): resp = client.post( "/simple", data=json.dumps({"key": "value"}), content_type="application/json", ) assert resp.status_code == 200 assert resp.json["ok"] is True def test_valid_dict_passes_status(self, client): resp = client.post( "/status", data=json.dumps({"key": "value"}), content_type="application/json", ) assert resp.status_code == 200 def test_valid_dict_passes_success(self, client): resp = client.post( "/success", data=json.dumps({"key": "value"}), content_type="application/json", ) assert resp.status_code == 200 def test_empty_dict_passes(self, client): """An empty dict {} is a valid JSON body.""" resp = client.post( "/simple", data=json.dumps({}), content_type="application/json", ) assert resp.status_code == 200 def test_nested_dict_passes(self, client): resp = client.post( "/simple", data=json.dumps({"nested": {"a": 1}, "list": [1, 2]}), content_type="application/json", ) assert resp.status_code == 200 # --------------------------------------------------------------------------- # Empty / missing body — should reject # --------------------------------------------------------------------------- class TestEmptyBody: def test_no_body_simple(self, client): resp = client.post("/simple") assert resp.status_code == 400 assert resp.json["error"] == "Request body must be valid JSON" def test_no_body_status(self, client): resp = client.post("/status") assert resp.status_code == 400 assert resp.json["status"] == "error" assert resp.json["message"] == "Request body must be valid JSON" def test_no_body_success(self, client): resp = client.post("/success") assert resp.status_code == 400 assert resp.json["success"] is False assert resp.json["error"] == "Request body must be valid JSON" def test_empty_string_body(self, client): resp = client.post("/simple", data="", content_type="application/json") assert resp.status_code == 400 # --------------------------------------------------------------------------- # Malformed JSON — should reject # --------------------------------------------------------------------------- class TestMalformedJSON: def test_invalid_json_string(self, client): resp = client.post( "/simple", data="{not valid json", content_type="application/json", ) assert resp.status_code == 400 assert "error" in resp.json def test_plain_text_body(self, client): resp = client.post( "/simple", data="hello world", content_type="text/plain" ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # Non-dict JSON values — should reject # --------------------------------------------------------------------------- class TestNonDictJSON: def test_json_null(self, client): resp = client.post( "/simple", data=json.dumps(None), content_type="application/json", ) assert resp.status_code == 400 def test_json_list(self, client): resp = client.post( "/simple", data=json.dumps([1, 2, 3]), content_type="application/json", ) assert resp.status_code == 400 def test_json_string(self, client): resp = client.post( "/simple", data=json.dumps("just a string"), content_type="application/json", ) assert resp.status_code == 400 def test_json_number(self, client): resp = client.post( "/simple", data=json.dumps(42), content_type="application/json", ) assert resp.status_code == 400 def test_json_boolean(self, client): resp = client.post( "/simple", data=json.dumps(True), content_type="application/json", ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # Custom error messages # --------------------------------------------------------------------------- class TestCustomMessages: def test_simple_custom_message(self, client): resp = client.post("/simple-custom") assert resp.status_code == 400 assert resp.json["error"] == "Query parameter is required" def test_status_custom_message(self, client): resp = client.post("/status-custom") assert resp.status_code == 400 assert resp.json["message"] == "No settings data provided" def test_success_custom_message(self, client): resp = client.post("/success-custom") assert resp.status_code == 400 assert resp.json["error"] == "Missing required fields" # --------------------------------------------------------------------------- # Error format response structure # --------------------------------------------------------------------------- class TestErrorFormatStructure: def test_simple_format_keys(self, client): resp = client.post("/simple") assert resp.status_code == 400 assert set(resp.json.keys()) == {"error"} def test_status_format_keys(self, client): resp = client.post("/status") assert resp.status_code == 400 assert set(resp.json.keys()) == {"status", "message"} def test_success_format_keys(self, client): resp = client.post("/success") assert resp.status_code == 400 assert set(resp.json.keys()) == {"success", "error"} # --------------------------------------------------------------------------- # Decorator preserves function metadata # --------------------------------------------------------------------------- class TestDecoratorMetadata: def test_wraps_preserves_name(self, app): """@wraps should preserve the original function name.""" with app.test_request_context(): for rule in app.url_map.iter_rules(): if rule.endpoint == "simple_route": view_func = app.view_functions[rule.endpoint] assert view_func.__name__ == "simple_route" return pytest.fail("simple_route endpoint not found")