Merge pull request proxmoxer#163 from jhollowe/bugfix/qemu-split

Fix improper spliting of non-exec QEMU commands

proxmoxer/backends/https.py

@@ -194,7 +194,7 @@ def request(
         total_file_size = 0
         for k, v in data.copy().items():
             # split qemu exec commands for proper parsing by PVE (issue#89)
-            if k == "command":
+            if k == "command" and url.endswith("agent/exec"):
                 if isinstance(v, list):
                     data[k] = v
                 elif "Windows" not in platform.platform():

tests/api_mock.py

@@ -100,6 +100,22 @@ def _generate_dynamic_responses(self):
             )
         )
 
+        resps.append(
+            responses.CallbackResponse(
+                method="GET",
+                url=re.compile(self.base_url + r"/nodes/[^/]+/qemu/[^/]+/agent/exec"),
+                callback=self._cb_echo,
+            )
+        )
+
+        resps.append(
+            responses.CallbackResponse(
+                method="GET",
+                url=re.compile(self.base_url + r"/nodes/[^/]+/qemu/[^/]+/monitor"),
+                callback=self._cb_qemu_monitor,
+            )
+        )
+
         resps.append(
             responses.CallbackResponse(
                 method="GET",
@@ -313,3 +329,32 @@ def _cb_url_metadata(self, request):
                     }
                 ),
             )
+
+    def _cb_qemu_monitor(self, request):
+        body = request.body
+        if body is not None:
+            body = body if isinstance(body, str) else str(body, "utf-8")
+
+        # if the command is an array, throw the type error PVE would throw
+        if "&" in body:
+            return (
+                400,
+                self.common_headers,
+                json.dumps(
+                    {
+                        "data": None,
+                        "errors": {"command": "type check ('string') failed - got ARRAY"},
+                    }
+                ),
+            )
+        else:
+            resp = {
+                "method": request.method,
+                "url": request.url,
+                "headers": dict(request.headers),
+                "cookies": request._cookies.get_dict(),
+                "body": body,
+                # "body_json": dict(parse_qsl(request.body)),
+            }
+            print(resp)
+            return (200, self.common_headers, json.dumps(resp))

tests/test_https.py

@@ -291,25 +291,51 @@ def test_request_data(self, mock_pve):
         assert content["body"] == "key=value"
         assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded"
 
-    def test_request_command_list(self, mock_pve):
+    def test_request_monitor_command_list(self, mock_pve):
         resp = self._session.request(
-            "GET", self.base_url + "/fake/echo", data={"command": ["echo", "hello", "world"]}
+            "GET",
+            self.base_url + "/nodes/node_name/qemu/100/monitor",
+            data={"command": ["info", "block"]},
+        )
+
+        assert resp.status_code == 400
+
+    def test_request_exec_command_list(self, mock_pve):
+        resp = self._session.request(
+            "GET",
+            self.base_url + "/nodes/node_name/qemu/100/agent/exec",
+            data={"command": ["echo", "hello", "world"]},
         )
         content = resp.json()
 
         assert content["method"] == "GET"
-        assert content["url"] == self.base_url + "/fake/echo"
+        assert content["url"] == self.base_url + "/nodes/node_name/qemu/100/agent/exec"
         assert content["body"] == "command=echo&command=hello&command=world"
         assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded"
 
-    def test_request_command_string(self, mock_pve):
+    def test_request_monitor_command_string(self, mock_pve):
         resp = self._session.request(
-            "GET", self.base_url + "/fake/echo", data={"command": "echo hello world"}
+            "GET",
+            self.base_url + "/nodes/node_name/qemu/100/monitor",
+            data={"command": "echo hello world"},
         )
         content = resp.json()
 
         assert content["method"] == "GET"
-        assert content["url"] == self.base_url + "/fake/echo"
+        assert content["url"] == self.base_url + "/nodes/node_name/qemu/100/monitor"
+        assert content["body"] == "command=echo+hello+world"
+        assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded"
+
+    def test_request_exec_command_string(self, mock_pve):
+        resp = self._session.request(
+            "GET",
+            self.base_url + "/nodes/node_name/qemu/100/agent/exec",
+            data={"command": "echo hello world"},
+        )
+        content = resp.json()
+
+        assert content["method"] == "GET"
+        assert content["url"] == self.base_url + "/nodes/node_name/qemu/100/agent/exec"
         assert content["body"] == "command=echo&command=hello&command=world"
         assert content["headers"]["Content-Type"] == "application/x-www-form-urlencoded"