def test_api_post_haveibeenpwned_not_reachable(self):
        payload = {
            "secret": "secret message",
            "passphrase": "Hello123",
            "haveibeenpwned": True
        }
        with self.client as c:
            with responses.RequestsMock() as rsps:
                rsps.add(
                    responses.GET,
                    re.compile(r"^(https:\/\/api\.pwnedpasswords\.com\/).*"),
                    body=Exception)
                response = json.loads(
                    c.post("/api/c", json=payload).get_data())

        # haveibeenpwned wasn't reachable, but secret still created if it has
        # all mandatory requirements.
        r = Parse(response)
        self.assertEqual(r.response.status, "created") 

def test_weather_city():
    """
    Check result when place is a city
    """
    filepath = os.path.join(os.path.dirname(__file__), "fixtures", "api_weather_response.json")

    with open(filepath, "r") as f:
        json_aq = json.load(f)

    with responses.RequestsMock() as rsps:
        rsps.add(
            "GET", re.compile(r"^http://api.openweathermap.org/data/2.5/"), status=200, json=json_aq
        )

        res = Weather.from_es(Admin(testee), lang="en")

    assert res == Weather(**{"temperature": 291.89, "icon": "01d",}) 

def mock_directions_car():
    with override_settings(
        {
            "QWANT_DIRECTIONS_API_BASE_URL": "http://api.qwant/directions",
            "MAPBOX_DIRECTIONS_ACCESS_TOKEN": None,
        }
    ):
        fixture_path = os.path.join(
            os.path.dirname(__file__), "fixtures/directions", "qwant_directions_car.json",
        )
        with responses.RequestsMock() as rsps:
            rsps.add(
                "GET",
                re.compile(r"^http://api.qwant/directions/"),
                status=200,
                json=json.load(open(fixture_path)),
            )
            yield rsps 

def test_rate_limiter_without_redis():
    """
    Test that Idunn doesn't stop external requests when
    no redis has been set: 10 requests to Idunn should
    generate 10 requests to Wikipedia API
    """
    client = TestClient(app)

    with responses.RequestsMock() as rsps:
        rsps.add(
            "GET", re.compile(r"^https://.*\.wikipedia.org/"), status=200, json={"test": "test"}
        )
        for i in range(10):
            response = client.get(url=f"http://localhost/v1/pois/osm:relation:7515426?lang=es",)

        assert len(rsps.calls) == 10 

def test_kuzzle_event_nok():
    """
    Check that an error  501 is raised when kuzzle port and address not set
    """
    # with pytest.raises(Exception):
    filepath = os.path.join(os.path.dirname(__file__), "fixtures", "kuzzle_event_response.json")
    with open(filepath, "r") as f:
        json_event = json.load(f)

    client = TestClient(app)
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
        rsps.add(
            "POST",
            re.compile(r"^http://localhost:7512/opendatasoft/events/"),
            status=200,
            json=json_event,
        )

        response = client.get(
            url=f"http://localhost/v1/events?bbox=2.0667651,48.432533,2.9384989,49.0349191& ,*&size=5",
        )

        assert response.status_code == 501 

def test_no_lang_WIKI_ES():
    """
    Test that when we don't have the lang available in the index
    we call Wikipedia API
    """
    client = TestClient(app)
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
        rsps.add(
            "GET", re.compile(r"^https://.*\.wikipedia.org/"), status=200, json={"test": "test"}
        )

        """
        We make a request in russian language ("ru")
        """
        response = client.get(url=f"http://localhost/v1/pois/osm:way:7777777?lang=ru",)

        assert len(rsps.calls) == 1 

def generate_and_save_stripe_fixture(decorated_function_name: str, mocked_function_name: str,
                                     mocked_function: CallableT) -> Callable[[Any, Any], Any]:  # nocoverage
    def _generate_and_save_stripe_fixture(*args: Any, **kwargs: Any) -> Any:
        # Note that mock is not the same as mocked_function, even though their
        # definitions look the same
        mock = operator.attrgetter(mocked_function_name)(sys.modules[__name__])
        fixture_path = stripe_fixture_path(decorated_function_name, mocked_function_name, mock.call_count)
        try:
            with responses.RequestsMock() as request_mock:
                request_mock.add_passthru("https://api.stripe.com")
                # Talk to Stripe
                stripe_object = mocked_function(*args, **kwargs)
        except stripe.error.StripeError as e:
            with open(fixture_path, 'w') as f:
                error_dict = e.__dict__
                error_dict["headers"] = dict(error_dict["headers"])
                f.write(json.dumps(error_dict, indent=2, separators=(',', ': '), sort_keys=True) + "\n")
            raise e
        with open(fixture_path, 'w') as f:
            if stripe_object is not None:
                f.write(str(stripe_object) + "\n")
            else:
                f.write("{}\n")
        return stripe_object
    return _generate_and_save_stripe_fixture 

def dataset_responses():
    """Authentication responses."""
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:

        def request_callback(request):
            return (200, {'Content-Type': 'application/text'}, '1234')

        rsps.add_callback(
            responses.GET,
            'http://example.com/file',
            callback=request_callback
        )
        rsps.add_callback(
            responses.GET,
            'https://example.com/file',
            callback=request_callback
        )
        rsps.add_callback(
            responses.GET,
            'http://example.com/file.ext?foo=bar',
            callback=request_callback
        )
        yield rsps 

def setUp(self):
        super(SessionTester, self).setUp()
        with open("tests/rets_responses/Login.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/Login.ashx",
                body=contents,
                status=200,
                headers={"Set-Cookie": "ASP.NET_SessionId=zacqcc1gjhkmazjznjmyrinq;"},
            )
            self.session = Session(
                login_url="http://server.rets.com/rets/Login.ashx",
                username="retsuser",
                version="RETS/1.7.2",
                session_id_cookie_name="ASP.NET_SessionId",
            )
            self.session.login() 

def test_system_metadata(self):

        with open("tests/rets_responses/COMPACT-DECODED/GetMetadata_system.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=contents,
                status=200,
            )
            sys_metadata = self.session.get_system_metadata()

        self.assertEqual(sys_metadata["version"], "1.11.76001")
        self.assertEqual(sys_metadata["system_id"], "MLS-RETS") 

def test_cache_metadata(self):
        with open("tests/rets_responses/COMPACT-DECODED/GetMetadata_table.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=contents,
                status=200,
            )
            self.session.get_table_metadata(resource="Property", resource_class="RES")

        self.assertIn(
            "METADATA-TABLE:Property:RES", list(self.session.metadata_responses.keys())
        )

        # Subsequent call without RequestMock should fail unless we get the saved response from metadata_responses
        table = self.session.get_table_metadata(
            resource="Property", resource_class="RES"
        )
        self.assertEqual(len(list(table)), 208) 

def test_table_metadata(self):
        with open("tests/rets_responses/COMPACT-DECODED/GetMetadata_table.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=contents,
                status=200,
            )
            table = self.session.get_table_metadata(
                resource="Property", resource_class="RES"
            )

        self.assertEqual(len(list(table)), 208) 

def test_change_parser_automatically(self):
        self.assertEqual(self.session.metadata_format, "COMPACT-DECODED")

        with open("tests/rets_responses/Errors/20514.xml") as f:
            dtd_error = "".join(f.readlines())

        with open("tests/rets_responses/STANDARD-XML/GetMetadata_system.xml") as f:
            content = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=dtd_error,
                status=200,
            )
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=content,
                status=200,
            )
            self.session.get_system_metadata()

        self.assertEqual(self.session.metadata_format, "STANDARD-XML") 

def test_wildcard_lookups(self):
        with open("tests/rets_responses/STANDARD-XML/GetMetadata_wildcard.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=contents,
                status=200,
            )
            format_hold = self.session.metadata_format
            try:
                self.session.metadata_format = "STANDARD-XML"
                lookup_values = self.session.get_lookup_values(
                    resource="Property", lookup_name="*"
                )
            finally:
                self.session.metadata_format = format_hold

        self.assertEqual(len(lookup_values), 40) 

def setUp(self):
        super(Session15Tester, self).setUp()
        with open("tests/rets_responses/Login.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/Login.ashx",
                body=contents,
                status=200,
            )
            self.session = Session(
                login_url="http://server.rets.com/rets/Login.ashx",
                username="retsuser",
                version="1.5",
            )
            self.session.metadata_format = "STANDARD-XML"
            self.session.login() 

def test_table_metadata(self):
        with open("tests/rets_responses/STANDARD-XML/GetMetadata_table.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=contents,
                status=200,
            )
            table = self.session.get_table_metadata(
                resource="Property", resource_class="1"
            )

        self.assertEqual(len(table), 162) 

def test_lookup_type_metadata(self):
        with open("tests/rets_responses/STANDARD-XML/GetMetadata_lookup.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=contents,
                status=200,
            )
            lookup_values = self.session.get_lookup_values(
                resource="Property", lookup_name="1_2"
            )

        self.assertEqual(len(lookup_values), 9) 

def test_alternative_lookup_type_metadata(self):
        with open("tests/rets_responses/STANDARD-XML/GetMetadata_lookup2.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com/rets/GetMetadata.ashx",
                body=contents,
                status=200,
            )
            lookup_values = self.session.get_lookup_values(
                resource="Property", lookup_name="mls_cooling"
            )

        self.assertEqual(len(lookup_values), 4) 

def test_port_added_to_actions(self):
        with open("tests/rets_responses/Login_relative_url.xml") as f:
            contents = "".join(f.readlines())

        with responses.RequestsMock() as resps:
            resps.add(
                resps.POST,
                "http://server.rets.com:9999/rets/Login.ashx",
                body=contents,
                status=200,
            )
            s = Session(
                login_url="http://server.rets.com:9999/rets/Login.ashx",
                username="retsuser",
                version="1.5",
            )
            s.login()

            for cap in s.capabilities.values():
                parts = urlparse(cap)
                self.assertEqual(parts.port, 9999) 

def test_downloader_conn_error(self):
        exception = ConnectionError()
        with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
            max_retry = 3
            for _ in range(max_retry + 1):
                rsps.add(responses.GET, self.TEST_MPD_URL, body=exception)

            dl = live.Downloader(
                mpd=self.TEST_MPD_URL,
                output_dir='output_connerror',
                duplicate_etag_retry=2,
                singlethreaded=True,
                max_connection_error_retry=max_retry)
            dl.run()
            dl.stream_id = '17875351285037717'
            output_file = 'output_connerror.mp4'
            dl.stitch(output_file, cleartempfiles=True)
            self.assertFalse(os.path.isfile(output_file), '{0!s} not generated'.format(output_file)) 

def setup(self):
        app.testing = True

        with responses.RequestsMock() as r:
            # mock provider discovery
            r.add(responses.GET, ISSUER1 + '/.well-known/openid-configuration', json=self.PROVIDER1_METADATA)
            r.add(responses.GET, ISSUER2 + '/.well-known/openid-configuration', json=self.PROVIDER2_METADATA)
            auth.init_app(app) 

def mock_responses():
    with responses.RequestsMock() as _responses:
        yield _responses 

def test_get_cities_bad_response(mock_responses):
        mock_responses.add(
            responses.RequestsMock.POST,
            mealpy.CITIES_URL,
            status=400,
        )

        with pytest.raises(requests.exceptions.HTTPError):
            mealpy.MealPal.get_cities() 

def test_login(mock_responses):
        mock_responses.add(
            responses.RequestsMock.POST,
            mealpy.LOGIN_URL,
            status=200,
            json={
                'id': 'GUID',
                'email': 'email',
                'status': 3,
                'firstName': 'first_name',
                'lastName': 'last_name',
                'sessionToken': 'r:GUID',
                'city': {
                    'id': 'GUID',
                    'name': 'San Francisco',
                    'city_code': 'SFO',
                    'countryCode': 'usa',
                    '__type': 'Pointer',
                    'className': 'City',
                    'objectId': 'GUID',
                },
            },
        )

        mealpal = mealpy.MealPal()

        assert mealpal.login('username', 'password') == 200 

def test_login_fail(mock_responses):
        mock_responses.add(
            method=responses.RequestsMock.POST,
            url=mealpy.LOGIN_URL,
            status=404,
            json={
                'code': 101,
                'error': 'An error occurred while blah blah, try agian.',
            },
        )

        mealpal = mealpy.MealPal()

        with pytest.raises(requests.HTTPError):
            mealpal.login('username', 'password') 

def menu_url_response(mock_responses, success_response, mock_city):
        mock_responses.add(
            responses.RequestsMock.GET,
            mealpy.MENU_URL.format(mock_city.objectId),
            status=200,
            json=success_response,
        )

        yield mock_responses 

def test_get_schedules_fail(mock_responses, mock_city):
        mock_responses.add(
            method=responses.RequestsMock.GET,
            url=mealpy.MENU_URL.format(mock_city.objectId),
            status=400,
        )

        with pytest.raises(requests.HTTPError):
            mealpy.MealPal.get_schedules(mock_city.name) 

def kitchen_url_response(mock_responses, success_response_no_reservation):
        mock_responses.add(
            responses.RequestsMock.POST,
            mealpy.KITCHEN_URL,
            status=200,
            json=success_response_no_reservation,
        )

        yield mock_responses 

def kitchen_url_response_with_reservation(mock_responses, success_response_no_reservation, current_meal):
        success_response_no_reservation['reservation'] = current_meal

        mock_responses.add(
            responses.RequestsMock.POST,
            mealpy.KITCHEN_URL,
            status=200,
            json=success_response_no_reservation,
        )

        yield mock_responses 

def reserve_response_failed(mock_responses):  # pragma: no cover
        # Current unused
        response = {'error': 'ERROR_RESERVATION_LIMIT'}
        mock_responses.add(
            responses.RequestsMock.POST,
            mealpy.KITCHEN_URL,
            status=400,
            json=response,
        )
        yield mock_responses