evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add dependency on google-photos-library-api: Change the Google Photos client library to a new external package (#125040)

Browse source 
* Change the Google Photos client library to a new external package

* Remove mime type guessing

* Update tests to mock out the client library and iterators

* Update homeassistant/components/google_photos/media_source.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Allen Porter 2024-09-03 04:54:43 -07:00 • committed by GitHub
parent b9db9eeab2
commit c07a9e9d59
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 281 additions and 412 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
homeassistant/components/google_photos/__init__.py
View file

@ -3,17 +3,17 @@
from __future__ import annotations from __future__ import annotations
from aiohttp import ClientError, ClientResponseError from aiohttp import ClientError, ClientResponseError
from google_photos_library_api.api import GooglePhotosLibraryApi
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_entry_oauth2_flow from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from . import api from . import api
from .const import DOMAIN from .const import DOMAIN
from .services import async_register_services from .services import async_register_services
from .types import GooglePhotosConfigEntry
type GooglePhotosConfigEntry = ConfigEntry[api.AsyncConfigEntryAuth]
__all__ = [ __all__ = [
"DOMAIN", "DOMAIN",
@ -29,8 +29,9 @@ async def async_setup_entry(
hass, entry hass, entry
) )
) )
session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation) web_session = async_get_clientsession(hass)
auth = api.AsyncConfigEntryAuth(hass, session) oauth_session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
auth = api.AsyncConfigEntryAuth(web_session, oauth_session)
try: try:
await auth.async_get_access_token() await auth.async_get_access_token()
except ClientResponseError as err: except ClientResponseError as err:
@ -41,7 +42,7 @@ async def async_setup_entry(
raise ConfigEntryNotReady from err raise ConfigEntryNotReady from err
except ClientError as err: except ClientError as err:
raise ConfigEntryNotReady from err raise ConfigEntryNotReady from err
entry.runtime_data = auth entry.runtime_data = GooglePhotosLibraryApi(auth)
async_register_services(hass) async_register_services(hass)

198
homeassistant/components/google_photos/api.py
View file

@ -1,216 +1,44 @@
"""API for Google Photos bound to Home Assistant OAuth.""" """API for Google Photos bound to Home Assistant OAuth."""
from abc import ABC, abstractmethod from typing import cast
from functools import partial
import logging
from typing import Any, cast
from aiohttp.client_exceptions import ClientError import aiohttp
from google.oauth2.credentials import Credentials from google_photos_library_api import api
from googleapiclient.discovery import Resource, build
from googleapiclient.errors import HttpError
from googleapiclient.http import HttpRequest
from homeassistant.const import CONF_ACCESS_TOKEN from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.core import HomeAssistant from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
from .exceptions import GooglePhotosApiError
_LOGGER = logging.getLogger(__name__)
DEFAULT_PAGE_SIZE = 20
# Only included necessary fields to limit response sizes
GET_MEDIA_ITEM_FIELDS = (
"id,baseUrl,mimeType,filename,mediaMetadata(width,height,photo,video)"
)
LIST_MEDIA_ITEM_FIELDS = f"nextPageToken,mediaItems({GET_MEDIA_ITEM_FIELDS})"
UPLOAD_API = "https://photoslibrary.googleapis.com/v1/uploads"
LIST_ALBUMS_FIELDS = (
"nextPageToken,albums(id,title,coverPhotoBaseUrl,coverPhotoMediaItemId)"
)
class AuthBase(ABC): class AsyncConfigEntryAuth(api.AbstractAuth):
"""Base class for Google Photos authentication library.
Provides an asyncio interface around the blocking client library.
"""
def __init__(
self,
hass: HomeAssistant,
) -> None:
"""Initialize Google Photos auth."""
self._hass = hass
@abstractmethod
async def async_get_access_token(self) -> str:
"""Return a valid access token."""
async def get_user_info(self) -> dict[str, Any]:
"""Get the user profile info."""
service = await self._get_profile_service()
cmd: HttpRequest = service.userinfo().get()
return await self._execute(cmd)
async def get_media_item(self, media_item_id: str) -> dict[str, Any]:
"""Get all MediaItem resources."""
service = await self._get_photos_service()
cmd: HttpRequest = service.mediaItems().get(
mediaItemId=media_item_id, fields=GET_MEDIA_ITEM_FIELDS
)
return await self._execute(cmd)
async def list_media_items(
self,
page_size: int | None = None,
page_token: str | None = None,
album_id: str | None = None,
favorites: bool = False,
) -> dict[str, Any]:
"""Get all MediaItem resources."""
service = await self._get_photos_service()
args: dict[str, Any] = {
"pageSize": (page_size or DEFAULT_PAGE_SIZE),
"pageToken": page_token,
}
cmd: HttpRequest
if album_id is not None or favorites:
if album_id is not None:
args["albumId"] = album_id
if favorites:
args["filters"] = {"featureFilter": {"includedFeatures": "FAVORITES"}}
cmd = service.mediaItems().search(body=args, fields=LIST_MEDIA_ITEM_FIELDS)
else:
cmd = service.mediaItems().list(**args, fields=LIST_MEDIA_ITEM_FIELDS)
return await self._execute(cmd)
async def list_albums(
self, page_size: int | None = None, page_token: str | None = None
) -> dict[str, Any]:
"""Get all Album resources."""
service = await self._get_photos_service()
cmd: HttpRequest = service.albums().list(
pageSize=(page_size or DEFAULT_PAGE_SIZE),
pageToken=page_token,
fields=LIST_ALBUMS_FIELDS,
)
return await self._execute(cmd)
async def upload_content(self, content: bytes, mime_type: str) -> str:
"""Upload media content to the API and return an upload token."""
token = await self.async_get_access_token()
session = aiohttp_client.async_get_clientsession(self._hass)
try:
result = await session.post(
UPLOAD_API, headers=_upload_headers(token, mime_type), data=content
)
result.raise_for_status()
return await result.text()
except ClientError as err:
raise GooglePhotosApiError(f"Failed to upload content: {err}") from err
async def create_media_items(self, upload_tokens: list[str]) -> list[str]:
"""Create a batch of media items and return the ids."""
service = await self._get_photos_service()
cmd: HttpRequest = service.mediaItems().batchCreate(
body={
"newMediaItems": [
{
"simpleMediaItem": {
"uploadToken": upload_token,
}
for upload_token in upload_tokens
}
]
}
)
result = await self._execute(cmd)
return [
media_item["mediaItem"]["id"]
for media_item in result["newMediaItemResults"]
]
async def _get_photos_service(self) -> Resource:
"""Get current photos library API resource."""
token = await self.async_get_access_token()
return await self._hass.async_add_executor_job(
partial(
build,
"photoslibrary",
"v1",
credentials=Credentials(token=token), # type: ignore[no-untyped-call]
static_discovery=False,
)
)
async def _get_profile_service(self) -> Resource:
"""Get current profile service API resource."""
token = await self.async_get_access_token()
return await self._hass.async_add_executor_job(
partial(build, "oauth2", "v2", credentials=Credentials(token=token)) # type: ignore[no-untyped-call]
)
async def _execute(self, request: HttpRequest) -> dict[str, Any]:
try:
result = await self._hass.async_add_executor_job(request.execute)
except HttpError as err:
raise GooglePhotosApiError(
f"Google Photos API responded with error ({err.status_code}): {err.reason}"
) from err
if not isinstance(result, dict):
raise GooglePhotosApiError(
f"Google Photos API replied with unexpected response: {result}"
)
if error := result.get("error"):
message = error.get("message", "Unknown Error")
raise GooglePhotosApiError(f"Google Photos API response: {message}")
return cast(dict[str, Any], result)
class AsyncConfigEntryAuth(AuthBase):
"""Provide Google Photos authentication tied to an OAuth2 based config entry.""" """Provide Google Photos authentication tied to an OAuth2 based config entry."""
def __init__( def __init__(
self, self,
hass: HomeAssistant, websession: aiohttp.ClientSession,
oauth_session: config_entry_oauth2_flow.OAuth2Session, oauth_session: config_entry_oauth2_flow.OAuth2Session,
) -> None: ) -> None:
"""Initialize AsyncConfigEntryAuth.""" """Initialize AsyncConfigEntryAuth."""
super().__init__(hass) super().__init__(websession)
self._oauth_session = oauth_session self._session = oauth_session
async def async_get_access_token(self) -> str: async def async_get_access_token(self) -> str:
"""Return a valid access token.""" """Return a valid access token."""
if not self._oauth_session.valid_token: await self._session.async_ensure_token_valid()
await self._oauth_session.async_ensure_token_valid() return cast(str, self._session.token[CONF_ACCESS_TOKEN])
return cast(str, self._oauth_session.token[CONF_ACCESS_TOKEN])
class AsyncConfigFlowAuth(AuthBase): class AsyncConfigFlowAuth(api.AbstractAuth):
"""An API client used during the config flow with a fixed token.""" """An API client used during the config flow with a fixed token."""
def __init__( def __init__(
self, self,
hass: HomeAssistant, websession: aiohttp.ClientSession,
token: str, token: str,
) -> None: ) -> None:
"""Initialize ConfigFlowAuth.""" """Initialize ConfigFlowAuth."""
super().__init__(hass) super().__init__(websession)
self._token = token self._token = token
async def async_get_access_token(self) -> str: async def async_get_access_token(self) -> str:
"""Return a valid access token.""" """Return a valid access token."""
return self._token return self._token
def _upload_headers(token: str, mime_type: str) -> dict[str, Any]:
"""Create the upload headers."""
return {
"Authorization": f"Bearer {token}",
"Content-Type": "application/octet-stream",
"X-Goog-Upload-Content-Type": mime_type,
"X-Goog-Upload-Protocol": "raw",
}

15
homeassistant/components/google_photos/config_flow.py
View file

@ -4,13 +4,15 @@ from collections.abc import Mapping
import logging import logging
from typing import Any from typing import Any
from google_photos_library_api.api import GooglePhotosLibraryApi
from google_photos_library_api.exceptions import GooglePhotosApiError
from homeassistant.config_entries import ConfigFlowResult from homeassistant.config_entries import ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN
from homeassistant.helpers import config_entry_oauth2_flow from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
from . import GooglePhotosConfigEntry, api from . import GooglePhotosConfigEntry, api
from .const import DOMAIN, OAUTH2_SCOPES from .const import DOMAIN, OAUTH2_SCOPES
from .exceptions import GooglePhotosApiError
class OAuth2FlowHandler( class OAuth2FlowHandler(
@ -39,7 +41,10 @@ class OAuth2FlowHandler(
async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult: async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
"""Create an entry for the flow.""" """Create an entry for the flow."""
client = api.AsyncConfigFlowAuth(self.hass, data[CONF_TOKEN][CONF_ACCESS_TOKEN]) session = aiohttp_client.async_get_clientsession(self.hass)
auth = api.AsyncConfigFlowAuth(session, data[CONF_TOKEN][CONF_ACCESS_TOKEN])
client = GooglePhotosLibraryApi(auth)
try: try:
user_resource_info = await client.get_user_info() user_resource_info = await client.get_user_info()
await client.list_media_items(page_size=1) await client.list_media_items(page_size=1)
@ -51,7 +56,7 @@ class OAuth2FlowHandler(
except Exception: except Exception:
self.logger.exception("Unknown error occurred") self.logger.exception("Unknown error occurred")
return self.async_abort(reason="unknown") return self.async_abort(reason="unknown")
user_id = user_resource_info["id"] user_id = user_resource_info.id
if self.reauth_entry: if self.reauth_entry:
if self.reauth_entry.unique_id == user_id: if self.reauth_entry.unique_id == user_id:
@ -62,7 +67,7 @@ class OAuth2FlowHandler(
await self.async_set_unique_id(user_id) await self.async_set_unique_id(user_id)
self._abort_if_unique_id_configured() self._abort_if_unique_id_configured()
return self.async_create_entry(title=user_resource_info["name"], data=data) return self.async_create_entry(title=user_resource_info.name, data=data)
async def async_step_reauth( async def async_step_reauth(
self, entry_data: Mapping[str, Any] self, entry_data: Mapping[str, Any]

7
homeassistant/components/google_photos/exceptions.py
View file

@ -1,7 +0,0 @@
"""Exceptions for Google Photos api calls."""
from homeassistant.exceptions import HomeAssistantError
class GooglePhotosApiError(HomeAssistantError):
"""Error talking to the Google Photos API."""

4
homeassistant/components/google_photos/manifest.json
View file

@ -6,6 +6,6 @@
"dependencies": ["application_credentials"], "dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/google_photos", "documentation": "https://www.home-assistant.io/integrations/google_photos",
"iot_class": "cloud_polling", "iot_class": "cloud_polling",
"loggers": ["googleapiclient"], "loggers": ["google_photos_library_api"],
"requirements": ["google-api-python-client==2.71.0"] "requirements": ["google-photos-library-api==0.8.0"]
} }

99
homeassistant/components/google_photos/media_source.py
View file

@ -5,6 +5,9 @@ from enum import Enum, StrEnum
import logging import logging
from typing import Any, Self, cast from typing import Any, Self, cast
from google_photos_library_api.exceptions import GooglePhotosApiError
from google_photos_library_api.model import Album, MediaItem
from homeassistant.components.media_player import MediaClass, MediaType from homeassistant.components.media_player import MediaClass, MediaType
from homeassistant.components.media_source import ( from homeassistant.components.media_source import (
BrowseError, BrowseError,
@ -17,17 +20,12 @@ from homeassistant.core import HomeAssistant
from . import GooglePhotosConfigEntry from . import GooglePhotosConfigEntry
from .const import DOMAIN, READ_SCOPES from .const import DOMAIN, READ_SCOPES
from .exceptions import GooglePhotosApiError
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
# Media Sources do not support paging, so we only show a subset of recent MAX_RECENT_PHOTOS = 100
# photos when displaying the users library. We fetch a minimum of 50 photos MEDIA_ITEMS_PAGE_SIZE = 100
# unless we run out, but in pages of 100 at a time given sometimes responses ALBUM_PAGE_SIZE = 50
# may only contain a handful of items Fetches at least 50 photos.
MAX_RECENT_PHOTOS = 50
MAX_ALBUMS = 50
PAGE_SIZE = 100
THUMBNAIL_SIZE = 256 THUMBNAIL_SIZE = 256
LARGE_IMAGE_SIZE = 2160 LARGE_IMAGE_SIZE = 2160
@ -158,14 +156,15 @@ class GooglePhotosMediaSource(MediaSource):
entry = self._async_config_entry(identifier.config_entry_id) entry = self._async_config_entry(identifier.config_entry_id)
client = entry.runtime_data client = entry.runtime_data
media_item = await client.get_media_item(media_item_id=identifier.media_id) media_item = await client.get_media_item(media_item_id=identifier.media_id)
is_video = media_item["mediaMetadata"].get("video") is not None if not media_item.mime_type:
raise BrowseError("Could not determine mime type of media item")
if media_item.media_metadata and (media_item.media_metadata.video is not None):
url = _video_url(media_item)
else:
url = _media_url(media_item, LARGE_IMAGE_SIZE)
return PlayMedia( return PlayMedia(
url=( url=url,
_video_url(media_item) mime_type=media_item.mime_type,
if is_video
else _media_url(media_item, LARGE_IMAGE_SIZE)
),
mime_type=media_item["mimeType"],
) )
async def async_browse_media(self, item: MediaSourceItem) -> BrowseMediaSource: async def async_browse_media(self, item: MediaSourceItem) -> BrowseMediaSource:
@ -199,7 +198,6 @@ class GooglePhotosMediaSource(MediaSource):
source = _build_account(entry, identifier) source = _build_account(entry, identifier)
if identifier.id_type is None: if identifier.id_type is None:
result = await client.list_albums(page_size=MAX_ALBUMS)
source.children = [ source.children = [
_build_album( _build_album(
special_album.value.title, special_album.value.title,
@ -208,17 +206,27 @@ class GooglePhotosMediaSource(MediaSource):
), ),
) )
for special_album in SpecialAlbum for special_album in SpecialAlbum
] + [ ]
albums: list[Album] = []
try:
async for album_result in await client.list_albums(
page_size=ALBUM_PAGE_SIZE
):
albums.extend(album_result.albums)
except GooglePhotosApiError as err:
raise BrowseError(f"Error listing albums: {err}") from err
source.children.extend(
_build_album( _build_album(
album["title"], album.title,
PhotosIdentifier.album( PhotosIdentifier.album(
identifier.config_entry_id, identifier.config_entry_id,
album["id"], album.id,
), ),
_cover_photo_url(album, THUMBNAIL_SIZE), _cover_photo_url(album, THUMBNAIL_SIZE),
) )
for album in result["albums"] for album in albums
] )
return source return source
if ( if (
@ -233,28 +241,24 @@ class GooglePhotosMediaSource(MediaSource):
else: else:
list_args = {"album_id": identifier.media_id} list_args = {"album_id": identifier.media_id}
media_items: list[dict[str, Any]] = [] media_items: list[MediaItem] = []
page_token: str | None = None
while (
not special_album
or (max_photos := special_album.value.max_photos) is None
or len(media_items) < max_photos
):
try: try:
result = await client.list_media_items( async for media_item_result in await client.list_media_items(
**list_args, page_size=PAGE_SIZE, page_token=page_token **list_args, page_size=MEDIA_ITEMS_PAGE_SIZE
) ):
media_items.extend(media_item_result.media_items)
if (
special_album
and (max_photos := special_album.value.max_photos)
and len(media_items) > max_photos
):
break
except GooglePhotosApiError as err: except GooglePhotosApiError as err:
raise BrowseError(f"Error listing media items: {err}") from err raise BrowseError(f"Error listing media items: {err}") from err
media_items.extend(result["mediaItems"])
page_token = result.get("nextPageToken")
if page_token is None:
break
# Render the grid of media item results
source.children = [ source.children = [
_build_media_item( _build_media_item(
PhotosIdentifier.photo(identifier.config_entry_id, media_item["id"]), PhotosIdentifier.photo(identifier.config_entry_id, media_item.id),
media_item, media_item,
) )
for media_item in media_items for media_item in media_items
@ -315,38 +319,41 @@ def _build_album(
def _build_media_item( def _build_media_item(
identifier: PhotosIdentifier, media_item: dict[str, Any] identifier: PhotosIdentifier,
media_item: MediaItem,
) -> BrowseMediaSource: ) -> BrowseMediaSource:
"""Build the node for an individual photo or video.""" """Build the node for an individual photo or video."""
is_video = media_item["mediaMetadata"].get("video") is not None is_video = media_item.media_metadata and (
media_item.media_metadata.video is not None
)
return BrowseMediaSource( return BrowseMediaSource(
domain=DOMAIN, domain=DOMAIN,
identifier=identifier.as_string(), identifier=identifier.as_string(),
media_class=MediaClass.IMAGE if not is_video else MediaClass.VIDEO, media_class=MediaClass.IMAGE if not is_video else MediaClass.VIDEO,
media_content_type=MediaType.IMAGE if not is_video else MediaType.VIDEO, media_content_type=MediaType.IMAGE if not is_video else MediaType.VIDEO,
title=media_item["filename"], title=media_item.filename,
can_play=is_video, can_play=is_video,
can_expand=False, can_expand=False,
thumbnail=_media_url(media_item, THUMBNAIL_SIZE), thumbnail=_media_url(media_item, THUMBNAIL_SIZE),
) )
def _media_url(media_item: dict[str, Any], max_size: int) -> str: def _media_url(media_item: MediaItem, max_size: int) -> str:
"""Return a media item url with the specified max thumbnail size on the longest edge. """Return a media item url with the specified max thumbnail size on the longest edge.
See https://developers.google.com/photos/library/guides/access-media-items#base-urls See https://developers.google.com/photos/library/guides/access-media-items#base-urls
""" """
return f"{media_item["baseUrl"]}=h{max_size}" return f"{media_item.base_url}=h{max_size}"
def _video_url(media_item: dict[str, Any]) -> str: def _video_url(media_item: MediaItem) -> str:
"""Return a video url for the item. """Return a video url for the item.
See https://developers.google.com/photos/library/guides/access-media-items#base-urls See https://developers.google.com/photos/library/guides/access-media-items#base-urls
""" """
return f"{media_item["baseUrl"]}=dv" return f"{media_item.base_url}=dv"
def _cover_photo_url(album: dict[str, Any], max_size: int) -> str: def _cover_photo_url(album: Album, max_size: int) -> str:
"""Return a media item url for the cover photo of the album.""" """Return a media item url for the cover photo of the album."""
return f"{album["coverPhotoBaseUrl"]}=h{max_size}" return f"{album.cover_photo_base_url}=h{max_size}"

44
homeassistant/components/google_photos/services.py
View file

@ -6,9 +6,10 @@ import asyncio
import mimetypes import mimetypes
from pathlib import Path from pathlib import Path
from google_photos_library_api.exceptions import GooglePhotosApiError
from google_photos_library_api.model import NewMediaItem, SimpleMediaItem
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_FILENAME from homeassistant.const import CONF_FILENAME
from homeassistant.core import ( from homeassistant.core import (
HomeAssistant, HomeAssistant,
@ -19,14 +20,8 @@ from homeassistant.core import (
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from . import api
from .const import DOMAIN, UPLOAD_SCOPE from .const import DOMAIN, UPLOAD_SCOPE
from .types import GooglePhotosConfigEntry
type GooglePhotosConfigEntry = ConfigEntry[api.AsyncConfigEntryAuth]
__all__ = [
"DOMAIN",
]
CONF_CONFIG_ENTRY_ID = "config_entry_id" CONF_CONFIG_ENTRY_ID = "config_entry_id"
@ -98,11 +93,38 @@ def async_register_services(hass: HomeAssistant) -> None:
) )
for mime_type, content in file_results: for mime_type, content in file_results:
upload_tasks.append(client_api.upload_content(content, mime_type)) upload_tasks.append(client_api.upload_content(content, mime_type))
upload_tokens = await asyncio.gather(*upload_tasks) try:
media_ids = await client_api.create_media_items(upload_tokens) upload_results = await asyncio.gather(*upload_tasks)
except GooglePhotosApiError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="upload_error",
translation_placeholders={"message": str(err)},
) from err
try:
upload_result = await client_api.create_media_items(
[
NewMediaItem(
SimpleMediaItem(upload_token=upload_result.upload_token)
)
for upload_result in upload_results
]
)
except GooglePhotosApiError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="api_error",
translation_placeholders={"message": str(err)},
) from err
if call.return_response: if call.return_response:
return { return {
"media_items": [{"media_item_id": media_id for media_id in media_ids}] "media_items": [
{
"media_item_id": item_result.media_item.id
for item_result in upload_result.new_media_item_results
if item_result.media_item and item_result.media_item.id
}
]
} }
return None return None

6
homeassistant/components/google_photos/strings.json
View file

@ -45,6 +45,12 @@
}, },
"missing_upload_permission": { "missing_upload_permission": {
"message": "Home Assistnt was not granted permission to upload to Google Photos" "message": "Home Assistnt was not granted permission to upload to Google Photos"
},
"upload_error": {
"message": "Failed to upload content: {message}"
},
"api_error": {
"message": "Google Photos API responded with error: {message}"
} }
}, },
"services": { "services": {

7
homeassistant/components/google_photos/types.py Normal file
View file

@ -0,0 +1,7 @@
"""Google Photos types."""
from google_photos_library_api.api import GooglePhotosLibraryApi
from homeassistant.config_entries import ConfigEntry
type GooglePhotosConfigEntry = ConfigEntry[GooglePhotosLibraryApi]

4
requirements_all.txt
View file

@ -979,7 +979,6 @@ goalzero==0.2.2
goodwe==0.3.6 goodwe==0.3.6
# homeassistant.components.google_mail # homeassistant.components.google_mail
# homeassistant.components.google_photos
# homeassistant.components.google_tasks # homeassistant.components.google_tasks
google-api-python-client==2.71.0 google-api-python-client==2.71.0
@ -995,6 +994,9 @@ google-generativeai==0.7.2
# homeassistant.components.nest # homeassistant.components.nest
google-nest-sdm==5.0.0 google-nest-sdm==5.0.0
# homeassistant.components.google_photos
google-photos-library-api==0.8.0
# homeassistant.components.google_travel_time # homeassistant.components.google_travel_time
googlemaps==2.5.1 googlemaps==2.5.1

4
requirements_test_all.txt
View file

@ -829,7 +829,6 @@ goalzero==0.2.2
goodwe==0.3.6 goodwe==0.3.6
# homeassistant.components.google_mail # homeassistant.components.google_mail
# homeassistant.components.google_photos
# homeassistant.components.google_tasks # homeassistant.components.google_tasks
google-api-python-client==2.71.0 google-api-python-client==2.71.0
@ -845,6 +844,9 @@ google-generativeai==0.7.2
# homeassistant.components.nest # homeassistant.components.nest
google-nest-sdm==5.0.0 google-nest-sdm==5.0.0
# homeassistant.components.google_photos
google-photos-library-api==0.8.0
# homeassistant.components.google_travel_time # homeassistant.components.google_travel_time
googlemaps==2.5.1 googlemaps==2.5.1

119
tests/components/google_photos/conftest.py
View file

@ -1,10 +1,18 @@
"""Test fixtures for Google Photos.""" """Test fixtures for Google Photos."""
from collections.abc import Awaitable, Callable, Generator from collections.abc import AsyncGenerator, Awaitable, Callable, Generator
import time import time
from typing import Any from typing import Any
from unittest.mock import Mock, patch from unittest.mock import AsyncMock, Mock, patch
from google_photos_library_api.api import GooglePhotosLibraryApi
from google_photos_library_api.model import (
Album,
ListAlbumResult,
ListMediaItemResult,
MediaItem,
UserInfoResult,
)
import pytest import pytest
from homeassistant.components.application_credentials import ( from homeassistant.components.application_credentials import (
@ -28,6 +36,12 @@ CLIENT_SECRET = "5678"
FAKE_ACCESS_TOKEN = "some-access-token" FAKE_ACCESS_TOKEN = "some-access-token"
FAKE_REFRESH_TOKEN = "some-refresh-token" FAKE_REFRESH_TOKEN = "some-refresh-token"
EXPIRES_IN = 3600 EXPIRES_IN = 3600
USERINFO_URL = "https://www.googleapis.com/oauth2/v1/userinfo"
PHOTOS_BASE_URL = "https://photoslibrary.googleapis.com"
MEDIA_ITEMS_URL = f"{PHOTOS_BASE_URL}/v1/mediaItems"
ALBUMS_URL = f"{PHOTOS_BASE_URL}/v1/albums"
UPLOADS_URL = f"{PHOTOS_BASE_URL}/v1/uploads"
CREATE_MEDIA_ITEMS_URL = f"{PHOTOS_BASE_URL}/v1/mediaItems:batchCreate"
@pytest.fixture(name="expires_at") @pytest.fixture(name="expires_at")
@ -100,56 +114,83 @@ def mock_user_identifier() -> str | None:
return USER_IDENTIFIER return USER_IDENTIFIER
@pytest.fixture(name="setup_api") @pytest.fixture(name="api_error")
def mock_setup_api( def mock_api_error() -> Exception | None:
fixture_name: str, user_identifier: str """Provide a json fixture file to load for list media item api responses."""
) -> Generator[Mock, None, None]:
"""Set up fake Google Photos API responses from fixtures."""
with patch("homeassistant.components.google_photos.api.build") as mock:
mock.return_value.userinfo.return_value.get.return_value.execute.return_value = {
"id": user_identifier,
"name": "Test Name",
}
responses = (
load_json_array_fixture(fixture_name, DOMAIN) if fixture_name else []
)
queue = list(responses)
def list_media_items(**kwargs: Any) -> Mock:
mock = Mock()
mock.execute.return_value = queue.pop(0)
return mock
mock.return_value.mediaItems.return_value.list = list_media_items
mock.return_value.mediaItems.return_value.search = list_media_items
# Mock a point lookup by reading contents of the fixture above
def get_media_item(mediaItemId: str, **kwargs: Any) -> Mock:
for response in responses:
for media_item in response["mediaItems"]:
if media_item["id"] == mediaItemId:
mock = Mock()
mock.execute.return_value = media_item
return mock
return None return None
mock.return_value.mediaItems.return_value.get = get_media_item
mock.return_value.albums.return_value.list.return_value.execute.return_value = ( @pytest.fixture(name="mock_api")
load_json_object_fixture("list_albums.json", DOMAIN) def mock_client_api(
fixture_name: str,
user_identifier: str,
api_error: Exception,
) -> Generator[Mock, None, None]:
"""Set up fake Google Photos API responses from fixtures."""
mock_api = AsyncMock(GooglePhotosLibraryApi, autospec=True)
mock_api.get_user_info.return_value = UserInfoResult(
id=user_identifier,
name="Test Name",
email="test.name@gmail.com",
) )
yield mock responses = load_json_array_fixture(fixture_name, DOMAIN) if fixture_name else []
async def list_media_items(
*args: Any,
) -> AsyncGenerator[ListMediaItemResult, None, None]:
for response in responses:
mock_list_media_items = Mock(ListMediaItemResult)
mock_list_media_items.media_items = [
MediaItem.from_dict(media_item) for media_item in response["mediaItems"]
]
yield mock_list_media_items
mock_api.list_media_items.return_value.__aiter__ = list_media_items
mock_api.list_media_items.return_value.__anext__ = list_media_items
mock_api.list_media_items.side_effect = api_error
# Mock a point lookup by reading contents of the fixture above
async def get_media_item(media_item_id: str, **kwargs: Any) -> Mock:
for response in responses:
for media_item in response["mediaItems"]:
if media_item["id"] == media_item_id:
return MediaItem.from_dict(media_item)
return None
mock_api.get_media_item = get_media_item
# Emulate an async iterator for returning pages of response objects. We just
# return a single page.
async def list_albums(
*args: Any, **kwargs: Any
) -> AsyncGenerator[ListAlbumResult, None, None]:
mock_list_album_result = Mock(ListAlbumResult)
mock_list_album_result.albums = [
Album.from_dict(album)
for album in load_json_object_fixture("list_albums.json", DOMAIN)["albums"]
]
yield mock_list_album_result
mock_api.list_albums.return_value.__aiter__ = list_albums
mock_api.list_albums.return_value.__anext__ = list_albums
mock_api.list_albums.side_effect = api_error
return mock_api
@pytest.fixture(name="setup_integration") @pytest.fixture(name="setup_integration")
async def mock_setup_integration( async def mock_setup_integration(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
mock_api: Mock,
) -> Callable[[], Awaitable[bool]]: ) -> Callable[[], Awaitable[bool]]:
"""Fixture to set up the integration.""" """Fixture to set up the integration."""
config_entry.add_to_hass(hass) config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.google_photos.GooglePhotosLibraryApi",
return_value=mock_api,
):
await hass.config_entries.async_setup(config_entry.entry_id) await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()

17
tests/components/google_photos/fixtures/api_not_enabled_response.json
View file

@ -1,17 +0,0 @@
[
{
"error": {
"code": 403,
"message": "Google Photos API has not been used in project 0 before or it is disabled. Enable it by visiting https://console.developers.google.com/apis/library/photoslibrary.googleapis.com/overview?project=0 then retry. If you enabled this API recently, wait a few minutes for the action to propagate to our systems and retry.",
"errors": [
{
"message": "Google Photos API has not been used in project 0 before or it is disabled. Enable it by visiting https://console.developers.google.com/apis/library/photoslibrary.googleapis.com/overview?project=0 then retry. If you enabled this API recently, wait a few minutes for the action to propagate to our systems and retry.",
"domain": "usageLimits",
"reason": "accessNotConfigured",
"extendedHelp": "https://console.developers.google.com"
}
],
"status": "PERMISSION_DENIED"
}
}
]

1
tests/components/google_photos/fixtures/list_albums.json
View file

@ -3,6 +3,7 @@
{ {
"id": "album-media-id-1", "id": "album-media-id-1",
"title": "Album title", "title": "Album title",
"productUrl": "http://photos.google.com/album-media-id-1",
"isWriteable": true, "isWriteable": true,
"mediaItemsCount": 7, "mediaItemsCount": 7,
"coverPhotoBaseUrl": "http://img.example.com/id3", "coverPhotoBaseUrl": "http://img.example.com/id3",

1
tests/components/google_photos/fixtures/not_dict.json
View file

@ -1 +0,0 @@
["not a dictionary"]

43
tests/components/google_photos/test_config_flow.py
View file

@ -4,8 +4,7 @@ from collections.abc import Generator
from typing import Any from typing import Any
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from googleapiclient.errors import HttpError from google_photos_library_api.exceptions import GooglePhotosApiError
from httplib2 import Response
import pytest import pytest
from homeassistant import config_entries from homeassistant import config_entries
@ -20,7 +19,7 @@ from homeassistant.helpers import config_entry_oauth2_flow
from .conftest import EXPIRES_IN, FAKE_ACCESS_TOKEN, FAKE_REFRESH_TOKEN, USER_IDENTIFIER from .conftest import EXPIRES_IN, FAKE_ACCESS_TOKEN, FAKE_REFRESH_TOKEN, USER_IDENTIFIER
from tests.common import MockConfigEntry, load_fixture from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator from tests.typing import ClientSessionGenerator
@ -37,6 +36,16 @@ def mock_setup_entry() -> Generator[Mock, None, None]:
yield mock_setup yield mock_setup
@pytest.fixture(autouse=True)
def mock_patch_api(mock_api: Mock) -> Generator[None, None, None]:
"""Fixture to patch the config flow api."""
with patch(
"homeassistant.components.google_photos.config_flow.GooglePhotosLibraryApi",
return_value=mock_api,
):
yield
@pytest.fixture(name="updated_token_entry", autouse=True) @pytest.fixture(name="updated_token_entry", autouse=True)
def mock_updated_token_entry() -> dict[str, Any]: def mock_updated_token_entry() -> dict[str, Any]:
"""Fixture to provide any test specific overrides to token data from the oauth token endpoint.""" """Fixture to provide any test specific overrides to token data from the oauth token endpoint."""
@ -60,7 +69,7 @@ def mock_token_request(
) )
@pytest.mark.usefixtures("current_request_with_host", "setup_api") @pytest.mark.usefixtures("current_request_with_host", "mock_api")
@pytest.mark.parametrize("fixture_name", ["list_mediaitems.json"]) @pytest.mark.parametrize("fixture_name", ["list_mediaitems.json"])
async def test_full_flow( async def test_full_flow(
hass: HomeAssistant, hass: HomeAssistant,
@ -126,11 +135,17 @@ async def test_full_flow(
@pytest.mark.usefixtures( @pytest.mark.usefixtures(
"current_request_with_host", "current_request_with_host",
"setup_credentials", "setup_credentials",
"mock_api",
)
@pytest.mark.parametrize(
"api_error",
[
GooglePhotosApiError("some error"),
],
) )
async def test_api_not_enabled( async def test_api_not_enabled(
hass: HomeAssistant, hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator, hass_client_no_auth: ClientSessionGenerator,
setup_api: Mock,
) -> None: ) -> None:
"""Check flow aborts if api is not enabled.""" """Check flow aborts if api is not enabled."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -160,24 +175,18 @@ async def test_api_not_enabled(
assert resp.status == 200 assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8" assert resp.headers["content-type"] == "text/html; charset=utf-8"
setup_api.return_value.mediaItems.return_value.list = Mock()
setup_api.return_value.mediaItems.return_value.list.return_value.execute.side_effect = HttpError(
Response({"status": "403"}),
bytes(load_fixture("google_photos/api_not_enabled_response.json"), "utf-8"),
)
result = await hass.config_entries.flow.async_configure(result["flow_id"]) result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "access_not_configured" assert result["reason"] == "access_not_configured"
assert result["description_placeholders"]["message"].endswith( assert result["description_placeholders"]["message"].endswith("some error")
"Google Photos API has not been used in project 0 before or it is disabled. Enable it by visiting https://console.developers.google.com/apis/library/photoslibrary.googleapis.com/overview?project=0 then retry. If you enabled this API recently, wait a few minutes for the action to propagate to our systems and retry."
)
@pytest.mark.usefixtures("current_request_with_host", "setup_credentials") @pytest.mark.usefixtures("current_request_with_host", "setup_credentials")
async def test_general_exception( async def test_general_exception(
hass: HomeAssistant, hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator, hass_client_no_auth: ClientSessionGenerator,
mock_api: Mock,
) -> None: ) -> None:
"""Check flow aborts if exception happens.""" """Check flow aborts if exception happens."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -206,17 +215,15 @@ async def test_general_exception(
assert resp.status == 200 assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8" assert resp.headers["content-type"] == "text/html; charset=utf-8"
with patch( mock_api.list_media_items.side_effect = Exception
"homeassistant.components.google_photos.api.build",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"]) result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown" assert result["reason"] == "unknown"
@pytest.mark.usefixtures("current_request_with_host", "setup_api", "setup_integration") @pytest.mark.usefixtures("current_request_with_host", "mock_api", "setup_integration")
@pytest.mark.parametrize("fixture_name", ["list_mediaitems.json"]) @pytest.mark.parametrize("fixture_name", ["list_mediaitems.json"])
@pytest.mark.parametrize( @pytest.mark.parametrize(
"updated_token_entry", "updated_token_entry",

58
tests/components/google_photos/test_media_source.py
View file

@ -1,10 +1,8 @@
"""Test the Google Photos media source.""" """Test the Google Photos media source."""
from typing import Any
from unittest.mock import Mock from unittest.mock import Mock
from googleapiclient.errors import HttpError from google_photos_library_api.exceptions import GooglePhotosApiError
from httplib2 import Response
import pytest import pytest
from homeassistant.components.google_photos.const import DOMAIN, UPLOAD_SCOPE from homeassistant.components.google_photos.const import DOMAIN, UPLOAD_SCOPE
@ -46,7 +44,7 @@ async def test_no_config_entries(
assert not browse.children assert not browse.children
@pytest.mark.usefixtures("setup_integration", "setup_api") @pytest.mark.usefixtures("setup_integration", "mock_api")
@pytest.mark.parametrize( @pytest.mark.parametrize(
("scopes"), ("scopes"),
[ [
@ -64,7 +62,7 @@ async def test_no_read_scopes(
assert not browse.children assert not browse.children
@pytest.mark.usefixtures("setup_integration", "setup_api") @pytest.mark.usefixtures("setup_integration", "mock_api")
@pytest.mark.parametrize( @pytest.mark.parametrize(
("album_path", "expected_album_title"), ("album_path", "expected_album_title"),
[ [
@ -135,14 +133,14 @@ async def test_browse_albums(
] == expected_medias ] == expected_medias
@pytest.mark.usefixtures("setup_integration", "setup_api") @pytest.mark.usefixtures("setup_integration", "mock_api")
async def test_invalid_config_entry(hass: HomeAssistant) -> None: async def test_invalid_config_entry(hass: HomeAssistant) -> None:
"""Test browsing to a config entry that does not exist.""" """Test browsing to a config entry that does not exist."""
with pytest.raises(BrowseError, match="Could not find config entry"): with pytest.raises(BrowseError, match="Could not find config entry"):
await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}/invalid-config-entry") await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}/invalid-config-entry")
@pytest.mark.usefixtures("setup_integration", "setup_api") @pytest.mark.usefixtures("setup_integration", "mock_api")
@pytest.mark.parametrize("fixture_name", ["list_mediaitems.json"]) @pytest.mark.parametrize("fixture_name", ["list_mediaitems.json"])
async def test_browse_invalid_path(hass: HomeAssistant) -> None: async def test_browse_invalid_path(hass: HomeAssistant) -> None:
"""Test browsing to a photo is not possible.""" """Test browsing to a photo is not possible."""
@ -161,8 +159,8 @@ async def test_browse_invalid_path(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("setup_integration") @pytest.mark.usefixtures("setup_integration")
@pytest.mark.parametrize("fixture_name", ["list_mediaitems.json"]) @pytest.mark.parametrize("api_error", [GooglePhotosApiError("some error")])
async def test_invalid_album_id(hass: HomeAssistant, setup_api: Mock) -> None: async def test_invalid_album_id(hass: HomeAssistant, mock_api: Mock) -> None:
"""Test browsing to an album id that does not exist.""" """Test browsing to an album id that does not exist."""
browse = await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}") browse = await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}")
assert browse.domain == DOMAIN assert browse.domain == DOMAIN
@ -172,11 +170,6 @@ async def test_invalid_album_id(hass: HomeAssistant, setup_api: Mock) -> None:
(CONFIG_ENTRY_ID, "Account Name") (CONFIG_ENTRY_ID, "Account Name")
] ]
setup_api.return_value.mediaItems.return_value.search = Mock()
setup_api.return_value.mediaItems.return_value.search.return_value.execute.side_effect = HttpError(
Response({"status": "404"}), b""
)
with pytest.raises(BrowseError, match="Error listing media items"): with pytest.raises(BrowseError, match="Error listing media items"):
await async_browse_media( await async_browse_media(
hass, f"{URI_SCHEME}{DOMAIN}/{CONFIG_ENTRY_ID}/a/invalid-album-id" hass, f"{URI_SCHEME}{DOMAIN}/{CONFIG_ENTRY_ID}/a/invalid-album-id"
@ -201,18 +194,9 @@ async def test_missing_photo_id(
await async_resolve_media(hass, f"{URI_SCHEME}{DOMAIN}/{identifier}", None) await async_resolve_media(hass, f"{URI_SCHEME}{DOMAIN}/{identifier}", None)
@pytest.mark.usefixtures("setup_integration", "setup_api") @pytest.mark.usefixtures("setup_integration", "mock_api")
@pytest.mark.parametrize( @pytest.mark.parametrize("api_error", [GooglePhotosApiError("some error")])
"side_effect", async def test_list_albums_failure(hass: HomeAssistant) -> None:
[
HttpError(Response({"status": "403"}), b""),
],
)
async def test_list_media_items_failure(
hass: HomeAssistant,
setup_api: Any,
side_effect: HttpError | Response,
) -> None:
"""Test browsing to an album id that does not exist.""" """Test browsing to an album id that does not exist."""
browse = await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}") browse = await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}")
assert browse.domain == DOMAIN assert browse.domain == DOMAIN
@ -222,24 +206,13 @@ async def test_list_media_items_failure(
(CONFIG_ENTRY_ID, "Account Name") (CONFIG_ENTRY_ID, "Account Name")
] ]
setup_api.return_value.mediaItems.return_value.list = Mock() with pytest.raises(BrowseError, match="Error listing albums"):
setup_api.return_value.mediaItems.return_value.list.return_value.execute.side_effect = side_effect await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}/{CONFIG_ENTRY_ID}")
with pytest.raises(BrowseError, match="Error listing media items"):
await async_browse_media(
hass, f"{URI_SCHEME}{DOMAIN}/{CONFIG_ENTRY_ID}/a/recent"
)
@pytest.mark.usefixtures("setup_integration", "setup_api") @pytest.mark.usefixtures("setup_integration", "mock_api")
@pytest.mark.parametrize( @pytest.mark.parametrize("api_error", [GooglePhotosApiError("some error")])
"fixture_name", async def test_list_media_items_failure(hass: HomeAssistant) -> None:
[
"api_not_enabled_response.json",
"not_dict.json",
],
)
async def test_media_items_error_parsing_response(hass: HomeAssistant) -> None:
"""Test browsing to an album id that does not exist.""" """Test browsing to an album id that does not exist."""
browse = await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}") browse = await async_browse_media(hass, f"{URI_SCHEME}{DOMAIN}")
assert browse.domain == DOMAIN assert browse.domain == DOMAIN
@ -248,6 +221,7 @@ async def test_media_items_error_parsing_response(hass: HomeAssistant) -> None:
assert [(child.identifier, child.title) for child in browse.children] == [ assert [(child.identifier, child.title) for child in browse.children] == [
(CONFIG_ENTRY_ID, "Account Name") (CONFIG_ENTRY_ID, "Account Name")
] ]
with pytest.raises(BrowseError, match="Error listing media items"): with pytest.raises(BrowseError, match="Error listing media items"):
await async_browse_media( await async_browse_media(
hass, f"{URI_SCHEME}{DOMAIN}/{CONFIG_ENTRY_ID}/a/recent" hass, f"{URI_SCHEME}{DOMAIN}/{CONFIG_ENTRY_ID}/a/recent"

51
tests/components/google_photos/test_services.py
View file

@ -1,45 +1,42 @@
"""Tests for Google Photos.""" """Tests for Google Photos."""
import http
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from googleapiclient.errors import HttpError from google_photos_library_api.exceptions import GooglePhotosApiError
from httplib2 import Response from google_photos_library_api.model import (
CreateMediaItemsResult,
MediaItem,
NewMediaItemResult,
Status,
)
import pytest import pytest
from homeassistant.components.google_photos.api import UPLOAD_API
from homeassistant.components.google_photos.const import DOMAIN, READ_SCOPES from homeassistant.components.google_photos.const import DOMAIN, READ_SCOPES
from homeassistant.config_entries import ConfigEntryState from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
@pytest.mark.usefixtures("setup_integration") @pytest.mark.usefixtures("setup_integration")
async def test_upload_service( async def test_upload_service(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker, mock_api: Mock,
setup_api: Mock,
) -> None: ) -> None:
"""Test service call to upload content.""" """Test service call to upload content."""
assert hass.services.has_service(DOMAIN, "upload") assert hass.services.has_service(DOMAIN, "upload")
aioclient_mock.post(UPLOAD_API, text="some-upload-token") mock_api.create_media_items.return_value = CreateMediaItemsResult(
setup_api.return_value.mediaItems.return_value.batchCreate.return_value.execute.return_value = { new_media_item_results=[
"newMediaItemResults": [ NewMediaItemResult(
{ upload_token="some-upload-token",
"status": { status=Status(code=200),
"code": 200, media_item=MediaItem(id="new-media-item-id-1"),
}, )
"mediaItem": {
"id": "new-media-item-id-1",
},
}
] ]
} )
with ( with (
patch( patch(
@ -62,6 +59,7 @@ async def test_upload_service(
blocking=True, blocking=True,
return_response=True, return_response=True,
) )
assert response == {"media_items": [{"media_item_id": "new-media-item-id-1"}]} assert response == {"media_items": [{"media_item_id": "new-media-item-id-1"}]}
@ -157,12 +155,11 @@ async def test_filename_does_not_exist(
async def test_upload_service_upload_content_failure( async def test_upload_service_upload_content_failure(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker, mock_api: Mock,
setup_api: Mock,
) -> None: ) -> None:
"""Test service call to upload content.""" """Test service call to upload content."""
aioclient_mock.post(UPLOAD_API, status=http.HTTPStatus.SERVICE_UNAVAILABLE) mock_api.upload_content.side_effect = GooglePhotosApiError()
with ( with (
patch( patch(
@ -192,15 +189,11 @@ async def test_upload_service_upload_content_failure(
async def test_upload_service_fails_create( async def test_upload_service_fails_create(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker, mock_api: Mock,
setup_api: Mock,
) -> None: ) -> None:
"""Test service call to upload content.""" """Test service call to upload content."""
aioclient_mock.post(UPLOAD_API, text="some-upload-token") mock_api.create_media_items.side_effect = GooglePhotosApiError()
setup_api.return_value.mediaItems.return_value.batchCreate.return_value.execute.side_effect = HttpError(
Response({"status": "403"}), b""
)
with ( with (
patch( patch(
@ -238,8 +231,6 @@ async def test_upload_service_fails_create(
async def test_upload_service_no_scope( async def test_upload_service_no_scope(
hass: HomeAssistant, hass: HomeAssistant,
config_entry: MockConfigEntry, config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker,
setup_api: Mock,
) -> None: ) -> None:
"""Test service call to upload content but the config entry is read-only.""" """Test service call to upload content but the config entry is read-only."""