files: import os
  from typing import Any, Dict
  
  from aiohttp import ClientSession, ClientTimeout
  from aiohttp.client_exceptions import ClientResponseError, ClientConnectorError
  from injector import inject
  from asyncio.exceptions import TimeoutError
  
  from ..util import AsyncHttpGetter
  from ..config import Config, Setting, Version
  from ..exceptions import HomeAssistantDeleteError, SupervisorConnectionError, SupervisorPermissionError, SupervisorTimeoutError, SupervisorUnexpectedError
  from ..model import HABackup
  from ..logger import getLogger
  from ..util import DataCache
  from backup.time import Time
  from backup.const import NECESSARY_OLD_BACKUP_PLURAL_NAME, NECESSARY_OLD_SUPERVISOR_URL
  from yarl import URL
  
  logger = getLogger(__name__)
  
  HEADER_TOKEN = "X-Supervisor-Token"
  
  NOTIFICATION_ID = "backup_broken"
  EVENT_BACKUP_START = "backup_started"
  EVENT_BACKUP_END = "backup_ended"
  
  VERSION_BACKUP_PATH = Version.parse("2021.8")
  
  
  def supervisor_call(func):
      async def wrap_and_call(*args, **kwargs):
          try:
              return await func(*args, **kwargs)
          except ClientConnectorError:
              raise SupervisorConnectionError()
          except TimeoutError:
              raise SupervisorConnectionError()
          except ClientResponseError as e:
              if e.status == 403:
                  raise SupervisorPermissionError()
              raise
      return wrap_and_call
  
  
  class HaRequests():
      """
      Stores logic for interacting with the supervisor add-on API
      """
      @inject
      def __init__(self, config: Config, session: ClientSession, time: Time, data_cache: DataCache):
          self.config: Config = config
          self.cache = {}
          self.session = session
          self._time = time
          self._data_cache = data_cache
  
          # default the supervisor versio to using the "most featured" when it can't be parsed.
          self._super_version = VERSION_BACKUP_PATH
  
      def getSupervisorURL(self) -> URL:
          if len(self.config.get(Setting.SUPERVISOR_URL)) > 0:
              return URL(self.config.get(Setting.SUPERVISOR_URL))
          if 'SUPERVISOR_TOKEN' in os.environ:
              return URL("http://supervisor")
          else:
              return URL(NECESSARY_OLD_SUPERVISOR_URL)
  
      def _getBackupPath(self):
          if self.supportsBackupPaths():
              return "backups"
          return NECESSARY_OLD_BACKUP_PLURAL_NAME
  
      def supportsBackupPaths(self):
          return not self._super_version or self._super_version >= VERSION_BACKUP_PATH
  
      @supervisor_call
      async def createBackup(self, info):
          if 'folders' in info or 'addons' in info:
              url = self.getSupervisorURL().with_path("{0}/new/partial".format(self._getBackupPath()))
          else:
              url = self.getSupervisorURL().with_path("{0}/new/full".format(self._getBackupPath()))
          return await self._postHassioData(url, info, timeout=ClientTimeout(total=self.config.get(Setting.PENDING_BACKUP_TIMEOUT_SECONDS)))
  
      @supervisor_call
      async def auth(self, user: str, password: str) -> None:
          await self._postHassioData(self.getSupervisorURL().with_path("auth"), {"username": user, "password": password}, headers=self._altAuthHeaders())
  
      @supervisor_call
      async def upload(self, stream):
          url = self.getSupervisorURL().with_path("{0}/new/upload".format(self._getBackupPath()))
          return await self._postHassioData(url, data=stream)
  
      @supervisor_call
      async def delete(self, slug) -> None:
          if slug in self.cache:
              del self.cache[slug]
          try:
              if self.supportsBackupPaths():
                  delete_url = self.getSupervisorURL().with_path("{1}/{0}".format(slug, self._getBackupPath()))
                  await self._sendHassioData("delete", delete_url, {})
              else:
                  delete_url = self.getSupervisorURL().with_path("{1}/{0}/remove".format(slug, self._getBackupPath()))
                  await self._sendHassioData("post", delete_url, {})
          except ClientResponseError as e:
              if e.status == 400:
                  raise HomeAssistantDeleteError()
              raise e
  
      @supervisor_call
      async def startAddon(self, slug) -> None:
          url = self.getSupervisorURL().with_path("addons/{0}/start".format(slug))
          await self._postHassioData(url, {})
  
      @supervisor_call
      async def stopAddon(self, slug) -> None:
          url = self.getSupervisorURL().with_path("addons/{0}/stop".format(slug))
          await self._postHassioData(url, {})
  
      @supervisor_call
      async def backup(self, slug):
          if slug in self.cache:
              info = self.cache[slug]
          else:
              info = await self._getHassioData(self.getSupervisorURL().with_path("{1}/{0}/info".format(slug, self._getBackupPath())))
              self.cache[slug] = info
          return HABackup(info, self._data_cache, self.config, self.config.isRetained(slug))
  
      @supervisor_call
      async def backups(self):
          return await self._getHassioData(self.getSupervisorURL().with_path(self._getBackupPath()))
  
      @supervisor_call
      async def haInfo(self):
          return await self._getHassioData(self.getSupervisorURL().with_path("core/info"))
  
      @supervisor_call
      async def selfInfo(self) -> Dict[str, Any]:
          return await self.getAddonInfo("self")
  
      @supervisor_call
      async def getAddonInfo(self, addon_slug) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("addons/{0}/info".format(addon_slug)))
  
      @supervisor_call
      async def getAddons(self) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("addons"))
  
      @supervisor_call
      async def hassosInfo(self) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("hassos/info"))
  
      @supervisor_call
      async def info(self) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("info"))
  
      @supervisor_call
      async def refreshBackups(self):
          url = self.getSupervisorURL().with_path("{0}/reload".format(self._getBackupPath()))
          return await self._postHassioData(url)
  
      @supervisor_call
      async def supervisorInfo(self):
          url = self.getSupervisorURL().with_path("supervisor/info")
          info = await self._getHassioData(url)
  
          # parse the supervisor version
          if 'version' in info:
              self._super_version = Version.parse(info['version'])
          return info
  
      @supervisor_call
      async def restore(self, slug: str, password: str = None) -> None:
          url = self.getSupervisorURL().with_path("{1}/{0}/restore/full".format(slug, self._getBackupPath()))
          if password:
              await self._postHassioData(url, {'password': password})
          else:
              await self._postHassioData(url, {})
  
      @supervisor_call
      async def download(self, slug) -> AsyncHttpGetter:
          url = self.getSupervisorURL().with_path("{1}/{0}/download".format(slug, self._getBackupPath()))
          ret = AsyncHttpGetter(url,
                                self._getAuthHeaders(),
                                self.session,
                                timeoutFactory=SupervisorTimeoutError.factory,
                                otherErrorFactory=SupervisorUnexpectedError.factory,
                                timeout=ClientTimeout(sock_connect=self.config.get(Setting.DOWNLOAD_TIMEOUT_SECONDS),
                                                      sock_read=self.config.get(Setting.DOWNLOAD_TIMEOUT_SECONDS)),
                                time=self._time)
          return ret
  
      @supervisor_call
      async def getSuperLogs(self):
          url = self.getSupervisorURL().with_path("supervisor/logs")
          async with self.session.get(url, headers=self._getAuthHeaders()) as resp:
              resp.raise_for_status()
              return await resp.text()
  
      @supervisor_call
      async def getCoreLogs(self):
          url = self.getSupervisorURL().with_path("core/logs")
          async with self.session.get(url, headers=self._getAuthHeaders()) as resp:
              resp.raise_for_status()
              return await resp.text()
  
      async def _validateHassioReply
  
  [... the rest is omitted ...]
files: hassio-google-drive-backup/backup/ha/harequests.py
files: import os
  from typing import Any, Dict
  
  from aiohttp import ClientSession, ClientTimeout
  from aiohttp.client_exceptions import ClientResponseError, ClientConnectorError
  from injector import inject
  from asyncio.exceptions import TimeoutError
  
  from ..util import AsyncHttpGetter
  from ..config import Config, Setting, Version
  from ..exceptions import HomeAssistantDeleteError, SupervisorConnectionError, SupervisorPermissionError, SupervisorTimeoutError, SupervisorUnexpectedError
  from ..model import HABackup
  from ..logger import getLogger
  from ..util import DataCache
  from backup.time import Time
  from backup.const import NECESSARY_OLD_BACKUP_PLURAL_NAME, NECESSARY_OLD_SUPERVISOR_URL
  from yarl import URL
  
  logger = getLogger(__name__)
  
  NOTIFICATION_ID = "backup_broken"
  EVENT_BACKUP_START = "backup_started"
  EVENT_BACKUP_END = "backup_ended"
  
  VERSION_BACKUP_PATH = Version.parse("2021.8")
  
  
  def supervisor_call(func):
      async def wrap_and_call(*args, **kwargs):
          try:
              return await func(*args, **kwargs)
          except ClientConnectorError:
              raise SupervisorConnectionError()
          except TimeoutError:
              raise SupervisorConnectionError()
          except ClientResponseError as e:
              if e.status == 403:
                  raise SupervisorPermissionError()
              raise
      return wrap_and_call
  
  
  class HaRequests():
      """
      Stores logic for interacting with the supervisor add-on API
      """
      @inject
      def __init__(self, config: Config, session: ClientSession, time: Time, data_cache: DataCache):
          self.config: Config = config
          self.cache = {}
          self.session = session
          self._time = time
          self._data_cache = data_cache
  
          # default the supervisor versio to using the "most featured" when it can't be parsed.
          self._super_version = VERSION_BACKUP_PATH
  
      def getSupervisorURL(self) -> URL:
          if len(self.config.get(Setting.SUPERVISOR_URL)) > 0:
              return URL(self.config.get(Setting.SUPERVISOR_URL))
          if 'SUPERVISOR_TOKEN' in os.environ:
              return URL("http://supervisor")
          else:
              return URL(NECESSARY_OLD_SUPERVISOR_URL)
  
      def _getBackupPath(self):
          if self.supportsBackupPaths():
              return "backups"
          return NECESSARY_OLD_BACKUP_PLURAL_NAME
  
      def supportsBackupPaths(self):
          return not self._super_version or self._super_version >= VERSION_BACKUP_PATH
  
      @supervisor_call
      async def createBackup(self, info):
          if 'folders' in info or 'addons' in info:
              url = self.getSupervisorURL().with_path("{0}/new/partial".format(self._getBackupPath()))
          else:
              url = self.getSupervisorURL().with_path("{0}/new/full".format(self._getBackupPath()))
          return await self._postHassioData(url, info, timeout=ClientTimeout(total=self.config.get(Setting.PENDING_BACKUP_TIMEOUT_SECONDS)))
  
      @supervisor_call
      async def auth(self, user: str, password: str) -> None:
          await self._postHassioData(self.getSupervisorURL().with_path("auth"), {"username": user, "password": password})
  
      @supervisor_call
      async def upload(self, stream):
          url = self.getSupervisorURL().with_path("{0}/new/upload".format(self._getBackupPath()))
          return await self._postHassioData(url, data=stream)
  
      @supervisor_call
      async def delete(self, slug) -> None:
          if slug in self.cache:
              del self.cache[slug]
          try:
              if self.supportsBackupPaths():
                  delete_url = self.getSupervisorURL().with_path("{1}/{0}".format(slug, self._getBackupPath()))
                  await self._sendHassioData("delete", delete_url, {})
              else:
                  delete_url = self.getSupervisorURL().with_path("{1}/{0}/remove".format(slug, self._getBackupPath()))
                  await self._sendHassioData("post", delete_url, {})
          except ClientResponseError as e:
              if e.status == 400:
                  raise HomeAssistantDeleteError()
              raise e
  
      @supervisor_call
      async def startAddon(self, slug) -> None:
          url = self.getSupervisorURL().with_path("addons/{0}/start".format(slug))
          await self._postHassioData(url, {})
  
      @supervisor_call
      async def stopAddon(self, slug) -> None:
          url = self.getSupervisorURL().with_path("addons/{0}/stop".format(slug))
          await self._postHassioData(url, {})
  
      @supervisor_call
      async def backup(self, slug):
          if slug in self.cache:
              info = self.cache[slug]
          else:
              info = await self._getHassioData(self.getSupervisorURL().with_path("{1}/{0}/info".format(slug, self._getBackupPath())))
              self.cache[slug] = info
          return HABackup(info, self._data_cache, self.config, self.config.isRetained(slug))
  
      @supervisor_call
      async def backups(self):
          return await self._getHassioData(self.getSupervisorURL().with_path(self._getBackupPath()))
  
      @supervisor_call
      async def haInfo(self):
          return await self._getHassioData(self.getSupervisorURL().with_path("core/info"))
  
      @supervisor_call
      async def selfInfo(self) -> Dict[str, Any]:
          return await self.getAddonInfo("self")
  
      @supervisor_call
      async def getAddonInfo(self, addon_slug) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("addons/{0}/info".format(addon_slug)))
  
      @supervisor_call
      async def getAddons(self) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("addons"))
  
      @supervisor_call
      async def hassosInfo(self) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("hassos/info"))
  
      @supervisor_call
      async def info(self) -> Dict[str, Any]:
          return await self._getHassioData(self.getSupervisorURL().with_path("info"))
  
      @supervisor_call
      async def refreshBackups(self):
          url = self.getSupervisorURL().with_path("{0}/reload".format(self._getBackupPath()))
          return await self._postHassioData(url)
  
      @supervisor_call
      async def supervisorInfo(self):
          url = self.getSupervisorURL().with_path("supervisor/info")
          info = await self._getHassioData(url)
  
          # parse the supervisor version
          if 'version' in info:
              self._super_version = Version.parse(info['version'])
          return info
  
      @supervisor_call
      async def restore(self, slug: str, password: str = None) -> None:
          url = self.getSupervisorURL().with_path("{1}/{0}/restore/full".format(slug, self._getBackupPath()))
          if password:
              await self._postHassioData(url, {'password': password})
          else:
              await self._postHassioData(url, {})
  
      @supervisor_call
      async def download(self, slug) -> AsyncHttpGetter:
          url = self.getSupervisorURL().with_path("{1}/{0}/download".format(slug, self._getBackupPath()))
          ret = AsyncHttpGetter(url,
                                self._getHassioHeaders(),
                                self.session,
                                timeoutFactory=SupervisorTimeoutError.factory,
                                otherErrorFactory=SupervisorUnexpectedError.factory,
                                timeout=ClientTimeout(sock_connect=self.config.get(Setting.DOWNLOAD_TIMEOUT_SECONDS),
                                                      sock_read=self.config.get(Setting.DOWNLOAD_TIMEOUT_SECONDS)),
                                time=self._time)
          return ret
  
      @supervisor_call
      async def getSuperLogs(self):
          url = self.getSupervisorURL().with_path("supervisor/logs")
          async with self.session.get(url, headers=self._getHassioHeaders()) as resp:
              resp.raise_for_status()
              return await resp.text()
  
      @supervisor_call
      async def getCoreLogs(self):
          url = self.getSupervisorURL().with_path("core/logs")
          async with self.session.get(url, headers=self._getHassioHeaders()) as resp:
              resp.raise_for_status()
              return await resp.text()
  
      async def _validateHassioReply(self, resp) -> Dict[str, Any]:
          async with
  
  [... the rest is omitted ...]
files: hassio-google-drive-backup/backup/ha/harequests.py
files: import asyncio
  from asyncio.tasks import sleep
  from datetime import timedelta
  import random
  import io
  
  from backup.config import Config, Version
  from backup.time import Time
  from aiohttp.web import (HTTPBadRequest, HTTPNotFound,
                           HTTPUnauthorized, Request, Response, get,
                           json_response, post, delete, FileResponse)
  from injector import inject, singleton
  from .base_server import BaseServer
  from .ports import Ports
  from typing import Any, Dict
  from tests.helpers import all_addons, createBackupTar, parseBackupInfo
  
  URL_MATCH_BACKUP_FULL = "^/backups/new/full$"
  URL_MATCH_BACKUP_DELETE = "^/backups/.*$"
  URL_MATCH_BACKUP_DOWNLOAD = "^/backups/.*/download$"
  URL_MATCH_MISC_INFO = "^/info$"
  URL_MATCH_CORE_API = "^/core/api.*$"
  URL_MATCH_START_ADDON = "^/addons/.*/start$"
  URL_MATCH_STOP_ADDON = "^/addons/.*/stop$"
  URL_MATCH_ADDON_INFO = "^/addons/.*/info$"
  URL_MATCH_SELF_OPTIONS = "^/addons/self/options$"
  
  URL_MATCH_SNAPSHOT = "^/snapshots.*$"
  URL_MATCH_BACKUPS = "^/backups.*$"
  
  
  @singleton
  class SimulatedSupervisor(BaseServer):
      @inject
      def __init__(self, config: Config, ports: Ports, time: Time):
          self._config = config
          self._time = time
          self._ports = ports
          self._auth_token = "test_header"
          self._backups: Dict[str, Any] = {}
          self._backup_data: Dict[str, bytearray] = {}
          self._backup_lock = asyncio.Lock()
          self._backup_inner_lock = asyncio.Lock()
          self._entities = {}
          self._events = []
          self._attributes = {}
          self._notification = None
          self._min_backup_size = 1024 * 1024 * 5
          self._max_backup_size = 1024 * 1024 * 5
          self._addon_slug = "self_slug"
          self._options = self.defaultOptions()
          self._username = "user"
          self._password = "pass"
          self._addons = all_addons.copy()
          self._super_version = Version(2021, 8)
  
          self.installAddon(self._addon_slug, "Home Assistant Google drive Backup")
          self.installAddon("42", "The answer")
          self.installAddon("sgadg", "sdgsagsdgsggsd")
  
      def defaultOptions(self):
          return {
              "max_backups_in_ha": 4,
              "max_backups_in_google_drive": 4,
              "days_between_backups": 3
          }
  
      def routes(self):
          return [
              post('/addons/{slug}/options', self._updateOptions),
              post("/core/api/services/persistent_notification/dismiss", self._dismissNotification),
              post("/core/api/services/persistent_notification/create", self._createNotification),
              post("/core/api/events/{name}", self._haEventUpdate),
              post("/core/api/states/{entity}", self._haStateUpdate),
              post('/auth', self._authenticate),
              get('/auth', self._authenticate),
              get('/info', self._miscInfo),
              get('/addons/self/info', self._selfInfo),
              get('/addons', self._allAddons),
              get('/addons/{slug}/info', self._addonInfo),
  
              post('/addons/{slug}/start', self._startAddon),
              post('/addons/{slug}/stop', self._stopAddon),
              get('/addons/{slug}/logo', self._logoAddon),
              get('/addons/{slug}/icon', self._logoAddon),
  
              get('/core/info', self._coreInfo),
              get('/supervisor/info', self._supervisorInfo),
              get('/supervisor/logs', self._supervisorLogs),
              get('/core/logs', self._coreLogs),
              get('/debug/insert/backup', self._debug_insert_backup),
              get('/debug/info', self._debugInfo),
  
              get('/backups', self._getBackups),
              delete('/backups/{slug}', self._deletebackup),
              post('/backups/new/upload', self._uploadbackup),
              post('/backups/new/partial', self._newbackup),
              post('/backups/new/full', self._newbackup),
              get('/backups/new/full', self._newbackup),
              get('/backups/{slug}/download', self._backupDownload),
              get('/backups/{slug}/info', self._backupDetail),
  
              # TODO: remove once the api path is fully deprecated
              get('/snapshots', self._getSnapshots),
              post('/snapshots/{slug}/remove', self._deletebackup),
              post('/snapshots/new/upload', self._uploadbackup),
              post('/snapshots/new/partial', self._newbackup),
              post('/snapshots/new/full', self._newbackup),
              get('/snapshots/new/full', self._newbackup),
              get('/snapshots/{slug}/download', self._backupDownload),
              get('/snapshots/{slug}/info', self._backupDetail),
          ]
  
      def getEvents(self):
          return self._events.copy()
  
      def getEntity(self, entity):
          return self._entities.get(entity)
  
      def clearEntities(self):
          self._entities = {}
  
      def addon(self, slug):
          for addon in self._addons:
              if addon["slug"] == slug:
                  return addon
          return None
  
      def getAttributes(self, attribute):
          return self._attributes.get(attribute)
  
      def getNotification(self):
          return self._notification
  
      def _formatErrorResponse(self, error: str) -> str:
          return json_response({'result': error})
  
      def _formatDataResponse(self, data: Any) -> str:
          return json_response({'result': 'ok', 'data': data})
  
      async def toggleBlockBackup(self):
          if self._backup_lock.locked():
              self._backup_lock.release()
          else:
              await self._backup_lock.acquire()
  
      async def _verifyHeader(self, request) -> bool:
          if request.headers.get("Authorization", None) == "Bearer " + self._auth_token:
              return
          if request.headers.get("X-Supervisor-Token", None) == self._auth_token:
              return
          raise HTTPUnauthorized()
  
      async def _getSnapshots(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({'snapshots': list(self._backups.values())})
  
      async def _getBackups(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({'backups': list(self._backups.values())})
  
      async def _stopAddon(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          for addon in self._addons:
              if addon.get("slug", "") == slug:
                  if addon.get("state") == "started":
                      addon["state"] = "stopped"
                      return self._formatDataResponse({})
          raise HTTPBadRequest()
  
      async def _logoAddon(self, request: Request):
          await self._verifyHeader(request)
          return FileResponse('hassio-google-drive-backup/backup/static/images/logo.png')
  
      async def _startAddon(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          for addon in self._addons:
              if addon.get("slug", "") == slug:
                  if addon.get("state") != "started":
                      addon["state"] = "started"
                      return self._formatDataResponse({})
          raise HTTPBadRequest()
  
      async def _addonInfo(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          for addon in self._addons:
              if addon.get("slug", "") == slug:
                  return self._formatDataResponse({
                      'boot': addon.get("boot"),
                      'watchdog': addon.get("watchdog"),
                      'state': addon.get("state"),
                  })
          raise HTTPBadRequest()
  
      async def _supervisorInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse(
              {
                  'version': str(self._super_version)
              }
          )
  
      async def _allAddons(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse(
              {
                  "add
  
  [... the rest is omitted ...]
files: hassio-google-drive-backup/dev/simulated_supervisor.py
files: import asyncio
  from asyncio.tasks import sleep
  from datetime import timedelta
  import random
  import io
  
  from backup.config import Config, Version
  from backup.time import Time
  from aiohttp.web import (HTTPBadRequest, HTTPNotFound,
                           HTTPUnauthorized, Request, Response, get,
                           json_response, post, delete, FileResponse)
  from injector import inject, singleton
  from .base_server import BaseServer
  from .ports import Ports
  from typing import Any, Dict
  from tests.helpers import all_addons, createBackupTar, parseBackupInfo
  
  URL_MATCH_BACKUP_FULL = "^/backups/new/full$"
  URL_MATCH_BACKUP_DELETE = "^/backups/.*$"
  URL_MATCH_BACKUP_DOWNLOAD = "^/backups/.*/download$"
  URL_MATCH_MISC_INFO = "^/info$"
  URL_MATCH_CORE_API = "^/core/api.*$"
  URL_MATCH_START_ADDON = "^/addons/.*/start$"
  URL_MATCH_STOP_ADDON = "^/addons/.*/stop$"
  URL_MATCH_ADDON_INFO = "^/addons/.*/info$"
  URL_MATCH_SELF_OPTIONS = "^/addons/self/options$"
  
  URL_MATCH_SNAPSHOT = "^/snapshots.*$"
  URL_MATCH_BACKUPS = "^/backups.*$"
  
  
  @singleton
  class SimulatedSupervisor(BaseServer):
      @inject
      def __init__(self, config: Config, ports: Ports, time: Time):
          self._config = config
          self._time = time
          self._ports = ports
          self._auth_token = "test_header"
          self._backups: Dict[str, Any] = {}
          self._backup_data: Dict[str, bytearray] = {}
          self._backup_lock = asyncio.Lock()
          self._backup_inner_lock = asyncio.Lock()
          self._entities = {}
          self._events = []
          self._attributes = {}
          self._notification = None
          self._min_backup_size = 1024 * 1024 * 5
          self._max_backup_size = 1024 * 1024 * 5
          self._addon_slug = "self_slug"
          self._options = self.defaultOptions()
          self._username = "user"
          self._password = "pass"
          self._addons = all_addons.copy()
          self._super_version = Version(2021, 8)
  
          self.installAddon(self._addon_slug, "Home Assistant Google drive Backup")
          self.installAddon("42", "The answer")
          self.installAddon("sgadg", "sdgsagsdgsggsd")
  
      def defaultOptions(self):
          return {
              "max_backups_in_ha": 4,
              "max_backups_in_google_drive": 4,
              "days_between_backups": 3
          }
  
      def routes(self):
          return [
              post('/addons/{slug}/options', self._updateOptions),
              post("/core/api/services/persistent_notification/dismiss", self._dismissNotification),
              post("/core/api/services/persistent_notification/create", self._createNotification),
              post("/core/api/events/{name}", self._haEventUpdate),
              post("/core/api/states/{entity}", self._haStateUpdate),
              post('/auth', self._authenticate),
              get('/auth', self._authenticate),
              get('/info', self._miscInfo),
              get('/addons/self/info', self._selfInfo),
              get('/addons', self._allAddons),
              get('/addons/{slug}/info', self._addonInfo),
  
              post('/addons/{slug}/start', self._startAddon),
              post('/addons/{slug}/stop', self._stopAddon),
              get('/addons/{slug}/logo', self._logoAddon),
              get('/addons/{slug}/icon', self._logoAddon),
  
              get('/core/info', self._coreInfo),
              get('/supervisor/info', self._supervisorInfo),
              get('/supervisor/logs', self._supervisorLogs),
              get('/core/logs', self._coreLogs),
              get('/debug/insert/backup', self._debug_insert_backup),
              get('/debug/info', self._debugInfo),
  
              get('/backups', self._getBackups),
              delete('/backups/{slug}', self._deletebackup),
              post('/backups/new/upload', self._uploadbackup),
              post('/backups/new/partial', self._newbackup),
              post('/backups/new/full', self._newbackup),
              get('/backups/new/full', self._newbackup),
              get('/backups/{slug}/download', self._backupDownload),
              get('/backups/{slug}/info', self._backupDetail),
  
              # TODO: remove once the api path is fully deprecated
              get('/snapshots', self._getSnapshots),
              post('/snapshots/{slug}/remove', self._deletebackup),
              post('/snapshots/new/upload', self._uploadbackup),
              post('/snapshots/new/partial', self._newbackup),
              post('/snapshots/new/full', self._newbackup),
              get('/snapshots/new/full', self._newbackup),
              get('/snapshots/{slug}/download', self._backupDownload),
              get('/snapshots/{slug}/info', self._backupDetail),
          ]
  
      def getEvents(self):
          return self._events.copy()
  
      def getEntity(self, entity):
          return self._entities.get(entity)
  
      def clearEntities(self):
          self._entities = {}
  
      def addon(self, slug):
          for addon in self._addons:
              if addon["slug"] == slug:
                  return addon
          return None
  
      def getAttributes(self, attribute):
          return self._attributes.get(attribute)
  
      def getNotification(self):
          return self._notification
  
      def _formatErrorResponse(self, error: str) -> str:
          return json_response({'result': error})
  
      def _formatDataResponse(self, data: Any) -> str:
          return json_response({'result': 'ok', 'data': data})
  
      async def toggleBlockBackup(self):
          if self._backup_lock.locked():
              self._backup_lock.release()
          else:
              await self._backup_lock.acquire()
  
      async def _verifyHeader(self, request) -> bool:
          if request.headers.get("Authorization", None) == "Bearer " + self._auth_token:
              return
          raise HTTPUnauthorized()
  
      async def _getSnapshots(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({'snapshots': list(self._backups.values())})
  
      async def _getBackups(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({'backups': list(self._backups.values())})
  
      async def _stopAddon(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          for addon in self._addons:
              if addon.get("slug", "") == slug:
                  if addon.get("state") == "started":
                      addon["state"] = "stopped"
                      return self._formatDataResponse({})
          raise HTTPBadRequest()
  
      async def _logoAddon(self, request: Request):
          await self._verifyHeader(request)
          return FileResponse('hassio-google-drive-backup/backup/static/images/logo.png')
  
      async def _startAddon(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          for addon in self._addons:
              if addon.get("slug", "") == slug:
                  if addon.get("state") != "started":
                      addon["state"] = "started"
                      return self._formatDataResponse({})
          raise HTTPBadRequest()
  
      async def _addonInfo(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          for addon in self._addons:
              if addon.get("slug", "") == slug:
                  return self._formatDataResponse({
                      'boot': addon.get("boot"),
                      'watchdog': addon.get("watchdog"),
                      'state': addon.get("state"),
                  })
          raise HTTPBadRequest()
  
      async def _supervisorInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse(
              {
                  'version': str(self._super_version)
              }
          )
  
      async def _allAddons(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse(
              {
                  "addons": list(self._addons).copy()
              }
          )
  
      async def _supervisorLogs(self,
  
  [... the rest is omitted ...]
files: hassio-google-drive-backup/dev/simulated_supervisor.py