python类GET的实例源码

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(
            httpretty.GET, "http://test.gf/users",
            body=ujson.dumps([
                sam_profile,
                jack_profile,
            ]),
            content_type="application/json"
        )
        httpretty.register_uri(
            httpretty.GET, "http://test.gf/blog/sam",
            body=ujson.dumps(sams_articles),
            content_type="application/json"
        )
        httpretty.register_uri(
            httpretty.GET, "http://test.gf/blog/jack",
            body=ujson.dumps(jacks_articles),
            content_type="application/json"
        )

def test_weibo_request(self):
        """
        :return:
        """

        body = """
        {"next_cursor": 4098746105417334, "hasvisible": false, "uve_blank": -1, "statuses": ["4098748692646689",
        "4098748689403910", "4098748688918202", "4098747640117551", "4098747434341769", "4098747430814353",
        "4098747430145399", "4098747052681634", "4098746357066702", "4098746322855226", "4098746297959121",
        "4098746251989195", "4098746226771596", "4098746205413046", "4098746180805829", "4098746175979920",
         "4098746172045575", "4098746172045403", "4098746160243131", "4098746139219122"], "ad": [],
         "advertises": [], "interval": 0, "previous_cursor": 0, "total_number": 1322}
        """
        httpretty.register_uri(
            httpretty.GET, "https://api.weibo.com/2/statuses/friends_timeline/ids.json",
            body=body,
            status=200,
            content_type='text/json'
        )

        self.assertDictEqual(self.client.get("statuses/friends_timeline/ids.json"), json.loads(body))

def test_weibo_api_error(self):
        """
        :return:
        """
        body = """
                {
                        "request": "/statuses/home_timeline.json",
                        "error_code": "20502",
                        "error": "Need you follow uid."
                }
                """
        httpretty.register_uri(
            httpretty.GET, "https://api.weibo.com/2/statuses/home_timeline.json",
            body=body,
            status=200,
            content_type='text/json'
        )

        self.assertRaises(WeiboAPIError, self.client.get, "statuses/home_timeline.json")

def test_project_edit(connection, project, project_payload):
    """Verifies project renaming."""
    httpretty.register_uri(
        httpretty.POST, '{}/project/{}/edit'.format(
            matchlight.MATCHLIGHT_API_URL_V2,
            project.upload_token,
        ),
        body='{}', status=200)
    new_name = 'Test Project 1'
    connection.projects.edit(project, new_name)
    assert project.name == new_name

    new_name = 'Test Project 2'
    httpretty.register_uri(
        httpretty.GET, '{}/project/{}'.format(
            matchlight.MATCHLIGHT_API_URL_V2,
            project.upload_token,
        ),
        body=json.dumps(project_payload),
        content_type='application/json', status=200)
    project = connection.projects.edit(project.upload_token, new_name)
    assert project.name == new_name

def test_project_get(connection, project_payload, project):
    """Verifies project retrieval."""
    httpretty.register_uri(
        httpretty.GET, '{}/project/{}'.format(
            matchlight.MATCHLIGHT_API_URL_V2,
            project.upload_token),
        responses=[
            httpretty.Response(body=json.dumps(project_payload),
                               content_type='application/json',
                               status=200),
            httpretty.Response(body='{}', content_type='application/json',
                               status=404),
            httpretty.Response(body='{}', content_type='application/json',
                               status=500),
        ])

    project_ = connection.projects.get(project.upload_token)
    assert project.upload_token == project_.upload_token

    project_ = connection.projects.get(project.upload_token)
    assert project_ is None

    with pytest.raises(matchlight.error.ConnectionError):
        connection.projects.get(project.upload_token)

def test_alert_filter_record(connection, alert, alert_payload,
                             document_record):
    """Verifies alert filtering on 'record_id'."""
    httpretty.register_uri(
        httpretty.GET, '{}/alerts?record_id_filter={}&limit=50'.format(
            matchlight.MATCHLIGHT_API_URL_V2,
            document_record.id
        ),
        body=json.dumps({'alerts': [alert_payload]}),
        content_type='application/json',
        status=200
    )

    alerts = connection.alerts.filter(limit=50, record=document_record)
    assert len(alerts) == 1
    assert alerts[0].id == alert_payload['id']

def test_alert_details(connection, alert, alert_details_pii_payload):
    """Verifies alert get details responses."""
    httpretty.register_uri(
        httpretty.GET, '{}/alert/{}/details'.format(
            matchlight.MATCHLIGHT_API_URL_V2,
            alert.id
        ),
        body=json.dumps(alert_details_pii_payload),
        content_type='application/json',
        status=200
    )

    details_ = connection.alerts.get_details(alert)
    assert details_ == alert_details_pii_payload

    details_ = connection.alerts.get_details(alert.id)
    assert details_ == alert_details_pii_payload

def test_usa_jobs_updater_deduplicated_items():
    httpretty.register_uri(
        httpretty.GET,
        USAJobsUpdater.base_url,
        body=callback
    )
    updater = USAJobsUpdater(
        auth_key=FAKE_KEY,
        key_email=FAKE_EMAIL,
    )
    postings = updater.deduplicated_postings()
    assert len(postings.keys()) == 3
    assert sorted(postings.keys()) == [5, 6, 7]
    assert postings == {
        5: {'k': 'v'},
        6: {'k': 'v2'},
        7: {'k': 'v3'},
    }

def test_place_ua():
    httpretty.register_uri(
        httpretty.GET,
        URL,
        body=RESPONSE,
        content_type='text/csv'
    )

    results = place_ua.__wrapped__(city_cleaner)
    assert results == {
        'LA': {'abbeville': '00037', 'ama': '62677'},
        'MD': {'aberdeen': '00199'},
        'NM': {'placitas': '01171'},
        'MO': {'st louis': '77770', 'grain valley': '43912'},
        'NC': {'winston-salem': '96670', 'winston': '96670', 'salem': '96670'},
        'ID': {'boise': '08785', 'boise city': '08785'}
    }

def test_ua_cbsa():
    httpretty.register_uri(
        httpretty.GET,
        URL,
        body=RESPONSE,
        content_type='text/csv'
    )

    results = ua_cbsa.__wrapped__()
    assert results == {
        '00037': [
            ('10020', 'Abbeville, LA Micro Area'),
            ('35340', 'New Iberia, LA Micro Area'),
        ],
        '00091': [('48140', 'Wausau, WI Metro Area')]
    }

def test_get_doc_with_no_doc_name(self):
        httpretty.register_uri(
            httpretty.GET,
            "http://example.com/api/resource/SomeDoc/",
            body='{"data": { "f1": "v1","f2": "v2"}}',
            content_type="application/json"
        )
        res = self.frappe.get_doc(
            "SomeDoc",
            filters=[["Note", "title", "LIKE", "S%"]],
            fields=["name", "foo"])

        self.assertEquals(res, {'f1': 'v1', 'f2': 'v2'})

        request = httpretty.last_request()
        url = urlparse.urlparse(request.path)
        query_dict = urlparse.parse_qs(url.query)
        self.assertEquals(query_dict['fields'],
                          [u'["name", "foo"]'])
        self.assertEquals(query_dict['filters'],
                          [u'[["Note", "title", "LIKE", "S%"]]'])

def test_httpretty_bypasses_a_unregistered_request(context):
    "httpretty should bypass a unregistered request by disabling it"

    httpretty.register_uri(
        httpretty.GET, "http://localhost:{0}/go-for-bubbles/".format(context.http_port),
        body="glub glub")

    fd = urllib2.urlopen('http://localhost:{0}/go-for-bubbles/'.format(context.http_port))
    got1 = fd.read()
    fd.close()

    expect(got1).to.equal(b'glub glub')

    fd = urllib2.urlopen('http://localhost:{0}/come-again/'.format(context.http_port))
    got2 = fd.read()
    fd.close()

    expect(got2).to.equal(b'<- HELLO WORLD ->')
    core.POTENTIAL_HTTP_PORTS.remove(context.http_port)

def test_disallow_net_connect_1(context):
    """
    When allow_net_connect = False, a request that otherwise
    would have worked results in UnmockedError.
    """
    httpretty.register_uri(httpretty.GET, "http://falcao.it/foo/",
                           body="BAR")

    def foo():
        fd = None
        try:
            fd = urllib2.urlopen('http://localhost:{0}/go-for-bubbles/'.format(context.http_port))
        finally:
            if fd:
                fd.close()

    foo.should.throw(httpretty.UnmockedError)

def mock_user_info_response(self, status=200, username=None):
        """ Mock the user info endpoint response of the OAuth2 provider. """

        username = username or USER_INFO['username']
        data = {
            'family_name': USER_INFO['last_name'],
            'preferred_username': username,
            'given_name': USER_INFO['first_name'],
            'email': USER_INFO['email'],
        }

        httpretty.register_uri(
            httpretty.GET,
            OAUTH2_USER_INFO_URL,
            body=json.dumps(data),
            content_type='application/json',
            status=status
        )

def test_items(self):
        dwarves = [{'name': 'Dori'}, {'name': 'Nori'}, {'name': 'Ori'},
                   {'name': 'Bifur'}, {'name': 'Bofur'}, {'name': 'Bombur'}]

        def request_callback(request, uri, headers):
            if uri.endswith('dwarves') or uri.endswith('?page=1'):
                items = dwarves[:3]
                page_id = 1
            elif uri.endswith('?page=2'):
                items = dwarves[3:]
                page_id = 2
            else:
                items = []
                page_id = uri.split('=')[-1]
            return 200, headers, json.dumps({'page': page_id, 'dwarves': items, 'total_pages': 2})

        self.register_url(self.GET, 'dwarves', body=request_callback)

        api = QuizletAPI()
        self.assertEqual(list(api.dwarves.items()), dwarves)

def test_request_timeout():
    timeout = 0.1
    http_scheme = 'http'
    host = 'coordinator'
    port = 8080
    url = http_scheme + '://' + host + ':' + str(port) + constants.URL_STATEMENT_PATH

    def long_call(request, uri, headers):
        time.sleep(timeout * 2)
        return (200, headers, "delayed success")

    httpretty.enable()
    for method in [httpretty.POST, httpretty.GET]:
        httpretty.register_uri(method, url, body=long_call)

    # timeout without retry
    for request_timeout in [timeout, (timeout, timeout)]:
        req = PrestoRequest(
            host=host,
            port=port,
            user='test',
            http_scheme=http_scheme,
            max_attempts=1,
            request_timeout=request_timeout,
        )

        with pytest.raises(requests.exceptions.Timeout):
            req.get(url)

        with pytest.raises(requests.exceptions.Timeout):
            req.post('select 1')

    httpretty.disable()
    httpretty.reset()

def setUp(self):
        with open("line_file.json", "w") as f:
            f.write(JSON_LINE_CONTENT)
        with open("array_file.json", "w") as f:
            f.write(JSON_ARRAY_CONTENT)
        with open("object_file.json", "w") as f:
            f.write(JSON_OBJECT_CONTENT)
        with open("block_file.json", "w") as f:
            f.write(JSON_BLOCK_CONTENT)

        httpretty.enable()

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/line",
            body=JSON_LINE_CONTENT,
            content_type="text/plain"
        )

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/array",
            body=JSON_ARRAY_CONTENT,
            content_type="application/json"
        )

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/object",
            body=JSON_OBJECT_CONTENT,
            content_type="application/json"
        )

def setUp(self):
        with open("test.csv", "w") as f:
            f.write(CSV_CONTENT)

        httpretty.enable()

        httpretty.register_uri(
            httpretty.GET,
            "http://test.gf/csv",
            body=CSV_CONTENT,
            content_type="text/plain"
        )

def test_req(self):
        http_server_fixture = HttpServerFixture()
        self.useFixture(http_server_fixture)
        context = self.workflow_context()
        result = Req(
            "GET",
            "http://test.gf/users",
            parser=_default_parser
        )(context, [])
        self.assertEquals(result, [sam_profile, jack_profile])

def test_graylog_api_search(self):
        httpretty.register_uri(
            httpretty.GET, "http://dummyhost:80/api/search/universal/absolute",
            body=self.generate_search_result(),
            content_type="application/json"
        )

        # More of some dummy tests now
        sr = api.SearchRange("10 minutes ago", arrow.now())
        q = api.SearchQuery(sr)
        result = self.api.search(q)
        self.assertEquals(len(result.messages), 1)
        self.assertEquals("*", result.query)

        q = api.SearchQuery(sr, fields=["level", "module", "message", "timestamp"], sort="level", ascending=True)
        qq = q.copy_with_range(sr)
        result = self.api.search(qq)
        self.assertEquals(len(result.messages), 1)
        self.assertEquals("*", result.query, )

        q = api.SearchQuery(sr, fields=["level", "module", "message", "timestamp"], sort="level", ascending=False)
        result = self.api.search(q)
        self.assertEquals(len(result.messages), 1)
        self.assertEquals("*", result.query)

        result = self.api.search(q, fetch_all=True)
        self.assertEquals(len(result.messages), 1)
        self.assertEquals("*", result.query)

def test_to_many_results(self):
        httpretty.register_uri(
            httpretty.GET, "http://dummyhost:80/api/search/universal/absolute",
            body=self.generate_search_result(1000000),
            content_type="application/json"
        )

        sr = api.SearchRange("10 minutes ago", arrow.now())
        q = api.SearchQuery(sr)

        with self.assertRaises(RuntimeError):
            self.api.search(q, fetch_all=True)

def test_userinfo(self):
        httpretty.register_uri(
            httpretty.GET, "http://dummyhost:80/api/users/dummy",
            body='{"someuser" : "info" }',
            content_type="application/json"
        )

        result = self.api.user_info()
        self.assertEquals({"someuser": "info"}, result)

def test_streams(self):
        httpretty.register_uri(
            httpretty.GET, "http://dummyhost:80/api/streams/enabled",
            body='[{"somestream" : "a" }]',
            content_type="application/json"
        )

        result = self.api.streams()
        self.assertEquals([{"somestream": "a"}], result)

def test_saved_searches(self):
        httpretty.register_uri(
            httpretty.GET, "http://dummyhost:80/api/search/saved",
            body='{"searches" : [{"query": {"query": "level:INFO", "fields":"timestamp,message"}}]}',
            content_type="application/json"
        )

        saved_searches = self.api.get_saved_queries()
        self.assertEquals("level:INFO", saved_searches['searches'][0]['query']['query'])
        self.assertEquals("timestamp,message", saved_searches['searches'][0]['query']['fields'])

def register_endpoint(endpoint, body, status=200, method=httpretty.GET):
    """Mock GitHub API response."""
    httpretty.register_uri(
        method,
        'https://api.github.com%s' % endpoint,
        body=json.dumps(body),
        status=status,
    )

def register_local_endpoint(endpoint, body, status=200, method=httpretty.GET):
    """Mock GitHub API response."""
    httpretty.register_uri(
        method,
        'https://api.github.com%s' % endpoint,
        body=json.dumps(body),
        status=status,
    )


#
# Fixture generators
#

def tfs_server_mock():
    for method in (httpretty.GET, httpretty.POST, httpretty.PATCH):
        httpretty.register_uri(method, re.compile(r"http://tfs.tfs.ru(.*)"),
                               body=request_callback_get,
                               content_type="application/json")

def test_download(self):
        # Onedrive requires the path in the URL, we use the chunk's uid as
        # path.
        httpretty.register_uri(
            httpretty.GET,
            OnedriveAPIClient.DOWNLOAD_URL[1].format(path=self._get_path()),
            body=TEST_CHUNK_BODY)

        self.assertEqual(TEST_CHUNK_BODY, self.client.download(self.chunk))

def test_download(self):
        # Box requires the file id in the URL, the file_id is assigned by Box,
        # and therefore is stored in ChunkStorage.attrs.
        httpretty.register_uri(
            httpretty.GET,
            BoxAPIClient.DOWNLOAD_URL[1].format(file_id='abc123'),
            body=TEST_CHUNK_BODY)

        self.assertEqual(TEST_CHUNK_BODY, self.client.download(self.chunk))

def test_download(self):
        # GDrive requires the file id in the URL, the file_id is assigned by
        # Google, and therefore is stored in ChunkStorage.attrs.
        httpretty.register_uri(
            httpretty.GET,
            GDriveAPIClient.DOWNLOAD_URL[1].format(file_id='abc123'),
            body=TEST_CHUNK_BODY)

        self.assertEqual(TEST_CHUNK_BODY, self.client.download(self.chunk))