Python httpretty 模块，register_uri() 实例源码

def test_get_saved_queries(mocked_url, mocked_rw_apikey, mocked_account_resource_id, capsys):
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.GET, MOCK_API_URL, status=200, content_type='application/json',
                           body=json.dumps({'saved_queries': [SAVED_QUERY_RESPONSE]}))

    api.get_saved_query()
    out, err = capsys.readouterr()

    assert "Name:" in out
    assert "Logs:" in out
    assert "ID:" in out
    assert "Statement:" in out
    assert "Time range:" in out
    assert "From:" in out
    assert "To:" in out

def test_get_saved_query(mocked_url, mocked_rw_apikey, mocked_account_resource_id, capsys):
    saved_query_id = str(uuid.uuid4())
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.GET, MOCK_API_URL, status=200,
                           content_type='application/json',
                           body=json.dumps({'saved_query': SAVED_QUERY_RESPONSE}))

    api.get_saved_query(saved_query_id)
    out, err = capsys.readouterr()

    assert "Name:" in out
    assert "Logs:" in out
    assert "ID:" in out
    assert "Statement:" in out
    assert "Time range:" in out
    assert "From:" in out
    assert "To:" in out

def test_failing_patch_saved_query(mocked_url, mocked_rw_apikey, mocked_account_resource_id,
                                   capsys):
    test_saved_query_id = str(uuid.uuid4())
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.PATCH, MOCK_API_URL, status=400,
                           content_type='application/json',
                           body=json.dumps(SAVED_QUERY_ERROR_RESPONSE))

    api.update_saved_query(
        test_saved_query_id, name="new_query_name", statement="new_statement", from_ts=123,
        to_ts=123456, time_range="last 10 days")
    out, err = capsys.readouterr()

    assert "Invalid field: time_range\n" in out
    assert "Message: Invalid query: time_range cannot be specified with from and/or to fields" in out

def test_rename_log(mocked_url, mocked_rw_apikey, mocked_ro_apikey, capsys):
    test_log_id = str(uuid.uuid4())
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = ID_WITH_VALID_LENGTH
    mocked_ro_apikey.return_value = ID_WITH_VALID_LENGTH

    request_body = '{"log": {"name": "test.log", "logsets_info": [], "source_type": "token"}}'
    expected_result = '{"log": {"name": "new_test_log_name", "logsets_info": [], "source_type": "token"}}'

    httpretty.register_uri(httpretty.GET, MOCK_API_URL,
                           status=200,
                           content_type='application/json',
                           body=request_body)

    httpretty.register_uri(httpretty.PUT, MOCK_API_URL, status=200,
                           body = expected_result, content_type='application/json')

    new_name_for_log = "new_test_log_name"
    api.rename_log(test_log_id, new_name_for_log)
    out, err = capsys.readouterr()

    assert new_name_for_log in out

def test_delete_api_key(mocked_url, mocked_owner_apikey, mocked_owner_apikey_id,
                        mocked_account_resource_id, capsys):
    api_key_id = str(uuid.uuid4())
    mocked_url.return_value = '', MOCK_API_URL + '/' + api_key_id
    mocked_owner_apikey.return_value = str(uuid.uuid4())
    mocked_owner_apikey_id.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.DELETE, MOCK_API_URL + '/' + api_key_id,
                           status=204,
                           content_type='application/json')

    api.delete(api_key_id)

    out, err = capsys.readouterr()
    assert not err
    assert 'Deleted api key with id: %s' % api_key_id in out

def test_enable_api_key(mocked_url, mocked_owner_apikey, mocked_owner_apikey_id,
                        mocked_account_resource_id, capsys):
    api_key_id = str(uuid.uuid4())
    mocked_url.return_value = '', MOCK_API_URL
    mocked_owner_apikey.return_value = str(uuid.uuid4())
    mocked_owner_apikey_id.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.PATCH, MOCK_API_URL,
                           status=200,
                           content_type='application/json',
                           body=json.dumps({}))

    api.update(api_key_id, True)

    out, err = capsys.readouterr()
    assert {'apikey': {'active': True}} == json.loads(httpretty.last_request().body)
    assert not err
    assert 'Enabled api key with id: %s' % api_key_id in out

def test_post_query_with_time(mocked_url, mocked_generate_headers, capsys):
    setup_httpretty()
    mocked_url.return_value = '', MOCK_API_URL

    httpretty.register_uri(httpretty.POST, MOCK_API_URL,
                           content_type='application/json',
                           body=json.dumps(SAMPLE_EVENTS_RESPONSE))
    api.query(query_string='foo', log_keys='foo', time_from='123456', time_to='123456')

    out, err = capsys.readouterr()

    assert mocked_generate_headers.called
    assert "Message contents1" in out
    assert "Message contents2" in out
    assert "Message contents3" in out

    teardown_httpretty()

def test_post_query_with_relative_range(mocked_url, mocked_generate_headers, capsys):
    setup_httpretty()
    mocked_url.return_value = '', MOCK_API_URL
    httpretty.register_uri(httpretty.POST, MOCK_API_URL,
                           content_type='application/json',
                           body=json.dumps(SAMPLE_EVENTS_RESPONSE))
    api.query(query_string='foo', log_keys='foo', relative_time_range='last 3 min')

    out, err = capsys.readouterr()

    assert mocked_generate_headers.called
    assert "Message contents1" in out
    assert "Message contents2" in out
    assert "Message contents3" in out

    teardown_httpretty()

def test_handle_response(capsys):
    setup_httpretty()

    httpretty.register_uri(httpretty.GET, MOCK_API_URL, content_type='application/json',
                           body=json.dumps(SAMPLE_EVENTS_RESPONSE))
    resp = requests.get(MOCK_API_URL)

    api.handle_response(resp, Mock())

    out, err = capsys.readouterr()

    assert "Message contents1" in out
    assert "Message contents2" in out
    assert "Message contents3" in out

    teardown_httpretty()

def test_create_logset_from_file(mocked_url, mocked_rw_apikey, capsys):
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())

    httpretty.register_uri(httpretty.POST, MOCK_API_URL, status=201,
                           content_type='application/json',
                           body=json.dumps(LOGSET_RESPONSE))

    params = {
        "logset": {
            "name": "Test Logset"
        }
    }

    api.create_logset(params=params)
    out, err = capsys.readouterr()

    assert 'Test Logset' in out

def test_create_logset_invalid_json(mocked_url, mocked_rw_apikey, capsys):
    with pytest.raises(SystemExit) as exit:
        mocked_url.return_value = '', MOCK_API_URL
        mocked_rw_apikey.return_value = str(uuid.uuid4())

        httpretty.register_uri(httpretty.POST, MOCK_API_URL, status=400,
                               content_type='application/json',
                               body='Client Error: Bad Request for url: https://rest.logentries.com/management/logsets')

        invalid_params = {
            "logset": {
                "id": "12341234-XXXX-YYYY-XXXX-12341234",
                "unknown_field": "unknown value"
            }
        }

        api.create_logset(params=invalid_params)
        out, err = capsys.readouterr()

        assert exit.code is 1
        assert "Creating logset failed, status code: 400" in out

def test_rename_logset(mocked_url, mocked_rw_apikey, mocked_ro_apikey, capsys):
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_ro_apikey.return_value = str(uuid.uuid4())

    response_body = '{"logset": {"id": "XXXXXXXX-XXXX-YYYY-XXXX-XXXXXXXX","logs_info": [],"name": "old logset name"}}'
    expected_result = '{"logset": {"id": "XXXXXXXX-XXXX-YYYY-XXXX-XXXXXXXX","logs_info": [],"name": "new logset name"}}'

    httpretty.register_uri(httpretty.GET, MOCK_API_URL, status=200,
                           content_type='application/json',
                           body=response_body)

    httpretty.register_uri(httpretty.PUT, MOCK_API_URL, status=200,
                           content_type='application/json',
                           body=expected_result)

    api.rename_logset('XXXXXXXX-XXXX-YYYY-XXXX-XXXXXXXX', 'new logset name')
    out, err = capsys.readouterr()

    assert "new logset name" in out

def test_add_unknown_log_to_logset(mocked_url, mocked_ro_apikey, mocked_rw_apikey, capsys):
    with pytest.raises(SystemExit) as exit:
        mocked_url.return_value = '', MOCK_API_URL
        mocked_rw_apikey.return_value = str(uuid.uuid4())
        mocked_ro_apikey.return_value = str(uuid.uuid4())

        httpretty.register_uri(httpretty.GET, MOCK_API_URL, status=200,
                               content_type='application/json', body=json.dumps(LOGSET_RESPONSE))

        httpretty.register_uri(httpretty.PUT, MOCK_API_URL, status=400,
                               content_type='application/json')

        api.add_log('XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXX', 'unknown_log')
        out, err = capsys.readouterr()

        assert "400" in out
        assert exit.code is 1

def test_remove_log_from_logset(mocked_url, mocked_ro_apikey, mocked_rw_apikey, capsys):
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_ro_apikey.return_value = str(uuid.uuid4())

    httpretty.register_uri(httpretty.GET, MOCK_API_URL ,
                           status=200,
                           content_type='application/json',
                           body=json.dumps(LOGSET_RESPONSE))

    httpretty.register_uri(httpretty.PUT, MOCK_API_URL,
                           status=200,
                           content_Type='application/json',
                           body=json.dumps(LOGSET_RESPONSE))
    api.delete_log('123', str(uuid.uuid4()))
    out, err = capsys.readouterr()

    assert not err

def test_delete_logset_with_log_in_another_logset(mocked_url, mocked_rw_apikey, mocked_ro_apikey, capsys):
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_ro_apikey.return_value = str(uuid.uuid4())

    httpretty.register_uri(httpretty.DELETE, MOCK_API_URL,
                           status=204, content_type='application/json')

    httpretty.register_uri(httpretty.GET, MOCK_API_URL,
                           status=200, content_type='application/json',
                           body=json.dumps({}))

    api.delete_logset('123')
    api.get_logset('456')
    out, err = capsys.readouterr()

    assert not err

def test_delete_user_from_team(mocked_url, mocked_rw_apikey, mocked_account_resource_id, capsys):
    test_team_id = str(uuid.uuid4())
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.GET, MOCK_API_URL, status=200,
                           body=json.dumps({'team': TEAM_RESPONSE}),
                           content_type='application/json')
    httpretty.register_uri(httpretty.PUT, MOCK_API_URL, status=200,
                           content_type='application/json')

    user_id_to_add = str(uuid.uuid4())
    api.delete_user_from_team(test_team_id, user_id_to_add)
    out, err = capsys.readouterr()

    assert "Deleted user with key: '%s' from team: %s\n" % (user_id_to_add, test_team_id) == out

def test_submite_POST_request(self):

        httpretty.register_uri(httpretty.POST,
            FACEBOOK_MESSAGES_POST_URL + 'users/',
            body='{ \
                "name": "ben", \
                "age": 12 \
            }', status=201)

        client = HttpClient('123123')
        response, error = client.submit_request('users/', 
                                                'POST', None)

        if error is None:
            assert response['name'] == 'ben'
            assert response['age'] == 12
        else:
            raise

def test_send_text_message(self):

        httpretty.register_uri(httpretty.POST,
            FACEBOOK_MESSAGES_POST_URL + '/me/messages',
            body='{ \
                "recipient_id": 1008, \
                "message_id": "mid.1" \
            }', status=201)

        bot = Bot("abc")
        message = Message('image', 'google.com')          

        response, error = bot.send_message(message, '123abc')
        assert response['recipient_id'] == 1008
        assert response['message_id'] == "mid.1"
        assert error is None

def test_send_button_template_message(self):
        httpretty.register_uri(httpretty.POST,
            FACEBOOK_MESSAGES_POST_URL + '/me/messages',
            body='{ \
                "recipient_id": "1008", \
                "message_id": "mid.1" \
            }', status=201)

        bot = Bot("abc")
        buttons = []
        for i in range(3):
            buttons.append(Button('web_url', str(i), "http://ex.com"))

        message = Template(Template.button_type,
                        title='Title',
                        buttons=buttons,)

        bot.send_message(message, '123abc')

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_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_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 testCanDeleteOwnedDomain(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.')

        url = reverse('domain-detail', args=(self.ownedDomains[1].pk,))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
        self.assertEqual(httpretty.last_request().method, 'DELETE')
        self.assertEqual(httpretty.last_request().headers['Host'], 'nsmaster:8081')

        httpretty.reset()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.')

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))

def testCanPostReverseDomains(self):
        name = '0.8.0.0.0.1.c.a.2.4.6.0.c.e.e.d.4.4.0.1.a.0.1.0.8.f.4.0.1.0.a.2.ip6.arpa'

        httpretty.enable()
        httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones', status=201)
        httpretty.register_uri(httpretty.GET,
                               settings.NSLORD_PDNS_API + '/zones/' + name + '.',
                               body='{"rrsets": []}',
                               content_type="application/json")
        httpretty.register_uri(httpretty.GET,
                               settings.NSLORD_PDNS_API + '/zones/' + name + './cryptokeys',
                               body='[]',
                               content_type="application/json")

        url = reverse('domain-list')
        data = {'name': name}
        response = self.client.post(url, data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(len(mail.outbox), 0)

def testCanDeleteOwnedDynDomain(self):
        httpretty.enable()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')

        url = reverse('domain-detail', args=(self.ownedDomains[1].pk,))
        response = self.client.delete(url)
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)

        # FIXME In this testing scenario, the parent domain dedyn.io does not
        # have the proper NS and DS records set up, so we cannot test their
        # deletion.

        httpretty.reset()
        httpretty.register_uri(httpretty.DELETE, settings.NSLORD_PDNS_API + '/zones/' + self.ownedDomains[1].name + '.')
        httpretty.register_uri(httpretty.DELETE, settings.NSMASTER_PDNS_API + '/zones/' + self.ownedDomains[1].name+ '.')

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertTrue(isinstance(httpretty.last_request(), httpretty.core.HTTPrettyRequestEmpty))

def setUp(self):
        if not hasattr(self, 'owner'):
            self.username = utils.generateRandomString(12)
            self.password = utils.generateRandomString(12)
            self.user = utils.createUser(self.username, self.password)
            self.token = utils.createToken(user=self.user)
            self.setCredentials(self.username, self.password)
            self.url = reverse('dyndns12update')

            self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)
            self.domain = utils.generateDynDomainname()
            url = reverse('domain-list')
            data = {'name': self.domain}
            response = self.client.post(url, data)
            self.assertEqual(response.status_code, status.HTTP_201_CREATED)

            httpretty.enable()
            httpretty.register_uri(httpretty.POST, settings.NSLORD_PDNS_API + '/zones')
            httpretty.register_uri(httpretty.GET,
                                   settings.NSLORD_PDNS_API + '/zones/' + self.domain + '.',
                                   body='{"rrsets": []}',
                                   content_type="application/json")
            httpretty.register_uri(httpretty.PATCH, settings.NSLORD_PDNS_API + '/zones/' + self.domain + '.')
            httpretty.register_uri(httpretty.PUT, settings.NSLORD_PDNS_API + '/zones/' + self.domain + './notify')

def _test_harvest_create(self, url, content, content_type, **kwargs):

        # Mock the GET request to get the file
        httpretty.register_uri(httpretty.GET, url,
                               body=content, content_type=content_type)

        # The harvester will try to do a HEAD request first so we need to mock
        # this as well
        httpretty.register_uri(httpretty.HEAD, url,
                               status=405, content_type=content_type)

        harvest_source = self._create_harvest_source(url, **kwargs)

        self._run_full_job(harvest_source['id'], num_objects=2)

        # Check that two datasets were created
        fq = "+type:dataset harvest_source_id:{0}".format(harvest_source['id'])
        results = h.call_action('package_search', {}, fq=fq)

        eq_(results['count'], 2)
        for result in results['results']:
            assert result['title'] in ('Example dataset 1',
                                       'Example dataset 2')

def test_harvest_after_download_extension_point_gets_called(self):

        plugin = p.get_plugin('test_rdf_harvester')

        # Mock the GET request to get the file
        httpretty.register_uri(httpretty.GET, self.rdf_mock_url)

        # The harvester will try to do a HEAD request first so we need to mock
        # this as well
        httpretty.register_uri(httpretty.HEAD, self.rdf_mock_url,
                               status=405)

        harvest_source = self._create_harvest_source(self.rdf_mock_url)
        self._create_harvest_job(harvest_source['id'])
        self._run_jobs(harvest_source['id'])
        self._gather_queue(1)

        eq_(plugin.calls['after_download'], 1)

def test_room_members(self):
        ''' test room members '''
        stream_id = 'HNmksPVAR6-f14WqKXmqHX___qu8LMLgdA'
        # register response
        httpretty.register_uri(httpretty.GET, self.__uri__ + 'pod/v2/room/' + str(stream_id) + '/membership/list',
                               body='[ \
                                       { \
                                         "id": 7078106103900, \
                                         "owner": false, \
                                         "joinDate": 1461430710531 \
                                       }, \
                                       { \
                                         "id": 7078106103809, \
                                         "owner": true, \
                                         "joinDate": 1461426797875 \
                                       } \
                                     ]',
                               status=200,
                               content_type='text/json')
        status_code, response = self.pod.room_members(stream_id)
        assert status_code == 200
        data = json.loads(response)
        for member in data:
            if member['id'] == 7078106103900:
                assert member['owner'] is False

def test_stream_info(self):
        ''' test stream_info '''
        stream_id = 'p9B316LKDto7iOECc8Xuz3___qeWsc0bdA'
        httpretty.register_uri(httpretty.GET, self.__uri__ + 'pod/v1/streams/' + stream_id + '/info',
                               body='{ \
                                       "id": "p9B316LKDto7iOECc8Xuz3___qeWsc0bdA", \
                                       "crossPod": false, \
                                       "active": true, \
                                       "streamType": { \
                                         "type": "IM" \
                                       }, \
                                       "streamAttributes": { \
                                         "members": [ \
                                           7627861917905, \
                                           7627861925698 \
                                         ] \
                                       } \
                                     }',
                               status=200,
                               content_type='text/json')
        status_code, response = self.pod.stream_info(stream_id)
        assert status_code == 200
        data = json.loads(response)
        assert data['id'] == stream_id

def test_ib_group_member_add(self):
        ''' test pod.ib_group_member_add '''
        # register response
        httpretty.register_uri(httpretty.POST, self.__uri__ + "pod/v1/admin/group/87654/membership/add",
                               body='{ \
                                      "overallResult": "SUCCESS", \
                                      "results": [ \
                                         "" \
                                      ] \
                                     }',
                               status=200,
                               content_type='text/json')
        # run query
        status_code, response = self.pod.ib_group_member_add('87654', ['123457', '567890'])
        # verify return
        response = json.loads(response)
        assert status_code == 200
        assert response['overallResult'] == "SUCCESS"

def test_list_connections(self):
        ''' test pod.list_connections '''
        # register response
        httpretty.register_uri(httpretty.GET, self.__uri__ + "pod/v1/connection/list",
                               body='[{ \
                                       "userId": 7078106126503, \
                                       "status": "PENDING_OUTGOING", \
                                       "updatedAt": 1471018076255 \
                                      }, \
                                      { \
                                       "userId": 7078106103809, \
                                       "status": "PENDING_INCOMING", \
                                       "updatedAt": 1467562406219 \
                                      } \
                                     ]',
                               status=200,
                               content_type='text/json')
        # run test query
        status_code, response = self.pod.list_connections()
        response = json.loads(response)
        # verify return
        assert status_code == 200
        assert response[0]['userId'] == 7078106126503

def test_connection_status(self):
        ''' test pod.connection_status '''
        # register response
        httpretty.register_uri(httpretty.GET, self.__uri__ + "pod/v1/connection/123456/info",
                               body='{ \
                                      "userId": 123456, \
                                      "status": "ACCEPTED" \
                                     }',
                               status=200,
                               content_type='text/json')
        # run test query
        status_code, response = self.pod.connection_status('123456')
        # verify return
        response = json.loads(response)
        assert status_code == 200
        assert response['status'] == "ACCEPTED"

def test_create_connection(self):
        ''' test pod.create_connection '''
        # register response
        httpretty.register_uri(httpretty.POST, self.__uri__ + "pod/v1/connection/create",
                               body='{ \
                                      "userId": 123456, \
                                      "status": "PENDING_OUTGOING", \
                                      "firstRequestedAt": 1471018076255, \
                                      "updatedAt": 1471018076255, \
                                      "requestCounter": 1 \
                                     }',
                               status=200,
                               content_type='text/json')
        # run test query
        status_code, response = self.pod.create_connection('123456')
        response = json.loads(response)
        # verify return
        assert status_code == 200
        assert response['userId'] == 123456
        assert response['status'] == 'PENDING_OUTGOING'

def test_send_message(self):
        ''' test agent.send_message '''
        # register response
        httpretty.register_uri(httpretty.POST, self.__uri__ + "agent/v3/stream/thread_id/message/create",
                               body='{"id": "9zJTiQBL98ZEPAkvtjcweH___qr9auZ9dA", \
                                      "timestamp": "1464627173769", \
                                      "v2messageType": "V2Message", \
                                      "streamId": "thread_id", \
                                      "message": "test string", \
                                      "attachments": [], \
                                      "fromUserId": 123456 \
                                    }',
                               status=200,
                               content_type='text/json')
        # run test query
        status_code, response = self.agent.send_message('thread_id', 'TEXT', 'test string')
        response = json.loads(response)
        # verify return
        assert status_code == 200
        assert response['streamId'] == 'thread_id'
        assert response['message'] == 'test string'

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_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_get_or_create_db():
    server = BootstrapServer("http://test.test:5984")
    httpretty.register_uri(
        httpretty.HEAD, "http://test.test:5984/test", status=404, body=""
    )
    def create_test_db(request, uri, headers):
        httpretty.reset()
        httpretty.register_uri(
            httpretty.HEAD, "http://test.test:5984/test", status=200
        )
        httpretty.register_uri(
            httpretty.PUT, "http://test.test:5984/test", status=500
        )
        return 201, headers, ""
    httpretty.register_uri(
        httpretty.PUT, "http://test.test:5984/test", body=create_test_db
    )
    assert "test" not in server
    test_db = server.get_or_create("test")
    assert "test" in server

def test_create_saved_query(mocked_url, mocked_rw_apikey, mocked_account_resource_id, capsys):
    saved_query_name = "my_saved_query"
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.POST, MOCK_API_URL, status=201,
                           content_type='application/json',
                           body=json.dumps({"saved_query": SAVED_QUERY_RESPONSE}))

    api.create_saved_query(saved_query_name, "where(/*/)")
    out, err = capsys.readouterr()

    assert "Saved query created with name: %s" % saved_query_name in out

def test_delete_saved_query(mocked_url, mocked_rw_apikey, mocked_account_resource_id, capsys):
    test_saved_query_id = str(uuid.uuid4())
    mocked_url.return_value = '', MOCK_API_URL
    mocked_rw_apikey.return_value = str(uuid.uuid4())
    mocked_account_resource_id.return_value = str(uuid.uuid4())
    httpretty.register_uri(httpretty.DELETE, MOCK_API_URL, status=204)

    api.delete_saved_query(test_saved_query_id)
    out, err = capsys.readouterr()

    assert "Deleted saved query with id: %s" % test_saved_query_id in out