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)
python类ANY的实例源码
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]
