Remove botocore dependency by transform any IO-stream (#5)

localstack_snapshot/snapshots/prototype.py

@@ -1,3 +1,4 @@
+import io
 import json
 import logging
 import os
@@ -7,7 +8,6 @@
 from re import Pattern
 from typing import Dict, List, Optional
 
-from botocore.response import StreamingBody
 from deepdiff import DeepDiff
 from jsonpath_ng import DatumInContext
 from jsonpath_ng.ext import parse
@@ -260,11 +260,10 @@ def _assert_all(
     def _transform_dict_to_parseable_values(self, original):
         """recursively goes through dict and tries to resolve values to strings (& parse them as json if possible)"""
         for k, v in original.items():
-            if isinstance(v, StreamingBody):
+            if isinstance(v, io.IOBase):
                 # update v for json parsing below
-                original[k] = v = v.read().decode(
-                    "utf-8"
-                )  # TODO: patch boto client so this doesn't break any further read() calls
+                # TODO: patch boto client so this doesn't break any further read() calls
+                original[k] = v = v.read().decode("utf-8")
             if isinstance(v, list) and v:
                 for item in v:
                     if isinstance(item, dict):

pyproject.toml

@@ -12,7 +12,6 @@ description = "Extracted snapshot testing lib for LocalStack"
 dependencies = [
     "jsonpath-ng>1.6",
     "deepdiff",
-    "botocore",
 ]
 requires-python = ">=3.10"
 license = {file = "LICENSE"}

tests/test_snapshots.py

@@ -1,3 +1,5 @@
+import io
+
 import pytest
 
 from localstack_snapshot.snapshots import SnapshotSession
@@ -20,6 +22,12 @@ def test_simple_diff_change(self):
             sm._assert_all()
         ctx.match("Parity snapshot failed")
 
+    def test_diff_with_io_stream(self):
+        sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False)
+        sm.recorded_state = {"key_a": {"a": "data"}}
+        sm.match("key_a", {"a": io.BytesIO(b"data")})
+        sm._assert_all()
+
     def test_multiple_assertmatch_with_same_key_fail(self):
         sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False)
         sm.recorded_state = {"key_a": {"a": 3}}