Cassettes mutations

README.rst

@@ -117,6 +117,33 @@ Run ``pytest``:
 
 The network blocking feature supports ``socket``-based transports and ``pycurl``.
 
+Cassettes mutation
+~~~~~~~~~~~~~~~~~~
+
+With static cassettes, the responses are stored and expected, but how to test unexpected behavior of a remote service?
+What if it will respond with plain text instead of JSON? unexpected status code? missing field or wrong value in JSON?
+
+If you already have cassettes recorded, then mutation testing could help you to test these unusual scenarios.
+To enable it you need to add ``pytest.mark.vcr_mutations`` to your test and specify ``mutators`` keyword argument:
+
+.. code:: python
+
+    from pytest_recording import mutations
+
+    def make_call():
+        response = requests.get("http://httpbin.org/get")
+        ...
+
+    @pytest.mark.vcr_mutations(
+        mutations=mutations.Body()
+    )
+    def test_malformed_body(mutation):
+        with pytest.raises(MalformedBodyError):
+            make_call()
+
+
+It will generate N(???) mutations for the cassettes and will run that test.
+
 Contributing
 ------------
 

docs/changelog.rst

@@ -6,6 +6,11 @@ Changelog
 `Unreleased`_
 -------------
 
+Added
+~~~~~
+
+- Add cassettes mutation support. `#13`_
+
 `0.3.2`_ - 2019-08-01
 ---------------------
 
@@ -49,6 +54,7 @@ Added
 .. _0.3.0: https://github.com/kiwicom/pytest-recording/compare/v0.2.0...v0.3.0
 .. _0.2.0: https://github.com/kiwicom/pytest-recording/compare/v0.1.0...v0.2.0
 
+.. _#13: https://github.com/kiwicom/pytest-recording/issues/13
 .. _#10: https://github.com/kiwicom/pytest-recording/issues/10
 .. _#8: https://github.com/kiwicom/pytest-recording/issues/8
 .. _#2: https://github.com/kiwicom/pytest-recording/issues/2

setup.py

@@ -10,7 +10,7 @@
 install_requires = ["vcrpy>=2.0.1", "attrs"]
 
 if sys.version_info[0] == 2:
-    install_requires.append("pytest>=3.5.0,<5.0")
+    install_requires.extend(["pytest>=3.5.0,<5.0", "funcsigs>=1.0.2"])
 else:
     install_requires.append("pytest>=3.5.0")
 

src/pytest_recording/_compat.py

@@ -0,0 +1,13 @@
+try:
+    from inspect import Signature
+
+    get_signature = Signature.from_callable  # pylint: disable=unused-import
+except ImportError:
+    from funcsigs import signature as get_signature  # pylint: disable=unused-import
+
+
+try:
+    from collections.abc import Iterable  # pylint: disable=unused-import
+except ImportError:
+    # Python 2
+    from collections import Iterable

src/pytest_recording/_vcr.py

@@ -9,25 +9,30 @@
 from .utils import unique, unpack
 
 
-def load_cassette(cassette_path, serializer):
+def load_cassette(cassette_path, serializer, mutation):
     try:
         with open(cassette_path) as f:
             cassette_content = f.read()
     except IOError:
         return [], []
-    return deserialize(cassette_content, serializer)
+    requests, cassettes = deserialize(cassette_content, serializer)
+    if mutation is not None:
+        for cassette in cassettes:
+            mutation.apply(cassette)
+    return requests, cassettes
 
 
 @attr.s(slots=True)
 class CombinedPersister(FilesystemPersister):
     """Load extra cassettes, but saves only the first one."""
 
     extra_paths = attr.ib()
+    mutation = attr.ib()
 
     def load_cassette(self, cassette_path, serializer):
         all_paths = chain.from_iterable(((cassette_path,), self.extra_paths))
         # Pairs of 2 lists per cassettes:
-        all_content = (load_cassette(path, serializer) for path in unique(all_paths))
+        all_content = (load_cassette(path, serializer, self.mutation) for path in unique(all_paths))
         # Two iterators from all pairs from above: all requests, all responses
         # Notes.
         # 1. It is possible to do it with accumulators, for loops and `extend` calls,
@@ -37,14 +42,14 @@ def load_cassette(self, cassette_path, serializer):
         return starmap(unpack, zip(*all_content))
 
 
-def use_cassette(vcr_cassette_dir, record_mode, markers, config):
+def use_cassette(vcr_cassette_dir, record_mode, markers, config, mutation):
     """Create a VCR instance and return an appropriate context manager for the given cassette configuration."""
     merged_config = merge_kwargs(config, markers)
     path_transformer = get_path_transformer(merged_config)
     vcr = VCR(path_transformer=path_transformer, cassette_library_dir=vcr_cassette_dir, record_mode=record_mode)
     # flatten the paths
     extra_paths = chain(*(marker[0] for marker in markers))
-    persister = CombinedPersister(extra_paths)
+    persister = CombinedPersister(extra_paths, mutation)
     vcr.register_persister(persister)
     return vcr.use_cassette(markers[0][0][0], **merged_config)
 

src/pytest_recording/mutations/__init__.py

@@ -0,0 +1,9 @@
+from .mutagens import status_code, malform_body, empty_body, reduce_body
+from .core import mutation, MutationGroup
+
+
+DEFAULT_STATUS_CODES = (400, 401, 403, 404, 500, 501, 502, 503)
+
+DEFAULT = MutationGroup((status_code(code=DEFAULT_STATUS_CODES), empty_body, malform_body, reduce_body))
+
+del MutationGroup  # Not needed in the API

src/pytest_recording/mutations/core.py

@@ -0,0 +1,52 @@
+import attr
+
+from .._compat import get_signature
+
+
+@attr.s(slots=True)
+class Mutation(object):
+    func = attr.ib()
+    context = attr.ib(factory=dict)
+    _signature = attr.ib(init=False)
+
+    @classmethod
+    def from_function(cls, func):
+        return cls(func)
+
+    @property
+    def signature(self):
+        if not hasattr(self, "_signature"):
+            self._signature = get_signature(self.func)
+        return self._signature
+
+    def __call__(self, **context):
+        self.validate_call(**context)
+        return self.__class__(self.func, context)
+
+    def validate_call(self, **context):
+        self.signature.bind(cassette={}, **context)
+
+    def generate(self):
+        yield self
+
+    def apply(self, cassette):
+        return self.func(cassette, **self.context)
+
+
+mutation = Mutation.from_function
+
+
+@attr.s(slots=True)
+class MutationGroup(object):
+    mutations = attr.ib()
+
+    def generate(self):
+        for mutation in self.mutations:
+            # change to "yield from" after dropping Python 2
+            for one in self.generate_one(mutation):
+                yield one
+
+    def generate_one(self, mutation):
+        # change to "yield from" after dropping Python 2
+        for one in mutation.generate():
+            yield one

src/pytest_recording/mutations/mutagens.py

@@ -0,0 +1,48 @@
+import json
+import random
+
+from .._compat import Iterable
+from .core import mutation, Mutation
+
+
+class ChangeStatusCode(Mutation):
+    def generate(self):
+        # "code" will always be in the context, because it is validated before
+        if isinstance(self.context["code"], Iterable):
+            for status_code in self.context["code"]:
+                yield Mutation(self.func, {"code": status_code})
+        else:
+            yield self
+
+
+@ChangeStatusCode.from_function
+def status_code(cassette, code):
+    cassette["status"]["code"] = code
+
+
+@mutation
+def empty_body(cassette):
+    """Make the body empty."""
+    cassette["body"]["string"] = b""
+
+
+@mutation
+def malform_body(cassette):
+    """Add chars to the body to make it not decodable."""
+    cassette["body"]["string"] = b"xX" + cassette["body"]["string"] + b"Xx"
+
+
+@mutation
+def reduce_body(cassette):
+    data = json.loads(cassette["body"]["string"])
+    # it could be a list or other json type
+    # take out random key
+    key = random.choice(list(data.keys()))
+    del data[key]
+    cassette["body"]["string"] = json.dumps(data)
+
+
+# TODO.
+# XML - Add / remove XML tags.
+# JSON - add keys, remove keys, replace values
+# JSON - shrink lists, extend lists

src/pytest_recording/plugin.py

@@ -4,13 +4,15 @@
 
 import pytest
 
-from . import network
+from . import mutations, network
+from ._compat import Iterable
 from ._vcr import use_cassette
 
 RECORD_MODES = ("once", "new_episodes", "none", "all")
 
 
 def pytest_configure(config):
+    # TODO. register mutation marker
     config.addinivalue_line("markers", "vcr: Mark the test as using VCR.py.")
     config.addinivalue_line("markers", "block_network: Block network access except for VCR recording.")
     network.install_pycurl_wrapper()
@@ -84,17 +86,55 @@ def block_network(request, record_mode):
         yield
 
 
+@pytest.fixture()
+def mutation():
+    """Default mutation applied to all cassettes if no `vcr_mutations` is specified.
+
+    Should be None by default. But, could be used to apply the same mutations for all cassettes in a certain context.
+    """
+    return
+
+
 @pytest.fixture(autouse=True)
 def vcr(request, vcr_markers, vcr_cassette_dir, record_mode):
     """Install a cassette if a test is marked with `pytest.mark.vcr`."""
     if vcr_markers:
         config = request.getfixturevalue("vcr_config")
-        with use_cassette(vcr_cassette_dir, record_mode, vcr_markers, config) as cassette:
+        mutation = request.getfixturevalue("mutation")
+        with use_cassette(vcr_cassette_dir, record_mode, vcr_markers, config, mutation) as cassette:
             yield cassette
     else:
         yield
 
 
+def pytest_generate_tests(metafunc):
+    """Generate tests if VCR mutations are applied."""
+    if should_apply_mutations(metafunc):
+        marker = metafunc.definition.get_closest_marker("vcr_mutations")
+        if not marker:
+            # No mutations applied
+            return
+        if not marker.kwargs:
+            targets = mutations.DEFAULT
+        elif not (len(marker.kwargs) == 1 and "mutations" in marker.kwargs):
+            pytest.fail("Some error message")
+        else:
+            targets = marker.kwargs["mutations"]
+            if isinstance(targets, Iterable):
+                targets = mutations.core.MutationGroup(targets)
+
+        metafunc.parametrize("mutation", targets.generate())
+
+
+def should_apply_mutations(metafunc):
+    """To perform mutations on a test we need:
+
+    - `none` record mode;
+    - `pytest.mark.vcr` applied to the test.
+    """
+    return metafunc.config.getoption("--record-mode") == "none" and list(metafunc.definition.iter_markers(name="vcr"))
+
+
 @pytest.fixture(scope="module")
 def vcr_cassette_dir(request):
     """Each test module has its own cassettes directory to avoid name collisions.