Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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()
async def _async_update(self, now=None):
"""Update info from Tile."""
try:
await self._client.async_init()
tiles = await self._client.tiles.all(
whitelist=self._types, show_inactive=self._show_inactive
)
except SessionExpiredError:
_LOGGER.info("Session expired; trying again shortly")
return
except TileError as err:
_LOGGER.error("There was an error while updating: %s", err)
return
if not tiles:
_LOGGER.warning("No Tiles found")
return
for tile in tiles:
await self._async_see(
dev_id="tile_{0}".format(slugify(tile["tile_uuid"])),
gps=(
tile["last_tile_state"]["latitude"],
tile["last_tile_state"]["longitude"],
async def _request(
self,
method: str,
endpoint: str,
*,
headers: Optional[dict] = None,
params: Optional[dict] = None,
data: Optional[dict] = None,
) -> dict:
"""Make a request against AirVisual."""
if self._session_expiry and self._session_expiry <= current_epoch_time():
raise SessionExpiredError("Session has expired; make a new one!")
_headers = headers or {}
_headers.update(
{
"Tile_app_id": DEFAULT_APP_ID,
"Tile_app_version": DEFAULT_APP_VERSION,
"Tile_client_uuid": self.client_uuid,
}
)
use_running_session = self._session and not self._session.closed
if use_running_session:
session = self._session
else:
session = ClientSession(timeout=ClientTimeout(total=DEFAULT_TIMEOUT))