Fix decompress_7z_base64_data (#765)

Author: edenhaus

deebot_client/commands/json/map.py

@@ -275,7 +275,7 @@ def _handle_body_data_dict(
             # This command is used by new and old bots
             if data.get("compress", 0) == 1:
                 # Newer bot's return coordinates as base64 decoded string
-                coordinates = decompress_7z_base64_data(data["value"])
+                coordinates = decompress_7z_base64_data(data["value"]).decode()
             else:
                 # Older bot's return coordinates direct as comma/semicolon separated list
                 coordinates = data["value"]
@@ -305,7 +305,7 @@ def _get_subset_ids(
     ) -> list[int] | None:
         """Return subset ids."""
         # subset is based64 7z compressed
-        subsets = json.loads(decompress_7z_base64_data(data["subsets"]))
+        subsets = json.loads(decompress_7z_base64_data(data["subsets"]).decode())
 
         match data["type"]:
             case MapSetType.ROOMS:

deebot_client/map.py

@@ -395,7 +395,7 @@ async def on_map_subset(event: MapSubsetEvent) -> None:
 
     def _update_trace_points(self, data: str) -> None:
         _LOGGER.debug("[_update_trace_points] Begin")
-        trace_points = decompress_7z_base64_data(data).encode()
+        trace_points = decompress_7z_base64_data(data)
 
         for i in range(0, len(trace_points), 5):
             position_x, position_y = struct.unpack("<hh", trace_points[i : i + 4])
@@ -622,7 +622,7 @@ def image(self) -> Image.Image:
 
     def update_points(self, base64_data: str) -> None:
         """Add map piece points."""
-        decoded = decompress_7z_base64_data(base64_data).encode()
+        decoded = decompress_7z_base64_data(base64_data)
         old_crc32 = self._crc32
         self._crc32 = zlib.crc32(decoded)
 

deebot_client/rs.pyi

@@ -1,2 +1,2 @@
-def decompress_7z_base64_data(value: str) -> str:
+def decompress_7z_base64_data(value: str) -> bytes:
     """Decompress base64 decoded 7z compressed string."""

src/lib.rs

@@ -4,21 +4,20 @@ use base64::{engine::general_purpose, Engine as _};
 use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 
-fn _decompress_7z_base64_data(value: String) -> Result<String, Box<dyn Error>> {
+fn _decompress_7z_base64_data(value: String) -> Result<Vec<u8>, Box<dyn Error>> {
     let mut bytes = general_purpose::STANDARD.decode(value)?;
 
     // Insert required 0 bytes
     for _ in 0..=3 {
         bytes.insert(8, 0);
     }
 
-    let decompressed = lzma::decompress(&bytes)?;
-    Ok(String::from_utf8(decompressed)?)
+    Ok(lzma::decompress(&bytes)?)
 }
 
 /// Decompress base64 decoded 7z compressed string.
 #[pyfunction]
-fn decompress_7z_base64_data(value: String) -> Result<String, PyErr> {
+fn decompress_7z_base64_data(value: String) -> Result<Vec<u8>, PyErr> {
     Ok(_decompress_7z_base64_data(value).map_err(|err| PyValueError::new_err(err.to_string()))?)
 }
 

tests/test_map.py

@@ -159,7 +159,7 @@ async def on_change() -> None:
 
 @patch(
     "deebot_client.map.decompress_7z_base64_data",
-    Mock(return_value="\x10\x00\x00\x01\x00"),
+    Mock(return_value=b"\x10\x00\x00\x01\x00"),
 )
 async def test_Map_svg_traces_path(
     execute_mock: AsyncMock, event_bus_mock: Mock

tests/test_rs.py

@@ -19,19 +19,24 @@
     [
         (
             "XQAABACZAAAAABaOQmW9Bsibxz42rKUpGlV7Rr4D1S/9x9mDa60v4J1BKrEsnk34EAt6X5gKkxwYzfOu3T8GAPpmIy5o4A==",
-            "-9125,3225;-9025,3225;-8975,3175;-8975,2475;-8925,2425;-8925,2375;-8325,2375;-8275,2425;-8225,2375;-8225,2425;-8174,2475;-8024,2475;-8024,4375;-9125,4375",
+            b"-9125,3225;-9025,3225;-8975,3175;-8975,2475;-8925,2425;-8925,2375;-8325,2375;-8275,2425;-8225,2375;-8225,2425;-8174,2475;-8024,2475;-8024,4375;-9125,4375",
         ),
         (
             "XQAABABBAAAAAC2WwEIwUhHX3vfFDfs1H1PUqtdWgakwVnMBz3Bb3yaoE5OYkdYA",
-            '[["4","-6217","3919","-6217","231","-2642","231","-2642","3919"]]',
+            b'[["4","-6217","3919","-6217","231","-2642","231","-2642","3919"]]',
         ),
         (
             "XQAABADHAAAAAC2WwEHwYhHX3vWwDK80QCnaQU0mwUd9Vk34ub6OxzOk6kdFfbFvpVp4iIlKisAvp0MznQNYEZ8koxFHnO+iM44GUKgujGQKgzl0bScbQgaon1jI3eyCRikWlkmrbwA=",
-            '[["0","-5195","-1059","-5195","-37","-5806","-37","-5806","-1059"],["1","-7959","220","-7959","1083","-9254","1083","-9254","220"],["2","-9437","347","-5387","410"],["3","-5667","317","-4888","-56"]]',
+            b'[["0","-5195","-1059","-5195","-37","-5806","-37","-5806","-1059"],["1","-7959","220","-7959","1083","-9254","1083","-9254","220"],["2","-9437","347","-5387","410"],["3","-5667","317","-4888","-56"]]',
+        ),
+        (
+            "XQAABACvAAAAAAAAAEINQkt4BfqEvt9Pow7YU9KWRVBcSBosIDAOtACCicHy+vmfexxcutQUhqkAPQlBawOeXo/VSrOqF7yhdJ1JPICUs3IhIebU62Qego0vdk8oObiLh3VY/PVkqQyvR4dHxUDzMhX7HAguZVn3yC17+cQ18N4kaydN3LfSUtV/zejrBM4=",
+            b'\x00\x00\x01\x00\x98\xf6\xff\x01\x00\x18\xf9\xff\xf8\xff@\x00\x00\xf1\xff@\x06\x00\xe9\xff@\x0b\x00\xe0\xff@\x15\x00\xe2\xff@\x1f\x00\xe2\xff@(\x00\xde\xff@.\x00\xd6\xff@5\x00\xcd\xff@4\x00\xc3\xff@0\x00\xba\xff@,\x00\xb1\xff@"\x00\xad\xff@\x18\x00\xad\xff@\x0e\x00\xae\xff@\x06\x00\xb4\xff@\x00\x00\xbc\xff@\xfe\xff\xc5\xff@\x00\x00\xd0\xff@\x03\x00\xda\xff@\x0b\x00\xe0\xff@\x15\x00\xe3\xff@\x15\x00\xed\xffH\x0e\x00\xf4\xffH\x05\x00\xf9\xffH\x0c\x00\xf2\xffH\x15\x00\xee\xffH\x1f\x00\xec\xffH)\x00\xec\xffH3\x00\xe8\xffH:\x00\xe1\xffH@\x00\xd9\xff@F\x00\xd1\xff@',
         ),
     ],
+    ids=["1", "2", "3", "4"],
 )
-def test_decompress_7z_base64_data(input: str, expected: str) -> None:
+def test_decompress_7z_base64_data(input: str, expected: bytes) -> None:
     """Test decompress_7z_base64_data function."""
     assert _decompress_7z_base64_data_python(input) == expected
     assert decompress_7z_base64_data(input) == expected
@@ -58,7 +63,7 @@ def test_decompress_7z_base64_data_errors(
         assert decompress_7z_base64_data(input)
 
 
-def _decompress_7z_base64_data_python(data: str) -> str:
+def _decompress_7z_base64_data_python(data: str) -> bytes:
     """Decompress base64 decoded 7z compressed string."""
     final_array = bytearray()
 
@@ -71,4 +76,4 @@ def _decompress_7z_base64_data_python(data: str) -> str:
         final_array.append(idx)
 
     dec = lzma.LZMADecompressor(lzma.FORMAT_AUTO, None, None)
-    return dec.decompress(final_array).decode()
+    return dec.decompress(final_array)