learningcircuit--local-deep-research/tests/security/test_decorators.py

"""
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")