Allow deletion of a single impostor from a running server. Also improve query_all_impostors() so that the impostor instances are attached to the server they are from.

Signed-off-by: Simon Brunning <simon@brunningonline.net>

Author: brunns

src/mbtest/matchers.py

@@ -112,7 +112,7 @@ def describe_mismatch(
         description.append_text(". All requests: ").append_description_of(self.all_requests)
 
     def _matches(self, actual: Union[Imposter, MountebankServer]) -> bool:
-        self.all_requests = cast(Sequence[HttpRequest], list(actual.get_actual_requests()))
+        self.all_requests = cast(Sequence[HttpRequest], actual.get_actual_requests())
         self.matching_requests = [
             request
             for request in self.all_requests

src/mbtest/server.py

@@ -3,9 +3,10 @@
 import subprocess  # nosec
 import time
 from collections import abc
+from operator import attrgetter
 from pathlib import Path
 from threading import Lock
-from typing import Iterable, Iterator, List, MutableSequence, Optional, Sequence, Set, Union
+from typing import Iterable, List, MutableSequence, Optional, Sequence, Set, Union
 
 import requests
 from _pytest.fixtures import FixtureRequest  # type: ignore
@@ -138,39 +139,59 @@ def add_imposters(self, definition: Union[Imposter, Iterable[Imposter]]) -> None
         """Add imposters to Mountebank server.
 
         :param definition: One or more Imposters.
-        :type definition: Imposter or list(Imposter)
         """
         if isinstance(definition, abc.Iterable):
             for imposter in definition:
                 self.add_imposters(imposter)
         else:
-            json = definition.as_structure()
-            post = requests.post(self.server_url, json=json, timeout=10)
-            post.raise_for_status()
-            definition.attach(self.host, post.json()["port"], self.server_url)
-            self._running_imposters.append(definition)
+            self.add_impostor(definition)
+
+    def add_impostor(self, definition):
+        """Add single imposter to Mountebank server.
+
+        :param definition: One or more Imposters."""
+        json = definition.as_structure()
+        post = requests.post(self.server_url, json=json, timeout=10)
+        post.raise_for_status()
+        definition.attach(self.host, post.json()["port"], self.server_url)
+        self._running_imposters.append(definition)
 
     def delete_imposters(self) -> None:
+        """Delete all impostors from server."""
         while self._running_imposters:
-            imposter = self._running_imposters.pop()
-            requests.delete(imposter.configuration_url).raise_for_status()
+            self.delete_impostor(self._running_imposters[0])
+
+    def delete_impostor(self, imposter):
+        """Delete impostor from server."""
+        requests.delete(imposter.configuration_url).raise_for_status()
+        self._running_imposters = [
+            i for i in self._running_imposters if i.configuration_url != imposter.configuration_url
+        ]
 
-    def get_actual_requests(self) -> Iterable[Request]:
+    def get_actual_requests(self) -> Sequence[Request]:
+        actual_requests: MutableSequence[Request] = []
         for imposter in self._running_imposters:
-            yield from imposter.get_actual_requests()
+            actual_requests += imposter.get_actual_requests()
+        return actual_requests
 
     @property
     def server_url(self) -> furl:
         return furl().set(
             scheme=self.scheme, host=self.host, port=self.server_port, path=self.imposters_path
         )
 
-    def query_all_imposters(self) -> Iterator[Imposter]:
+    def query_all_imposters(self) -> Sequence[Imposter]:
         """Yield all imposters running on the server, including those defined elsewhere."""
         server_info = requests.get(self.server_url)
-        imposters = server_info.json()["imposters"]
-        for imposter in imposters:
-            yield Imposter.from_structure(requests.get(imposter["_links"]["self"]["href"]).json())
+        imposters_structure = server_info.json()["imposters"]
+        all_imposters: MutableSequence[Imposter] = []
+        for imposter_structure in imposters_structure:
+            impostor_url = imposter_structure["_links"]["self"]["href"]
+            imposter = Imposter.from_structure(requests.get(impostor_url).json())
+            imposter.host = self.host
+            imposter.server_url = self.server_url
+            all_imposters.append(imposter)
+        return sorted(all_imposters, key=attrgetter("port"))
 
 
 class ExecutingMountebankServer(MountebankServer):

tests/integration/test_server.py

@@ -98,17 +98,64 @@ def test_query_all_imposters(mock_server):
     imposter2 = Imposter(Stub(Predicate(path="/test2"), Response(body="egg")))
 
     with mock_server([imposter1, imposter2]) as server:
-        actual = list(server.query_all_imposters())
+        actual = server.query_all_imposters()
         assert_that(
             actual,
             contains_inanyorder(
-                has_identical_properties_to(
-                    imposter1,
-                    ignoring={"host", "url", "server_url", "configuration_url", "attached"},
-                ),
-                has_identical_properties_to(
-                    imposter2,
-                    ignoring={"host", "url", "server_url", "configuration_url", "attached"},
-                ),
+                has_identical_properties_to(imposter1), has_identical_properties_to(imposter2)
+            ),
+        )
+
+
+def test_removing_impostor_from_running_server(mock_server):
+    # Set up server
+    with mock_server(
+        [
+            Imposter(Stub(Predicate(path="/test"), Response(body="sausage")), name="sausage"),
+            Imposter(Stub(Predicate(path="/test"), Response(body="egg")), name="egg"),
+            Imposter(Stub(Predicate(path="/test"), Response(body="chips")), name="chips"),
+        ]
+    ) as server:
+
+        # Retrieve impostor details from running server, and check they work
+        initial = server.query_all_imposters()
+
+        responses = [requests.get(f"{initial[i].url}/test") for i in range(3)]
+        assert_that(
+            responses,
+            contains_inanyorder(
+                is_response().with_body("sausage"),
+                is_response().with_body("egg"),
+                is_response().with_body("chips"),
+            ),
+        )
+
+        # Delete one impostor, make sure it's gone, and that the rest still work
+        egg_impostor = [i for i in initial if i.name == "egg"][0]
+        other_impostors = [i for i in initial if i.name != "egg"]
+        server.delete_impostor(egg_impostor)
+
+        with pytest.raises(requests.exceptions.ConnectionError):
+            requests.get(f"{egg_impostor.url}/test")
+        responses = [requests.get(f"{i.url}/test") for i in other_impostors]
+        assert_that(
+            responses,
+            contains_inanyorder(
+                is_response().with_body("sausage"),
+                is_response().with_body("chips"),
+            ),
+        )
+
+        # Reset the server from the initial impostors, and check it's back to normal
+        server.delete_imposters()
+        server.add_imposters(initial)
+
+        responses = [requests.get(f"{initial[i].url}/test") for i in range(3)]
+        assert_that(
+            responses,
+            contains_inanyorder(
+                is_response().with_body("sausage"),
+                is_response().with_body("egg"),
+                is_response().with_body("chips"),
             ),
         )