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(self, resp) -> Dict[str, Any]:
          async with resp:
              resp.raise_for_status()
              details: Dict[str, Any] = await resp.json()
              if "result" not in details or details["result"] != "ok":
                  if "result" in details:
                      raise Exception("Hassio said: " + details["result"])
                  else:
                      raise Exception(
                          "Malformed response from Hassio: " + str(details))
  
              if "data" not in details:
                  return {}
              logger.trace("Hassio replied: %s", details)
              return details["data"]
  
      async def getAddonLogo(self, slug: str):
          url = self.getSupervisorURL().with_path("addons/{0}/icon".format(slug))
          async with self.session.get(url, headers=self._getAuthHeaders()) as resp:
              resp.raise_for_status()
              return (resp.headers['Content-Type'], await resp.read())
  
      def _getToken(self):
          configured = self.config.get(Setting.SUPERVISOR_TOKEN)
          if configured and len(configured) > 0:
              return configured
          if "SUPERVISOR_TOKEN" in os.environ:
              return os.environ.get("SUPERVISOR_TOKEN")
          # Older versions of the supervisor use a different name for the token.
          return os.environ.get("HASSIO_TOKEN")
  
      def _getAuthHeaders(self):
          return {
              'Authorization': 'Bearer ' + self._getToken()
          }
  
      def _altAuthHeaders(self):
          return {
              HEADER_TOKEN: self._getToken()
          }
  
      @supervisor_call
      async def _getHassioData(self, url: URL) -> Dict[str, Any]:
          logger.debug("Making Hassio request: " + str(url))
          return await self._validateHassioReply(await self.session.get(url, headers=self._getAuthHeaders()))
  
      async def _postHassioData(self, url: URL, json=None, file=None, data=None, timeout=None, headers=None) -> Dict[str, Any]:
          return await self._sendHassioData("post", url, json, file, data, timeout, headers)
  
      @supervisor_call
      async def _sendHassioData(self, method: str, url: URL, json=None, file=None, data=None, timeout=None, headers=None) -> Dict[str, Any]:
          if headers is None:
              headers = self._getAuthHeaders()
          logger.debug("Making Hassio request: " + str(url))
          return await self._validateHassioReply(await self.session.request(method, url, headers=headers, json=json, data=data, timeout=timeout))
  
      async def _postHaData(self, path: str, data: Dict[str, Any]) -> None:
          url = self.getSupervisorURL().with_path("/core/api/" + path)
          async with self.session.post(url, headers=self._getAuthHeaders(), json=data) as resp:
              resp.raise_for_status()
  
      async def sendNotification(self, title: str, message: str) -> None:
          data: Dict[str, str] = {
              "title": title,
              "message": message,
              "notification_id": NOTIFICATION_ID
          }
          await self._postHaData("services/persistent_notification/create", data)
  
      async def eventBackupStart(self, name, time):
          await self._sendEvent(EVENT_BACKUP_START, {
              'backup_name': name,
              'backup_time': str(time)
          })
  
      async def eventBackupEnd(self, name, time, completed):
          await self._sendEvent(EVENT_BACKUP_END, {
              'completed': completed,
              'backup_name': name,
              'backup_time': str(time)
          })
  
      async def _sendEvent(self, event_name: str, data: Dict[str, str]) -> None:
          await self._postHaData("events/" + event_name, data)
  
      async def dismissNotification(self) -> None:
          data: Dict[str, str] = {
              "notification_id": NOTIFICATION_ID
          }
          await self._postHaData("services/persistent_notification/dismiss", data)
  
      async def updateBackupStaleSensor(self, state: bool) -> None:
          if self.config.get(Setting.CALL_BACKUP_SNAPSHOT):
              data: Dict[str, Any] = {
                  "state": state,
                  "attributes": {
                      "friendly_name": "Snapshots Stale",
                      "device_class": "problem"
                  }
              }
              await self._postHaData("states/binary_sensor." + NECESSARY_OLD_BACKUP_PLURAL_NAME + "_stale", data)
          else:
              data: Dict[str, Any] = {
                  "state": state,
                  "attributes": {
                      "friendly_name": "Backups Stale",
                      "device_class": "problem"
                  }
              }
              await self._postHaData("states/binary_sensor.backups_stale", data)
  
      @supervisor_call
      async def updateConfig(self, config) -> None:
          return await self._postHassioData(self.getSupervisorURL().with_path("addons/self/options"), {'options': config})
  
      @supervisor_call
      async def updateAddonOptions(self, slug, options):
          return await self._postHassioData(self.getSupervisorURL().with_path("addons/{0}/options".format(slug)), options)
  
      async def updateEntity(self, entity, data):
          await self._postHaData("states/" + entity, data)
  
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 resp:
              resp.raise_for_status()
              details: Dict[str, Any] = await resp.json()
              if "result" not in details or details["result"] != "ok":
                  if "result" in details:
                      raise Exception("Hassio said: " + details["result"])
                  else:
                      raise Exception(
                          "Malformed response from Hassio: " + str(details))
  
              if "data" not in details:
                  return {}
              logger.trace("Hassio replied: %s", details)
              return details["data"]
  
      async def getAddonLogo(self, slug: str):
          url = self.getSupervisorURL().with_path("addons/{0}/icon".format(slug))
          async with self.session.get(url, headers=self._getHassioHeaders()) as resp:
              resp.raise_for_status()
              return (resp.headers['Content-Type'], await resp.read())
  
      def _getToken(self):
          configured = self.config.get(Setting.SUPERVISOR_TOKEN)
          if configured and len(configured) > 0:
              return configured
          if "SUPERVISOR_TOKEN" in os.environ:
              return os.environ.get("SUPERVISOR_TOKEN")
          # Older versions of the supervisor use a different name for the token.
          return os.environ.get("HASSIO_TOKEN")
  
      def _getHassioHeaders(self):
          return self._getHaHeaders()
  
      def _getHaHeaders(self):
          return {
              'Authorization': 'Bearer ' + self._getToken()
          }
  
      @supervisor_call
      async def _getHassioData(self, url: URL) -> Dict[str, Any]:
          logger.debug("Making Hassio request: " + str(url))
          return await self._validateHassioReply(await self.session.get(url, headers=self._getHassioHeaders()))
  
      async def _postHassioData(self, url: URL, json=None, file=None, data=None, timeout=None) -> Dict[str, Any]:
          return await self._sendHassioData("post", url, json, file, data, timeout)
  
      @supervisor_call
      async def _sendHassioData(self, method: str, url: URL, json=None, file=None, data=None, timeout=None) -> Dict[str, Any]:
          logger.debug("Making Hassio request: " + str(url))
          return await self._validateHassioReply(await self.session.request(method, url, headers=self._getHassioHeaders(), json=json, data=data, timeout=timeout))
  
      async def _postHaData(self, path: str, data: Dict[str, Any]) -> None:
          url = self.getSupervisorURL().with_path("/core/api/" + path)
          async with self.session.post(url, headers=self._getHaHeaders(), json=data) as resp:
              resp.raise_for_status()
  
      async def sendNotification(self, title: str, message: str) -> None:
          data: Dict[str, str] = {
              "title": title,
              "message": message,
              "notification_id": NOTIFICATION_ID
          }
          await self._postHaData("services/persistent_notification/create", data)
  
      async def eventBackupStart(self, name, time):
          await self._sendEvent(EVENT_BACKUP_START, {
              'backup_name': name,
              'backup_time': str(time)
          })
  
      async def eventBackupEnd(self, name, time, completed):
          await self._sendEvent(EVENT_BACKUP_END, {
              'completed': completed,
              'backup_name': name,
              'backup_time': str(time)
          })
  
      async def _sendEvent(self, event_name: str, data: Dict[str, str]) -> None:
          await self._postHaData("events/" + event_name, data)
  
      async def dismissNotification(self) -> None:
          data: Dict[str, str] = {
              "notification_id": NOTIFICATION_ID
          }
          await self._postHaData("services/persistent_notification/dismiss", data)
  
      async def updateBackupStaleSensor(self, state: bool) -> None:
          if self.config.get(Setting.CALL_BACKUP_SNAPSHOT):
              data: Dict[str, Any] = {
                  "state": state,
                  "attributes": {
                      "friendly_name": "Snapshots Stale",
                      "device_class": "problem"
                  }
              }
              await self._postHaData("states/binary_sensor." + NECESSARY_OLD_BACKUP_PLURAL_NAME + "_stale", data)
          else:
              data: Dict[str, Any] = {
                  "state": state,
                  "attributes": {
                      "friendly_name": "Backups Stale",
                      "device_class": "problem"
                  }
              }
              await self._postHaData("states/binary_sensor.backups_stale", data)
  
      @supervisor_call
      async def updateConfig(self, config) -> None:
          return await self._postHassioData(self.getSupervisorURL().with_path("addons/self/options"), {'options': config})
  
      @supervisor_call
      async def updateAddonOptions(self, slug, options):
          return await self._postHassioData(self.getSupervisorURL().with_path("addons/{0}/options".format(slug)), options)
  
      async def updateEntity(self, entity, data):
          await self._postHaData("states/" + entity, data)
  
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(
              {
                  "addons": list(self._addons).copy()
              }
          )
  
      async def _supervisorLogs(self, request: Request):
          await self._verifyHeader(request)
          return Response(body="Supervisor Log line 1\nSupervisor Log Line 2")
  
      async def _coreLogs(self, request: Request):
          await self._verifyHeader(request)
          return Response(body="Core Log line 1\nCore Log Line 2")
  
      async def _coreInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse(
              {
                  "version": "1.3.3.7",
                  "last_version": "1.3.3.8",
                  "machine": "VS Dev",
                  "ip_address": "127.0.0.1",
                  "arch": "x86",
                  "image": "image",
                  "custom": "false",
                  "boot": "true",
                  "port": self._ports.server,
                  "ssl": "false",
                  "watchdog": "what is this",
                  "wait_boot": "so many arguments"
              }
          )
  
      async def _internalNewBackup(self, request: Request, input_json, date=None, verify_header=True) -> str:
          async with self._backup_lock:
              async with self._backup_inner_lock:
                  if 'wait' in input_json:
                      await sleep(input_json['wait'])
                  if verify_header:
                      await self._verifyHeader(request)
                  slug = self.generateId(8)
                  password = input_json.get('password', None)
                  data = createBackupTar(
                      slug,
                      input_json.get('name', "Default name"),
                      date=date or self._time.now(),
                      padSize=int(random.uniform(self._min_backup_size, self._max_backup_size)),
                      included_folders=input_json.get('folders', None),
                      included_addons=input_json.get('addons', None),
                      password=password)
                  backup_info = parseBackupInfo(data)
                  self._backups[slug] = backup_info
                  self._backup_data[slug] = bytearray(data.getbuffer())
                  return slug
  
      async def createBackup(self, input_json, date=None):
          return await self._internalNewBackup(None, input_json, date=date, verify_header=False)
  
      async def _newbackup(self, request: Request):
          if self._backup_lock.locked():
              raise HTTPBadRequest()
          input_json = await request.json()
          task = asyncio.shield(asyncio.create_task(self._internalNewBackup(request, input_json)))
          return self._formatDataResponse({"slug": await task})
  
      async def _uploadbackup(self, request: Request):
          await self._verifyHeader(request)
          try:
              reader = await request.multipart()
              contents = await reader.next()
              received_bytes = bytearray()
              while True:
                  chunk = await contents.read_chunk()
                  if not chunk:
                      break
                  received_bytes.extend(chunk)
              info = parseBackupInfo(io.BytesIO(received_bytes))
              self._backups[info['slug']] = info
              self._backup_data[info['slug']] = received_bytes
              return self._formatDataResponse({"slug": info['slug']})
          except Exception as e:
              print(str(e))
              return self._formatErrorResponse("Bad backup")
  
      async def _deletebackup(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          if slug not in self._backups:
              raise HTTPNotFound()
          del self._backups[slug]
          del self._backup_data[slug]
          return self._formatDataResponse("deleted")
  
      async def _backupDetail(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          if slug not in self._backups:
              raise HTTPNotFound()
          return self._formatDataResponse(self._backups[slug])
  
      async def _backupDownload(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          if slug not in self._backup_data:
              raise HTTPNotFound()
          return self.serve_bytes(request, self._backup_data[slug])
  
      async def _selfInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({
              "webui": "http://some/address",
              'ingress_url': "fill me in later",
              "slug": self._addon_slug,
              "options": self._options
          })
  
      async def _debugInfo(self, request: Request):
          return self._formatDataResponse({
              "config": {
                  "   webui": "http://some/address",
                  'ingress_url': "fill me in later",
                  "slug": self._addon_slug,
                  "options": self._options
              }
          })
  
      async def _miscInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({
              "supervisor": "super version",
              "homeassistant": "ha version",
              "hassos": "hassos version",
              "hostname": "hostname",
              "machine": "machine",
              "arch": "Arch",
              "supported_arch": "supported arch",
              "channel": "channel"
          })
  
      def installAddon(self, slug, name, version="v1.0", boot=True, started=True):
          self._addons.append({
              "name": 'Name for ' + name,
              "slug": slug,
              "description": slug + " description",
              "version": version,
              "watchdog": False,
              "boot": "auto" if boot else "manual",
              "logo": True,
              "ingress_entry": "/api/hassio_ingress/" + slug,
              "state": "started" if started else "stopped"
          })
  
      async def _authenticate(self, request: Request):
          await self._verifyHeader(request)
          input_json = await request.json()
          if input_json.get("username") != self._username or input_json.get("password") != self._password:
              raise HTTPBadRequest()
          return self._formatDataResponse({})
  
      async def _updateOptions(self, request: Request):
          slug = request.match_info.get('slug')
  
          if slug == "self":
              await self._verifyHeader(request)
              self._options = (await request.json())['options'].copy()
          else:
              self.addon(slug).update(await request.json())
          return self._formatDataResponse({})
  
      async def _haStateUpdate(self, request: Request):
          await self._verifyHeader(request)
          entity = request.match_info.get('entity')
          json = await request.json()
          self._entities[entity] = json['state']
          self._attributes[entity] = json['attributes']
          return Response()
  
      async def _haEventUpdate(self, request: Request):
          await self._verifyHeader(request)
          name = request.match_info.get('name')
          self._events.append((name, await request.json()))
          return Response()
  
      async def _createNotification(self, request: Request):
          await self._verifyHeader(request)
          notification = await request.json()
          print("Created notification with: {}".format(notification))
          self._notification = notification.copy()
          return Response()
  
      async def _dismissNotification(self, request: Request):
          await self._verifyHeader(request)
          print("Dismissed notification with: {}".format(await request.json()))
          self._notification = None
          return Response()
  
      async def _debug_insert_backup(self, request: Request) -> bool:
          days_back = int(request.query.get("days"))
          date = self._time.now() - timedelta(days=days_back)
          name = date.strftime("Full Backup %Y-%m-%d %H:%M-%S")
          wait = int(request.query.get("wait", 0))
          slug = await self._internalNewBackup(request, {'name': name, 'wait': wait}, date=date, verify_header=False)
          return self._formatDataResponse({'slug': slug})
  
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, request: Request):
          await self._verifyHeader(request)
          return Response(body="Supervisor Log line 1\nSupervisor Log Line 2")
  
      async def _coreLogs(self, request: Request):
          await self._verifyHeader(request)
          return Response(body="Core Log line 1\nCore Log Line 2")
  
      async def _coreInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse(
              {
                  "version": "1.3.3.7",
                  "last_version": "1.3.3.8",
                  "machine": "VS Dev",
                  "ip_address": "127.0.0.1",
                  "arch": "x86",
                  "image": "image",
                  "custom": "false",
                  "boot": "true",
                  "port": self._ports.server,
                  "ssl": "false",
                  "watchdog": "what is this",
                  "wait_boot": "so many arguments"
              }
          )
  
      async def _internalNewBackup(self, request: Request, input_json, date=None, verify_header=True) -> str:
          async with self._backup_lock:
              async with self._backup_inner_lock:
                  if 'wait' in input_json:
                      await sleep(input_json['wait'])
                  if verify_header:
                      await self._verifyHeader(request)
                  slug = self.generateId(8)
                  password = input_json.get('password', None)
                  data = createBackupTar(
                      slug,
                      input_json.get('name', "Default name"),
                      date=date or self._time.now(),
                      padSize=int(random.uniform(self._min_backup_size, self._max_backup_size)),
                      included_folders=input_json.get('folders', None),
                      included_addons=input_json.get('addons', None),
                      password=password)
                  backup_info = parseBackupInfo(data)
                  self._backups[slug] = backup_info
                  self._backup_data[slug] = bytearray(data.getbuffer())
                  return slug
  
      async def createBackup(self, input_json, date=None):
          return await self._internalNewBackup(None, input_json, date=date, verify_header=False)
  
      async def _newbackup(self, request: Request):
          if self._backup_lock.locked():
              raise HTTPBadRequest()
          input_json = await request.json()
          task = asyncio.shield(asyncio.create_task(self._internalNewBackup(request, input_json)))
          return self._formatDataResponse({"slug": await task})
  
      async def _uploadbackup(self, request: Request):
          await self._verifyHeader(request)
          try:
              reader = await request.multipart()
              contents = await reader.next()
              received_bytes = bytearray()
              while True:
                  chunk = await contents.read_chunk()
                  if not chunk:
                      break
                  received_bytes.extend(chunk)
              info = parseBackupInfo(io.BytesIO(received_bytes))
              self._backups[info['slug']] = info
              self._backup_data[info['slug']] = received_bytes
              return self._formatDataResponse({"slug": info['slug']})
          except Exception as e:
              print(str(e))
              return self._formatErrorResponse("Bad backup")
  
      async def _deletebackup(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          if slug not in self._backups:
              raise HTTPNotFound()
          del self._backups[slug]
          del self._backup_data[slug]
          return self._formatDataResponse("deleted")
  
      async def _backupDetail(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          if slug not in self._backups:
              raise HTTPNotFound()
          return self._formatDataResponse(self._backups[slug])
  
      async def _backupDownload(self, request: Request):
          await self._verifyHeader(request)
          slug = request.match_info.get('slug')
          if slug not in self._backup_data:
              raise HTTPNotFound()
          return self.serve_bytes(request, self._backup_data[slug])
  
      async def _selfInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({
              "webui": "http://some/address",
              'ingress_url': "fill me in later",
              "slug": self._addon_slug,
              "options": self._options
          })
  
      async def _debugInfo(self, request: Request):
          return self._formatDataResponse({
              "config": {
                  "   webui": "http://some/address",
                  'ingress_url': "fill me in later",
                  "slug": self._addon_slug,
                  "options": self._options
              }
          })
  
      async def _miscInfo(self, request: Request):
          await self._verifyHeader(request)
          return self._formatDataResponse({
              "supervisor": "super version",
              "homeassistant": "ha version",
              "hassos": "hassos version",
              "hostname": "hostname",
              "machine": "machine",
              "arch": "Arch",
              "supported_arch": "supported arch",
              "channel": "channel"
          })
  
      def installAddon(self, slug, name, version="v1.0", boot=True, started=True):
          self._addons.append({
              "name": 'Name for ' + name,
              "slug": slug,
              "description": slug + " description",
              "version": version,
              "watchdog": False,
              "boot": "auto" if boot else "manual",
              "logo": True,
              "ingress_entry": "/api/hassio_ingress/" + slug,
              "state": "started" if started else "stopped"
          })
  
      async def _authenticate(self, request: Request):
          await self._verifyHeader(request)
          input_json = await request.json()
          if input_json.get("username") != self._username or input_json.get("password") != self._password:
              raise HTTPBadRequest()
          return self._formatDataResponse({})
  
      async def _updateOptions(self, request: Request):
          slug = request.match_info.get('slug')
  
          if slug == "self":
              await self._verifyHeader(request)
              self._options = (await request.json())['options'].copy()
          else:
              self.addon(slug).update(await request.json())
          return self._formatDataResponse({})
  
      async def _haStateUpdate(self, request: Request):
          await self._verifyHeader(request)
          entity = request.match_info.get('entity')
          json = await request.json()
          self._entities[entity] = json['state']
          self._attributes[entity] = json['attributes']
          return Response()
  
      async def _haEventUpdate(self, request: Request):
          await self._verifyHeader(request)
          name = request.match_info.get('name')
          self._events.append((name, await request.json()))
          return Response()
  
      async def _createNotification(self, request: Request):
          await self._verifyHeader(request)
          notification = await request.json()
          print("Created notification with: {}".format(notification))
          self._notification = notification.copy()
          return Response()
  
      async def _dismissNotification(self, request: Request):
          await self._verifyHeader(request)
          print("Dismissed notification with: {}".format(await request.json()))
          self._notification = None
          return Response()
  
      async def _debug_insert_backup(self, request: Request) -> bool:
          days_back = int(request.query.get("days"))
          date = self._time.now() - timedelta(days=days_back)
          name = date.strftime("Full Backup %Y-%m-%d %H:%M-%S")
          wait = int(request.query.get("wait", 0))
          slug = await self._internalNewBackup(request, {'name': name, 'wait': wait}, date=date, verify_header=False)
          return self._formatDataResponse({'slug': slug})
  
files: hassio-google-drive-backup/dev/simulated_supervisor.py