Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"production.tile-api.com",
client_pattern,
"put",
aresponses.Response(
text=load_fixture("create_client_response.json"), status=200
),
)
aresponses.add(
"production.tile-api.com",
session_pattern,
"post",
aresponses.Response(text=json.dumps(fixture_create_session), status=200),
)
async with aiohttp.ClientSession() as session:
client = await async_login(TILE_EMAIL, TILE_PASSWORD, session=session)
assert isinstance(client.client_uuid, str)
assert client.client_uuid != TILE_CLIENT_UUID
assert client.user_uuid == TILE_USER_UUID
async def main():
"""Run."""
async with ClientSession() as session:
try:
# Create a client:
client = await async_login("", "", session=session)
print("Showing active Tiles:")
print(await client.tiles.all())
print("Showing all Tiles:")
print(await client.tiles.all(show_inactive=True))
except TileError as err:
print(err)
"production.tile-api.com",
f"/api/v1/users/{TILE_USER_UUID}/user_tiles",
"get",
aresponses.Response(text=load_fixture("tile_list_response.json"), status=200),
)
aresponses.add(
"production.tile-api.com",
"/api/v1/tiles",
"get",
aresponses.Response(
text=load_fixture("tile_details_response.json"), status=200
),
)
async with aiohttp.ClientSession() as session:
client = await async_login(
TILE_EMAIL, TILE_PASSWORD, client_uuid=TILE_CLIENT_UUID, session=session
)
tiles = await client.tiles.all()
assert len(tiles) == 1
assert tiles[TILE_TILE_UUID]["name"] == TILE_TILE_NAME
aresponses.add(
"production.tile-api.com",
f"/api/v1/clients/{TILE_CLIENT_UUID}/sessions",
"post",
aresponses.Response(text=json.dumps(fixture_create_session), status=200),
)
aresponses.add(
"production.tile-api.com",
"/api/v1/bad_endpoint",
"get",
aresponses.Response(text="", status=404),
)
with pytest.raises(RequestError):
async with aiohttp.ClientSession() as session:
client = await async_login(
TILE_EMAIL, TILE_PASSWORD, client_uuid=TILE_CLIENT_UUID, session=session
)
await client._request("get", "bad_endpoint")
aresponses.add(
"production.tile-api.com",
f"/api/v1/users/{TILE_USER_UUID}/user_tiles",
"get",
aresponses.Response(text=load_fixture("tile_list_response.json"), status=200),
)
aresponses.add(
"production.tile-api.com",
"/api/v1/tiles",
"get",
aresponses.Response(
text=load_fixture("tile_details_response.json"), status=200
),
)
client = await async_login(TILE_EMAIL, TILE_PASSWORD, client_uuid=TILE_CLIENT_UUID)
tiles = await client.tiles.all()
assert len(tiles) == 1
assert tiles[TILE_TILE_UUID]["name"] == TILE_TILE_NAME
async def main():
"""Run."""
async with ClientSession() as session:
try:
# Create a client:
client = await async_login("", "", session=session)
print("Showing active Tiles:")
print(await client.tiles.all())
print("Showing all Tiles:")
print(await client.tiles.all(show_inactive=True))
except TileError as err:
print(err)
aresponses.add(
"production.tile-api.com",
f"/api/v1/clients/{TILE_CLIENT_UUID}",
"put",
aresponses.Response(
text=load_fixture("create_client_response.json"), status=200
),
)
aresponses.add(
"production.tile-api.com",
f"/api/v1/clients/{TILE_CLIENT_UUID}/sessions",
"post",
aresponses.Response(text=json.dumps(fixture_expired_session), status=200),
)
with pytest.raises(SessionExpiredError):
async with aiohttp.ClientSession() as session:
client = await async_login(
TILE_EMAIL, TILE_PASSWORD, client_uuid=TILE_CLIENT_UUID, session=session
)
await client.tiles.all()
),
)
aresponses.add(
"production.tile-api.com",
f"/api/v1/clients/{TILE_CLIENT_UUID}/sessions",
"post",
aresponses.Response(text=json.dumps(fixture_create_session), status=200),
)
aresponses.add(
"production.tile-api.com",
"/api/v1/bad_endpoint",
"get",
aresponses.Response(text="", status=404),
)
with pytest.raises(RequestError):
async with aiohttp.ClientSession() as session:
client = await async_login(
TILE_EMAIL, TILE_PASSWORD, client_uuid=TILE_CLIENT_UUID, session=session
)
await client._request("get", "bad_endpoint")
"""Validate the configuration and return a Tile scanner."""
websession = aiohttp_client.async_get_clientsession(hass)
config_file = hass.config.path(
".{}{}".format(slugify(config[CONF_USERNAME]), CLIENT_UUID_CONFIG_FILE)
)
config_data = await hass.async_add_job(load_json, config_file)
if config_data:
client = await async_login(
config[CONF_USERNAME],
config[CONF_PASSWORD],
websession,
client_uuid=config_data["client_uuid"],
)
else:
client = await async_login(
config[CONF_USERNAME], config[CONF_PASSWORD], websession
)
config_data = {"client_uuid": client.client_uuid}
await hass.async_add_job(save_json, config_file, config_data)
scanner = TileScanner(
client,
hass,
async_see,
config[CONF_MONITORED_VARIABLES],
config[CONF_SHOW_INACTIVE],
)
return await scanner.async_init()
async def async_init(self):
"""Further initialize connection to the Tile servers."""
try:
await self._client.async_init()
except TileError as err:
_LOGGER.error("Unable to set up Tile scanner: %s", err)
return False
await self._async_update()
async_track_time_interval(self._hass, self._async_update, DEFAULT_SCAN_INTERVAL)
return True