Add PS Vita support to PlayStation Network integration (#148186)

This commit is contained in:
Manu 2025-07-14 21:24:50 +02:00 committed by GitHub
parent 80eb4fb2f6
commit 5ec9c4e6e3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 614 additions and 60 deletions

View File

@ -6,7 +6,12 @@ from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from .const import CONF_NPSSO from .const import CONF_NPSSO
from .coordinator import PlaystationNetworkConfigEntry, PlaystationNetworkCoordinator from .coordinator import (
PlaystationNetworkConfigEntry,
PlaystationNetworkRuntimeData,
PlaystationNetworkTrophyTitlesCoordinator,
PlaystationNetworkUserDataCoordinator,
)
from .helpers import PlaystationNetwork from .helpers import PlaystationNetwork
PLATFORMS: list[Platform] = [ PLATFORMS: list[Platform] = [
@ -23,9 +28,12 @@ async def async_setup_entry(
psn = PlaystationNetwork(hass, entry.data[CONF_NPSSO]) psn = PlaystationNetwork(hass, entry.data[CONF_NPSSO])
coordinator = PlaystationNetworkCoordinator(hass, psn, entry) coordinator = PlaystationNetworkUserDataCoordinator(hass, psn, entry)
await coordinator.async_config_entry_first_refresh() await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
trophy_titles = PlaystationNetworkTrophyTitlesCoordinator(hass, psn, entry)
entry.runtime_data = PlaystationNetworkRuntimeData(coordinator, trophy_titles)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True return True

View File

@ -49,7 +49,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback, async_add_entities: AddConfigEntryEntitiesCallback,
) -> None: ) -> None:
"""Set up the binary sensor platform.""" """Set up the binary sensor platform."""
coordinator = config_entry.runtime_data coordinator = config_entry.runtime_data.user_data
async_add_entities( async_add_entities(
PlaystationNetworkBinarySensorEntity(coordinator, description) PlaystationNetworkBinarySensorEntity(coordinator, description)
for description in BINARY_SENSOR_DESCRIPTIONS for description in BINARY_SENSOR_DESCRIPTIONS

View File

@ -10,7 +10,6 @@ from psnawp_api.core.psnawp_exceptions import (
PSNAWPInvalidTokenError, PSNAWPInvalidTokenError,
PSNAWPNotFoundError, PSNAWPNotFoundError,
) )
from psnawp_api.models.user import User
from psnawp_api.utils.misc import parse_npsso_token from psnawp_api.utils.misc import parse_npsso_token
import voluptuous as vol import voluptuous as vol
@ -42,7 +41,7 @@ class PlaystationNetworkConfigFlow(ConfigFlow, domain=DOMAIN):
else: else:
psn = PlaystationNetwork(self.hass, npsso) psn = PlaystationNetwork(self.hass, npsso)
try: try:
user: User = await psn.get_user() user = await psn.get_user()
except PSNAWPAuthenticationError: except PSNAWPAuthenticationError:
errors["base"] = "invalid_auth" errors["base"] = "invalid_auth"
except PSNAWPNotFoundError: except PSNAWPNotFoundError:
@ -98,7 +97,7 @@ class PlaystationNetworkConfigFlow(ConfigFlow, domain=DOMAIN):
try: try:
npsso = parse_npsso_token(user_input[CONF_NPSSO]) npsso = parse_npsso_token(user_input[CONF_NPSSO])
psn = PlaystationNetwork(self.hass, npsso) psn = PlaystationNetwork(self.hass, npsso)
user: User = await psn.get_user() user = await psn.get_user()
except PSNAWPAuthenticationError: except PSNAWPAuthenticationError:
errors["base"] = "invalid_auth" errors["base"] = "invalid_auth"
except (PSNAWPNotFoundError, PSNAWPInvalidTokenError): except (PSNAWPNotFoundError, PSNAWPInvalidTokenError):

View File

@ -8,9 +8,10 @@ DOMAIN = "playstation_network"
CONF_NPSSO: Final = "npsso" CONF_NPSSO: Final = "npsso"
SUPPORTED_PLATFORMS = { SUPPORTED_PLATFORMS = {
PlatformType.PS5, PlatformType.PS_VITA,
PlatformType.PS4,
PlatformType.PS3, PlatformType.PS3,
PlatformType.PS4,
PlatformType.PS5,
PlatformType.PSPC, PlatformType.PSPC,
} }

View File

@ -2,6 +2,8 @@
from __future__ import annotations from __future__ import annotations
from abc import abstractmethod
from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
import logging import logging
@ -10,6 +12,7 @@ from psnawp_api.core.psnawp_exceptions import (
PSNAWPClientError, PSNAWPClientError,
PSNAWPServerError, PSNAWPServerError,
) )
from psnawp_api.models.trophies import TrophyTitle
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -21,13 +24,22 @@ from .helpers import PlaystationNetwork, PlaystationNetworkData
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
type PlaystationNetworkConfigEntry = ConfigEntry[PlaystationNetworkCoordinator] type PlaystationNetworkConfigEntry = ConfigEntry[PlaystationNetworkRuntimeData]
class PlaystationNetworkCoordinator(DataUpdateCoordinator[PlaystationNetworkData]): @dataclass
"""Data update coordinator for PSN.""" class PlaystationNetworkRuntimeData:
"""Dataclass holding PSN runtime data."""
user_data: PlaystationNetworkUserDataCoordinator
trophy_titles: PlaystationNetworkTrophyTitlesCoordinator
class PlayStationNetworkBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""Base coordinator for PSN."""
config_entry: PlaystationNetworkConfigEntry config_entry: PlaystationNetworkConfigEntry
_update_inverval: timedelta
def __init__( def __init__(
self, self,
@ -41,16 +53,43 @@ class PlaystationNetworkCoordinator(DataUpdateCoordinator[PlaystationNetworkData
name=DOMAIN, name=DOMAIN,
logger=_LOGGER, logger=_LOGGER,
config_entry=config_entry, config_entry=config_entry,
update_interval=timedelta(seconds=30), update_interval=self._update_interval,
) )
self.psn = psn self.psn = psn
@abstractmethod
async def update_data(self) -> _DataT:
"""Update coordinator data."""
async def _async_update_data(self) -> _DataT:
"""Get the latest data from the PSN."""
try:
return await self.update_data()
except PSNAWPAuthenticationError as error:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="not_ready",
) from error
except (PSNAWPServerError, PSNAWPClientError) as error:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_failed",
) from error
class PlaystationNetworkUserDataCoordinator(
PlayStationNetworkBaseCoordinator[PlaystationNetworkData]
):
"""Data update coordinator for PSN."""
_update_interval = timedelta(seconds=30)
async def _async_setup(self) -> None: async def _async_setup(self) -> None:
"""Set up the coordinator.""" """Set up the coordinator."""
try: try:
await self.psn.get_user() await self.psn.async_setup()
except PSNAWPAuthenticationError as error: except PSNAWPAuthenticationError as error:
raise ConfigEntryAuthFailed( raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_domain=DOMAIN,
@ -62,17 +101,22 @@ class PlaystationNetworkCoordinator(DataUpdateCoordinator[PlaystationNetworkData
translation_key="update_failed", translation_key="update_failed",
) from error ) from error
async def _async_update_data(self) -> PlaystationNetworkData: async def update_data(self) -> PlaystationNetworkData:
"""Get the latest data from the PSN.""" """Get the latest data from the PSN."""
try:
return await self.psn.get_data() return await self.psn.get_data()
except PSNAWPAuthenticationError as error:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, class PlaystationNetworkTrophyTitlesCoordinator(
translation_key="not_ready", PlayStationNetworkBaseCoordinator[list[TrophyTitle]]
) from error ):
except (PSNAWPServerError, PSNAWPClientError) as error: """Trophy titles data update coordinator for PSN."""
raise UpdateFailed(
translation_domain=DOMAIN, _update_interval = timedelta(days=1)
translation_key="update_failed",
) from error async def update_data(self) -> list[TrophyTitle]:
"""Update trophy titles data."""
self.psn.trophy_titles = await self.hass.async_add_executor_job(
lambda: list(self.psn.user.trophy_titles())
)
await self.config_entry.runtime_data.user_data.async_request_refresh()
return self.psn.trophy_titles

View File

@ -10,7 +10,7 @@ from psnawp_api.models.trophies import PlatformType
from homeassistant.components.diagnostics import async_redact_data from homeassistant.components.diagnostics import async_redact_data
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from .coordinator import PlaystationNetworkConfigEntry, PlaystationNetworkCoordinator from .coordinator import PlaystationNetworkConfigEntry
TO_REDACT = { TO_REDACT = {
"account_id", "account_id",
@ -27,12 +27,12 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: PlaystationNetworkConfigEntry hass: HomeAssistant, entry: PlaystationNetworkConfigEntry
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Return diagnostics for a config entry.""" """Return diagnostics for a config entry."""
coordinator: PlaystationNetworkCoordinator = entry.runtime_data coordinator = entry.runtime_data.user_data
return { return {
"data": async_redact_data( "data": async_redact_data(
_serialize_platform_types(asdict(coordinator.data)), TO_REDACT _serialize_platform_types(asdict(coordinator.data)), TO_REDACT
), )
} }
@ -46,10 +46,12 @@ def _serialize_platform_types(data: Any) -> Any:
for platform, record in data.items() for platform, record in data.items()
} }
if isinstance(data, set): if isinstance(data, set):
return [ return sorted(
[
record.value if isinstance(record, PlatformType) else record record.value if isinstance(record, PlatformType) else record
for record in data for record in data
] ]
)
if isinstance(data, PlatformType): if isinstance(data, PlatformType):
return data.value return data.value
return data return data

View File

@ -7,17 +7,19 @@ from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN from .const import DOMAIN
from .coordinator import PlaystationNetworkCoordinator from .coordinator import PlaystationNetworkUserDataCoordinator
class PlaystationNetworkServiceEntity(CoordinatorEntity[PlaystationNetworkCoordinator]): class PlaystationNetworkServiceEntity(
CoordinatorEntity[PlaystationNetworkUserDataCoordinator]
):
"""Common entity class for PlayStationNetwork Service entities.""" """Common entity class for PlayStationNetwork Service entities."""
_attr_has_entity_name = True _attr_has_entity_name = True
def __init__( def __init__(
self, self,
coordinator: PlaystationNetworkCoordinator, coordinator: PlaystationNetworkUserDataCoordinator,
entity_description: EntityDescription, entity_description: EntityDescription,
) -> None: ) -> None:
"""Initialize PlayStation Network Service Entity.""" """Initialize PlayStation Network Service Entity."""

View File

@ -8,7 +8,7 @@ from typing import Any
from psnawp_api import PSNAWP from psnawp_api import PSNAWP
from psnawp_api.models.client import Client from psnawp_api.models.client import Client
from psnawp_api.models.trophies import PlatformType, TrophySummary from psnawp_api.models.trophies import PlatformType, TrophySummary, TrophyTitle
from psnawp_api.models.user import User from psnawp_api.models.user import User
from pyrate_limiter import Duration, Rate from pyrate_limiter import Duration, Rate
@ -16,7 +16,7 @@ from homeassistant.core import HomeAssistant
from .const import SUPPORTED_PLATFORMS from .const import SUPPORTED_PLATFORMS
LEGACY_PLATFORMS = {PlatformType.PS3, PlatformType.PS4} LEGACY_PLATFORMS = {PlatformType.PS3, PlatformType.PS4, PlatformType.PS_VITA}
@dataclass @dataclass
@ -52,10 +52,22 @@ class PlaystationNetwork:
"""Initialize the class with the npsso token.""" """Initialize the class with the npsso token."""
rate = Rate(300, Duration.MINUTE * 15) rate = Rate(300, Duration.MINUTE * 15)
self.psn = PSNAWP(npsso, rate_limit=rate) self.psn = PSNAWP(npsso, rate_limit=rate)
self.client: Client | None = None self.client: Client
self.hass = hass self.hass = hass
self.user: User self.user: User
self.legacy_profile: dict[str, Any] | None = None self.legacy_profile: dict[str, Any] | None = None
self.trophy_titles: list[TrophyTitle] = []
self._title_icon_urls: dict[str, str] = {}
def _setup(self) -> None:
"""Setup PSN."""
self.user = self.psn.user(online_id="me")
self.client = self.psn.me()
self.trophy_titles = list(self.user.trophy_titles())
async def async_setup(self) -> None:
"""Setup PSN."""
await self.hass.async_add_executor_job(self._setup)
async def get_user(self) -> User: async def get_user(self) -> User:
"""Get the user object from the PlayStation Network.""" """Get the user object from the PlayStation Network."""
@ -68,9 +80,6 @@ class PlaystationNetwork:
"""Bundle api calls to retrieve data from the PlayStation Network.""" """Bundle api calls to retrieve data from the PlayStation Network."""
data = PlaystationNetworkData() data = PlaystationNetworkData()
if not self.client:
self.client = self.psn.me()
data.registered_platforms = { data.registered_platforms = {
PlatformType(device["deviceType"]) PlatformType(device["deviceType"])
for device in self.client.get_account_devices() for device in self.client.get_account_devices()
@ -123,7 +132,7 @@ class PlaystationNetwork:
presence = self.legacy_profile["profile"].get("presences", []) presence = self.legacy_profile["profile"].get("presences", [])
if (game_title_info := presence[0] if presence else {}) and game_title_info[ if (game_title_info := presence[0] if presence else {}) and game_title_info[
"onlineStatus" "onlineStatus"
] == "online": ] != "offline":
platform = PlatformType(game_title_info["platform"]) platform = PlatformType(game_title_info["platform"])
if platform is PlatformType.PS4: if platform is PlatformType.PS4:
@ -135,6 +144,10 @@ class PlaystationNetwork:
account_id="me", account_id="me",
np_communication_id="", np_communication_id="",
).get_title_icon_url() ).get_title_icon_url()
elif platform is PlatformType.PS_VITA and game_title_info.get(
"npTitleId"
):
media_image_url = self.get_psvita_title_icon_url(game_title_info)
else: else:
media_image_url = None media_image_url = None
@ -147,3 +160,28 @@ class PlaystationNetwork:
status=game_title_info["onlineStatus"], status=game_title_info["onlineStatus"],
) )
return data return data
def get_psvita_title_icon_url(self, game_title_info: dict[str, Any]) -> str | None:
"""Look up title_icon_url from trophy titles data."""
if url := self._title_icon_urls.get(game_title_info["npTitleId"]):
return url
url = next(
(
title.title_icon_url
for title in self.trophy_titles
if game_title_info["titleName"]
== normalize_title(title.title_name or "")
and next(iter(title.title_platform)) == PlatformType.PS_VITA
),
None,
)
if url is not None:
self._title_icon_urls[game_title_info["npTitleId"]] = url
return url
def normalize_title(name: str) -> str:
"""Normalize trophy title."""
return name.removesuffix("Trophies").removesuffix("Trophy Set").strip()

View File

@ -17,13 +17,18 @@ from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import PlaystationNetworkConfigEntry, PlaystationNetworkCoordinator from . import (
PlaystationNetworkConfigEntry,
PlaystationNetworkTrophyTitlesCoordinator,
PlaystationNetworkUserDataCoordinator,
)
from .const import DOMAIN, SUPPORTED_PLATFORMS from .const import DOMAIN, SUPPORTED_PLATFORMS
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
PLATFORM_MAP = { PLATFORM_MAP = {
PlatformType.PS_VITA: "PlayStation Vita",
PlatformType.PS5: "PlayStation 5", PlatformType.PS5: "PlayStation 5",
PlatformType.PS4: "PlayStation 4", PlatformType.PS4: "PlayStation 4",
PlatformType.PS3: "PlayStation 3", PlatformType.PS3: "PlayStation 3",
@ -38,7 +43,8 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback, async_add_entities: AddConfigEntryEntitiesCallback,
) -> None: ) -> None:
"""Media Player Entity Setup.""" """Media Player Entity Setup."""
coordinator = config_entry.runtime_data coordinator = config_entry.runtime_data.user_data
trophy_titles = config_entry.runtime_data.trophy_titles
devices_added: set[PlatformType] = set() devices_added: set[PlatformType] = set()
device_reg = dr.async_get(hass) device_reg = dr.async_get(hass)
entities = [] entities = []
@ -50,10 +56,12 @@ async def async_setup_entry(
if not SUPPORTED_PLATFORMS - devices_added: if not SUPPORTED_PLATFORMS - devices_added:
remove_listener() remove_listener()
new_platforms = set(coordinator.data.active_sessions.keys()) - devices_added new_platforms = (
set(coordinator.data.active_sessions.keys()) & SUPPORTED_PLATFORMS
) - devices_added
if new_platforms: if new_platforms:
async_add_entities( async_add_entities(
PsnMediaPlayerEntity(coordinator, platform_type) PsnMediaPlayerEntity(coordinator, platform_type, trophy_titles)
for platform_type in new_platforms for platform_type in new_platforms
) )
devices_added |= new_platforms devices_added |= new_platforms
@ -64,7 +72,7 @@ async def async_setup_entry(
(DOMAIN, f"{coordinator.config_entry.unique_id}_{platform.value}") (DOMAIN, f"{coordinator.config_entry.unique_id}_{platform.value}")
} }
): ):
entities.append(PsnMediaPlayerEntity(coordinator, platform)) entities.append(PsnMediaPlayerEntity(coordinator, platform, trophy_titles))
devices_added.add(platform) devices_added.add(platform)
if entities: if entities:
async_add_entities(entities) async_add_entities(entities)
@ -74,7 +82,7 @@ async def async_setup_entry(
class PsnMediaPlayerEntity( class PsnMediaPlayerEntity(
CoordinatorEntity[PlaystationNetworkCoordinator], MediaPlayerEntity CoordinatorEntity[PlaystationNetworkUserDataCoordinator], MediaPlayerEntity
): ):
"""Media player entity representing currently playing game.""" """Media player entity representing currently playing game."""
@ -86,7 +94,10 @@ class PsnMediaPlayerEntity(
_attr_name = None _attr_name = None
def __init__( def __init__(
self, coordinator: PlaystationNetworkCoordinator, platform: PlatformType self,
coordinator: PlaystationNetworkUserDataCoordinator,
platform: PlatformType,
trophy_titles: PlaystationNetworkTrophyTitlesCoordinator,
) -> None: ) -> None:
"""Initialize PSN MediaPlayer.""" """Initialize PSN MediaPlayer."""
super().__init__(coordinator) super().__init__(coordinator)
@ -101,15 +112,21 @@ class PsnMediaPlayerEntity(
model=PLATFORM_MAP[platform], model=PLATFORM_MAP[platform],
via_device=(DOMAIN, coordinator.config_entry.unique_id), via_device=(DOMAIN, coordinator.config_entry.unique_id),
) )
self.trophy_titles = trophy_titles
@property @property
def state(self) -> MediaPlayerState: def state(self) -> MediaPlayerState:
"""Media Player state getter.""" """Media Player state getter."""
session = self.coordinator.data.active_sessions.get(self.key) session = self.coordinator.data.active_sessions.get(self.key)
if session and session.status == "online": if session:
if session.title_id is not None: if session.status == "online":
return MediaPlayerState.PLAYING return (
return MediaPlayerState.ON MediaPlayerState.PLAYING
if session.title_id is not None
else MediaPlayerState.ON
)
if session.status == "standby":
return MediaPlayerState.STANDBY
return MediaPlayerState.OFF return MediaPlayerState.OFF
@property @property
@ -129,3 +146,12 @@ class PsnMediaPlayerEntity(
"""Media image url getter.""" """Media image url getter."""
session = self.coordinator.data.active_sessions.get(self.key) session = self.coordinator.data.active_sessions.get(self.key)
return session.media_image_url if session else None return session.media_image_url if session else None
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if self.key is PlatformType.PS_VITA:
self.async_on_remove(
self.trophy_titles.async_add_listener(self._handle_coordinator_update)
)

View File

@ -131,7 +131,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback, async_add_entities: AddConfigEntryEntitiesCallback,
) -> None: ) -> None:
"""Set up the sensor platform.""" """Set up the sensor platform."""
coordinator = config_entry.runtime_data coordinator = config_entry.runtime_data.user_data
async_add_entities( async_add_entities(
PlaystationNetworkSensorEntity(coordinator, description) PlaystationNetworkSensorEntity(coordinator, description)
for description in SENSOR_DESCRIPTIONS for description in SENSOR_DESCRIPTIONS

View File

@ -1,9 +1,15 @@
"""Common fixtures for the Playstation Network tests.""" """Common fixtures for the Playstation Network tests."""
from collections.abc import Generator from collections.abc import Generator
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from psnawp_api.models.trophies import TrophySet, TrophySummary from psnawp_api.models.trophies import (
PlatformType,
TrophySet,
TrophySummary,
TrophyTitle,
)
import pytest import pytest
from homeassistant.components.playstation_network.const import CONF_NPSSO, DOMAIN from homeassistant.components.playstation_network.const import CONF_NPSSO, DOMAIN
@ -83,13 +89,14 @@ def mock_psnawpapi(mock_user: MagicMock) -> Generator[MagicMock]:
client.user.return_value = mock_user client.user.return_value = mock_user
client.me.return_value.get_account_devices.return_value = [ client.me.return_value.get_account_devices.return_value = [
{"deviceType": "PSVITA"},
{ {
"deviceId": "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234", "deviceId": "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234",
"deviceType": "PS5", "deviceType": "PS5",
"activationType": "PRIMARY", "activationType": "PRIMARY",
"activationDate": "2021-01-14T18:00:00.000Z", "activationDate": "2021-01-14T18:00:00.000Z",
"accountDeviceVector": "abcdefghijklmnopqrstuv", "accountDeviceVector": "abcdefghijklmnopqrstuv",
} },
] ]
client.me.return_value.trophy_summary.return_value = TrophySummary( client.me.return_value.trophy_summary.return_value = TrophySummary(
PSN_ID, 1079, 19, 10, TrophySet(14450, 8722, 11754, 1398) PSN_ID, 1079, 19, 10, TrophySet(14450, 8722, 11754, 1398)
@ -118,7 +125,37 @@ def mock_psnawpapi(mock_user: MagicMock) -> Generator[MagicMock]:
"isOfficiallyVerified": False, "isOfficiallyVerified": False,
"isMe": True, "isMe": True,
} }
client.user.return_value.trophy_titles.return_value = [
TrophyTitle(
np_service_name="trophy",
np_communication_id="NPWR03134_00",
trophy_set_version="01.03",
title_name="Assassin's Creed® III Liberation",
title_detail="Assassin's Creed® III Liberation",
title_icon_url="https://image.api.playstation.com/trophy/np/NPWR03134_00_0008206095F67FD3BB385E9E00A7C9CFE6F5A4AB96/5F87A6997DD23D1C4D4CC0D1F958ED79CB905331.PNG",
title_platform=frozenset({PlatformType.PS_VITA}),
has_trophy_groups=False,
progress=28,
hidden_flag=False,
earned_trophies=TrophySet(bronze=4, silver=8, gold=0, platinum=0),
defined_trophies=TrophySet(bronze=22, silver=21, gold=1, platinum=1),
last_updated_datetime=datetime(2016, 10, 6, 18, 5, 8, tzinfo=UTC),
np_title_id=None,
)
]
client.me.return_value.get_profile_legacy.return_value = {
"profile": {
"presences": [
{
"onlineStatus": "online",
"platform": "PSVITA",
"npTitleId": "PCSB00074_00",
"titleName": "Assassin's Creed® III Liberation",
"hasBroadcastData": False,
}
]
}
}
yield client yield client

View File

@ -12,6 +12,14 @@
'title_id': 'PPSA07784_00', 'title_id': 'PPSA07784_00',
'title_name': 'STAR WARS Jedi: Survivor™', 'title_name': 'STAR WARS Jedi: Survivor™',
}), }),
'PSVITA': dict({
'format': 'PSVITA',
'media_image_url': 'https://image.api.playstation.com/trophy/np/NPWR03134_00_0008206095F67FD3BB385E9E00A7C9CFE6F5A4AB96/5F87A6997DD23D1C4D4CC0D1F958ED79CB905331.PNG',
'platform': 'PSVITA',
'status': 'online',
'title_id': 'PCSB00074_00',
'title_name': "Assassin's Creed® III Liberation",
}),
}), }),
'availability': 'availableToPlay', 'availability': 'availableToPlay',
'presence': dict({ 'presence': dict({
@ -61,6 +69,7 @@
}), }),
'registered_platforms': list([ 'registered_platforms': list([
'PS5', 'PS5',
'PSVITA',
]), ]),
'trophy_summary': dict({ 'trophy_summary': dict({
'account_id': '**REDACTED**', 'account_id': '**REDACTED**',

View File

@ -1,4 +1,166 @@
# serializer version: 1 # serializer version: 1
# name: test_media_player_psvita[presence_payload0][media_player.playstation_vita-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'media_player',
'entity_category': None,
'entity_id': 'media_player.playstation_vita',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <MediaPlayerDeviceClass.RECEIVER: 'receiver'>,
'original_icon': None,
'original_name': None,
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'playstation',
'unique_id': 'my-psn-id_PSVITA',
'unit_of_measurement': None,
})
# ---
# name: test_media_player_psvita[presence_payload0][media_player.playstation_vita-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'receiver',
'entity_picture_local': None,
'friendly_name': 'PlayStation Vita',
'media_content_type': <MediaType.GAME: 'game'>,
'supported_features': <MediaPlayerEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'media_player.playstation_vita',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'standby',
})
# ---
# name: test_media_player_psvita[presence_payload1][media_player.playstation_vita-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'media_player',
'entity_category': None,
'entity_id': 'media_player.playstation_vita',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <MediaPlayerDeviceClass.RECEIVER: 'receiver'>,
'original_icon': None,
'original_name': None,
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'playstation',
'unique_id': 'my-psn-id_PSVITA',
'unit_of_measurement': None,
})
# ---
# name: test_media_player_psvita[presence_payload1][media_player.playstation_vita-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'receiver',
'entity_picture': 'https://image.api.playstation.com/trophy/np/NPWR03134_00_0008206095F67FD3BB385E9E00A7C9CFE6F5A4AB96/5F87A6997DD23D1C4D4CC0D1F958ED79CB905331.PNG',
'entity_picture_local': '/api/media_player_proxy/media_player.playstation_vita?token=123456789&cache=c7c916a6e18aec3d',
'friendly_name': 'PlayStation Vita',
'media_content_id': 'PCSB00074_00',
'media_content_type': <MediaType.GAME: 'game'>,
'media_title': "Assassin's Creed® III Liberation",
'supported_features': <MediaPlayerEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'media_player.playstation_vita',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'playing',
})
# ---
# name: test_media_player_psvita[presence_payload2][media_player.playstation_vita-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'media_player',
'entity_category': None,
'entity_id': 'media_player.playstation_vita',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <MediaPlayerDeviceClass.RECEIVER: 'receiver'>,
'original_icon': None,
'original_name': None,
'platform': 'playstation_network',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'playstation',
'unique_id': 'my-psn-id_PSVITA',
'unit_of_measurement': None,
})
# ---
# name: test_media_player_psvita[presence_payload2][media_player.playstation_vita-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'receiver',
'entity_picture_local': None,
'friendly_name': 'PlayStation Vita',
'media_content_type': <MediaType.GAME: 'game'>,
'supported_features': <MediaPlayerEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'media_player.playstation_vita',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_platform[PS4_idle][media_player.playstation_4-entry] # name: test_platform[PS4_idle][media_player.playstation_4-entry]
EntityRegistryEntrySnapshot({ EntityRegistryEntrySnapshot({
'aliases': set({ 'aliases': set({

View File

@ -1,7 +1,9 @@
"""Tests for PlayStation Network.""" """Tests for PlayStation Network."""
from datetime import timedelta
from unittest.mock import MagicMock from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from psnawp_api.core import ( from psnawp_api.core import (
PSNAWPAuthenticationError, PSNAWPAuthenticationError,
PSNAWPClientError, PSNAWPClientError,
@ -11,10 +13,13 @@ from psnawp_api.core import (
import pytest import pytest
from homeassistant.components.playstation_network.const import DOMAIN from homeassistant.components.playstation_network.const import DOMAIN
from homeassistant.components.playstation_network.coordinator import (
PlaystationNetworkRuntimeData,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -107,3 +112,154 @@ async def test_coordinator_update_auth_failed(
assert "context" in flow assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == config_entry.entry_id assert flow["context"].get("entry_id") == config_entry.entry_id
async def test_trophy_title_coordinator(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trophy title coordinator updates when PS Vita is registered."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(mock_psnawpapi.user.return_value.trophy_titles.mock_calls) == 1
freezer.tick(timedelta(days=1))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(mock_psnawpapi.user.return_value.trophy_titles.mock_calls) == 2
async def test_trophy_title_coordinator_auth_failed(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trophy title coordinator starts reauth on authentication error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
mock_psnawpapi.user.return_value.trophy_titles.side_effect = (
PSNAWPAuthenticationError
)
freezer.tick(timedelta(days=1))
async_fire_time_changed(hass)
await hass.async_block_till_done()
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow.get("step_id") == "reauth_confirm"
assert flow.get("handler") == DOMAIN
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == config_entry.entry_id
@pytest.mark.parametrize(
"exception", [PSNAWPNotFoundError, PSNAWPServerError, PSNAWPClientError]
)
async def test_trophy_title_coordinator_update_data_failed(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
exception: Exception,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trophy title coordinator update failed."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
mock_psnawpapi.user.return_value.trophy_titles.side_effect = exception
freezer.tick(timedelta(days=1))
async_fire_time_changed(hass)
await hass.async_block_till_done()
await hass.async_block_till_done()
runtime_data: PlaystationNetworkRuntimeData = config_entry.runtime_data
assert runtime_data.trophy_titles.last_update_success is False
async def test_trophy_title_coordinator_doesnt_update(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test trophy title coordinator does not update if no PS Vita is registered."""
mock_psnawpapi.me.return_value.get_account_devices.return_value = [
{"deviceType": "PS5"},
{"deviceType": "PS3"},
]
mock_psnawpapi.me.return_value.get_profile_legacy.return_value = {
"profile": {"presences": []}
}
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(mock_psnawpapi.user.return_value.trophy_titles.mock_calls) == 1
freezer.tick(timedelta(days=1))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(mock_psnawpapi.user.return_value.trophy_titles.mock_calls) == 1
async def test_trophy_title_coordinator_play_new_game(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_psnawpapi: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test we play a new game and get a title image on next trophy titles update."""
_tmp = mock_psnawpapi.user.return_value.trophy_titles.return_value
mock_psnawpapi.user.return_value.trophy_titles.return_value = []
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert (state := hass.states.get("media_player.playstation_vita"))
assert state.attributes.get("entity_picture") is None
mock_psnawpapi.user.return_value.trophy_titles.return_value = _tmp
freezer.tick(timedelta(days=1))
async_fire_time_changed(hass)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert len(mock_psnawpapi.user.return_value.trophy_titles.mock_calls) == 2
assert (state := hass.states.get("media_player.playstation_vita"))
assert (
state.attributes["entity_picture"]
== "https://image.api.playstation.com/trophy/np/NPWR03134_00_0008206095F67FD3BB385E9E00A7C9CFE6F5A4AB96/5F87A6997DD23D1C4D4CC0D1F958ED79CB905331.PNG"
)

View File

@ -114,6 +114,76 @@ async def test_platform(
"""Test setup of the PlayStation Network media_player platform.""" """Test setup of the PlayStation Network media_player platform."""
mock_psnawpapi.user().get_presence.return_value = presence_payload mock_psnawpapi.user().get_presence.return_value = presence_payload
mock_psnawpapi.me.return_value.get_profile_legacy.return_value = {
"profile": {"presences": []}
}
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
@pytest.mark.parametrize(
"presence_payload",
[
{
"profile": {
"presences": [
{
"onlineStatus": "standby",
"platform": "PSVITA",
"hasBroadcastData": False,
}
]
}
},
{
"profile": {
"presences": [
{
"onlineStatus": "online",
"platform": "PSVITA",
"npTitleId": "PCSB00074_00",
"titleName": "Assassin's Creed® III Liberation",
"hasBroadcastData": False,
}
]
}
},
{
"profile": {
"presences": [
{
"onlineStatus": "online",
"platform": "PSVITA",
"hasBroadcastData": False,
}
]
}
},
],
)
@pytest.mark.usefixtures("mock_psnawpapi", "mock_token")
async def test_media_player_psvita(
hass: HomeAssistant,
config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_psnawpapi: MagicMock,
presence_payload: dict[str, Any],
) -> None:
"""Test setup of the PlayStation Network media_player for PlayStation Vita."""
mock_psnawpapi.user().get_presence.return_value = {
"basicPresence": {
"availability": "unavailable",
"primaryPlatformInfo": {"onlineStatus": "offline", "platform": ""},
}
}
mock_psnawpapi.me.return_value.get_profile_legacy.return_value = presence_payload
config_entry.add_to_hass(hass) config_entry.add_to_hass(hass)
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()