test(General): implement unit test for data parser, rabbitmq client and other python files

This commit bump the coverage rate to 77%

tests/adapter/test_data_parser.py

@@ -1,3 +1,165 @@
-class DataParserTest:
-    def test_multipart_form_data_parser(self):
-        assert False
+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")