From d4ace43414f163c2abe8cafccab6155d27404e7d Mon Sep 17 00:00:00 2001 From: Maksim Beliaev Date: Mon, 12 Sep 2022 16:25:34 +0200 Subject: [PATCH] [BETA] replay responses from toml files. (#581) Added beta option for replaying responses from toml files. Co-authored-by: Mark Story --- CHANGES | 2 + responses/__init__.py | 19 +++++ responses/tests/test_recorder.py | 138 +++++++++++++++++++++---------- 3 files changed, 114 insertions(+), 45 deletions(-) diff --git a/CHANGES b/CHANGES index e9e3ce07..662cfc3f 100644 --- a/CHANGES +++ b/CHANGES @@ -3,6 +3,8 @@ * Update `requests` dependency to the version of 2.22.0 or higher. See #584. * [BETA] Added possibility to record responses to TOML files via `@_recorder.record(file_path="out.toml")` decorator. +* [BETA] Added possibility to replay responses (populate registry) from TOML files + via `responses._add_from_file(file_path="out.toml")` method. * Fix type for the `mock`'s patcher object. See #556 * Fix type annotation for `CallList` * Add `passthrough` argument to `BaseResponse` object. See #557 diff --git a/responses/__init__.py b/responses/__init__.py index 4910996a..24d664be 100644 --- a/responses/__init__.py +++ b/responses/__init__.py @@ -24,6 +24,7 @@ from typing import overload from warnings import warn +import toml as _toml from requests.adapters import HTTPAdapter from requests.adapters import MaxRetryError from requests.exceptions import ConnectionError @@ -64,6 +65,7 @@ if TYPE_CHECKING: # pragma: no cover # import only for linter run + import os from unittest.mock import _patch as _mock_patcher from requests import PreparedRequest @@ -769,6 +771,21 @@ def add( response = Response(method=method, url=url, body=body, **kwargs) return self._registry.add(response) + def _add_from_file(self, file_path: "Union[str, bytes, os.PathLike[Any]]") -> None: + with open(file_path) as file: + data = _toml.load(file) + + for rsp in data["responses"]: + rsp = rsp["response"] + self.add( + method=rsp["method"], + url=rsp["url"], + body=rsp["body"], + status=rsp["status"], + content_type=rsp["content_type"], + auto_calculate_content_length=rsp["auto_calculate_content_length"], + ) + def delete(self, *args: Any, **kwargs: Any) -> BaseResponse: return self.add(DELETE, *args, **kwargs) @@ -1146,6 +1163,7 @@ def assert_call_count(self, url: str, count: int) -> bool: # Exposed by the RequestsMock class: "activate", "add", + "_add_from_file", "add_callback", "add_passthru", "_deprecated_assert_all_requests_are_fired", @@ -1180,6 +1198,7 @@ def assert_call_count(self, url: str, count: int) -> bool: # expose only methods and/or read-only methods activate = _default_mock.activate add = _default_mock.add +_add_from_file = _default_mock._add_from_file add_callback = _default_mock.add_callback add_passthru = _default_mock.add_passthru _deprecated_assert_all_requests_are_fired = _default_mock.assert_all_requests_are_fired diff --git a/responses/tests/test_recorder.py b/responses/tests/test_recorder.py index 0d733793..d70d60f6 100644 --- a/responses/tests/test_recorder.py +++ b/responses/tests/test_recorder.py @@ -3,9 +3,58 @@ import requests import toml +import responses from responses import _recorder +def get_data(host, port): + data = { + "responses": [ + { + "response": { + "method": "GET", + "url": f"http://{host}:{port}/404", + "body": "404 Not Found", + "status": 404, + "content_type": "text/plain", + "auto_calculate_content_length": False, + } + }, + { + "response": { + "method": "GET", + "url": f"http://{host}:{port}/status/wrong", + "body": "Invalid status code", + "status": 400, + "content_type": "text/plain", + "auto_calculate_content_length": False, + } + }, + { + "response": { + "method": "GET", + "url": f"http://{host}:{port}/500", + "body": "500 Internal Server Error", + "status": 500, + "content_type": "text/plain", + "auto_calculate_content_length": False, + } + }, + { + "response": { + "method": "PUT", + "url": f"http://{host}:{port}/202", + "body": "OK", + "status": 202, + "content_type": "text/plain", + "auto_calculate_content_length": False, + } + }, + ] + } + return data + + class TestRecord: def setup(self): self.out_file = Path("out.toml") @@ -35,7 +84,7 @@ def test_recorder(self, httpserver): def another(): requests.get(url500) - requests.get(url202) + requests.put(url202) @_recorder.record(file_path=self.out_file) def run(): @@ -48,47 +97,46 @@ def run(): with open(self.out_file) as file: data = toml.load(file) - assert data == { - "responses": [ - { - "response": { - "method": "GET", - "url": f"http://{httpserver.host}:{httpserver.port}/404", - "body": "404 Not Found", - "status": 404, - "content_type": "text/plain", - "auto_calculate_content_length": False, - } - }, - { - "response": { - "method": "GET", - "url": f"http://{httpserver.host}:{httpserver.port}/status/wrong", - "body": "Invalid status code", - "status": 400, - "content_type": "text/plain", - "auto_calculate_content_length": False, - } - }, - { - "response": { - "method": "GET", - "url": f"http://{httpserver.host}:{httpserver.port}/500", - "body": "500 Internal Server Error", - "status": 500, - "content_type": "text/plain", - "auto_calculate_content_length": False, - } - }, - { - "response": { - "method": "GET", - "url": f"http://{httpserver.host}:{httpserver.port}/202", - "body": "OK", - "status": 202, - "content_type": "text/plain", - "auto_calculate_content_length": False, - } - }, - ] - } + assert data == get_data(httpserver.host, httpserver.port) + + +class TestReplay: + def teardown(self): + out_file = Path("out.toml") + if out_file.exists(): + out_file.unlink() + + assert not out_file.exists() + + def test_add_from_file(self): + with open("out.toml", "w") as file: + toml.dump(get_data("example.com", "8080"), file) + + @responses.activate + def run(): + responses.patch("http://httpbin.org") + responses._add_from_file(file_path="out.toml") + responses.post("http://httpbin.org/form") + + assert responses.registered()[0].url == "http://httpbin.org/" + assert responses.registered()[1].url == "http://example.com:8080/404" + assert ( + responses.registered()[2].url == "http://example.com:8080/status/wrong" + ) + assert responses.registered()[3].url == "http://example.com:8080/500" + assert responses.registered()[4].url == "http://example.com:8080/202" + assert responses.registered()[5].url == "http://httpbin.org/form" + + assert responses.registered()[0].method == "PATCH" + assert responses.registered()[2].method == "GET" + assert responses.registered()[4].method == "PUT" + assert responses.registered()[5].method == "POST" + + assert responses.registered()[2].status == 400 + assert responses.registered()[3].status == 500 + + assert responses.registered()[3].body == "500 Internal Server Error" + + assert responses.registered()[3].content_type == "text/plain" + + run()