Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add possibility of using an address list instead #202

Merged
merged 5 commits into from
Dec 7, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 38 additions & 1 deletion glocaltokens/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
ACCESS_TOKEN_DURATION,
ACCESS_TOKEN_SERVICE,
ANDROID_ID_LENGTH,
DEFAULT_DISCOVERY_PORT,
DISCOVERY_TIMEOUT,
GOOGLE_HOME_FOYER_API,
HOMEGRAPH_DURATION,
Expand All @@ -26,6 +27,7 @@
from .types import DeviceDict
from .utils import network as net_utils, token as token_utils
from .utils.logs import censor
from .utils.network import is_valid_ipv4_address

logging.basicConfig(level=logging.ERROR)
LOGGER = logging.getLogger(__name__)
Expand Down Expand Up @@ -335,6 +337,7 @@ def get_google_devices(
self,
models_list: list[str] | None = None,
disable_discovery: bool = False,
addresses: dict[str, str] | None = None,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's better to add new arguments in the end, to not break reverse compatibility.

zeroconf_instance: Zeroconf | None = None,
force_homegraph_reload: bool = False,
discovery_timeout: int = DISCOVERY_TIMEOUT,
Expand All @@ -346,6 +349,9 @@ def get_google_devices(
models_list: The list of accepted model names.
disable_discovery: Whether or not the device's IP and port should
be searched for in the network.
addresses: Dict of network devices from the local network
({"name": "ip_address"}). If set to `None` will try to automatically
discover network devices. Disable discovery by setting to `{}`.
zeroconf_instance: If you already have an initialized zeroconf instance,
use it here.
force_homegraph_reload: If the stored homegraph should be generated again.
Expand All @@ -365,13 +371,29 @@ def get_google_devices(

devices: list[Device] = []

def is_dict_with_valid_ipv4_addresses(data: dict[str, str]) -> bool:
# Validate the data structure is correct and that each entry contains a
# valid IPv4 address.
return isinstance(data, dict) and all(
isinstance(x, str) and is_valid_ipv4_address(x) for x in data.values()
)

if addresses and not is_dict_with_valid_ipv4_addresses(addresses):
KapJI marked this conversation as resolved.
Show resolved Hide resolved
# We need to disable flake8-use-fstring because of the brackets,
# it causes a false positive.
LOGGER.error(
"Invalid dictionary structure for addresses dictionary "
"argument. Correct structure is {'device_name': 'ipaddress'}" # noqa
)
return devices

if homegraph is None:
LOGGER.debug("Failed to fetch homegraph")
return devices

network_devices: list[NetworkDevice] = []
if disable_discovery is False:
LOGGER.debug("Getting network devices...")
LOGGER.debug("Automatically discovering network devices...")
network_devices = discover_devices(
models_list,
timeout=discovery_timeout,
Expand All @@ -385,6 +407,8 @@ def find_device(unique_id: str) -> NetworkDevice | None:
return device
return None

address_dict = addresses if addresses else {}
KapJI marked this conversation as resolved.
Show resolved Hide resolved

LOGGER.debug("Iterating in %d homegraph devices", len(homegraph.home.devices))
for item in homegraph.home.devices:
if item.local_auth_token != "":
Expand All @@ -405,6 +429,14 @@ def find_device(unique_id: str) -> NetworkDevice | None:
unique_id,
)
network_device = find_device(unique_id)
elif item.device_name in address_dict:
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This means if discovery is enabled, address_dict won't be used. That contradicts the idea to have separate disable_discovery flag.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also if there's no item with such name in address_dict, it's worth to print a warning message.

network_device = NetworkDevice(
name=item.device_name,
ip_address=address_dict[item.device_name],
port=DEFAULT_DISCOVERY_PORT,
model=item.hardware.model,
unique_id=item.device_info.device_id,
)

device = Device(
device_id=item.device_info.device_id,
Expand Down Expand Up @@ -438,6 +470,7 @@ def get_google_devices_json(
models_list: list[str] | None = None,
indent: int = 2,
disable_discovery: bool = False,
addresses: dict[str, str] | None = None,
zeroconf_instance: Zeroconf | None = None,
force_homegraph_reload: bool = False,
) -> str:
Expand All @@ -449,6 +482,9 @@ def get_google_devices_json(
indent: The indentation for the json formatting.
disable_discovery: Whether or not the device's IP and port should
be searched for in the network.
addresses: Dict of network devices from the local network
({"name": "ip_address"}). If set to `None` will try to automatically
discover network devices. Disable discovery by setting to `{}`.
zeroconf_instance: If you already have an initialized zeroconf instance,
use it here.
force_homegraph_reload: If the stored homegraph should be generated again.
Expand All @@ -457,6 +493,7 @@ def get_google_devices_json(
google_devices = self.get_google_devices(
models_list=models_list,
disable_discovery=disable_discovery,
addresses=addresses,
zeroconf_instance=zeroconf_instance,
force_homegraph_reload=force_homegraph_reload,
)
Expand Down
1 change: 1 addition & 0 deletions glocaltokens/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
HOMEGRAPH_DURATION: Final = 24 * 60 * 60

DISCOVERY_TIMEOUT: Final = 2
DEFAULT_DISCOVERY_PORT: Final = 0

GOOGLE_HOME_MODELS: Final = [
"Google Home",
Expand Down
8 changes: 6 additions & 2 deletions tests/factory/providers.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,14 +43,18 @@ def local_auth_token(self) -> str:
class HomegraphProvider(TokenProvider):
"""Homegraph provider"""

def homegraph_device(self) -> GetHomeGraphResponse.Home.Device:
def homegraph_device(
self, device_name: str | None = None
) -> GetHomeGraphResponse.Home.Device:
"""Using the content from test requests as reference"""
device_name = device_name if device_name else faker.word()

return GetHomeGraphResponse.Home.Device(
local_auth_token=self.local_auth_token(),
device_info=GetHomeGraphResponse.Home.Device.DeviceInfo(
device_id=str(faker.uuid4)
),
device_name=faker.word(),
device_name=device_name,
hardware=GetHomeGraphResponse.Home.Device.Hardware(model=faker.word()),
)

Expand Down
24 changes: 20 additions & 4 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -322,19 +322,27 @@ def test_get_homegraph_retries(
def test_get_google_devices(self, m_get_homegraph: NonCallableMock) -> None:
"""Test getting google devices"""
# With just one device returned from homegraph
homegraph_device = faker.homegraph_device()
fake_device_name = faker.word()
fake_ip_address = faker.ipv4()
homegraph_device = faker.homegraph_device(device_name=fake_device_name)
m_get_homegraph.return_value.home.devices = [homegraph_device]

# With no discover_devices, with no model_list
google_devices = self.client.get_google_devices(disable_discovery=True)
google_devices = self.client.get_google_devices(
disable_discovery=True,
addresses={fake_device_name: fake_ip_address},
)
self.assertEqual(len(google_devices), 1)
DurgNomis-drol marked this conversation as resolved.
Show resolved Hide resolved

google_device = google_devices[0]
self.assertDevice(google_device, homegraph_device)
self.assertIsNotNone(google_device.network_device)
if google_device.network_device is not None:
self.assertEqual(google_device.network_device.ip_address, fake_ip_address)
DurgNomis-drol marked this conversation as resolved.
Show resolved Hide resolved

# With two devices returned from homegraph
# but one device having the invalid token
homegraph_device_valid = faker.homegraph_device()
homegraph_device_valid = faker.homegraph_device(device_name=fake_device_name)
homegraph_device_invalid = faker.homegraph_device()
homegraph_device_invalid.local_auth_token = (
faker.word()
Expand All @@ -345,9 +353,17 @@ def test_get_google_devices(self, m_get_homegraph: NonCallableMock) -> None:
homegraph_device_invalid,
homegraph_device_valid,
]
google_devices = self.client.get_google_devices(disable_discovery=True)
google_devices = self.client.get_google_devices(
disable_discovery=True,
addresses={fake_device_name: fake_ip_address},
)
self.assertEqual(len(google_devices), 1)
self.assertDevice(google_devices[0], homegraph_device_valid)
self.assertIsNotNone(google_devices[0].network_device)
if google_devices[0].network_device is not None:
self.assertEqual(
google_devices[0].network_device.ip_address, fake_ip_address
)

@patch("glocaltokens.client.GLocalAuthenticationTokens.get_google_devices")
def test_get_google_devices_json(
Expand Down