Python源码示例:responses.GET

示例1
def test_session_get():
    responses.add(
        responses.GET,
        'http://foo.bar.baz/api/v1.0/test',
        body='okay',
        status=200)
    host = 'foo.bar.baz'
    token = '5f1e08b6-38ec-4a99-9d0f-00d29c4e325b'
    marker = '40c3eaf6-6a8a-11e7-a4bd-080027ef795a'

    def auth_gen():
        return [('X-Auth-Token', token)]

    dd_ses = dc_session.DrydockSession(host, auth_gen=auth_gen, marker=marker)

    resp = dd_ses.get('v1.0/test')
    req = resp.request

    assert req.headers.get('X-Auth-Token', None) == token
    assert req.headers.get('X-Context-Marker', None) == marker 
示例2
def test_session_get_returns_401(*args):
    responses.add(
        responses.GET,
        'http://foo.bar.baz/api/v1.0/test',
        body='okay',
        status=401)
    host = 'foo.bar.baz'
    token = '5f1e08b6-38ec-4a99-9d0f-00d29c4e325b'
    marker = '40c3eaf6-6a8a-11e7-a4bd-080027ef795a'

    def auth_gen():
        return [('X-Auth-Token', dc_session.KeystoneClient.get_token())]

    dd_ses = dc_session.DrydockSession(host, auth_gen=auth_gen, marker=marker)

    resp = dd_ses.get('v1.0/test')
    req = resp.request

    assert req.headers.get('X-Auth-Token', None) == token
    assert req.headers.get('X-Context-Marker', None) == marker
    assert dc_session.KeystoneClient.get_token.call_count == 2 
示例3
def setUp(self):
        self.client = self.app.test_client()
        self.app_context = self.app.app_context()
        self.app_context.push()
        for table in reversed(self.db.metadata.sorted_tables):
            self.db.session.execute(table.delete())
        # Mock responses from haveibeenpwned.
        responses.add(
            responses.GET,
            re.compile(
                r"^(https:\/\/api\.pwnedpasswords\.com\/range\/836BA).*"),
            body=("BDDC66080E01D52B8272AA9461C69EE0496:12145\n"
                  "00d4f6e8fa6eecad2a3aa415eec418d38ec:2"))
        responses.add(
            responses.GET,
            re.compile(
                r"^(https:\/\/api\.pwnedpasswords\.com\/)(?!.*836BA).*"),
            body=("BDDC66080E01D52B8272AA9461C69EE0496:12145\n"
                  "00d4f6e8fa6eecad2a3aa415eec418d38ec:2")) 
示例4
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") 
示例5
def test_list_policies(nomad_setup):
    responses.add(
        responses.GET,
        "http://{ip}:{port}/v1/sentinel/policies".format(ip=common.IP, port=common.NOMAD_PORT),
        status=200,
        json=[
            {
                "Name": "foo",
                "Description": "test policy",
                "Scope": "submit-job",
                "EnforcementLevel": "advisory",
                "Hash": "CIs8aNX5OfFvo4D7ihWcQSexEJpHp+Za+dHSncVx5+8=",
                "CreateIndex": 8,
                "ModifyIndex": 8
            }
        ]
    )

    policies = nomad_setup.sentinel.get_policies()
    assert isinstance(policies, list)
    assert "foo" in nomad_setup.sentinel.get_policies()[0]["Name"] 
示例6
def test_get_policy(nomad_setup):
    responses.add(
        responses.GET,
        "http://{ip}:{port}/v1/sentinel/policy/foo".format(ip=common.IP, port=common.NOMAD_PORT),
        status=200,
        json={
            "Name": "foo",
            "Description": "test policy",
            "Scope": "submit-job",
            "EnforcementLevel": "advisory",
            "Policy": "main = rule { true }\n",
            "Hash": "CIs8aNX5OfFvo4D7ihWcQSexEJpHp+Za+dHSncVx5+8=",
            "CreateIndex": 8,
            "ModifyIndex": 8
        }
    )

    policy = nomad_setup.sentinel.get_policy("foo")
    assert "advisory" in policy["EnforcementLevel"] 
示例7
def test_get_namespaces(nomad_setup):
    responses.add(
        responses.GET,
        "http://{ip}:{port}/v1/namespaces".format(ip=common.IP, port=common.NOMAD_PORT),
        status=200,
        json=[
                {
                    "CreateIndex": 31,
                    "Description": "Production API Servers",
                    "ModifyIndex": 31,
                    "Name": "api-prod"
                },
                {
                    "CreateIndex": 5,
                    "Description": "Default shared namespace",
                    "ModifyIndex": 5,
                    "Name": "default"
                }
            ]
    )

    assert isinstance(nomad_setup.namespaces.get_namespaces(), list) == True 
示例8
def test_get_namespaces_prefix(nomad_setup):
    responses.add(
        responses.GET,
        "http://{ip}:{port}/v1/namespaces?prefix=api-".format(ip=common.IP, port=common.NOMAD_PORT),
        status=200,
        json=[
                {
                    "CreateIndex": 31,
                    "Description": "Production API Servers",
                    "ModifyIndex": 31,
                    "Name": "api-prod"
                },
            ]
    )

    assert isinstance(nomad_setup.namespaces.get_namespaces(prefix="api-"), list) == True 
示例9
def test_namespaces_iter(nomad_setup):
    responses.add(
        responses.GET,
        "http://{ip}:{port}/v1/namespaces".format(ip=common.IP, port=common.NOMAD_PORT),
        status=200,
        json=[
                {
                    "CreateIndex": 31,
                    "Description": "Production API Servers",
                    "ModifyIndex": 31,
                    "Name": "api-prod"
                },
                {
                    "CreateIndex": 5,
                    "Description": "Default shared namespace",
                    "ModifyIndex": 5,
                    "Name": "default"
                }
            ]
    )

    assert "api-prod" in nomad_setup.namespaces 
示例10
def test_cli_dataset_list_features_stdout():
    id = "cii9dtexw0039uelz7nzk1lq3"
    collection='{"type":"FeatureCollection","features":[]}'
    responses.add(
        responses.GET,
        'https://api.mapbox.com/datasets/v1/{0}/{1}/features?access_token={2}'.format(username, id, access_token),
        match_querystring=True,
        status=200, body=collection,
        content_type='application/json'
    )

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', access_token,
         'datasets',
         'list-features', id])

    assert result.exit_code == 0
    assert result.output.strip() == collection.strip() 
示例11
def test_cli_dataset_list_features_tofile(tmpdir):
    tmpfile=str(tmpdir.join('test.list-features.json'))
    id = "dataset-1"
    collection='{"type":"FeatureCollection","features":[]}'
    responses.add(
        responses.GET,
        'https://api.mapbox.com/datasets/v1/{0}/{1}/features?access_token={2}'.format(username, id, access_token),
        match_querystring=True,
        status=200, body=collection,
        content_type='application/json'
    )

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', access_token,
         'datasets',
         'list-features', id,
         '--output', tmpfile])

    assert result.exit_code == 0
    assert result.output.strip() == ""
    assert open(tmpfile).read().strip() == collection.strip() 
示例12
def test_cli_static():

    responses.add(
        responses.GET,
        'https://api.mapbox.com/v4/mapbox.satellite/-61.7,12.1,12/600x600.png256?access_token=bogus',
        match_querystring=True,
        body='.PNG...',
        status=200,
        content_type='image/png')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', 'bogus',
         'staticmap',
         '--lon', '-61.7',
         '--lat', '12.1',
         '--zoom', '12',
         'mapbox.satellite',
         '/dev/null'])

    assert result.exit_code == 0 
示例13
def test_cli_geocode_fwd():

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/1600%20pennsylvania%20ave%20nw.json?access_token=bogus',
        match_querystring=True,
        body='{"query": ["1600", "pennsylvania", "ave", "nw"]}', status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', 'bogus', 'geocoding', '--forward', '1600 pennsylvania ave nw'],
        catch_exceptions=False)
    print(result.output)
    print(result.exception)
    print(result.exc_info)
    assert result.exit_code == 0
    assert result.output == '{"query": ["1600", "pennsylvania", "ave", "nw"]}\n' 
示例14
def test_cli_geocode_fwd_bbox():

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/1600%20pennsylvania%20ave%20nw.json?access_token=bogus&bbox=-78.3284%2C38.6039%2C-78.0428%2C38.7841',
        match_querystring=True,
        body='{"query": ["1600", "pennsylvania", "ave", "nw"]}', status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', 'bogus', 'geocoding', '--forward', '1600 pennsylvania ave nw', '--bbox', '[-78.3284,38.6039,-78.0428,38.7841]'],
        catch_exceptions=False)
    print(result.output)
    print(result.exception)
    print(result.exc_info)
    assert result.exit_code == 0
    assert result.output == '{"query": ["1600", "pennsylvania", "ave", "nw"]}\n' 
示例15
def test_cli_geocode_fwd_env_token():

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/1600%20pennsylvania%20ave%20nw.json?access_token=bogus',
        match_querystring=True,
        body='{"query": ["1600", "pennsylvania", "ave", "nw"]}', status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['geocoding', '--forward', '1600 pennsylvania ave nw'],
        env={'MAPBOX_ACCESS_TOKEN': 'bogus'})
    assert result.exit_code == 0
    assert result.output == '{"query": ["1600", "pennsylvania", "ave", "nw"]}\n' 
示例16
def test_cli_geocode_reverse():

    lon, lat = -77.4371, 37.5227
    body = json.dumps({"query": [lon, lat]})

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/{0},{1}.json?access_token=pk.test'.format(lon, lat),
        match_querystring=True,
        body=body,
        status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', 'pk.test', 'geocoding', '--reverse'],
        input='{0},{1}'.format(lon, lat))
    assert result.exit_code == 0
    assert result.output.strip() == body 
示例17
def test_cli_geocode_rev_unauthorized():

    lon, lat = -77.4371, 37.5227

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/{0},{1}.json'.format(lon, lat),
        body='{"message":"Not Authorized - Invalid Token"}', status=401,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['geocoding', '--reverse'],
        input='{0},{1}'.format(lon, lat))
    assert result.exit_code == 1
    assert result.output == 'Error: {"message":"Not Authorized - Invalid Token"}\n' 
示例18
def test_cli_geocode_rev_headers():

    lon, lat = -77.4371, 37.5227
    body = json.dumps({"query": [lon, lat]})

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/{0},{1}.json'.format(lon, lat),
        body=body,
        status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['geocoding', '-i', '--reverse'],
        input='{0},{1}'.format(lon, lat))
    assert result.exit_code == 0
    assert result.output.startswith('Content-Type') 
示例19
def test_cli_geocode_fwd_limit():

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/1600%20pennsylvania%20ave%20nw.json?access_token=bogus&limit=2',
        match_querystring=True,
        body='{"features": [{"name": "first"}, {"name": "second"}]}', status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', 'bogus', 'geocoding', '--limit', '2',
         '--forward', '1600 pennsylvania ave nw'],
        catch_exceptions=False)
    assert result.exit_code == 0
    assert result.output == '{"features": [{"name": "first"}, {"name": "second"}]}\n' 
示例20
def test_cli_geocode_reverse_features():

    lon, lat = -77.4371, 37.5227
    res = {"query": [lon, lat],
           "features": [{"name": "first"}, {"name": "second"}]}
    body = json.dumps(res)

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/{0},{1}.json?access_token=pk.test'.format(lon, lat),
        match_querystring=True,
        body=body,
        status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', 'pk.test', 'geocoding', '--reverse', '--features'],
        input='{0},{1}'.format(lon, lat))
    assert result.exit_code == 0
    assert result.output == '{"name": "first"}\n{"name": "second"}\n' 
示例21
def test_cli_geocode_fwd_features():

    responses.add(
        responses.GET,
        'https://api.mapbox.com/geocoding/v5/mapbox.places/1600%20pennsylvania%20ave%20nw.json?access_token=bogus&limit=2',
        match_querystring=True,
        body='{"features": [{"name": "first"}, {"name": "second"}]}', status=200,
        content_type='application/json')

    runner = CliRunner()
    result = runner.invoke(
        main_group,
        ['--access-token', 'bogus', 'geocoding', '--limit', '2', '--features',
         '--forward', '1600 pennsylvania ave nw'],
        catch_exceptions=False)
    assert result.exit_code == 0
    assert result.output == '{"name": "first"}\n{"name": "second"}\n' 
示例22
def test_los_angeles(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/timezone/json",
            body='{"status":"OK"}',
            status=200,
            content_type="application/json",
        )

        ts = 1331766000
        timezone = self.client.timezone((39.603481, -119.682251), ts)
        self.assertIsNotNone(timezone)

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/timezone/json"
            "?location=39.603481,-119.682251&timestamp=%d"
            "&key=%s" % (ts, self.key),
            responses.calls[0].request.url,
        ) 
示例23
def test_los_angeles_with_no_timestamp(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/timezone/json",
            body='{"status":"OK"}',
            status=200,
            content_type="application/json",
        )

        timezone = self.client.timezone((39.603481, -119.682251))
        self.assertIsNotNone(timezone)

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/timezone/json"
            "?location=39.603481,-119.682251&timestamp=%d"
            "&key=%s" % (1608, self.key),
            responses.calls[0].request.url,
        ) 
示例24
def test_autocomplete_query(self):
        url = "https://maps.googleapis.com/maps/api/place/queryautocomplete/json"
        responses.add(
            responses.GET,
            url,
            body='{"status": "OK", "predictions": []}',
            status=200,
            content_type="application/json",
        )

        self.client.places_autocomplete_query("pizza near New York")

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "%s?input=pizza+near+New+York&key=%s" % (url, self.key),
            responses.calls[0].request.url,
        ) 
示例25
def test_elevation_single(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/elevation/json",
            body='{"status":"OK","results":[]}',
            status=200,
            content_type="application/json",
        )

        results = self.client.elevation((40.714728, -73.998672))

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/elevation/json?"
            "locations=enc:abowFtzsbM&key=%s" % self.key,
            responses.calls[0].request.url,
        ) 
示例26
def test_elevation_single_list(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/elevation/json",
            body='{"status":"OK","results":[]}',
            status=200,
            content_type="application/json",
        )

        results = self.client.elevation([(40.714728, -73.998672)])

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/elevation/json?"
            "locations=enc:abowFtzsbM&key=%s" % self.key,
            responses.calls[0].request.url,
        ) 
示例27
def test_elevation_along_path(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/elevation/json",
            body='{"status":"OK","results":[]}',
            status=200,
            content_type="application/json",
        )

        path = [(40.714728, -73.998672), (-34.397, 150.644)]

        results = self.client.elevation_along_path(path, 5)

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/elevation/json?"
            "path=enc:abowFtzsbMhgmiMuobzi@&"
            "key=%s&samples=5" % self.key,
            responses.calls[0].request.url,
        ) 
示例28
def test_short_latlng(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/elevation/json",
            body='{"status":"OK","results":[]}',
            status=200,
            content_type="application/json",
        )

        results = self.client.elevation((40, -73))

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/elevation/json?"
            "locations=40,-73&key=%s" % self.key,
            responses.calls[0].request.url,
        ) 
示例29
def test_mixed_params(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/distancematrix/json",
            body='{"status":"OK","rows":[]}',
            status=200,
            content_type="application/json",
        )

        origins = ["Bobcaygeon ON", [41.43206, -81.38992]]
        destinations = [
            (43.012486, -83.6964149),
            {"lat": 42.8863855, "lng": -78.8781627},
        ]

        matrix = self.client.distance_matrix(origins, destinations)

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/distancematrix/json?"
            "key=%s&origins=Bobcaygeon+ON%%7C41.43206%%2C-81.38992&"
            "destinations=43.012486%%2C-83.6964149%%7C42.8863855%%2C"
            "-78.8781627" % self.key,
            responses.calls[0].request.url,
        ) 
示例30
def test_simple_geocode(self):
        responses.add(
            responses.GET,
            "https://maps.googleapis.com/maps/api/geocode/json",
            body='{"status":"OK","results":[]}',
            status=200,
            content_type="application/json",
        )

        results = self.client.geocode("Sydney")

        self.assertEqual(1, len(responses.calls))
        self.assertURLEqual(
            "https://maps.googleapis.com/maps/api/geocode/json?"
            "key=%s&address=Sydney" % self.key,
            responses.calls[0].request.url,
        )