diff --git a/.circleci/config.yml b/.circleci/config.yml index d8106b6..090853c 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -1,13 +1,13 @@ version: 2.1 orbs: - python: circleci/python@2.1.1 - codecov: codecov/codecov@3.2.3 + python: circleci/python@3.3.0 + codecov: codecov/codecov@5.4.3 jobs: build: docker: - - image: cimg/python:3.10.2 + - image: cimg/python:3.14.2 resource_class: small steps: @@ -23,7 +23,7 @@ jobs: unittest: docker: - - image: cimg/python:3.10.2 + - image: cimg/python:3.14.2 resource_class: small steps: @@ -47,7 +47,7 @@ jobs: build-and-publish: docker: - - image: cimg/python:3.10.2 + - image: cimg/python:3.14.2 resource_class: small diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 1dd4a11..e49f9d0 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -30,11 +30,11 @@ jobs: # Initializes the CodeQL tools for scanning. - name: Initialize CodeQL - uses: github/codeql-action/init@v2 + uses: github/codeql-action/init@v3 with: languages: ${{ matrix.language }} - name: Perform CodeQL Analysis - uses: github/codeql-action/analyze@v2 + uses: github/codeql-action/analyze@v3 with: category: "/language:${{matrix.language}}" diff --git a/PyStageLinQ/PyStageLinQ.py b/PyStageLinQ/PyStageLinQ.py index 6f1801c..9b8c414 100644 --- a/PyStageLinQ/PyStageLinQ.py +++ b/PyStageLinQ/PyStageLinQ.py @@ -8,7 +8,9 @@ import time import asyncio import logging +import platform import psutil +import ipaddress from typing import Callable from . import Device @@ -21,6 +23,91 @@ logger = logging.getLogger("PyStageLinQ") +@dataclass +class PyStageLinQ_interface_info: + name: str + id: int + addr: int + addr_str: str + mask: int + status: psutil._common.snicstats + n_disc_msg_send: int + + +class PyStageLinQ_network_interface: + def __init__(self, ip=None, discovery_port=51337): + self.target_interfaces = [] + self.discovery_port = discovery_port + self.get_interface_from_ip(ip) + + def get_interface_from_ip(self, ip): + if ip is None: + ip_list = ["any"] + elif type(ip) is str: + ip_list = [ip] + elif type(ip) is list: + for entry in ip: + if type(entry) is not str: + raise TypeError + ip_list = ip + else: + raise TypeError + + for interface in psutil.net_if_stats().items(): + for interface_info in psutil.net_if_addrs()[interface[0]]: + # Only look for IPV4 binds + if socket.AF_INET == interface_info.family and ( + interface_info.address in ip_list or ip_list[0] == "any" + ): + self.target_interfaces.append( + PyStageLinQ_interface_info( + interface[0], + len(self.target_interfaces), + int(ipaddress.IPv4Address(interface_info.address)), + interface_info.address, + int(ipaddress.IPv4Address(interface_info.netmask)), + interface[1], + 0, + ) + ) + + def send_discovery_frame(self, discovery_frame): + for interface in self.target_interfaces: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as discovery_socket: + discovery_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + discovery_socket.bind((interface.addr_str, 0)) + try: + discovery_socket.sendto( + discovery_frame, + ("255.255.255.255", self.discovery_port), + ) + self.target_interfaces[interface.id].n_disc_msg_send += 1 + except PermissionError: + logger.warning( + f"Cannot send message on interface {interface.name}, " + f"this error could be due to that there is no network interface set up with this IP range" + ) + raise PermissionError + + def determine_interface_of_remote_ip(self, ip): + for interface in self.target_interfaces: + if (int(ipaddress.IPv4Address(ip)) & interface.mask) == ( + interface.mask & interface.addr + ): + return interface + + return None + + def send_desc_on_all_if(self): + for interface in self.target_interfaces: + # Wait until a few discovery frames have been sent to make sure the other devices have seen us. If they have + # not and we are asking for services it will be denied. + if interface.n_disc_msg_send < 3: + return False + + return True + + class PyStageLinQ: """ The main object for PyStageLinQ. Use this object to first initialize and then start PyStageLinq @@ -64,27 +151,12 @@ def __init__( self.device_list = Device.DeviceList() - self.ip = [] - if ip is None: - interfaces = psutil.net_if_addrs() - for interface in interfaces.items(): - for interface_address in interface[1]: - if socket.AF_INET == interface_address.family: - self.ip.append(interface_address.address) - - else: - self.ip = [ip] + self.network_interface = PyStageLinQ_network_interface(ip) self.tasks = set() - self.found_services = [] - self.new_services_available = False self.active_services = [] - self.devices_with_services_pending_list = [] - self.devices_with_services_pending = False - self.devices_with_services_lock = asyncio.Lock() - self.new_device_found_callback = new_device_found_callback logger.debug(f"Initialized!") @@ -115,32 +187,16 @@ def _stop(self): discovery_info.ConnectionType = ConnectionTypes.EXIT discovery_frame = discovery.encode_frame(discovery_info) - self._send_discovery_frame(discovery_frame) + self.network_interface.send_discovery_frame(discovery_frame) logger.info(f"Gracefully shutdown complete") - except: + except Exception as e: logger.debug('Could not send "EXIT" discovery frame during shutdown') + raise e def _announce_self(self): discovery = StageLinQDiscovery() discovery_frame = discovery.encode_frame(self.discovery_info) - self._send_discovery_frame(discovery_frame) - - def _send_discovery_frame(self, discovery_frame): - for ip in self.ip: - with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as discovery_socket: - discovery_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) - discovery_socket.bind((ip, 0)) - try: - discovery_socket.sendto( - discovery_frame, - ("255.255.255.255", self.StageLinQ_discovery_port), - ) - except PermissionError: - logger.warning( - f"Cannot send message on interface {ip}, " - f"this error could be due to that there is no network interface set up with this IP range" - ) - raise PermissionError + self.network_interface.send_discovery_frame(discovery_frame) def get_loop_condition(self) -> bool: return self._loopcondition @@ -149,6 +205,7 @@ async def _discover_stagelinq_device(self, host_ip, timeout=10): """ This function is used to find StageLinQ device announcements. """ + logger.info(f"Trying to discover StageLinQ devices.") # Local Constants @@ -156,26 +213,27 @@ async def _discover_stagelinq_device(self, host_ip, timeout=10): # Create socket discover_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: discover_socket.bind( - ("255.255.255.255", self.StageLinQ_discovery_port) - ) # bind socket StageLinQ interface - except: + (host_ip, self.StageLinQ_discovery_port) + ) # bind socket to broadcast + except Exception as e: # Cannot bind to socket, check if IP is correct and link is up logger.warning( f"Cannot bind to IP socket: {host_ip} on port {self.StageLinQ_discovery_port}" ) - return PyStageLinQError.CANNOTBINDSOCKET + raise e discover_socket.setblocking(False) loop_timeout = time.time() + timeout logger.debug( - f"Socket bound to IP {host_ip} and Port {self.StageLinQ_discovery_port} successfully. Starting to look for discovery frames" + f"Socket bound to IP {host_ip} and Port {self.StageLinQ_discovery_port} successfully. Starting to look for " + f"discovery frames" ) while self.get_loop_condition(): - await asyncio.sleep(0.1) data_available = select.select([discover_socket], [], [], 0) if data_available[0]: data, addr = discover_socket.recvfrom(discover_buffer_size) @@ -186,17 +244,18 @@ async def _discover_stagelinq_device(self, host_ip, timeout=10): # something went wrong continue - # Devices found, setting new timeout - loop_timeout = time.time() + timeout + if self.name == discovery_frame.device_name: + # Ourselves, ignore message + continue if 0 == discovery_frame.Port: - # If port is 0 there are no services to request + # If port is 0 there are no services to request, and for our use an invalid device continue - if self.name == discovery_frame.device_name: - # Ourselves, ignore - continue + # External Device present, setting new timeout + loop_timeout = time.time() + timeout + # Check if we have already registered this device, and if so ignore it for now device_registered = self.device_list.find_registered_device( discovery_frame.get() ) @@ -209,9 +268,10 @@ async def _discover_stagelinq_device(self, host_ip, timeout=10): if time.time() > loop_timeout: # No devices found within timeout logger.info( - "No discovery frames found on {host_ip} last {timeout} seconds." + f"No discovery frames found on {host_ip} last {timeout} seconds." ) return PyStageLinQError.DISCOVERYTIMEOUT + await asyncio.sleep(0.1) async def _register_new_device(self, discovery_frame, ip): stagelinq_device = StageLinQService(ip, discovery_frame, self.OwnToken, None) @@ -265,6 +325,10 @@ async def _start_stagelinq(self, standalone=False): # Start the initial tasks of the library self.tasks.add(asyncio.create_task(self._periodic_announcement())) + # wait for discovery message to be send on all interfaces + while not self.network_interface.send_desc_on_all_if(): + await asyncio.sleep(0.1) + self.tasks.add(asyncio.create_task(self._py_stagelinq_strapper())) if standalone: @@ -279,6 +343,7 @@ async def _wait_for_exit(self): f"Exception: {task.exception()} occured in task: {task.get_coro()}, stopping " f"PyStageLinQ." ) + logger.debug(f"Stacktrace to error: {task.print_stack()}") self.stop() raise task.exception() return @@ -296,15 +361,11 @@ async def _periodic_announcement(self): async def _py_stagelinq_strapper(self): strapper_tasks = set() - logger.info( - f"Looking for discovery frames on {len(self.ip)} IP local IP addresses:" - ) + logger.info(f"Starting to look for StageLinQ discovery frames:") - for ip in self.ip: - logger.info(f"{ip}") - strapper_tasks.add( - asyncio.create_task(self._discover_stagelinq_device(ip, timeout=2)) - ) + strapper_tasks.add( + asyncio.create_task(self._discover_stagelinq_device("", timeout=2)) + ) while self.get_loop_condition(): all_tasks_done = True diff --git a/PyStageLinQ/Token.py b/PyStageLinQ/Token.py index b1e117a..4cc827f 100644 --- a/PyStageLinQ/Token.py +++ b/PyStageLinQ/Token.py @@ -25,11 +25,14 @@ def generate_token(self): """ randomized_bytes = self._get_randomized_bytes(self.TOKENLENGTH) - # check if first bit is set and set bit to 0 if so. - if randomized_bytes[0] >= 128: - randomized_bytes[0] = randomized_bytes[0] - 128 - - self.token = int.from_bytes(randomized_bytes, byteorder="big") + # The following checks are to solve issues with invalid token values I've found when testing. The actual token + # structure is currently unknown, so a random value is generated as this seems to be what's going on from the + # devices I've tested against. + invalid_token_bit_mask = ~(0x1 << 127 | 0xC << 60) + + self.token = ( + int.from_bytes(randomized_bytes, byteorder="big") & invalid_token_bit_mask + ) @staticmethod def _get_randomized_bytes(length: int) -> bytes: diff --git a/README.md b/README.md index e9612fc..2790aa1 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,7 @@ isn't properly tested and may cause unexpected issues. As this file is not part be found on [GitHub](https://github.com/Jaxc/PyStageLinQ/blob/main/tools/StageLinQ.lua) # Compatability -PyStageLinQ has been tested with a Denon DJ Prime Go on Windows 10 and Linux (Mint 20.3) with Python 3.10. +PyStageLinQ has been tested with a Denon DJ Prime Go on Windows 11 and Linux (Mint 22.2) with Python 3.10. # Acknowledgements Big thanks to icedream for his implementation of StageLinQ in go: diff --git a/changelog.md b/changelog.md index 89736e7..b65c4b0 100644 --- a/changelog.md +++ b/changelog.md @@ -3,15 +3,25 @@ Here follows a log of released versions of PyStageLinQ. ## [0.2.2] ### Fixed -Problems on Linux should now be solved. +Problems on Linux should now be solved. The soluton has been tested on both Windows and Linux and seems to be +working on both systems! + PyStageLinQ will now listen to address "255.255.255.255" on all interfaces, and transmit discovery frames on either the interface specified by `PyStageLinQ(..., ip=)` or on all interfaces if `ip` is not set. PyStageLinQ will now use its release version when sending discovery frames. +Invalid Tokens should not be generated anymore, causing PyStageLinQ to fail to get services. + ### Added More logging output in PyStageLinQ.py. +A short delay when PyStageLinQ is started between sending discovery frames and trying to request services. +This is due to an issue with the StageLinQ device rejecting the service request due to it not having seen the discovery +message. + +Release instructions to help to make releases quicker and more consistent. + ## [0.2.0] ### Fixed diff --git a/release_instructions.md b/release_instructions.md new file mode 100644 index 0000000..eb51ed8 --- /dev/null +++ b/release_instructions.md @@ -0,0 +1,32 @@ +# Commit checklist +* Make sure unit tests are successful +* make sure changelog.md has been updated +* Check if the update affects the documentation, and if so update it +* If CI has stopped working it is fine to update it in the commit + +# Release checklist +* Recheck changelog.md and docs/ to make sure they are updated properly +* Before a real release is to be created, create a dev release to make sure the release process works as intended. +It is not possible to re-release a version number so better to first release to a test. +* This is done by adding a tag in git. The CI will then automatically build and publish the release. +* Once the dev release is out, do additional testing. Currently, I'll test it for a few streams to make sure it seems +stable but this process could be done better. +* Once the dev release seems stable, create another tag on the same commit with the real release version +* Once the release is deployed, create a release in GitHub, use changelog.md to make sure the changelog is correct on +the release +* IMMEDIATELY after release is done, create a new commit that adds a new section to changelog.md to the new version. The new release shall only have the patch incremented by one, if at a later date the +next release gets upgraded to a new minor or major version this has to be changed again. + +# Versioning +PyStageLinQ uses the standard major.minor.patch version convention. + +The major version is to indicate incompatibility with previous software, except for version 1.0.0. +The plan is to push for release 1.0.0 to avoid +getting stuck in the 0.x.y pitfall with having stable software but not a version 1.0.0 release. After that no new major +release is planned. + +Minors versions are ment to indicate new functionality, while a patch version is a smaller fix that should not affect +the usage of the code. + +There is also the possibility for dev-releases, these are a test version released before a real version, either to test +a fix or to test the release flow. These are indicated by having devx in the end, e.g. 0.2.2.dev1 \ No newline at end of file diff --git a/tests/Main.py b/tests/Main.py index 0c66912..a3f372a 100644 --- a/tests/Main.py +++ b/tests/Main.py @@ -53,12 +53,22 @@ def state_map_data_print(data): def main(): logging.basicConfig(level=logging.INFO) global PrimeGo - """PrimeGo = PyStageLinQ.PyStageLinQ( - new_device_found_callback, name="Jaxcie StageLinQ", ip="169.254.13.37" - )""" - PrimeGo = PyStageLinQ.PyStageLinQ( - new_device_found_callback, name="Jaxcie StageLinQ" - ) + ip_choice = 1 + match ip_choice: + case 0: + PrimeGo = PyStageLinQ.PyStageLinQ( + new_device_found_callback, name="Jaxcie StageLinQ" + ) + case 1: + PrimeGo = PyStageLinQ.PyStageLinQ( + new_device_found_callback, name="Jaxcie StageLinQ", ip="169.254.13.37" + ) + case 2: + PrimeGo = PyStageLinQ.PyStageLinQ( + new_device_found_callback, + name="Jaxcie StageLinQ", + ip=["169.254.13.37", "127.0.0.1"], + ) PrimeGo.start_standalone() diff --git a/tests/unit/test_unit_PyStageLinQ.py b/tests/unit/test_unit_PyStageLinQ.py index a71f664..12bb445 100644 --- a/tests/unit/test_unit_PyStageLinQ.py +++ b/tests/unit/test_unit_PyStageLinQ.py @@ -2,8 +2,6 @@ import PyStageLinQ.PyStageLinQ from PyStageLinQ.ErrorCodes import * from unittest.mock import AsyncMock, Mock, MagicMock -from unittest import mock -from socket import AF_INET import random @@ -26,7 +24,17 @@ def dummy_socket(): @pytest.fixture() -def dummy_pystagelinq(dummy_ip): +def dummy_PyStageLinQ_network_interface(): + return MagicMock() + + +@pytest.fixture() +def dummy_pystagelinq(dummy_ip, monkeypatch, dummy_PyStageLinQ_network_interface): + monkeypatch.setattr( + PyStageLinQ.PyStageLinQ, + "PyStageLinQ_network_interface", + dummy_PyStageLinQ_network_interface, + ) return PyStageLinQ.PyStageLinQ.PyStageLinQ(None, name=name, ip=dummy_ip) @@ -41,13 +49,14 @@ def ensure_cleanup(dummy_socket): gc.collect() -def test_init_values(dummy_pystagelinq, dummy_ip): +def test_init_values(dummy_pystagelinq, dummy_ip, dummy_PyStageLinQ_network_interface): assert dummy_pystagelinq.REQUESTSERVICEPORT == 0 + assert dummy_pystagelinq._loopcondition is True assert dummy_pystagelinq.name == name assert dummy_pystagelinq.OwnToken.get_token() != 0 assert dummy_pystagelinq.discovery_info.Token is dummy_pystagelinq.OwnToken assert dummy_pystagelinq.discovery_info.DeviceName == name - assert dummy_pystagelinq.discovery_info.ConnectionType is "DISCOVERER_HOWDY_" + assert dummy_pystagelinq.discovery_info.ConnectionType == "DISCOVERER_HOWDY_" assert dummy_pystagelinq.discovery_info.SwName == "Python" assert dummy_pystagelinq.discovery_info.SwVersion == "0.0.1" assert ( @@ -58,48 +67,15 @@ def test_init_values(dummy_pystagelinq, dummy_ip): assert ( type(dummy_pystagelinq.device_list) is PyStageLinQ.PyStageLinQ.Device.DeviceList ) - assert dummy_pystagelinq.ip == [dummy_ip] + dummy_PyStageLinQ_network_interface.assert_called_once_with(dummy_ip) assert dummy_pystagelinq.tasks == set() - assert dummy_pystagelinq.found_services == [] - assert dummy_pystagelinq.new_services_available is False - assert dummy_pystagelinq.active_services == [] - assert dummy_pystagelinq.devices_with_services_pending_list == [] - assert dummy_pystagelinq.devices_with_services_pending is False - assert ( - type(dummy_pystagelinq.devices_with_services_lock) - == PyStageLinQ.PyStageLinQ.asyncio.Lock - ) - assert dummy_pystagelinq.new_device_found_callback is None -def test_init_values_ip_none(monkeypatch, dummy_socket): - class ifutils_net_if_addrs: - def __init__(self, ip=[]): - self.address = ip - self.family = AF_INET - - dummy_psutil = MagicMock() - dummy_ips = { - "interface1": [ifutils_net_if_addrs(ip="1.2.3.4")], - "interface2": [ifutils_net_if_addrs(ip="5.6.7.8")], - "interface3": [ifutils_net_if_addrs(ip="9.10.11.12")], - } - - dummy_ips["interface3"][0].family = None - - dummy_psutil.net_if_addrs.return_value = dummy_ips - monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil) - - dummy_pystagelinq = PyStageLinQ.PyStageLinQ.PyStageLinQ(None, ip=None) - - assert dummy_pystagelinq.ip == ["1.2.3.4", "5.6.7.8"] - - def test_start_standalone(dummy_pystagelinq, monkeypatch): run_mock = Mock() start_stagelinq_mock = Mock() @@ -124,16 +100,14 @@ def test_start(dummy_pystagelinq, monkeypatch): def test_internal_stop(dummy_pystagelinq, monkeypatch): - send_discovery_frame_mock = Mock() + network_interface_mock = Mock() dummy_discovery = PyStageLinQ.PyStageLinQ.StageLinQDiscovery() - monkeypatch.setattr( - dummy_pystagelinq, "_send_discovery_frame", send_discovery_frame_mock - ) + monkeypatch.setattr(dummy_pystagelinq, "network_interface", network_interface_mock) dummy_pystagelinq._stop() - send_discovery_frame_mock.assert_called_once_with( + network_interface_mock.send_discovery_frame.assert_called_once_with( dummy_discovery.encode_frame( PyStageLinQ.PyStageLinQ.StageLinQDiscoveryData( Token=dummy_pystagelinq.OwnToken, @@ -148,73 +122,18 @@ def test_internal_stop(dummy_pystagelinq, monkeypatch): def test_announce_self(dummy_pystagelinq, monkeypatch): - send_discovery_frame_mock = Mock() + network_interface_mock = Mock() dummy_discovery = PyStageLinQ.PyStageLinQ.StageLinQDiscovery() - monkeypatch.setattr( - dummy_pystagelinq, "_send_discovery_frame", send_discovery_frame_mock - ) + monkeypatch.setattr(dummy_pystagelinq, "network_interface", network_interface_mock) dummy_pystagelinq._announce_self() - send_discovery_frame_mock.assert_called_once_with( + network_interface_mock.send_discovery_frame.assert_called_once_with( dummy_discovery.encode_frame(dummy_pystagelinq.discovery_info) ) -def test_send_discovery_frame(dummy_pystagelinq, monkeypatch, dummy_socket): - - dummy_discovery_frame = "AAAA" - - discovery_socket = MagicMock() - - dummy_socket.socket.side_effect = discovery_socket - - dummy_socket.getaddrinfo.side_effect = [[["255.255.255.255"]]] - - monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "socket", dummy_socket) - - dummy_pystagelinq._send_discovery_frame(dummy_discovery_frame) - - dummy_socket.socket.assert_called_once_with( - dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM - ) - discovery_socket.return_value.__enter__.return_value.setsockopt.assert_called_once_with( - dummy_socket.SOL_SOCKET, dummy_socket.SO_BROADCAST, 1 - ) - discovery_socket.return_value.__enter__.return_value.sendto.assert_called_once_with( - dummy_discovery_frame, - ("255.255.255.255", dummy_pystagelinq.StageLinQ_discovery_port), - ) - - -def test_send_discovery_frame_permission_error( - dummy_pystagelinq, monkeypatch, dummy_ip, dummy_socket -): - dummy_discovery_frame = "AAAA" - - discovery_socket = MagicMock() - - dummy_socket.getaddrinfo.side_effect = [[["255.255.255.255"]]] - - dummy_socket.socket.side_effect = discovery_socket - discovery_socket.return_value.__enter__.return_value.sendto.side_effect = ( - PermissionError - ) - - monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "socket", dummy_socket) - - with pytest.raises(PermissionError) as exception: - dummy_pystagelinq._send_discovery_frame(dummy_discovery_frame) - - dummy_socket.socket.assert_called_once_with( - dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM - ) - discovery_socket.return_value.__enter__.return_value.setsockopt.assert_called_once_with( - dummy_socket.SOL_SOCKET, dummy_socket.SO_BROADCAST, 1 - ) - - @pytest.mark.asyncio async def test_discover_stagelinq_device_bind_error( dummy_pystagelinq, monkeypatch, dummy_ip, dummy_socket @@ -223,10 +142,10 @@ async def test_discover_stagelinq_device_bind_error( dummy_socket.socket.return_value.bind.side_effect = Exception() - assert ( + with pytest.raises(Exception) as exception: await dummy_pystagelinq._discover_stagelinq_device(dummy_ip) - == PyStageLinQError.CANNOTBINDSOCKET - ) + + assert exception.type is Exception def test_get_loop_condition(dummy_pystagelinq): @@ -257,7 +176,7 @@ async def test_discover_stagelinq_check_initialization( dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM ) dummy_socket.socket.return_value.bind.assert_called_once_with( - ("255.255.255.255", dummy_pystagelinq.StageLinQ_discovery_port) + (dummy_ip, dummy_pystagelinq.StageLinQ_discovery_port) ) dummy_socket.socket.return_value.setblocking.assert_called_once_with(False) @@ -291,7 +210,6 @@ async def test_discover_stagelinq_timeout( == PyStageLinQError.DISCOVERYTIMEOUT ) - sleep_mock.assert_called_once_with(0.1) select_mock.select.assert_called_once_with( [dummy_socket.socket.return_value], [], [], 0 ) @@ -329,7 +247,6 @@ async def test_discover_stagelinq_bad_frame( assert await dummy_pystagelinq._discover_stagelinq_device(dummy_ip) is None - sleep_mock.assert_called_once_with(0.1) select_mock.select.assert_called_once_with( [dummy_socket.socket.return_value], [], [], 0 ) @@ -353,10 +270,10 @@ async def test_discover_stagelinq_bad_port( class discovery_dummy: decode_frame = Mock(side_effect=[PyStageLinQError.STAGELINQOK]) + device_name = "Not Python" get = Mock() Port = 0 - stagelinq_discovery_mock = discovery_dummy() monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "StageLinQDiscovery", discovery_dummy) get_loop_condition_mock = Mock() @@ -371,7 +288,7 @@ class discovery_dummy: assert await dummy_pystagelinq._discover_stagelinq_device(dummy_ip) is None - assert time_mock.time.call_count == 2 + assert time_mock.time.call_count == 1 @pytest.mark.asyncio @@ -395,7 +312,6 @@ class discovery_dummy: Port = dummy_port device_name = name - stagelinq_discovery_mock = discovery_dummy() monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "StageLinQDiscovery", discovery_dummy) get_loop_condition_mock = Mock() @@ -410,7 +326,7 @@ class discovery_dummy: assert await dummy_pystagelinq._discover_stagelinq_device(dummy_ip) is None - assert time_mock.time.call_count == 2 + assert time_mock.time.call_count == 1 @pytest.mark.asyncio @@ -698,6 +614,34 @@ async def test_start_stagelinq(dummy_pystagelinq, monkeypatch): create_task_mock.assert_called_with(_py_stagelinq_strapper_mock.return_value) +@pytest.mark.asyncio +async def test_start_stagelinq_disc_msg_not_sent( + dummy_pystagelinq, monkeypatch, dummy_PyStageLinQ_network_interface +): + create_task_mock = Mock() + monkeypatch.setattr( + PyStageLinQ.PyStageLinQ.asyncio, "create_task", create_task_mock + ) + + _periodic_announcement_mock = Mock() + monkeypatch.setattr( + dummy_pystagelinq, "_periodic_announcement", _periodic_announcement_mock + ) + + _py_stagelinq_strapper_mock = Mock() + monkeypatch.setattr( + dummy_pystagelinq, "_py_stagelinq_strapper", _py_stagelinq_strapper_mock + ) + + dummy_pystagelinq.network_interface.send_desc_on_all_if = MagicMock( + side_effect=[False, False, True] + ) + + await dummy_pystagelinq._start_stagelinq() + + assert dummy_pystagelinq.network_interface.send_desc_on_all_if.call_count == 3 + + @pytest.mark.asyncio async def test_start_stagelinq_standalone(dummy_pystagelinq, monkeypatch): create_task_mock = Mock() @@ -883,7 +827,7 @@ async def test_periodic_announcement(dummy_pystagelinq, monkeypatch): @pytest.mark.asyncio -async def test_py_stagelinq_strapper(dummy_pystagelinq, monkeypatch, dummy_ip): +async def test_py_stagelinq_strapper(dummy_pystagelinq, monkeypatch): discover_device_mock = AsyncMock() monkeypatch.setattr( dummy_pystagelinq, "_discover_stagelinq_device", discover_device_mock @@ -891,12 +835,12 @@ async def test_py_stagelinq_strapper(dummy_pystagelinq, monkeypatch, dummy_ip): await dummy_pystagelinq._py_stagelinq_strapper() - discover_device_mock.assert_called_once_with(dummy_ip, timeout=2) + discover_device_mock.assert_called_once_with("", timeout=2) @pytest.mark.asyncio async def test_py_stagelinq_strapper_loop_condition_false( - dummy_pystagelinq, monkeypatch, dummy_ip + dummy_pystagelinq, monkeypatch ): get_loop_condition_mock = Mock(side_effect=[False]) monkeypatch.setattr( @@ -909,7 +853,7 @@ async def test_py_stagelinq_strapper_loop_condition_false( await dummy_pystagelinq._py_stagelinq_strapper() - discover_device_mock.assert_called_once_with(dummy_ip, timeout=2) + discover_device_mock.assert_called_once_with("", timeout=2) @pytest.mark.asyncio @@ -960,6 +904,21 @@ def test_stop(dummy_pystagelinq, monkeypatch): stop_mock.assert_called_once_with() +def test__stop_fail_to_send(dummy_pystagelinq, monkeypatch): + send_discovery_frame_mock = Mock(side_effect=Exception()) + + monkeypatch.setattr( + dummy_pystagelinq.network_interface, + "send_discovery_frame", + send_discovery_frame_mock, + ) + + with pytest.raises(Exception) as exception: + dummy_pystagelinq._stop() + + assert exception.type is Exception + + def test___del__(dummy_pystagelinq, monkeypatch): stop_mock = MagicMock() stop_all_tasks_mock = MagicMock() diff --git a/tests/unit/test_unit_PyStageLinQ_interface_info.py b/tests/unit/test_unit_PyStageLinQ_interface_info.py new file mode 100644 index 0000000..d8b1615 --- /dev/null +++ b/tests/unit/test_unit_PyStageLinQ_interface_info.py @@ -0,0 +1,491 @@ +import pytest +import PyStageLinQ.PyStageLinQ +from PyStageLinQ.ErrorCodes import * +from unittest.mock import AsyncMock, Mock, MagicMock +import ipaddress +from socket import AF_INET + +import random + + +class dummy_net_if_stats: + def __init__(self): + self.isup = True + + +class dummy_net_if_addrs: + def __init__(self, ip="0.0.0.0"): + self.family = AF_INET + self.address = ip + self.netmask = "255.255.255.0" + + +class ifutils_net_if_addrs: + def __init__(self, ip=[], netmask="255.255.255.0"): + self.address = ip + self.netmask = netmask + self.family = AF_INET + + +@pytest.fixture() +def dummy_socket(): + return MagicMock() + + +@pytest.fixture(autouse=True) +def ensure_cleanup(dummy_socket): + """Ensure that everything is cleaned up between tests.""" + yield + + # Force garbage collection to trigger __del__ if necessary + import gc + + gc.collect() + + +@pytest.fixture() +def dummy_port(): + return random.randint(1, 65535) + + +@pytest.fixture() +def dummy_ip(): + return ".".join(map(str, (random.randint(0, 255) for _ in range(4)))) + + +@pytest.fixture() +def dummy_PyStageLinQ_network_interface(dummy_ip, monkeypatch, dummy_port): + dummy_psutil = MagicMock() + dummy_psutil.net_if_addrs.return_value = { + "interface1": [dummy_net_if_addrs(ip=dummy_ip)] + } + dummy_psutil.net_if_stats.return_value = {"interface1": dummy_net_if_stats()} + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil) + + return PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface(dummy_ip, dummy_port) + + +def test_init_values(dummy_ip, monkeypatch): + + def dummy_get_interface_from_ip(_, ip=None, discovery_port=None): + assert ip == dummy_ip + + monkeypatch.setattr( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface, + "get_interface_from_ip", + dummy_get_interface_from_ip, + ) + + test_PyStageLinQ_network_interface = ( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface( + ip=dummy_ip, discovery_port=dummy_port + ) + ) + + assert test_PyStageLinQ_network_interface.discovery_port == dummy_port + + +def test_init_values_ip_none(monkeypatch, dummy_socket): + dummy_psutil = MagicMock() + dummy_ips = { + "interface1": [ifutils_net_if_addrs(ip="1.2.3.4", netmask="255.0.0.0")], + "interface2": [ifutils_net_if_addrs(ip="5.6.7.8", netmask="255.255.0.0")], + "interface3": [ifutils_net_if_addrs(ip="9.10.11.12", netmask="255.255.255.0")], + } + + dummy_ips["interface3"][0].family = None + + dummy_stats = { + "interface1": [None], + "interface2": [None], + "interface3": [None], + } + + dummy_psutil.net_if_addrs.return_value = dummy_ips + dummy_psutil.net_if_stats.return_value = dummy_stats + + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil) + + discovery_port = 12345 + + dummy_pystagelinq_network_interface = ( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface( + ip=None, discovery_port=discovery_port + ) + ) + + assert dummy_pystagelinq_network_interface.discovery_port == discovery_port + + assert ( + dummy_pystagelinq_network_interface.target_interfaces[0].addr_str + == dummy_ips["interface1"][0].address + ) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[0].addr + ) == ipaddress.IPv4Address(dummy_ips["interface1"][0].address) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[0].mask + ) == ipaddress.IPv4Address(dummy_ips["interface1"][0].netmask) + assert ( + dummy_pystagelinq_network_interface.target_interfaces[0].status + is dummy_stats["interface1"] + ) + + assert ( + dummy_pystagelinq_network_interface.target_interfaces[1].addr_str + == dummy_ips["interface2"][0].address + ) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[1].addr + ) == ipaddress.IPv4Address(dummy_ips["interface2"][0].address) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[1].mask + ) == ipaddress.IPv4Address(dummy_ips["interface2"][0].netmask) + assert ( + dummy_pystagelinq_network_interface.target_interfaces[1].status + is dummy_stats["interface2"] + ) + + assert len(dummy_pystagelinq_network_interface.target_interfaces) == 2 + + +def test_init_values_ip_single(monkeypatch, dummy_socket): + dummy_psutil = MagicMock() + dummy_ips = { + "interface1": [ifutils_net_if_addrs(ip="1.2.3.4", netmask="255.0.0.0")], + "interface2": [ifutils_net_if_addrs(ip="5.6.7.8", netmask="255.255.0.0")], + "interface3": [ifutils_net_if_addrs(ip="9.10.11.12", netmask="255.255.255.0")], + } + + dummy_ips["interface3"][0].family = None + + dummy_stats = { + "interface1": [None], + "interface2": [None], + "interface3": [None], + } + + dummy_psutil.net_if_addrs.return_value = dummy_ips + dummy_psutil.net_if_stats.return_value = dummy_stats + + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil) + + discovery_port = 12345 + + dummy_pystagelinq_network_interface = ( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface( + ip="1.2.3.4", discovery_port=discovery_port + ) + ) + + assert dummy_pystagelinq_network_interface.discovery_port == discovery_port + + assert ( + dummy_pystagelinq_network_interface.target_interfaces[0].addr_str + == dummy_ips["interface1"][0].address + ) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[0].addr + ) == ipaddress.IPv4Address(dummy_ips["interface1"][0].address) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[0].mask + ) == ipaddress.IPv4Address(dummy_ips["interface1"][0].netmask) + assert ( + dummy_pystagelinq_network_interface.target_interfaces[0].status + is dummy_stats["interface1"] + ) + + assert len(dummy_pystagelinq_network_interface.target_interfaces) == 1 + + +def test_init_values_ip_multiple(monkeypatch, dummy_socket): + dummy_psutil = MagicMock() + dummy_ips = { + "interface1": [ifutils_net_if_addrs(ip="1.2.3.4", netmask="255.0.0.0")], + "interface2": [ifutils_net_if_addrs(ip="5.6.7.8", netmask="255.255.0.0")], + "interface3": [ifutils_net_if_addrs(ip="9.10.11.12", netmask="255.255.255.0")], + "interface4": [ifutils_net_if_addrs(ip="13.14.15.16", netmask="255.255.255.0")], + } + + dummy_ips["interface3"][0].family = None + + dummy_stats = { + "interface1": [None], + "interface2": [None], + "interface3": [None], + "interface4": [None], + } + + dummy_psutil.net_if_addrs.return_value = dummy_ips + dummy_psutil.net_if_stats.return_value = dummy_stats + + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil) + + discovery_port = 12345 + + dummy_pystagelinq_network_interface = ( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface( + ip=["1.2.3.4", "5.6.7.8", "13.14.15.16"], discovery_port=discovery_port + ) + ) + + assert dummy_pystagelinq_network_interface.discovery_port == discovery_port + + assert ( + dummy_pystagelinq_network_interface.target_interfaces[0].addr_str + == dummy_ips["interface1"][0].address + ) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[0].addr + ) == ipaddress.IPv4Address(dummy_ips["interface1"][0].address) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[0].mask + ) == ipaddress.IPv4Address(dummy_ips["interface1"][0].netmask) + assert ( + dummy_pystagelinq_network_interface.target_interfaces[0].status + is dummy_stats["interface1"] + ) + + assert ( + dummy_pystagelinq_network_interface.target_interfaces[1].addr_str + == dummy_ips["interface2"][0].address + ) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[1].addr + ) == ipaddress.IPv4Address(dummy_ips["interface2"][0].address) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[1].mask + ) == ipaddress.IPv4Address(dummy_ips["interface2"][0].netmask) + assert ( + dummy_pystagelinq_network_interface.target_interfaces[1].status + is dummy_stats["interface2"] + ) + + assert ( + dummy_pystagelinq_network_interface.target_interfaces[2].addr_str + == dummy_ips["interface4"][0].address + ) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[2].addr + ) == ipaddress.IPv4Address(dummy_ips["interface4"][0].address) + assert ipaddress.IPv4Address( + dummy_pystagelinq_network_interface.target_interfaces[2].mask + ) == ipaddress.IPv4Address(dummy_ips["interface4"][0].netmask) + assert ( + dummy_pystagelinq_network_interface.target_interfaces[2].status + is dummy_stats["interface4"] + ) + + assert len(dummy_pystagelinq_network_interface.target_interfaces) == 3 + + +def test_init_values_ip_invalid(monkeypatch, dummy_socket): + dummy_psutil = MagicMock() + dummy_ips = { + "interface1": [ifutils_net_if_addrs(ip="1.2.3.4", netmask="255.0.0.0")], + "interface2": [ifutils_net_if_addrs(ip="5.6.7.8", netmask="255.255.0.0")], + "interface3": [ifutils_net_if_addrs(ip="9.10.11.12", netmask="255.255.255.0")], + "interface4": [ifutils_net_if_addrs(ip="13.14.15.16", netmask="255.255.255.0")], + } + + dummy_ips["interface3"][0].family = None + + dummy_stats = { + "interface1": [None], + "interface2": [None], + "interface3": [None], + "interface4": [None], + } + + dummy_psutil.net_if_addrs.return_value = dummy_ips + dummy_psutil.net_if_stats.return_value = dummy_stats + + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil) + + discovery_port = 12345 + + with pytest.raises(Exception) as exception: + dummy_pystagelinq_network_interface = ( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface( + ip=["1.2.3.4", "5.6.7.8", 5], discovery_port=discovery_port + ) + ) + assert exception.type is TypeError + + with pytest.raises(Exception) as exception: + dummy_pystagelinq_network_interface = ( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface( + ip=5, discovery_port=discovery_port + ) + ) + assert exception.type is TypeError + + +def test_send_discovery_frame( + dummy_PyStageLinQ_network_interface, monkeypatch, dummy_socket +): + dummy_discovery_frame = "AAAA" + + discovery_socket = MagicMock() + + dummy_socket.socket.side_effect = discovery_socket + + dummy_socket.getaddrinfo.side_effect = [[["255.255.255.255"]]] + + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "socket", dummy_socket) + + dummy_PyStageLinQ_network_interface.send_discovery_frame(dummy_discovery_frame) + + dummy_socket.socket.assert_called_once_with( + dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM + ) + discovery_socket.return_value.__enter__.return_value.setsockopt.assert_called_once_with( + dummy_socket.SOL_SOCKET, dummy_socket.SO_BROADCAST, 1 + ) + discovery_socket.return_value.__enter__.return_value.sendto.assert_called_once_with( + dummy_discovery_frame, + ("255.255.255.255", dummy_PyStageLinQ_network_interface.discovery_port), + ) + + +def test_send_discovery_frame_permission_error( + dummy_PyStageLinQ_network_interface, monkeypatch, dummy_ip, dummy_socket +): + dummy_discovery_frame = "AAAA" + + discovery_socket = MagicMock() + + dummy_socket.getaddrinfo.side_effect = [[["255.255.255.255"]]] + + dummy_socket.socket.side_effect = discovery_socket + discovery_socket.return_value.__enter__.return_value.sendto.side_effect = ( + PermissionError + ) + + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "socket", dummy_socket) + + with pytest.raises(PermissionError) as exception: + dummy_PyStageLinQ_network_interface.send_discovery_frame(dummy_discovery_frame) + + dummy_socket.socket.assert_called_once_with( + dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM + ) + discovery_socket.return_value.__enter__.return_value.setsockopt.assert_called_once_with( + dummy_socket.SOL_SOCKET, dummy_socket.SO_BROADCAST, 1 + ) + + +def test_determine_interface_of_remote_ip(monkeypatch, dummy_socket): + dummy_psutil = MagicMock() + dummy_ips = { + "interface1": [ifutils_net_if_addrs(ip="1.2.3.4", netmask="255.0.0.0")], + "interface2": [ifutils_net_if_addrs(ip="5.6.7.8", netmask="255.255.0.0")], + "interface3": [ifutils_net_if_addrs(ip="9.10.11.12", netmask="255.255.255.0")], + "interface4": [ifutils_net_if_addrs(ip="13.14.15.16", netmask="255.255.255.0")], + } + + dummy_ips["interface3"][0].family = None + + dummy_stats = { + "interface1": [None], + "interface2": [None], + "interface3": [None], + "interface4": [None], + } + + dummy_psutil.net_if_addrs.return_value = dummy_ips + dummy_psutil.net_if_stats.return_value = dummy_stats + + monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil) + + discovery_port = 12345 + + dummy_pystagelinq_network_interface = ( + PyStageLinQ.PyStageLinQ.PyStageLinQ_network_interface( + ip=["1.2.3.4", "5.6.7.8", "13.14.15.16"], discovery_port=discovery_port + ) + ) + + # Try valid IPs for each interface + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip("1.1.1.1") + == dummy_pystagelinq_network_interface.target_interfaces[0] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip("1.2.1.1") + == dummy_pystagelinq_network_interface.target_interfaces[0] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip("1.1.2.1") + == dummy_pystagelinq_network_interface.target_interfaces[0] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip("1.2.3.3") + == dummy_pystagelinq_network_interface.target_interfaces[0] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip( + "1.255.255.255" + ) + == dummy_pystagelinq_network_interface.target_interfaces[0] + ) + + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip("5.6.5.5") + == dummy_pystagelinq_network_interface.target_interfaces[1] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip("5.6.6.5") + == dummy_pystagelinq_network_interface.target_interfaces[1] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip( + "5.6.255.255" + ) + == dummy_pystagelinq_network_interface.target_interfaces[1] + ) + + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip( + "13.14.15.5" + ) + == dummy_pystagelinq_network_interface.target_interfaces[2] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip( + "13.14.15.1" + ) + == dummy_pystagelinq_network_interface.target_interfaces[2] + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip( + "13.14.15.255" + ) + == dummy_pystagelinq_network_interface.target_interfaces[2] + ) + + # Try some IPs that has no interface + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip("5.1.1.1") + is None + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip( + "253.196.1.1" + ) + is None + ) + assert ( + dummy_pystagelinq_network_interface.determine_interface_of_remote_ip( + "2.255.255.255" + ) + is None + ) + + +def test_send_desc_on_all_if(dummy_PyStageLinQ_network_interface): + assert dummy_PyStageLinQ_network_interface.send_desc_on_all_if() is False + + dummy_PyStageLinQ_network_interface.target_interfaces[0].n_disc_msg_send = 10 + + assert dummy_PyStageLinQ_network_interface.send_desc_on_all_if() is True diff --git a/tests/unit/test_unit_Token.py b/tests/unit/test_unit_Token.py index 3933231..2f6e4fc 100644 --- a/tests/unit/test_unit_Token.py +++ b/tests/unit/test_unit_Token.py @@ -33,14 +33,12 @@ def mock_random_0(length): def test_generate_token_msb1(token, monkeypatch): def mock_random_msb1(length): - return bytearray( - int("80000000000000000000000000000001", length).to_bytes(length, "big") - ) + return bytearray(0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF.to_bytes(length, "big")) monkeypatch.setattr(token, "_get_randomized_bytes", mock_random_msb1) token.generate_token() - assert token.get_token() == 1 + assert token.get_token() == 0x7FFFFFFFFFFFFFFF3FFFFFFFFFFFFFFF def test_set_token_wrong_input_type(token):