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_build_with_port():
with pytest.raises(ValueError):
URL.build(port=8000)
u = URL.build(scheme="http", host="127.0.0.1", port=8000)
assert str(u) == "http://127.0.0.1:8000"
return st.builds (lambda x: URL.build (**x), args)
def test_build_with_host():
u = URL.build(host="127.0.0.1")
assert str(u) == "//127.0.0.1"
assert u == URL("//127.0.0.1")
def test_build_with_scheme_and_host():
u = URL.build(scheme="http", host="127.0.0.1")
assert str(u) == "http://127.0.0.1"
assert u == URL("http://127.0.0.1")
def test_build_with_port():
with pytest.raises(ValueError):
URL.build(port=8000)
u = URL.build(scheme="http", host="127.0.0.1", port=8000)
assert str(u) == "http://127.0.0.1:8000"
def name(self) -> str:
scheme = 'https' if self._ssl_context else 'http'
return str(URL.build(scheme=scheme, host=self._host, port=self._port))
self.version = __version__
if token is None:
token = os.environ['OANDA_TOKEN']
self.account_id = account_id
self.format_order_requests = format_order_requests
self.max_transaction_history = max_transaction_history
# V20 REST API URL
rest_host = partial(URL.build, host=rest_host, port=rest_port, scheme=rest_scheme)
# v20 STREAM API URL
stream_host = partial(URL.build, host=stream_host, port=stream_port, scheme=stream_scheme)
# V20 API health URL
health_host = partial(URL.build, host=health_host, port=health_port, scheme=health_scheme)
self._hosts = {'REST': rest_host, 'STREAM': stream_host, 'HEALTH': health_host}
# The timeout to use when making a polling request with the
# v20 REST server
self.rest_timeout = rest_timeout
# The timeout to use when waiting for the next object when wait for a stream response
self.stream_timeout = stream_timeout
self.max_requests_per_second = max_requests_per_second
self.max_simultaneous_connections = max_simultaneous_connections
account_id = await next(iter(client.get_accounts()['accounts'])).id
# This parameters is placed here for easy access
client.account_id = account_id
# This is the default parameter dictionary. Client Methods that require certain parameters
# that are not explicitly passed will try to find it in this dict
client.default_parameters = {Authorization: 'Bearer {}'.format(token),
AccountID: client.account_id,
AcceptDatetimeFormat: datetime_format}
# V20 REST API URL
client.rest_url = URL.build(host=rest_host, port=rest_port, scheme='https')
# v20 STREAM API URL
client.stream_url = URL.build(host=stream_host, port=stream_port, scheme='https')
# The size of each chunk to read when processing a stream
# response
client.stream_chunk_size = stream_chunk_size
# The timeout to use when making a stream request with the
# v20 REST server
client.stream_timeout = stream_timeout
# The timeout to use when making a polling request with the
# v20 REST server
client.poll_timeout = poll_timeout
def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None:
"""Initialize the switch."""
self._close_session = False
self._host = host
self._session = session
self._consumption = 0
self._state = None
self._temperature = None
self._firmware = None
self._mac = None
self.uri = URL.build(scheme="http", host=self._host)
def __init__(self, token):
self.agent_url = yarl.URL.build(
scheme="http", host="127.0.0.1", port="8500", path="/v1/agent"
)
if token:
self.headers = {"X-Consul-Token": token}
else:
self.headers = {}
self.session_factory = partial(
aiohttp.ClientSession, headers=self.headers
)