mizan/examples/django-react-desktop-app/tests/test_desktop_rpc.py

"""
REAL integration tests for the mizan RPC framework layer.

Tests the actual HTTP stack: CSRF, middleware, error codes, validation.
Every test makes a real HTTP request — no mocks, no RequestFactory.
"""

import json
from urllib.request import urlopen, Request
from urllib.error import HTTPError

from django.test import LiveServerTestCase


class RealHTTPMixin:
    def _session_init(self):
        url = f"{self.live_server_url}/api/mizan/session/"
        resp = urlopen(Request(url))
        cookies = resp.headers.get_all("Set-Cookie") or []
        for cookie in cookies:
            if "csrftoken=" in cookie:
                self._csrf_token = cookie.split("csrftoken=")[1].split(";")[0]
                self._cookies = f"csrftoken={self._csrf_token}"
                return
        self._csrf_token = None
        self._cookies = ""

    def _call(self, fn: str, args: dict | None = None):
        url = f"{self.live_server_url}/api/mizan/call/"
        body = json.dumps({"fn": fn, "args": args or {}}).encode()
        req = Request(url, data=body, method="POST")
        req.add_header("Content-Type", "application/json")
        if self._csrf_token:
            req.add_header("X-CSRFToken", self._csrf_token)
        if self._cookies:
            req.add_header("Cookie", self._cookies)
        resp = urlopen(req)
        return json.loads(resp.read())

    def _raw_post(
        self,
        path: str,
        body: bytes | str,
        content_type: str = "application/json",
        include_csrf: bool = False,
    ):
        """Raw POST without the call() envelope — for testing malformed requests."""
        url = f"{self.live_server_url}{path}"
        if isinstance(body, str):
            body = body.encode()
        req = Request(url, data=body, method="POST")
        req.add_header("Content-Type", content_type)
        if include_csrf and self._csrf_token:
            req.add_header("X-CSRFToken", self._csrf_token)
            req.add_header("Cookie", self._cookies)
        return urlopen(req)


class CSRFTests(RealHTTPMixin, LiveServerTestCase):
    """CSRF handling over real HTTP — the thing that was broken."""

    def test_session_endpoint_sets_csrf_cookie(self):
        """GET /session/ must return a Set-Cookie with csrftoken."""
        url = f"{self.live_server_url}/api/mizan/session/"
        resp = urlopen(Request(url))
        cookies = resp.headers.get_all("Set-Cookie") or []

        csrf_cookies = [c for c in cookies if "csrftoken=" in c]
        self.assertGreater(len(csrf_cookies), 0, "No csrftoken cookie set by /session/")

    def test_call_without_csrf_is_rejected(self):
        """POST /call/ without CSRF token must fail."""
        url = f"{self.live_server_url}/api/mizan/call/"
        body = json.dumps({"fn": "system_info", "args": {}}).encode()
        req = Request(url, data=body, method="POST")
        req.add_header("Content-Type", "application/json")

        try:
            resp = urlopen(req)
            data = json.loads(resp.read())
            # If it doesn't raise, the response should indicate an error
            self.assertTrue(data.get("error"), "POST without CSRF should be rejected")
        except HTTPError as e:
            self.assertEqual(e.code, 403, f"Expected 403, got {e.code}")

    def test_call_with_csrf_succeeds(self):
        """POST /call/ with valid CSRF token must work."""
        self._session_init()
        data = self._call("system_info")

        self.assertFalse(data["error"])
        self.assertIn("os_name", data["data"])


class ValidationTests(RealHTTPMixin, LiveServerTestCase):
    """Pydantic validation errors over real HTTP."""

    def setUp(self):
        self._session_init()

    def test_missing_required_field(self):
        """Calling create_note without title should return VALIDATION_ERROR."""
        data = self._call("create_note", {})

        self.assertTrue(data["error"])
        self.assertEqual(data["code"], "VALIDATION_ERROR")

    def test_wrong_type(self):
        """Calling delete_note with string id should return VALIDATION_ERROR."""
        data = self._call("delete_note", {"id": "not-an-int"})

        self.assertTrue(data["error"])
        self.assertEqual(data["code"], "VALIDATION_ERROR")

    def test_missing_multiple_fields(self):
        """write_file with no args should list all missing fields."""
        data = self._call("write_file", {})

        self.assertTrue(data["error"])
        self.assertEqual(data["code"], "VALIDATION_ERROR")


class ErrorCodeTests(RealHTTPMixin, LiveServerTestCase):
    """Error codes over real HTTP."""

    def setUp(self):
        self._session_init()

    def test_not_found_function(self):
        data = self._call("this_does_not_exist")

        self.assertTrue(data["error"])
        self.assertEqual(data["code"], "NOT_FOUND")

    def test_forbidden_write_outside_home(self):
        data = self._call("write_file", {"path": "/etc/nope.txt", "content": "x"})

        self.assertTrue(data["error"])
        self.assertEqual(data["code"], "FORBIDDEN")

    def test_get_method_rejected(self):
        """GET to /call/ should be rejected."""
        url = f"{self.live_server_url}/api/mizan/call/"
        try:
            resp = urlopen(Request(url))
            data = json.loads(resp.read())
            self.assertTrue(data.get("error"))
        except HTTPError as e:
            self.assertIn(e.code, [403, 405])

    def test_invalid_json_body(self):
        """Malformed JSON should return BAD_REQUEST."""
        self._session_init()
        try:
            resp = self._raw_post(
                "/api/mizan/call/",
                body="not valid json{{{",
                include_csrf=True,
            )
            data = json.loads(resp.read())
            self.assertTrue(data["error"])
            self.assertEqual(data["code"], "BAD_REQUEST")
        except HTTPError as e:
            self.assertIn(e.code, [400, 403])

    def test_missing_fn_field(self):
        """POST with valid JSON but no 'fn' field should return BAD_REQUEST."""
        self._session_init()
        try:
            resp = self._raw_post(
                "/api/mizan/call/",
                body=json.dumps({"not_fn": "hello"}),
                include_csrf=True,
            )
            data = json.loads(resp.read())
            self.assertTrue(data["error"])
            self.assertEqual(data["code"], "BAD_REQUEST")
        except HTTPError as e:
            self.assertIn(e.code, [400, 403])