handle binary data accordingly

Author: sohankshirsagar

drift/instrumentation/redis/e2e-tests/src/app.py

@@ -134,6 +134,79 @@ def test_pipeline_no_transaction():
     except Exception as e:
         return jsonify({"error": str(e)}), 500
 
+@app.route("/test/async-pipeline", methods=["GET"])
+def test_async_pipeline():
+    """Test async pipeline operations using asyncio."""
+    import asyncio
+    import redis.asyncio as aioredis
+
+    async def run_async_pipeline():
+        # Create async Redis client
+        async_client = aioredis.Redis(
+            host=os.getenv("REDIS_HOST", "redis"),
+            port=int(os.getenv("REDIS_PORT", "6379")),
+            db=0,
+            decode_responses=True
+        )
+
+        try:
+            # Create async pipeline
+            pipe = async_client.pipeline()
+            pipe.set("test:async:pipe:key1", "async_value1")
+            pipe.set("test:async:pipe:key2", "async_value2")
+            pipe.get("test:async:pipe:key1")
+            pipe.get("test:async:pipe:key2")
+            results = await pipe.execute()
+
+            # Clean up
+            await async_client.delete("test:async:pipe:key1", "test:async:pipe:key2")
+
+            return results
+        finally:
+            await async_client.aclose()
+
+    try:
+        results = asyncio.run(run_async_pipeline())
+        return jsonify({
+            "success": True,
+            "results": results
+        })
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+@app.route("/test/binary-data", methods=["GET"])
+def test_binary_data():
+    """Test binary data that cannot be decoded as UTF-8."""
+    try:
+        # Create a Redis client without decode_responses for binary data
+        binary_client = redis.Redis(
+            host=os.getenv("REDIS_HOST", "redis"),
+            port=int(os.getenv("REDIS_PORT", "6379")),
+            db=0,
+            decode_responses=False
+        )
+
+        # Binary data that cannot be decoded as UTF-8
+        binary_value = bytes([0x80, 0x81, 0x82, 0xFF, 0xFE, 0xFD])
+
+        # Set binary data
+        binary_client.set("test:binary:key", binary_value)
+
+        # Get binary data back
+        retrieved = binary_client.get("test:binary:key")
+
+        # Clean up
+        binary_client.delete("test:binary:key")
+
+        return jsonify({
+            "success": True,
+            "original_hex": binary_value.hex(),
+            "retrieved_hex": retrieved.hex() if retrieved else None,
+            "match": binary_value == retrieved
+        })
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
 
 if __name__ == "__main__":
     sdk.mark_app_as_ready()

drift/instrumentation/redis/e2e-tests/src/test_requests.py

@@ -55,4 +55,10 @@ def make_request(method, endpoint, **kwargs):
     make_request("GET", "/test/pipeline-basic")
     make_request("GET", "/test/pipeline-no-transaction")
 
+    # Async Pipeline operations
+    make_request("GET", "/test/async-pipeline")
+
+    # Binary data handling
+    make_request("GET", "/test/binary-data")
+
     print("\nAll requests completed successfully")

drift/instrumentation/redis/instrumentation.py

@@ -817,9 +817,10 @@ def _serialize_value(self, value: Any) -> Any:
         """Serialize a single value for JSON."""
         if isinstance(value, bytes):
             try:
-                return value.decode("utf-8")
+                decoded = value.decode("utf-8")
+                return {"__bytes__": True, "encoding": "utf8", "value": decoded}
             except UnicodeDecodeError:
-                return value.hex()
+                return {"__bytes__": True, "encoding": "hex", "value": value.hex()}
         elif isinstance(value, (str, int, float, bool, type(None))):
             return value
         elif isinstance(value, (list, tuple)):
@@ -835,6 +836,24 @@ def _serialize_response(self, response: Any) -> Any:
         """Serialize Redis response for recording."""
         return self._serialize_value(response)
 
+    def _deserialize_value(self, value: Any) -> Any:
+        """Deserialize a value, converting typed wrappers back to original types."""
+        if isinstance(value, dict):
+            # Check for bytes wrapper
+            if value.get("__bytes__") is True:
+                encoding = value.get("encoding")
+                data = value.get("value", "")
+                if encoding == "utf8":
+                    return data.encode("utf-8")
+                elif encoding == "hex":
+                    return bytes.fromhex(data)
+                return data  # fallback
+            # Recursively deserialize dict values
+            return {k: self._deserialize_value(v) for k, v in value.items()}
+        elif isinstance(value, list):
+            return [self._deserialize_value(v) for v in value]
+        return value
+
     def _deserialize_response(self, mock_data: dict[str, Any]) -> Any:
         """Deserialize mocked response data from CLI.
 
@@ -845,9 +864,9 @@ def _deserialize_response(self, mock_data: dict[str, Any]) -> Any:
 
         if isinstance(mock_data, dict):
             if "result" in mock_data:
-                return mock_data["result"]
+                return self._deserialize_value(mock_data["result"])
             elif "results" in mock_data:
-                return mock_data["results"]
+                return [self._deserialize_value(r) for r in mock_data["results"]]
 
         logger.warning(f"Could not deserialize mock_data structure: {mock_data}")
         return None