test: add multipart tests

test/test_api_rql.py

@@ -198,3 +198,103 @@
             "specification.",
             "title": "OpenApiValidationError",
         }
+
+
+class ApiRqlMultipartTC(ApiRqlTC):
+    @property
+    def content_type(self):
+        return "multipart/form-data"
+
+    def get_body(self, queries: list):
+        return {"queries": json.dumps(queries)}
+
+    def test_upload_file(self):
+        self.login_request()
+        queries = [
+            {
+                "query": "Insert EntityWithBinary X: X binary %(binary_ref)s",
+                "params": {
+                    "binary_ref": {"type": "binary_reference", "ref": "test_file"}
+                },
+            }
+        ]
+        response = self.webapp.post(
+            self.get_api_path("rql"),
+            params={
+                "queries": json.dumps(queries),
+                "test_file": webtest.Upload("filename.txt", b"content"),
+            },
+            content_type=self.content_type,
+        )
+        eid = response.json[0][0][0]
+        with self.admin_access.repo_cnx() as cnx:
+            rset = cnx.execute("Any X, B WHERE X eid %(eid)s, X binary B", {"eid": eid})
+            binary: Binary = rset[0][1]
+            self.assertEqual(binary.read(), b"content")
+
+    def test_upload_file_missing_reference(self):
+        self.login_request()
+        queries = [
+            {
+                "query": "Insert EntityWithBinary X: X binary %(binary_ref)s",
+                "params": {
+                    "binary_ref": {"type": "binary_reference", "ref": "wrong_ref"}
+                },
+            }
+        ]
+        response = self.webapp.post(
+            self.get_api_path("rql"),
+            params={
+                "queries": json.dumps(queries),
+                "test_file": webtest.Upload("filename.txt", b"content"),
+            },
+            content_type=self.content_type,
+            status=400,
+        ).json
+
+        assert response == {
+            "message": "Could not find binary of id wrong_ref",
+            "data": None,
+            "title": "InvalidTransaction",
+        }
+
+    def test_upload_file_malformed_reference(self):
+        self.login_request()
+        queries = [
+            {
+                "query": "Insert EntityWithBinary X: X binary %(binary_ref)s",
+                "params": {
+                    "binary_ref": {
+                        "type": "binary_reference",
+                    }
+                },
+            }
+        ]
+        response = self.webapp.post(
+            self.get_api_path("rql"),
+            params={
+                "queries": json.dumps(queries),
+                "test_file": webtest.Upload("filename.txt", b"content"),
+            },
+            content_type=self.content_type,
+            status=400,
+        ).json
+
+        assert response == {
+            "data": [
+                {
+                    "exception": "ValidationError",
+                    "message": "{'type': 'binary_reference'} is valid under each of "
+                    "{'additionalProperties': False, 'properties': {'ref': "
+                    "{'type': 'string'}, 'type': {'type': 'string'}}, "
+                    "'type': 'object'}, {'additionalProperties': False, "
+                    "'properties': {'column': {'type': 'number'}, "
+                    "'queryIndex': {'type': 'number'}, 'row': {'type': "
+                    "'number'}, 'type': {'type': 'string'}}, 'type': "
+                    "'object'}",
+                }
+            ],
+            "message": "Your request could not be validated against the openapi "
+            "specification.",
+            "title": "OpenApiValidationError",
+        }