Python requests_mock 模块，ANY 实例源码

def setUp(self, mock):
        mock.register_uri(requests_mock.ANY, self.installation,
                          json=self.get_installation)
        mock.register_uri(requests_mock.ANY, self.device_server)

        username = self.fake.name()
        self.password = self.fake.password()
        self.user = User.objects.create_user(username=username,
                                             password=self.password,
                                             email=None)
        i = Installation(user=self.user, api_key=self.fake.sha1(),
                         password=self.password)
        i.status
        key = self.user.tokens.file_token
        file_path = SessionStore(session_key=key)['file_path']
        with open(file_path, 'r') as f:
            file_contents = f.read()
        self.store_in_session(data=file_contents)

        self.c = callback(self.user)

        os.remove(file_path)

def test_load_file(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.installation,
                          json=self.get_installation)
        mock.register_uri(requests_mock.ANY, self.attachemt_pubilc,
                          content=self.get_avatar)
        mock.register_uri(requests_mock.ANY, self.session_server,
                          json=self.get_start_session)
        mock.register_uri(requests_mock.ANY, self.accounts,
                          json=self.get_accounts)
        mock.register_uri(requests_mock.ANY, self.payment,
                          json=self.get_payments)

        # c = callback(user=self.user)
        self.c.load_file()
        self.c.account_id = self.fake.random_number()
        self.c.payment_id = self.fake.random_number()
        self.c.load_file()  # NOTE: need to find an assertion for this.
        # assertEqual doesnt do the job due to the order
        # of the JSON's

def test_generate_view(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.installation,
                          json=self.get_installation)
        mock.register_uri(requests_mock.ANY, self.device_server)
        data = {
            'user_password': self.password,
            'API': self.fake.sha1()
        }

        request = self.request.post('/generate', data=data)
        request.user = self.user
        request.session = {}

        self.client.login(username=self.user.username, password=self.password)

        response = views.GenerateView.as_view()(request)
        self.assertEqual(response.status_code, 200)

def test_filter_equals(self):
        with requests_mock.mock() as m:
            m.get(requests_mock.ANY)
            url = "http://test/api/2.3/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks"
            opts = TSC.RequestOptions(pagesize=13, pagenumber=13)
            opts.filter.add(TSC.Filter(TSC.RequestOptions.Field.Name,
                                       TSC.RequestOptions.Operator.Equals,
                                       'Superstore'))

            resp = self.server.workbooks._make_request(requests.get,
                                                       url,
                                                       content=None,
                                                       request_object=opts,
                                                       auth_token='j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM',
                                                       content_type='text/xml')

            self.assertEqual(resp.request.query, 'pagenumber=13&pagesize=13&filter=name:eq:superstore')

def test_filter_in(self):
        with requests_mock.mock() as m:
            m.get(requests_mock.ANY)
            url = "http://test/api/2.3/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks"
            opts = TSC.RequestOptions(pagesize=13, pagenumber=13)

            opts.filter.add(TSC.Filter(TSC.RequestOptions.Field.Tags,
                                       TSC.RequestOptions.Operator.In,
                                       ['stocks', 'market']))

            resp = self.server.workbooks._make_request(requests.get,
                                                       url,
                                                       content=None,
                                                       request_object=opts,
                                                       auth_token='j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM',
                                                       content_type='text/xml')

            self.assertEqual(resp.request.query, 'pagenumber=13&pagesize=13&filter=tags:in:[stocks,market]')

def test_sort_asc(self):
        with requests_mock.mock() as m:
            m.get(requests_mock.ANY)
            url = "http://test/api/2.3/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks"
            opts = TSC.RequestOptions(pagesize=13, pagenumber=13)
            opts.sort.add(TSC.Sort(TSC.RequestOptions.Field.Name,
                                   TSC.RequestOptions.Direction.Asc))

            resp = self.server.workbooks._make_request(requests.get,
                                                       url,
                                                       content=None,
                                                       request_object=opts,
                                                       auth_token='j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM',
                                                       content_type='text/xml')

            self.assertEqual(resp.request.query, 'pagenumber=13&pagesize=13&sort=name:asc')

def test_obtain_access_token(self, rmock):
        rmock.post(requests_mock.ANY, text='{"access_token": "ANY_TOKEN"}')
        cmock = Mock()
        cmock.username = "ANY_USERNAME"
        cmock.auth_host = "ANY_URL.example"
        result = obtain_access_token(cmock, 'ANY_PASSWORD')
        self.assertEqual('ANY_TOKEN', result)
        received_post_data = parse_qs(rmock.request_history[0].text)
        expected_post_data = {u'username': [u'ANY_USERNAME'],
                              u'password': [u'ANY_PASSWORD'],
                              u'client_id': [u'jumpauth'],
                              u'grant_type': [u'password']}
        self.assertEqual(received_post_data, expected_post_data)

def test_fetch_query_results(settings, jira):
    """Return the results."""
    settings.JIRA_AUTH = ("user", "password")
    expected = dict(foo="bar", baz="bat")
    expected_json = json.dumps(expected)
    with requests_mock.mock() as m:
        m.register_uri(requests_mock.ANY, requests_mock.ANY, text=expected_json)
        result = jira.fetch_query_results(43035)
        assert result == expected

def test_fetch_query_results_with_errors(settings, jira):
    """JIRA responses with error messages should be logged and returned."""
    settings.JIRA_AUTH = ("user", "password")
    expected = dict(
        errors=[],
        errorMessages=["Was a boring conversation anyway... Luke we're gonna have company!", ],
    )
    expected_json = json.dumps(expected)
    with requests_mock.mock() as m:
        with patch("dashboard.services.jira.LOGGER") as mocked_logger:
            m.register_uri(requests_mock.ANY, requests_mock.ANY, text=expected_json)
            result = jira.fetch_query_results(43035, logger=mocked_logger)
            assert mocked_logger.warning.called
            assert result == expected

def setUp(self, m):
        super(TestCloudFeedClient, self).setUp()
        self.example_url = 'http://example.com'
        self.tokens_url = 'http://example.com/tokens'
        self.config_fixture = self.useFixture(config_fixture.Config(CONF))
        self.config(self.config_files())
        self.config_overrides()
        m.post(requests_mock.ANY, status_code=200,
               json={'access': {'token': {'id': 'id'}}})
        self.ci = cache_invalidator.CloudFeedClient(self.example_url)
        filename = os.path.join(FIXTURES_DIR, 'rackspace_feed_response.json')
        with open(filename) as f:
            self.feed_response_body = json.load(f)

def test_installation(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.installation,
                          json=self.get_installation)
        mock.register_uri(requests_mock.ANY, self.device_server)
        i = Installation(user=self.user, api_key=self.fake.sha1(),
                         password=self.password)

        self.assertTrue(i.status)

def test_delete_session(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.attachemt_pubilc,
                          content=self.get_avatar)
        mock.register_uri(requests_mock.ANY, self.session_server,
                          json=self.get_start_session)
        mock.register_uri(requests_mock.ANY, self.session)

        self.c.start_session()
        self.assertTrue(self.c.delete_session())

def test_users(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.users,
                          json=self.get_users)
        mock.register_uri(requests_mock.ANY, self.attachemt_pubilc,
                          content=self.get_avatar)
        # c = callback(user=self.user)
        self.c.users()
        self.c.user_id = self.fake.random_number()
        self.c.users()

def test_invoice(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.invoice,
                          json=self.get_invoice_data)
        mock.register_uri(requests_mock.ANY, self.invoice_api,
                          text=self.get_invoice_pdf)

        self.c.user_id = self.fake.random_number()
        self.c.invoice()

def test_payment_pdf(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.payment,
                          json=self.get_payments)

        self.c.payment_id = self.fake.random_number()
        self.c.get_payment_pdf()

def test_cusomer_statement_pdf(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.customer_statement,
                          json=self.get_customer_statment)

        self.c.user_id = self.fake.random_number()
        self.c.account_id = self.fake.random_number()
        self.c.date_start = self.fake.date()
        self.c.date_end = self.fake.date()
        self.c.statement_format = 'PDF'

        self.c.customer_statement()

def test_cusomer_statement_csv(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.customer_statement,
                          json=self.get_customer_statment)

        self.c.user_id = self.fake.random_number()
        self.c.account_id = self.fake.random_number()
        self.c.date_start = self.fake.date()
        self.c.date_end = self.fake.date()
        self.c.statement_format = 'CSV'

        self.c.customer_statement()

def test_cusomer_statement_mt940(self, mock, _):
        mock.register_uri(requests_mock.ANY, self.customer_statement,
                          json=self.get_customer_statment)

        self.c.user_id = self.fake.random_number()
        self.c.account_id = self.fake.random_number()
        self.c.date_start = self.fake.date()
        self.c.date_end = self.fake.date()
        self.c.statement_format = 'MT940'

        self.c.customer_statement()

def test_delete(self):
        with requests_mock.Mocker() as m:
            m.delete(requests_mock.ANY, text='{}')
            self.comment.delete()

def test_delete(self):
        with requests_mock.Mocker() as m:
            m.delete(requests_mock.ANY, text='{}')
            self.comment.delete()

def setUp(self):
        self.session = Session()
        self.adapter = Adapter()
        self.session.mount('http://', self.adapter)
        response_text = resource_string("parsers.test.mocks", "IN_AP.html")
        self.adapter.register_uri(ANY, ANY, text=response_text)

def test_fetch_exchange(self):
        json_data = resource_string("parsers.test.mocks", "ESIOS_ES_MA.json")
        self.adapter.register_uri(ANY, ANY, json=loads(json_data))
        try:
            data_list = ESIOS.fetch_exchange('ES', 'MA', self.session, 'ESIOS_MOCK_TOKEN')
            self.assertIsNotNone(data_list)
            for data in data_list:
                self.assertEqual(data['sortedCountryCodes'], 'ES->MA')
                self.assertEqual(data['source'], 'api.esios.ree.es')
                self.assertIsNotNone(data['datetime'])
                self.assertIsNotNone(data['netFlow'])
        except Exception as ex:
            self.fail(ex.message)

def test_stickup_cam_controls(self, mock):
        mock.get('https://api.ring.com/clients_api/ring_devices',
                 text=load_fixture('ring_devices.json'))
        mock.get('https://api.ring.com/clients_api/doorbots/987652/health',
                 text=load_fixture('ring_doorboot_health_attrs.json'))
        mock.put(requests_mock.ANY, text='ok')

        data = self.ring_persistent
        for dev in data.stickup_cams:
            dev.lights = 'off'
            dev.lights = 'on'
            dev.siren = 0
            dev.siren = 30

            history = list(filter(lambda x: x.method == 'PUT',
                                  mock.request_history))
            self.assertEqual(
                '/clients_api/doorbots/987652/floodlight_light_off',
                history[0].path)
            self.assertEqual(
                '/clients_api/doorbots/987652/floodlight_light_on',
                history[1].path)
            self.assertEqual(
                '/clients_api/doorbots/987652/siren_off',
                history[2].path)
            self.assertNotIn('duration', history[2].qs)
            self.assertEqual(
                '/clients_api/doorbots/987652/siren_on',
                history[3].path)
            self.assertEqual('30', history[3].qs['duration'][0])

def test_withdraw_api_exception():
    """Test Withdraw API response Exception"""

    with pytest.raises(BinanceWithdrawException):

        with requests_mock.mock() as m:
            json_obj = {"success": False, "msg": "Insufficient funds"}
            m.register_uri('POST', requests_mock.ANY, json=json_obj, status_code=200)
            client.withdraw(asset='BTC', address='BTCADDRESS', amount=100)

def test_bad_URLs(self, id,**kwargs):
        processor = StepProcessor(recipe=self.r, stepid=id, variables={})
        self.r.steps[id] = processor.prepare()
        kwargs['mock'].request(  # set up the mock
                           requests_mock.GET,
                           requests_mock.ANY,
                           status_code=200,
                           headers={'content-type': 'application/json'},
                           json={'msg': 'hello'}
                           )

        with self.assertRaises(ValueError):
            prepared_statement = processor.call()

def test_small_change(self):
        provider = PowerDnsProvider('test', 'non.existant', 'api-key')

        expected = Zone('unit.tests.', [])
        source = YamlProvider('test', join(dirname(__file__), 'config'))
        source.populate(expected)
        self.assertEquals(18, len(expected.records))

        # A small change to a single record
        with requests_mock() as mock:
            mock.get(ANY, status_code=200, text=FULL_TEXT)

            missing = Zone(expected.name, [])
            # Find and delete the SPF record
            for record in expected.records:
                if record._type != 'SPF':
                    missing.add_record(record)

            def assert_delete_callback(request, context):
                self.assertEquals({
                    'rrsets': [{
                        'records': [
                            {'content': '"v=spf1 ip4:192.168.0.1/16-all"',
                             'disabled': False}
                        ],
                        'changetype': 'DELETE',
                        'type': 'SPF',
                        'name': 'spf.unit.tests.',
                        'ttl': 600
                    }]
                }, loads(request.body))
                return ''

            mock.patch(ANY, status_code=201, text=assert_delete_callback)

            plan = provider.plan(missing)
            self.assertEquals(1, len(plan.changes))
            self.assertEquals(1, provider.apply(plan))

def test_make_get_request(self):
        with requests_mock.mock() as m:
            m.get(requests_mock.ANY)
            url = "http://test/api/2.3/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks"
            opts = TSC.RequestOptions(pagesize=13, pagenumber=13)
            resp = self.server.workbooks._make_request(requests.get,
                                                       url,
                                                       content=None,
                                                       request_object=opts,
                                                       auth_token='j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM',
                                                       content_type='text/xml')

            self.assertEqual(resp.request.query, 'pagenumber=13&pagesize=13')
            self.assertEqual(resp.request.headers['x-tableau-auth'], 'j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM')
            self.assertEqual(resp.request.headers['content-type'], 'text/xml')

def test_make_post_request(self):
        with requests_mock.mock() as m:
            m.post(requests_mock.ANY)
            url = "http://test/api/2.3/sites/dad65087-b08b-4603-af4e-2887b8aafc67/workbooks"
            resp = self.server.workbooks._make_request(requests.post,
                                                       url,
                                                       content=b'1337',
                                                       request_object=None,
                                                       auth_token='j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM',
                                                       content_type='multipart/mixed')
            self.assertEqual(resp.request.headers['x-tableau-auth'], 'j80k54ll2lfMZ0tv97mlPvvSCRyD0DOM')
            self.assertEqual(resp.request.headers['content-type'], 'multipart/mixed')
            self.assertEqual(resp.request.body, b'1337')

def register_uris(requirements, requests_mocker):
    """
    Given a list of required fixtures and an requests_mocker object,
    register each fixture as a uri with the mocker.

    :param base_url: str
    :param requirements: dict
    :param requests_mocker: requests_mock.mocker.Mocker
    """
    for fixture, objects in requirements.items():

        try:
            with open('tests/fixtures/{}.json'.format(fixture)) as file:
                data = json.loads(file.read())
        except IOError:
            raise ValueError('Fixture {}.json contains invalid JSON.'.format(fixture))

        if not isinstance(objects, list):
            raise TypeError('{} is not a list.'.format(objects))

        for obj_name in objects:
            obj = data.get(obj_name)

            if obj is None:
                raise ValueError('{} does not exist in {}.json'.format(
                    obj_name.__repr__(),
                    fixture
                ))

            method = requests_mock.ANY if obj['method'] == 'ANY' else obj['method']
            if obj['endpoint'] == 'ANY':
                url = requests_mock.ANY
            else:
                url = settings.BASE_URL + obj['endpoint']

            try:
                requests_mocker.register_uri(
                    method,
                    url,
                    json=obj.get('data'),
                    status_code=obj.get('status_code', 200),
                    headers=obj.get('headers', {})
                )
            except Exception as e:
                print(e)

def test_populate(self):
        provider = DigitalOceanProvider('test', 'token')

        # Bad auth
        with requests_mock() as mock:
            mock.get(ANY, status_code=401,
                     text='{"id":"unauthorized",'
                     '"message":"Unable to authenticate you."}')

            with self.assertRaises(Exception) as ctx:
                zone = Zone('unit.tests.', [])
                provider.populate(zone)
            self.assertEquals('Unauthorized', ctx.exception.message)

        # General error
        with requests_mock() as mock:
            mock.get(ANY, status_code=502, text='Things caught fire')

            with self.assertRaises(HTTPError) as ctx:
                zone = Zone('unit.tests.', [])
                provider.populate(zone)
            self.assertEquals(502, ctx.exception.response.status_code)

        # Non-existant zone doesn't populate anything
        with requests_mock() as mock:
            mock.get(ANY, status_code=404,
                     text='{"id":"not_found","message":"The resource you '
                     'were accessing could not be found."}')

            zone = Zone('unit.tests.', [])
            provider.populate(zone)
            self.assertEquals(set(), zone.records)

        # No diffs == no changes
        with requests_mock() as mock:
            base = 'https://api.digitalocean.com/v2/domains/unit.tests/' \
                'records?page='
            with open('tests/fixtures/digitalocean-page-1.json') as fh:
                mock.get('{}{}'.format(base, 1), text=fh.read())
            with open('tests/fixtures/digitalocean-page-2.json') as fh:
                mock.get('{}{}'.format(base, 2), text=fh.read())

            zone = Zone('unit.tests.', [])
            provider.populate(zone)
            self.assertEquals(12, len(zone.records))
            changes = self.expected.changes(zone, provider)
            self.assertEquals(0, len(changes))

        # 2nd populate makes no network calls/all from cache
        again = Zone('unit.tests.', [])
        provider.populate(again)
        self.assertEquals(12, len(again.records))

        # bust the cache
        del provider._zone_records[zone.name]