diff --git a/localstack_snapshot/snapshots/prototype.py b/localstack_snapshot/snapshots/prototype.py index 30f2e1f..3ddc7b5 100644 --- a/localstack_snapshot/snapshots/prototype.py +++ b/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): diff --git a/pyproject.toml b/pyproject.toml index 16255bf..bd70059 100644 --- a/pyproject.toml +++ b/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"} diff --git a/tests/test_snapshots.py b/tests/test_snapshots.py index 49d1034..4b986b4 100644 --- a/tests/test_snapshots.py +++ b/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}}