python类POST的实例源码

def test_auth_access(self):
        """
        :return:
        """

        body = """
           {
               "access_token": "ACCESS_TOKEN",
               "expires_in": 1234,
               "remind_in":"798114",
               "uid":"12341234"
         }
        """

        httpretty.register_uri(
            httpretty.POST, "https://api.weibo.com/oauth2/access_token",
            body=body,
            status=200,
            content_type='text/json'
        )

        response = self.oauth2_client.auth_access("abcde")

        self.assertDictEqual(response, json.loads(body))

def test_revoke_auth_access(self):
        """
        :return:
        """

        body = '{"result":true}'

        httpretty.register_uri(
            httpretty.POST, "https://api.weibo.com/oauth2/revokeoauth2",
            body=body,
            status=200,
            content_type='text/json'
        )

        response = self.oauth2_client.revoke_auth_access(self.access_token)
        self.assertTrue(response)

def test_refresh_token(self):
        """
        :return:
        """

        body = """
         {
            "access_token": "SlAV32hkKG",
            "expires_in": 3600
        }
        """

        httpretty.register_uri(
            httpretty.POST, "https://api.weibo.com/oauth2/access_token",
            body=body,
            status=200,
            content_type='text/json'
        )

        response = self.oauth2_client.refresh_token("xxxxxxxx")
        self.assertTrue(response, json.loads(body))

def test_get_token_info(self):
        """
        :return:
        """
        body = """
          {
           "uid": 1073880650,
           "appkey": 1352222456,
           "scope": null,
           "create_at": 1352267591,
           "expire_in": 157679471
         }
        """

        httpretty.register_uri(
            httpretty.POST, "https://api.weibo.com/oauth2/get_token_info",
            body=body,
            status=200,
            content_type='text/json'
        )

        response = self.oauth2_client.get_token_info(self.access_token)

        self.assertDictEqual(response, json.loads(body))

def test_assert_api_error(self):
        """
        :return:
        """
        body = """
        {
                "error": "unsupported_response_type",
                "error_code": 21329,
                "error_description": "Unsupport ResponseType."
            }
        """

        httpretty.register_uri(
            httpretty.POST, "https://api.weibo.com/oauth2/get_token_info",
            body=body,
            status=200,
            content_type='text/json'
        )

        self.assertRaises(WeiboOauth2Error, self.oauth2_client.get_token_info, "dbdsds")

def test_record_add_document(min_score, connection, project, document):
    """Verifies adding document records to a project."""
    httpretty.register_uri(
        httpretty.POST, '{}/records/upload/document/{}'.format(
            matchlight.MATCHLIGHT_API_URL_V2, project.upload_token),
        body=json.dumps({
            'id': uuid.uuid4().hex,
            'name': 'name',
            'description': '',
            'ctime': time.time(),
            'mtime': time.time(),
            'metadata': '{}',
        }),
        content_type='application/json', status=200)
    connection.records.add_document(
        project=project,
        name=document['name'],
        description=document['description'],
        document_path=DOCUMENT_RECORD_PATH,
        min_score=min_score)

def test_feed_counts(connection, feed, start_time, end_time):
    """Verifies requesting feed counts."""
    expected = {
        int(time.time()): random.randint(0, 1000),
        int(time.time()): random.randint(0, 1000),
        int(time.time()): random.randint(0, 1000),
        int(time.time()): random.randint(0, 1000),
    }
    httpretty.register_uri(
        httpretty.POST, '{}/feeds/{}'.format(
            matchlight.MATCHLIGHT_API_URL_V2,
            feed.name),
        body=json.dumps(expected),
        content_type='application/json', status=200)
    result = connection.feeds.counts(feed, start_time, end_time)
    assert result == connection.feeds._format_count(expected)

    result = connection.feeds.counts(feed.name, start_time, end_time)
    assert result == connection.feeds._format_count(expected)

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_delete(connection, project, project_payload):
    """Verifies project deletion."""
    httpretty.register_uri(
        httpretty.POST, '{}/project/{}/delete'.format(
            matchlight.MATCHLIGHT_API_URL_V2,
            project.upload_token,
        ),
        body='{}', status=200)
    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)
    connection.projects.delete(project)

def test_search_with_results():
    httpretty.register_uri(
        httpretty.POST, 'https://pypi.python.org/pypi',
        body=RESULTS_RESPONSE,
        content_type='text/xml'
    )

    app = Application()

    command = app.find('search')
    command_tester = CommandTester(command)
    command_tester.execute([
        ('command', command.get_name()),
        ('tokens', ['pendulum'])
    ])

    output = command_tester.get_display()

    assert '\npendulum (0.6.0)\n Python datetimes made easy.\n' == output

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 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):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.DOWNLOAD_URL[1],
            body=TEST_CHUNK_BODY)

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

def test_upload(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.UPLOAD_URL[1],
            body='')

        self.client.upload(self.chunk, TEST_CHUNK_BODY)

def test_delete(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.DELETE_URL[1],
            body=TEST_CHUNK_BODY)

        self.client.delete(self.chunk)

def test_upload(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.POST,
            BoxAPIClient.UPLOAD_URL[1].format(file_id='abc123'),
            body='{"entries": [{"id":1}]}', content_type='application/json')

        self.client.upload(self.chunk, TEST_CHUNK_BODY)

def test_upload(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.POST,
            GDriveAPIClient.UPLOAD_URL[1].format(file_id='abc123'),
            body='{"id":1234}')

        self.client.upload(self.chunk, TEST_CHUNK_BODY)

def setUp(self):
        super().setUp()
        httpretty.register_uri(
            METHODS.get(self.oauth2_client.USER_PROFILE_URL[0]),
            self.oauth2_client.USER_PROFILE_URL[1],
            body=json.dumps(USER_PROFILE_BOX),
            content_type='application/json')
        httpretty.register_uri(httpretty.POST, self.oauth2_client.CREATE_URL,
                               body=json.dumps(CREATE_BOX),
                               content_type='application/json')

def test_download(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.DOWNLOAD_URL[1],
            body=TEST_CHUNK_BODY)

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

def test_upload(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.UPLOAD_URL[1],
            body='')

        self.client.upload(self.chunk, TEST_CHUNK_BODY)

def test_delete(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.DELETE_URL[1],
            body=TEST_CHUNK_BODY)

        self.client.delete(self.chunk)

def test_upload(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.POST,
            BoxAPIClient.UPLOAD_URL[1].format(file_id='abc123'),
            body='{"entries": [{"id":1}]}', content_type='application/json')

        self.client.upload(self.chunk, TEST_CHUNK_BODY)

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(
            httpretty.POST, self.oauth2_client.ACCESS_TOKEN_URL,
            body=json.dumps(ACCESS_TOKEN),
            content_type='application/json')

def setUp(self):
        super().setUp()
        httpretty.register_uri(
            METHODS.get(self.oauth2_client.USER_PROFILE_URL[0]),
            self.oauth2_client.USER_PROFILE_URL[1],
            body=json.dumps(USER_PROFILE_BOX),
            content_type='application/json')
        httpretty.register_uri(httpretty.POST, self.oauth2_client.CREATE_URL,
                               body=json.dumps(CREATE_BOX),
                               content_type='application/json')

def test_download(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.DOWNLOAD_URL[1],
            body=TEST_CHUNK_BODY)

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

def test_upload(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.UPLOAD_URL[1],
            body='')

        self.client.upload(self.chunk, TEST_CHUNK_BODY)

def test_delete(self):
        httpretty.register_uri(
            httpretty.POST, DropboxAPIClient.DELETE_URL[1],
            body=TEST_CHUNK_BODY)

        self.client.delete(self.chunk)

def test_upload(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.POST,
            GDriveAPIClient.UPLOAD_URL[1].format(file_id='abc123'),
            body='{"id":1234}')

        self.client.upload(self.chunk, TEST_CHUNK_BODY)

def setUp(self):
        httpretty.enable()
        httpretty.register_uri(
            httpretty.POST, self.oauth2_client.ACCESS_TOKEN_URL,
            body=json.dumps(ACCESS_TOKEN),
            content_type='application/json')

def setUp(self):
        super().setUp()
        httpretty.register_uri(
            METHODS.get(self.oauth2_client.USER_PROFILE_URL[0]),
            self.oauth2_client.USER_PROFILE_URL[1],
            body=json.dumps(USER_PROFILE_BOX),
            content_type='application/json')
        httpretty.register_uri(httpretty.POST, self.oauth2_client.CREATE_URL,
                               body=json.dumps(CREATE_BOX),
                               content_type='application/json')