diff --git a/examples/cameraServiceExample.py b/examples/cameraServiceExample.py index 3e4fad0..1c946f9 100644 --- a/examples/cameraServiceExample.py +++ b/examples/cameraServiceExample.py @@ -3,7 +3,7 @@ from context_logger import get_logger, setup_logging from examples import setup_shutdown -from hello import ServiceInfo, Hello, Group +from hello import ServiceInfo, Hello, Group, GroupUrl setup_logging('hello') @@ -13,8 +13,11 @@ def main() -> None: shutdown_event = setup_shutdown() + # Define the group URL optionally specifying the network interface + url = GroupUrl(address='239.0.1.1', port=5555, interface='wlan0') + # Define the group to advertise the camera service on - group = Group(name='effective-range/sniper', url='udp://239.0.1.1:5555') + group = Group.create(name='effective-range/sniper', url=url) # Define the service information for the camera hostname = 'er-sniper-camera-1' diff --git a/hello/group.py b/hello/group.py index f553510..4c53663 100644 --- a/hello/group.py +++ b/hello/group.py @@ -5,6 +5,25 @@ from dataclasses import dataclass from enum import Enum +import netifaces + + +@dataclass +class GroupUrl: + address: str + port: int + protocol: str = 'udp' + interface: str | None = None + + def resolve(self) -> str: + if self.interface and self.interface in netifaces.interfaces(): + inet_address = netifaces.ifaddresses(self.interface).get(netifaces.AF_INET) + if inet_address and len(inet_address) > 0: + address = inet_address[0].get('addr') + return f'{self.protocol}://{address};{self.address}:{self.port}' + + return f'{self.protocol}://{self.address}:{self.port}' + class GroupPrefix(Enum): HELLO = 'hello' @@ -22,6 +41,10 @@ def hello(self) -> 'PrefixedGroup': def query(self) -> 'PrefixedGroup': return PrefixedGroup(GroupPrefix.QUERY, self) + @staticmethod + def create(name: str, url: GroupUrl) -> 'Group': + return Group(name, url.resolve()) + @dataclass class PrefixedGroup: diff --git a/pyproject.toml b/pyproject.toml index 3f5fe54..715bb09 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ maintainers = [ { name = "Ferenc Nandor Janky & Attila Gombos", email = "info@effective-range.com" } ] dependencies = [ + "netifaces", "pyzmq @ git+https://github.com/EffectiveRange/pyzmq.git@v27.1.0+drafts1", "python-context-logger @ git+https://github.com/EffectiveRange/python-context-logger.git@latest", "python-common-utility @ git+https://github.com/EffectiveRange/python-common-utility.git@latest" diff --git a/tests/groupTest.py b/tests/groupTest.py new file mode 100644 index 0000000..02bcdc6 --- /dev/null +++ b/tests/groupTest.py @@ -0,0 +1,82 @@ +import unittest +from unittest import TestCase + +from context_logger import setup_logging + +from hello import GroupUrl, Group + + +class GroupTest(TestCase): + + @classmethod + def setUpClass(cls): + setup_logging('hello', 'DEBUG', warn_on_overwrite=False) + + def setUp(self): + print() + + def test_url_resolves_when_interface_is_not_provided(self): + # Given + group_url = GroupUrl(protocol='udp', address='239.0.1.1', port=5555) + + # When + url = group_url.resolve() + + # Then + self.assertEqual('udp://239.0.1.1:5555', url) + + def test_url_resolves_when_interface_is_provided(self): + # Given + group_url = GroupUrl(protocol='udp', address='239.0.1.1', port=5555, interface='lo') + + # When + url = group_url.resolve() + + # Then + self.assertEqual('udp://127.0.0.1;239.0.1.1:5555', url) + + def test_url_resolves_when_interface_is_provided_but_not_exists(self): + # Given + group_url = GroupUrl(protocol='udp', address='239.0.1.1', port=5555, interface='nonexistent0') + + # When + url = group_url.resolve() + + # Then + self.assertEqual('udp://239.0.1.1:5555', url) + + def test_group_created_with_resolved_url(self): + # Given + group_url = GroupUrl(address='239.0.1.1', port=5555, interface='lo') + + # When + group = Group.create('test-group', group_url) + + # Then + self.assertEqual(Group(name='test-group', url='udp://127.0.0.1;239.0.1.1:5555'), group) + + def test_prefixed_group_created_with_hello_prefix(self): + # Given + group = Group('test-group', 'udp://239.0.1.1:5555') + + # When + prefixed_group = group.hello() + + # Then + self.assertEqual('hello:test-group', prefixed_group.name) + self.assertEqual(group.url, prefixed_group.url) + + def test_prefixed_group_created_with_query_prefix(self): + # Given + group = Group('test-group', 'udp://239.0.1.1:5555') + + # When + prefixed_group = group.query() + + # Then + self.assertEqual('query:test-group', prefixed_group.name) + self.assertEqual(group.url, prefixed_group.url) + + +if __name__ == '__main__': + unittest.main()