amq-adapter-python/tests/adapter/test_data_parser.py

import json
import unittest
from unittest.mock import MagicMock, patch

from amqp.adapter.data_parser import AMQDataParser, MultipartFormDataParser


class DummyAMQMessage:
    def __init__(self, headers=None, body=b"", extra_data=None):
        self._headers = headers or {"Content-Type": ["application/json"]}
        self._body = body
        self.extra_data = extra_data if extra_data is not None else {}

    def headers(self):
        return self._headers

    def body(self):
        return self._body


class TestAMQDataParser(unittest.TestCase):
    def test_process_form_data(self):
        cases = [
            (
                '{"form": {"foo": ["bar"]}, "files": {"file1": ["data"]}}',
                {"foo": "bar", "file1": "data"},
            ),
            ({"form": {"a": ["1"]}, "files": {"b": ["2"]}}, {"a": "1", "b": "2"}),
            ({"form": {}, "files": {}}, {}),
        ]
        for json_input, expected in cases:
            result = AMQDataParser.process_form_data(json_input)
            self.assertEqual(result, expected)

    def test_parse_get_url(self):
        url = "http://example.com/path?foo=1&bar=2&foo=3"
        path, params = AMQDataParser.parse_get_url(url)
        self.assertEqual(path, "/path")
        self.assertTrue(params["foo"] == ["1", "3"] or params["foo"] == "1" or params["foo"] == "3")
        self.assertEqual(params["bar"], "2")

    def test_parse_request_body_json(self):
        data = {"foo": "bar"}
        msg = DummyAMQMessage(
            headers={"Content-Type": ["application/json"]}, body=json.dumps(data).encode()
        )
        result = AMQDataParser.parse_request_body(msg)
        self.assertEqual(result, data)

    def test_parse_request_body_invalid_json(self):
        msg = DummyAMQMessage(headers={"Content-Type": ["application/json"]}, body=b"notjson")
        with patch("amqp.adapter.data_parser.logging_error") as log_err:
            with self.assertRaises(ValueError):
                AMQDataParser.parse_request_body(msg)
            self.assertTrue(log_err.called)

    def test_parse_request_body_urlencoded(self):
        msg = DummyAMQMessage(
            headers={"Content-Type": ["application/x-www-form-urlencoded"]}, body=b"foo=bar&baz=qux"
        )
        result = AMQDataParser.parse_request_body(msg)
        self.assertEqual(result, {b"foo": b"bar", b"baz": b"qux"})

    def test_parse_request_body_urlencoded_json(self):
        data = {"form": {"foo": ["bar"]}, "files": {}}
        msg = DummyAMQMessage(
            headers={"Content-Type": ["application/x-www-form-urlencoded"]},
            body=json.dumps(data).encode(),
        )
        result = AMQDataParser.parse_request_body(msg)
        self.assertEqual(result, {"foo": "bar"})

    def test_parse_request_body_invalid_urlencoded(self):
        msg = DummyAMQMessage(
            headers={"Content-Type": ["application/x-www-form-urlencoded"]}, body=b"not a json"
        )
        with patch("amqp.adapter.data_parser.logging_error") as log_err:
            with self.assertRaises(ValueError):
                AMQDataParser.parse_request_body(msg)
            self.assertTrue(log_err.called)

    def test_parse_request_body_text(self):
        msg = DummyAMQMessage(headers={"Content-Type": ["text/plain"]}, body=b"hello world")
        result = AMQDataParser.parse_request_body(msg)
        self.assertEqual(result, {"text": b"hello world"})

    def test_parse_request_body_xml(self):
        xml = b"<root><foo>bar</foo><baz>qux</baz></root>"
        msg = DummyAMQMessage(headers={"Content-Type": ["application/xml"]}, body=xml)
        result = AMQDataParser.parse_request_body(msg)
        self.assertEqual(result, {"foo": "bar", "baz": "qux"})

    def test_parse_request_body_invalid_xml(self):
        msg = DummyAMQMessage(
            headers={"Content-Type": ["application/xml"]}, body=b"<root><foo></bar></root>"
        )
        with self.assertRaises(ValueError):
            AMQDataParser.parse_request_body(msg)

    def test_parse_request_body_unsupported(self):
        msg = DummyAMQMessage(headers={"Content-Type": ["application/unknown"]}, body=b"")
        with self.assertRaises(ValueError):
            AMQDataParser.parse_request_body(msg)


class TestMultipartFormDataParser(unittest.TestCase):
    def test_multipart_form_data_parser_json(self):
        data = {"form": {"foo": ["bar"]}, "files": {}}
        msg = DummyAMQMessage(
            headers={"Content-Type": ["multipart/form-data"]},
            body=json.dumps(data).encode(),
            extra_data={"some": "data"},
        )
        parser = MultipartFormDataParser(msg)
        self.assertTrue(parser.is_json)
        self.assertEqual(parser.form_data, {"foo": "bar"})
        self.assertTrue(parser.is_valid)

    def test_multipart_form_data_parser_invalid_json(self):
        msg = DummyAMQMessage(
            headers={"Content-Type": ["multipart/form-data"]},
            body=b"notjson",
            extra_data={"some": "data"},
        )
        with patch("amqp.adapter.data_parser.logging_error") as log_err:
            parser = MultipartFormDataParser(msg)
            self.assertFalse(parser.is_valid)
            self.assertTrue(log_err.called)

    def test_multipart_form_data_parser_fallback_to_json(self):
        msg = DummyAMQMessage(
            headers={"Content-Type": ["multipart/form-data"]},
            body=json.dumps({"form": {"foo": ["bar"]}, "files": {}}).encode(),
        )
        with patch(
            "amqp.adapter.data_parser.python_multipart.parse_form", side_effect=Exception("fail")
        ):
            parser = MultipartFormDataParser(msg)
            self.assertTrue(parser.is_json)
            self.assertEqual(parser.form_data, {"foo": "bar"})
            self.assertTrue(parser.is_valid)

    def test_multipart_form_data_parser_on_field_and_file(self):
        msg = DummyAMQMessage(
            headers={"Content-Type": ["multipart/form-data"]}, body=b"", extra_data={}
        )
        parser = MultipartFormDataParser(msg)

        class DummyField:
            field_name = b"foo"
            value = b"bar"

        class DummyFile:
            field_name = b"file"
            file_object = MagicMock()
            file_object.seek = MagicMock()
            size = 123
            file_name = b"filename.txt"

        parser.on_field(DummyField())
        self.assertEqual(parser.form_data["foo"], b"bar")
        parser.on_file(DummyFile())
        self.assertIn("file", parser.form_data)
        self.assertTrue(hasattr(parser.form_data["file"], "filename"))
        self.assertEqual(parser.form_data["file"].filename, "filename.txt")