Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_token_params(self, client):
token = client.oauth2.get_token(client_id='client_id', client_secret='client_secret', scope=['account', 'events'])
client.request.assert_called_once_with('post', '/oauth2/token/', auth=('client_id', 'client_secret'), data={'scope': 'account events', 'grant_type': 'client_credentials'})
assert isinstance(token, AccessToken)
assert token.to_primitive() == client.request.return_value
with pytest.raises(ValidationError):
client.oauth2.get_token(client_id=None)
with pytest.raises(ValidationError):
client.oauth2.get_token(invalid_arg=None)
def test_get_params(self, client):
client.signals.get(id="signal-id")
client.request.assert_called_once_with('get', '/v1/signals/signal-id/')
with self.assertRaises(ValidationError):
client.signals.get(id=None)
arg1 = schemas.StringType(required=True)
arg2 = schemas.ListType(schemas.IntType)
class EndpointExample(BaseEndpoint):
@decorators.returns(SchemaExample)
def func(self, **kwargs):
return kwargs
endpoint = EndpointExample(None)
self.assertEqual(endpoint.func(arg1="test", arg2=[1, 2]), SchemaExample({'arg1': 'test', 'arg2': [1, 2]}))
with self.assertRaises(ValidationError):
endpoint.func(arg2=[1, 2])
with self.assertRaises(ValidationError):
endpoint.func(arg1="value", arg2="invalid")
arg2 = schemas.ListType(schemas.IntType)
class EndpointExample(BaseEndpoint):
@decorators.accepts(SchemaExample)
def func(self, **kwargs):
return kwargs
endpoint = EndpointExample(None)
self.assertDictEqual(endpoint.func(arg1="test", arg2=[1, 2]), {'arg1': 'test', 'arg2': '1,2'})
self.assertDictEqual(endpoint.func(SchemaExample({"arg1": "test", "arg2": [1, 2]})), {'arg1': 'test', 'arg2': '1,2'})
self.assertDictEqual(endpoint.func({"arg1": "test", "arg2": [1, 2]}), {'arg1': 'test', 'arg2': '1,2'})
with self.assertRaises(ValidationError):
endpoint.func(arg2=[1, 2])
with self.assertRaises(ValidationError):
endpoint.func(arg1="value", arg2="invalid")
def test_update_params(self, client):
client.signals.update(id="signal-id", name="Test", dimensions=[{"name": "dimension", "type": "category"}], country="NZ")
client.request.assert_called_once_with('put', '/v1/signals/signal-id/', json={'country':'NZ', 'dimensions': [{'type':'category', 'name':'dimension'}], 'name':'Test'})
with self.assertRaises(ValidationError):
client.signals.update(name="Test", dimensions=[{"name": "dimension", "type": "category"}], country="NZ")
def test_revoke_token_params(self, client):
result = client.oauth2.revoke_token(client_id='client_id', client_secret='client_secret', token='token123')
client.request.assert_called_once_with('post', '/oauth2/revoke/', auth=('client_id', 'client_secret'), data={'token': 'token123', 'token_type_hint': 'access_token'})
assert result is None
with pytest.raises(ValidationError):
client.oauth2.revoke_token(token=None)
with pytest.raises(ValidationError):
client.oauth2.revoke_token(invalid_arg=None)
new_args = tuple(a for a in args if not isinstance(a, (schema, dict)))
if args != new_args:
instance = next(a for a in args if isinstance(a, (schema, dict)))
if isinstance(instance, dict):
kwargs = instance
else:
kwargs = instance.to_dict()
args = new_args
try:
data = _process_kwargs(kwargs)
model = schema()
model.import_data(data, strict=True, partial=False)
model.validate()
except SchematicsDataError as e:
raise ValidationError(e.messages)
if query_string:
params = _to_url_params(model.to_primitive(role=role))
else:
params = model.to_primitive(role=role)
return f(endpoint, *args, **params)
# if results are of type Model, make sure to set the endpoint on each item
if data is not None and 'results' in data \
and hasattr(model._fields['results'], 'model_class') \
and issubclass(model._fields['results'].model_class, Model):
def initialize_result_type(item_data):
item = model._fields['results'].model_class(item_data, strict=False)
item._endpoint = endpoint
return item
# Use generator so results are not iterated over more than necessary
data['results'] = (initialize_result_type(item_data) for item_data in data['results'])
model.import_data(data, strict=False)
model.validate()
except SchematicsDataError as e:
raise ValidationError(e.messages)
return model