From 492041d6238f81e37ccb66b67cf3cead25baccef Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 00:45:31 +0200 Subject: [PATCH 01/21] Fix 4.8.1 wireguard integration --- gli4py/glinet.py | 28 +++++++++++++++++++++------- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 0477edc..739f4d7 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -310,21 +310,35 @@ async def wireguard_client_list(self) -> dict: ) return configs - async def wireguard_client_state(self) -> dict: + async def wireguard_client_state(self) -> list: """ - {"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"status":0,"proxy":True,"log":"","ipv4":""} - status 0:not start 1:connected 2:connecting + {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} """ - return await self._request( - self.gen_sid_payload("call", ["wg-client", "get_status"], self.sid) + response = await self._request( + self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) ) + return response.get("status_list", []) - async def wireguard_client_start(self, group_id: int, peer_id: int) -> dict: + async def wireguard_client_start(self, tunnel_id: int) -> dict: """Starts a WireGuard client with the specified group ID and peer ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, True) + + async def wireguard_client_stop(self, tunnel_id: int) -> dict: + """Stops the WireGuard client.""" + return await self._wireguard_set_client_enabled(tunnel_id, False) + + async def _wireguard_set_client_enabled( + self, tunnel_id: int, enabled: bool + ) -> dict: + """Sets the WireGuard client enabled state.""" return await self._request( self.gen_sid_payload( "call", - ["wg-client", "start", {"group_id": group_id, "peer_id": peer_id}], + [ + "vpn-client", + "set_tunnel", + {"enabled": enabled, "tunnel_id": tunnel_id}, + ], self.sid, ) ) From 7e358f4e4608bd8f926ef3350b69e9431e05739e Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 00:47:11 +0200 Subject: [PATCH 02/21] Fix outdated comments --- gli4py/glinet.py | 75 +++++++++++++++++++++++++----------------------- 1 file changed, 39 insertions(+), 36 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 739f4d7..4a6f080 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -9,7 +9,8 @@ from gli4py.enums import TailscaleConnection -from .error_handling import APIClientError, AuthenticationError, raise_for_status # , timeout_error +# , timeout_error +from .error_handling import APIClientError, AuthenticationError, raise_for_status # typical base url http://192.168.8.1/rpc @@ -107,7 +108,9 @@ async def login(self, username: str, password: str) -> None: password ) else: - raise ValueError("Router requested unsupported hashing algorithm for cipher password") + raise ValueError( + "Router requested unsupported hashing algorithm for cipher password" + ) # Step3: Generate hash values for login data = f"{username}:{cipher_password}:{nonce}" @@ -118,7 +121,9 @@ async def login(self, username: str, password: str) -> None: elif hash_method == "sha256": # SHA-512 hsh = hashlib.sha512(data.encode()).hexdigest() else: - raise ValueError("Router requested unsupported hashing algorithm for hash") + raise ValueError( + "Router requested unsupported hashing algorithm for hash" + ) # Step4: Get sid by login res = await self._get_sid(username, hsh) @@ -133,7 +138,9 @@ async def login(self, username: str, password: str) -> None: except AuthenticationError as e: raise AuthenticationError("Authentication failed during login") from e except APIClientError as e: - raise APIClientError(f"An unexpected error of type {type(e).__name__} has occurred during login") from e + raise APIClientError( + f"An unexpected error of type {type(e).__name__} has occurred during login" + ) from e async def router_info(self) -> dict: """Retrieves information about the router, requires authentication.""" @@ -303,52 +310,46 @@ async def wireguard_client_list(self) -> dict: for peer in item["peers"]: configs.append( { - "name": f'{item["group_name"]}/{peer["name"]}', + "name": f"{item['group_name']}/{peer['name']}", "group_id": item["group_id"], "peer_id": peer["peer_id"], } ) return configs - async def wireguard_client_state(self) -> list: + async def wireguard_client_state(self) -> list: """ - {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} + {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} """ - response = await self._request( - self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) + response = await self._request( + self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) ) - return response.get("status_list", []) - - async def wireguard_client_start(self, tunnel_id: int) -> dict: - """Starts a WireGuard client with the specified group ID and peer ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, True) - - async def wireguard_client_stop(self, tunnel_id: int) -> dict: - """Stops the WireGuard client.""" - return await self._wireguard_set_client_enabled(tunnel_id, False) - - async def _wireguard_set_client_enabled( - self, tunnel_id: int, enabled: bool - ) -> dict: - """Sets the WireGuard client enabled state.""" + return response.get("status_list", []) + + async def wireguard_client_start(self, tunnel_id: int) -> dict: + """Starts a WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, True) + + async def wireguard_client_stop(self, tunnel_id: int) -> dict: + """Stops the WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, False) + + async def _wireguard_set_client_enabled( + self, tunnel_id: int, enabled: bool + ) -> dict: + """Sets the WireGuard client enabled state.""" return await self._request( self.gen_sid_payload( "call", - [ - "vpn-client", - "set_tunnel", - {"enabled": enabled, "tunnel_id": tunnel_id}, - ], + [ + "vpn-client", + "set_tunnel", + {"enabled": enabled, "tunnel_id": tunnel_id}, + ], self.sid, ) ) - async def wireguard_client_stop(self) -> dict: - """Stops the WireGuard client.""" - return await self._request( - self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) - ) - async def _tailscale_get_config(self) -> dict | bool: """ {'wan_enabled': False, 'lan_ip': '192.168.0.0/24', 'enabled': False, 'lan_enabled': True} @@ -424,11 +425,13 @@ async def tailscale_start(self, depth: int = 0) -> True: status = (await self._tailscale_status())["status"] if status != 3: raise ConnectionError( - f"Did not try to start tailscale as device reported 'Connecting' and then 3 seconds later {TailscaleConnection[status].name}") + f"Did not try to start tailscale as device reported 'Connecting' and then 3 seconds later {TailscaleConnection[status].name}" + ) return True if status in [1, 2]: raise ConnectionAbortedError( - f"Connection not attempted as authorisation is not complete, due to {TailscaleConnection[status].name}") + f"Connection not attempted as authorisation is not complete, due to {TailscaleConnection[status].name}" + ) raise ConnectionError(f"Unknown connection status: {status}") From 431a52fb1860224bc5122e557ce92f410aae69e6 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 00:50:05 +0200 Subject: [PATCH 03/21] Revert "Fix outdated comments" This reverts commit 7e358f4e4608bd8f926ef3350b69e9431e05739e. --- gli4py/glinet.py | 75 +++++++++++++++++++++++------------------------- 1 file changed, 36 insertions(+), 39 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 4a6f080..739f4d7 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -9,8 +9,7 @@ from gli4py.enums import TailscaleConnection -# , timeout_error -from .error_handling import APIClientError, AuthenticationError, raise_for_status +from .error_handling import APIClientError, AuthenticationError, raise_for_status # , timeout_error # typical base url http://192.168.8.1/rpc @@ -108,9 +107,7 @@ async def login(self, username: str, password: str) -> None: password ) else: - raise ValueError( - "Router requested unsupported hashing algorithm for cipher password" - ) + raise ValueError("Router requested unsupported hashing algorithm for cipher password") # Step3: Generate hash values for login data = f"{username}:{cipher_password}:{nonce}" @@ -121,9 +118,7 @@ async def login(self, username: str, password: str) -> None: elif hash_method == "sha256": # SHA-512 hsh = hashlib.sha512(data.encode()).hexdigest() else: - raise ValueError( - "Router requested unsupported hashing algorithm for hash" - ) + raise ValueError("Router requested unsupported hashing algorithm for hash") # Step4: Get sid by login res = await self._get_sid(username, hsh) @@ -138,9 +133,7 @@ async def login(self, username: str, password: str) -> None: except AuthenticationError as e: raise AuthenticationError("Authentication failed during login") from e except APIClientError as e: - raise APIClientError( - f"An unexpected error of type {type(e).__name__} has occurred during login" - ) from e + raise APIClientError(f"An unexpected error of type {type(e).__name__} has occurred during login") from e async def router_info(self) -> dict: """Retrieves information about the router, requires authentication.""" @@ -310,46 +303,52 @@ async def wireguard_client_list(self) -> dict: for peer in item["peers"]: configs.append( { - "name": f"{item['group_name']}/{peer['name']}", + "name": f'{item["group_name"]}/{peer["name"]}', "group_id": item["group_id"], "peer_id": peer["peer_id"], } ) return configs - async def wireguard_client_state(self) -> list: + async def wireguard_client_state(self) -> list: """ - {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} + {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} """ - response = await self._request( - self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) + response = await self._request( + self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) ) - return response.get("status_list", []) - - async def wireguard_client_start(self, tunnel_id: int) -> dict: - """Starts a WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, True) - - async def wireguard_client_stop(self, tunnel_id: int) -> dict: - """Stops the WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, False) - - async def _wireguard_set_client_enabled( - self, tunnel_id: int, enabled: bool - ) -> dict: - """Sets the WireGuard client enabled state.""" + return response.get("status_list", []) + + async def wireguard_client_start(self, tunnel_id: int) -> dict: + """Starts a WireGuard client with the specified group ID and peer ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, True) + + async def wireguard_client_stop(self, tunnel_id: int) -> dict: + """Stops the WireGuard client.""" + return await self._wireguard_set_client_enabled(tunnel_id, False) + + async def _wireguard_set_client_enabled( + self, tunnel_id: int, enabled: bool + ) -> dict: + """Sets the WireGuard client enabled state.""" return await self._request( self.gen_sid_payload( "call", - [ - "vpn-client", - "set_tunnel", - {"enabled": enabled, "tunnel_id": tunnel_id}, - ], + [ + "vpn-client", + "set_tunnel", + {"enabled": enabled, "tunnel_id": tunnel_id}, + ], self.sid, ) ) + async def wireguard_client_stop(self) -> dict: + """Stops the WireGuard client.""" + return await self._request( + self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) + ) + async def _tailscale_get_config(self) -> dict | bool: """ {'wan_enabled': False, 'lan_ip': '192.168.0.0/24', 'enabled': False, 'lan_enabled': True} @@ -425,13 +424,11 @@ async def tailscale_start(self, depth: int = 0) -> True: status = (await self._tailscale_status())["status"] if status != 3: raise ConnectionError( - f"Did not try to start tailscale as device reported 'Connecting' and then 3 seconds later {TailscaleConnection[status].name}" - ) + f"Did not try to start tailscale as device reported 'Connecting' and then 3 seconds later {TailscaleConnection[status].name}") return True if status in [1, 2]: raise ConnectionAbortedError( - f"Connection not attempted as authorisation is not complete, due to {TailscaleConnection[status].name}" - ) + f"Connection not attempted as authorisation is not complete, due to {TailscaleConnection[status].name}") raise ConnectionError(f"Unknown connection status: {status}") From 1adc9f0472db0b3979c86c8c5903b6fa7215f50c Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 00:50:54 +0200 Subject: [PATCH 04/21] Again fix outdated comments without changing everything else --- gli4py/glinet.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 739f4d7..3ba30e5 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -318,13 +318,13 @@ async def wireguard_client_state(self) -> list: self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) ) return response.get("status_list", []) - + async def wireguard_client_start(self, tunnel_id: int) -> dict: - """Starts a WireGuard client with the specified group ID and peer ID.""" + """Starts a WireGuard client with the specified tunnel ID.""" return await self._wireguard_set_client_enabled(tunnel_id, True) async def wireguard_client_stop(self, tunnel_id: int) -> dict: - """Stops the WireGuard client.""" + """Stops the WireGuard client with the specified tunnel ID.""" return await self._wireguard_set_client_enabled(tunnel_id, False) async def _wireguard_set_client_enabled( From d7eca024e9c6310d9fbbbfe1711a8f8366bd2a2d Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 14:07:01 +0200 Subject: [PATCH 05/21] Fix duplicate method that just appeared? --- gli4py/glinet.py | 52 +++++++++++++++++++++--------------------------- 1 file changed, 23 insertions(+), 29 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 3ba30e5..636be13 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -310,45 +310,39 @@ async def wireguard_client_list(self) -> dict: ) return configs - async def wireguard_client_state(self) -> list: + async def wireguard_client_state(self) -> list: """ - {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} + {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} """ - response = await self._request( - self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) + response = await self._request( + self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) ) - return response.get("status_list", []) - - async def wireguard_client_start(self, tunnel_id: int) -> dict: - """Starts a WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, True) - - async def wireguard_client_stop(self, tunnel_id: int) -> dict: - """Stops the WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, False) - - async def _wireguard_set_client_enabled( - self, tunnel_id: int, enabled: bool - ) -> dict: - """Sets the WireGuard client enabled state.""" + return response.get("status_list", []) + + async def wireguard_client_start(self, tunnel_id: int) -> dict: + """Starts a WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, True) + + async def wireguard_client_stop(self, tunnel_id: int) -> dict: + """Stops the WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, False) + + async def _wireguard_set_client_enabled( + self, tunnel_id: int, enabled: bool + ) -> dict: + """Sets the WireGuard client enabled state.""" return await self._request( self.gen_sid_payload( "call", - [ - "vpn-client", - "set_tunnel", - {"enabled": enabled, "tunnel_id": tunnel_id}, - ], + [ + "vpn-client", + "set_tunnel", + {"enabled": enabled, "tunnel_id": tunnel_id}, + ], self.sid, ) ) - async def wireguard_client_stop(self) -> dict: - """Stops the WireGuard client.""" - return await self._request( - self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) - ) - async def _tailscale_get_config(self) -> dict | bool: """ {'wan_enabled': False, 'lan_ip': '192.168.0.0/24', 'enabled': False, 'lan_enabled': True} From 1bb98975b8960046ecdf6200e29f64b0340c33bc Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 14:08:01 +0200 Subject: [PATCH 06/21] Revert "Fix duplicate method that just appeared?" This reverts commit d7eca024e9c6310d9fbbbfe1711a8f8366bd2a2d. --- gli4py/glinet.py | 52 +++++++++++++++++++++++++++--------------------- 1 file changed, 29 insertions(+), 23 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 636be13..3ba30e5 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -310,39 +310,45 @@ async def wireguard_client_list(self) -> dict: ) return configs - async def wireguard_client_state(self) -> list: + async def wireguard_client_state(self) -> list: """ - {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} + {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} """ - response = await self._request( - self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) + response = await self._request( + self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) ) - return response.get("status_list", []) - - async def wireguard_client_start(self, tunnel_id: int) -> dict: - """Starts a WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, True) - - async def wireguard_client_stop(self, tunnel_id: int) -> dict: - """Stops the WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, False) - - async def _wireguard_set_client_enabled( - self, tunnel_id: int, enabled: bool - ) -> dict: - """Sets the WireGuard client enabled state.""" + return response.get("status_list", []) + + async def wireguard_client_start(self, tunnel_id: int) -> dict: + """Starts a WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, True) + + async def wireguard_client_stop(self, tunnel_id: int) -> dict: + """Stops the WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled(tunnel_id, False) + + async def _wireguard_set_client_enabled( + self, tunnel_id: int, enabled: bool + ) -> dict: + """Sets the WireGuard client enabled state.""" return await self._request( self.gen_sid_payload( "call", - [ - "vpn-client", - "set_tunnel", - {"enabled": enabled, "tunnel_id": tunnel_id}, - ], + [ + "vpn-client", + "set_tunnel", + {"enabled": enabled, "tunnel_id": tunnel_id}, + ], self.sid, ) ) + async def wireguard_client_stop(self) -> dict: + """Stops the WireGuard client.""" + return await self._request( + self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) + ) + async def _tailscale_get_config(self) -> dict | bool: """ {'wan_enabled': False, 'lan_ip': '192.168.0.0/24', 'enabled': False, 'lan_enabled': True} From fd7dfd18f2c11947843b7c165936b597a06c1245 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 14:10:04 +0200 Subject: [PATCH 07/21] oh my god I hate python and vscode --- gli4py/glinet.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 3ba30e5..d15a6c1 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -343,12 +343,6 @@ async def _wireguard_set_client_enabled( ) ) - async def wireguard_client_stop(self) -> dict: - """Stops the WireGuard client.""" - return await self._request( - self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) - ) - async def _tailscale_get_config(self) -> dict | bool: """ {'wan_enabled': False, 'lan_ip': '192.168.0.0/24', 'enabled': False, 'lan_enabled': True} From f220edb49deea9fca3174ee99d1cffc96d79e4b3 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 15:26:09 +0200 Subject: [PATCH 08/21] Add versioning to wireguard clients --- gli4py/__init__.py | 1 + gli4py/glinet.py | 66 ++++++++++---- gli4py/version.py | 108 +++++++++++++++++++++++ tests/test_version.py | 198 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 356 insertions(+), 17 deletions(-) create mode 100644 gli4py/version.py create mode 100644 tests/test_version.py diff --git a/gli4py/__init__.py b/gli4py/__init__.py index 6b8cd1d..74ba59f 100644 --- a/gli4py/__init__.py +++ b/gli4py/__init__.py @@ -1,5 +1,6 @@ """gli4py - A Python library for GL.iNet routers""" from .glinet import GLinet +from .version import Version if __name__ == "__main__": pass diff --git a/gli4py/glinet.py b/gli4py/glinet.py index d15a6c1..0f7d82a 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -8,11 +8,13 @@ from passlib.hash import md5_crypt, sha256_crypt, sha512_crypt from gli4py.enums import TailscaleConnection +from gli4py.version import Version from .error_handling import APIClientError, AuthenticationError, raise_for_status # , timeout_error # typical base url http://192.168.8.1/rpc +NEW_VPN_CLIENT_VERSION = Version(4, 8, 0, 0) @response_handler(raise_for_status) @@ -310,38 +312,68 @@ async def wireguard_client_list(self) -> dict: ) return configs - async def wireguard_client_state(self) -> list: + async def wireguard_client_state(self, version_string: str) -> list: """ {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} """ + parsed_version = Version.parse(version_string) + # If version is 4.8 or greater use vpn-client otherwise use wg-client + target_call = "vpn-client" if parsed_version >= NEW_VPN_CLIENT_VERSION else "wg-client" + response = await self._request( - self.gen_sid_payload("call", ["vpn-client", "get_status"], self.sid) + self.gen_sid_payload("call", [target_call, "get_status"], self.sid) ) + + if parsed_version < NEW_VPN_CLIENT_VERSION: + # If the version is less than 4.8 we need to adjust the response to match the new format + # The old format does not return an array, but just a single object. + # We will wrap it in an array to match the new format. + response = {"status_list": [response]} + return response.get("status_list", []) - async def wireguard_client_start(self, tunnel_id: int) -> dict: - """Starts a WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, True) + async def wireguard_client_start(self, group_id: int, peer_id: int, tunnel_id: int, version_string: str) -> dict: + """Starts a WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled(group_id, peer_id, tunnel_id, True, version_string) - async def wireguard_client_stop(self, tunnel_id: int) -> dict: + async def wireguard_client_stop(self, tunnel_id: int, version_string: str) -> dict: """Stops the WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(tunnel_id, False) + # Pass -1 for group_id and peer_id as they are not needed to stop the client + return await self._wireguard_set_client_enabled(-1, -1, tunnel_id, False, version_string) async def _wireguard_set_client_enabled( - self, tunnel_id: int, enabled: bool + self, group_id: int, peer_id: int, tunnel_id: int, enabled: bool, version_string: str ) -> dict: """Sets the WireGuard client enabled state.""" - return await self._request( + parsed_version = Version.parse(version_string) + # If version is 4.8 or greater use vpn-client otherwise use wg-client + if parsed_version >= NEW_VPN_CLIENT_VERSION: + return await self._request( self.gen_sid_payload( - "call", - [ - "vpn-client", - "set_tunnel", - {"enabled": enabled, "tunnel_id": tunnel_id}, - ], - self.sid, + "call", + [ + "vpn-client", + "set_tunnel", + {"enabled": enabled, "tunnel_id": tunnel_id}, + ], + self.sid, + ) ) - ) + else: + if enabled: + return await self._request( + self.gen_sid_payload( + "call", + ["wg-client", "start", {"group_id": group_id, "peer_id": peer_id}], + self.sid, + ) + ) + else: + return await self._request( + self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) + ) + + async def _tailscale_get_config(self) -> dict | bool: """ diff --git a/gli4py/version.py b/gli4py/version.py new file mode 100644 index 0000000..3fcfb46 --- /dev/null +++ b/gli4py/version.py @@ -0,0 +1,108 @@ +"""Version utility class for handling semantic versioning.""" + +import re + + +class Version: + """A class to represent and parse semantic version numbers. + + Attributes: + major (int): The major version number + minor (int): The minor version number + patch (int): The patch version number + build (int): The build version number + """ + + def __init__(self, major: int = 0, minor: int = 0, patch: int = 0, build: int = 0): + """Initialize a Version instance. + + Args: + major (int): The major version number (default: 0) + minor (int): The minor version number (default: 0) + patch (int): The patch version number (default: 0) + build (int): The build version number (default: 0) + """ + self.major = int(major) + self.minor = int(minor) + self.patch = int(patch) + self.build = int(build) + + @classmethod + def parse(cls, version_string: str) -> 'Version': + """Parse a version string into a Version object. + + Args: + version_string (str): A version string in the format "major.minor.patch" or "major.minor.patch.build" + (e.g., "1.2.3", "12.34.56.78") + + Returns: + Version: A Version instance with the parsed values + + Raises: + ValueError: If the version string format is invalid + """ + if not isinstance(version_string, str): + raise ValueError("Version string must be a string") + + # Remove any leading 'v' if present (e.g., "v1.2.3" -> "1.2.3") + version_string = version_string.lstrip('v') + + # Regular expression to match 3-part or 4-part version pattern + pattern_3_part = r'^(\d+)\.(\d+)\.(\d+)$' + pattern_4_part = r'^(\d+)\.(\d+)\.(\d+)\.(\d+)$' + + match_3 = re.match(pattern_3_part, version_string) + match_4 = re.match(pattern_4_part, version_string) + + if match_4: + # 4-part version: major.minor.patch.build + major, minor, patch, build = match_4.groups() + return cls(int(major), int(minor), int(patch), int(build)) + elif match_3: + # 3-part version: major.minor.patch (build defaults to 0) + major, minor, patch = match_3.groups() + return cls(int(major), int(minor), int(patch), 0) + else: + raise ValueError( + f"Invalid version string format: '{version_string}'. Expected format: 'major.minor.patch' or 'major.minor.patch.build'") + + def __str__(self) -> str: + """Return the string representation of the version.""" + if self.build == 0: + return f"{self.major}.{self.minor}.{self.patch}" + else: + return f"{self.major}.{self.minor}.{self.patch}.{self.build}" + + def __repr__(self) -> str: + """Return the detailed string representation of the version.""" + return f"Version(major={self.major}, minor={self.minor}, patch={self.patch}, build={self.build})" + + def __eq__(self, other) -> bool: + """Check if two versions are equal.""" + if not isinstance(other, Version): + return False + return (self.major, self.minor, self.patch, self.build) == (other.major, other.minor, other.patch, other.build) + + def __lt__(self, other) -> bool: + """Check if this version is less than another version.""" + if not isinstance(other, Version): + return NotImplemented + return (self.major, self.minor, self.patch, self.build) < (other.major, other.minor, other.patch, other.build) + + def __le__(self, other) -> bool: + """Check if this version is less than or equal to another version.""" + return self == other or self < other + + def __gt__(self, other) -> bool: + """Check if this version is greater than another version.""" + if not isinstance(other, Version): + return NotImplemented + return (self.major, self.minor, self.patch, self.build) > (other.major, other.minor, other.patch, other.build) + + def __ge__(self, other) -> bool: + """Check if this version is greater than or equal to another version.""" + return self == other or self > other + + def to_tuple(self) -> tuple[int, int, int, int]: + """Return the version as a tuple (major, minor, patch, build).""" + return (self.major, self.minor, self.patch, self.build) diff --git a/tests/test_version.py b/tests/test_version.py new file mode 100644 index 0000000..8726e4c --- /dev/null +++ b/tests/test_version.py @@ -0,0 +1,198 @@ +"""Tests for the Version class.""" + +import pytest +from gli4py.version import Version + + +class TestVersion: + """Test cases for the Version class.""" + + def test_init_default(self): + """Test Version initialization with default values.""" + version = Version() + assert version.major == 0 + assert version.minor == 0 + assert version.patch == 0 + assert version.build == 0 + + def test_init_with_values(self): + """Test Version initialization with specific values.""" + version = Version(1, 2, 3, 4) + assert version.major == 1 + assert version.minor == 2 + assert version.patch == 3 + assert version.build == 4 + + def test_init_with_three_values(self): + """Test Version initialization with three values (build defaults to 0).""" + version = Version(1, 2, 3) + assert version.major == 1 + assert version.minor == 2 + assert version.patch == 3 + assert version.build == 0 + + def test_parse_valid_version_3_part(self): + """Test parsing valid 3-part version strings.""" + version = Version.parse("1.2.3") + assert version.major == 1 + assert version.minor == 2 + assert version.patch == 3 + assert version.build == 0 + + def test_parse_valid_version_4_part(self): + """Test parsing valid 4-part version strings.""" + version = Version.parse("12.34.56.78") + assert version.major == 12 + assert version.minor == 34 + assert version.patch == 56 + assert version.build == 78 + + def test_parse_with_v_prefix_3_part(self): + """Test parsing 3-part version strings with 'v' prefix.""" + version = Version.parse("v2.5.1") + assert version.major == 2 + assert version.minor == 5 + assert version.patch == 1 + assert version.build == 0 + + def test_parse_with_v_prefix_4_part(self): + """Test parsing 4-part version strings with 'v' prefix.""" + version = Version.parse("v2.5.1.9") + assert version.major == 2 + assert version.minor == 5 + assert version.patch == 1 + assert version.build == 9 + + def test_parse_zero_values_3_part(self): + """Test parsing 3-part version strings with zero values.""" + version = Version.parse("0.0.0") + assert version.major == 0 + assert version.minor == 0 + assert version.patch == 0 + assert version.build == 0 + + def test_parse_zero_values_4_part(self): + """Test parsing 4-part version strings with zero values.""" + version = Version.parse("0.0.0.0") + assert version.major == 0 + assert version.minor == 0 + assert version.patch == 0 + assert version.build == 0 + + def test_parse_large_numbers_3_part(self): + """Test parsing 3-part version strings with large numbers.""" + version = Version.parse("123.456.789") + assert version.major == 123 + assert version.minor == 456 + assert version.patch == 789 + assert version.build == 0 + + def test_parse_large_numbers_4_part(self): + """Test parsing 4-part version strings with large numbers.""" + version = Version.parse("123.456.789.101112") + assert version.major == 123 + assert version.minor == 456 + assert version.patch == 789 + assert version.build == 101112 + + def test_parse_invalid_format(self): + """Test parsing invalid version string formats.""" + with pytest.raises(ValueError, match="Invalid version string format"): + Version.parse("1.2") + + with pytest.raises(ValueError, match="Invalid version string format"): + Version.parse("1.2.3.4.5") + + with pytest.raises(ValueError, match="Invalid version string format"): + Version.parse("1.2.a") + + with pytest.raises(ValueError, match="Invalid version string format"): + Version.parse("a.b.c") + + with pytest.raises(ValueError, match="Invalid version string format"): + Version.parse("1-2-3") + + with pytest.raises(ValueError, match="Invalid version string format"): + Version.parse("") + + def test_parse_non_string(self): + """Test parsing non-string inputs.""" + with pytest.raises(ValueError, match="Version string must be a string"): + Version.parse(123) + + with pytest.raises(ValueError, match="Version string must be a string"): + Version.parse(None) + + def test_str_representation_3_part(self): + """Test string representation of Version with build=0.""" + version = Version(1, 2, 3, 0) + assert str(version) == "1.2.3" + + def test_str_representation_4_part(self): + """Test string representation of Version with build>0.""" + version = Version(1, 2, 3, 4) + assert str(version) == "1.2.3.4" + + def test_repr_representation(self): + """Test repr representation of Version.""" + version = Version(1, 2, 3, 4) + assert repr(version) == "Version(major=1, minor=2, patch=3, build=4)" + + def test_equality(self): + """Test version equality comparison.""" + version1 = Version(1, 2, 3, 0) + version2 = Version(1, 2, 3, 0) + version3 = Version(1, 2, 4, 0) + version4 = Version(1, 2, 3, 1) + + assert version1 == version2 + assert version1 != version3 + assert version1 != version4 + assert version1 != "1.2.3" # Different type + + def test_comparison_operators(self): + """Test version comparison operators.""" + v1_0_0_0 = Version(1, 0, 0, 0) + v1_2_3_0 = Version(1, 2, 3, 0) + v1_2_3_1 = Version(1, 2, 3, 1) + v1_2_4_0 = Version(1, 2, 4, 0) + v2_0_0_0 = Version(2, 0, 0, 0) + + # Less than + assert v1_0_0_0 < v1_2_3_0 + assert v1_2_3_0 < v1_2_3_1 + assert v1_2_3_1 < v1_2_4_0 + assert v1_2_4_0 < v2_0_0_0 + + # Less than or equal + assert v1_0_0_0 <= v1_2_3_0 + assert v1_2_3_0 <= Version(1, 2, 3, 0) # Equal case + + # Greater than + assert v2_0_0_0 > v1_2_4_0 + assert v1_2_4_0 > v1_2_3_1 + assert v1_2_3_1 > v1_2_3_0 + assert v1_2_3_0 > v1_0_0_0 + + # Greater than or equal + assert v2_0_0_0 >= v1_2_4_0 + assert v1_2_3_0 >= Version(1, 2, 3, 0) # Equal case + + def test_to_tuple(self): + """Test conversion to tuple.""" + version = Version(1, 2, 3, 4) + assert version.to_tuple() == (1, 2, 3, 4) + + def test_parse_and_str_roundtrip_3_part(self): + """Test that parsing a 3-part version and converting back to string works.""" + original = "1.2.3" + version = Version.parse(original) + result = str(version) + assert result == original + + def test_parse_and_str_roundtrip_4_part(self): + """Test that parsing a 4-part version and converting back to string works.""" + original = "1.2.3.4" + version = Version.parse(original) + result = str(version) + assert result == original From f5226d401f52c43606b6c4fbbf3a4017ad1abbc3 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 15:28:33 +0200 Subject: [PATCH 09/21] Add wireguard tests --- tests/test_glinet.py | 602 ++++++++++++++++++++++++------------------- 1 file changed, 344 insertions(+), 258 deletions(-) diff --git a/tests/test_glinet.py b/tests/test_glinet.py index db75fb2..0b41bcf 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -1,258 +1,344 @@ -"""Tests for the GLinet router API using gli4py, must be run against a GLinet router.""" - -import asyncio -import pytest -from gli4py.enums import TailscaleConnection -from gli4py.error_handling import NonZeroResponse -from gli4py.glinet import GLinet - -router = GLinet(base_url="http://192.168.0.1/rpc") -PERFORM_DISTRUPTIVE_TESTS = False - -models = [ - "mt1300", - "x3000", - "mt2500", - "mt2500a", - "axt1800", - "a1300", - "ax1800", - "sft1200", - "e750", - "mv100", - "mv1000w", - "s10", - "s200", - "s1300", - "sf1200", - "b1300", - "b2200", - "ap1300", - "ap1300lte", - "x1200", - "x750", - "x300b", - "xe300", - "ar750s", - "ar750", - "ar300m", - "n300", -] - - -@pytest.fixture(scope="session") -def event_loop(): - """Create a new event loop for each test session.""" - policy = asyncio.get_event_loop_policy() - loop = policy.new_event_loop() - yield loop - loop.close() - - -@pytest.mark.asyncio -async def test_router_reachable() -> None: - """Test if the router is reachable.""" - response = await router.router_reachable() - assert response - print(response) - - -@pytest.mark.asyncio -async def test_login() -> None: - """Test logging into the router.""" - with open("router_pwd", "r", encoding="utf-8") as file: - pwd = str(file.read()) - assert not router.logged_in - await router.login("root", pwd) - assert router.logged_in - print(router.sid) - - -@pytest.mark.asyncio -async def test_router_info() -> None: - """Test retrieving router information.""" - response = await router.router_info() - assert "model" in response - assert "firmware_version" in response - assert "mac" in response - print(response) - - -@pytest.mark.asyncio -async def test_router_get_status() -> None: - """Test retrieving router status.""" - response = await router.router_get_status() - assert "service" in response - assert "network" in response - assert "system" in response - assert "wifi" in response - system = response.get("system") - assert "uptime" in system - assert "load_average" in system - print(response) - - -@pytest.mark.asyncio -async def test_router_get_load() -> None: - """Test retrieving router load information.""" - response = await router.router_get_load() - assert "load_average" in response - assert "memory_free" in response - assert "memory_total" in response - print(response) - - -@pytest.mark.asyncio -async def test_router_mac() -> None: - """Test retrieving the router's MAC address.""" - response = await router.router_mac() - assert "factory_mac" in response - print(response) - - -@pytest.mark.asyncio -async def test_connected_clients() -> None: - """Test retrieving connected clients.""" - clients = await router.connected_clients() - print(len(clients)) - assert len(clients) > 0 - - -@pytest.mark.asyncio -async def test_wifi_ifaces_get() -> None: - """Test retrieving WiFi interfaces.""" - wifi_ifaces = await router.wifi_ifaces_get() - print(wifi_ifaces) - for iface in wifi_ifaces.values(): - assert "enabled" in iface - assert "ssid" in iface - assert "name" in iface - assert "key" in iface - - -@pytest.mark.asyncio -async def test_wifi_ifaces_set_enabled() -> None: - """Test enabling/disabling a WiFi interface.""" - - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - wifi_ifaces = await router.wifi_ifaces_get() - iface = next(iter(wifi_ifaces.values())) - iface_enabled = iface.get("enabled") - - response = await router.wifi_iface_set_enabled(iface.get("name"), not iface_enabled) - print(response) - await asyncio.sleep(1) - - wifi_ifaces2 = await router.wifi_ifaces_get() - iface_enabled_after = wifi_ifaces2.get(iface.get("name")).get("enabled") - assert iface_enabled_after != iface_enabled - - -@pytest.mark.asyncio -async def test_connected_to_internet() -> None: - """Test checking if the router is connected to the internet.""" - response = await router.connected_to_internet() - print(response) - assert response["detected"] in [0, 1, 2, 3] - assert "ip" in response - - -@pytest.mark.asyncio -async def test_ping() -> None: - """Test pinging a host.""" - response = await router.ping("google.com") - assert response - print(response) - response = await router.ping("8.8.8.8") - assert response - response = await router.ping("0.0.0.1") - assert not response - - -@pytest.mark.asyncio -async def test_wireguard_client_list() -> None: - """Test retrieving the list of WireGuard clients.""" - response = await router.wireguard_client_list() - print(response) - # assert(response['enable'] in [True,False]) - - -@pytest.mark.asyncio -async def test_wireguard_client_state() -> None: - """Test retrieving the state of the WireGuard client.""" - response = await router.wireguard_client_state() - print(response) - assert response["status"] in [0, 1, 2] - - -@pytest.mark.asyncio -async def test_tailscale_status() -> None: - """Test retrieving the Tailscale status.""" - response = await router._tailscale_status() # pylint: disable=protected-access - print(response) - assert dict(response).get("status", 0) in [1, 2, 3, 4] or response == [] - - -@pytest.mark.asyncio -async def test_tailscale_connection() -> None: - """Test retrieving the Tailscale connection state.""" - response = await router.tailscale_connection_state() - print(response) - assert response in [TailscaleConnection.DISCONNECTED, TailscaleConnection.CONNECTED] - - -@pytest.mark.asyncio -async def test_tailscale_configured() -> None: - """Test checking if Tailscale is configured.""" - response = await router.tailscale_configured() - print("Tailscale configured:", response) - assert response in [True, False] - - -@pytest.mark.asyncio -async def test_tailscale_get_config() -> None: - """Test retrieving the Tailscale configuration.""" - response = await router._tailscale_get_config() # pylint: disable=protected-access - print(response["enabled"]) - assert response["enabled"] in [True, False] - - -@pytest.mark.asyncio -async def test_tailscale_start() -> None: - """Test starting Tailscale.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - result = await router.tailscale_start() - print(result) - assert result in [True, False] - - -@pytest.mark.asyncio -async def test_tailscale_stop() -> None: - """Test stopping Tailscale.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - result = await router.tailscale_stop() - print(result) - assert result in [True, False] - - -@pytest.mark.asyncio -async def test_router_reboot() -> None: - """Test rebooting the router.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - response = await router.router_reboot() - print(response) - print("waiting `15s` for router to shutdown") - await asyncio.sleep(15) - while not await router.router_reachable(): - print("waiting for router to wake") - await asyncio.sleep(1) - with pytest.raises(NonZeroResponse): - await router.router_info() +"""Tests for the GLinet router API using gli4py, must be run against a GLinet router.""" + +import asyncio +import pytest +from gli4py.enums import TailscaleConnection +from gli4py.error_handling import NonZeroResponse +from gli4py.glinet import GLinet, NEW_VPN_CLIENT_VERSION +from gli4py.version import Version + +router = GLinet(base_url="http://192.168.0.1/rpc") +PERFORM_DISTRUPTIVE_TESTS = False + +models = [ + "mt1300", + "x3000", + "mt2500", + "mt2500a", + "axt1800", + "a1300", + "ax1800", + "sft1200", + "e750", + "mv100", + "mv1000w", + "s10", + "s200", + "s1300", + "sf1200", + "b1300", + "b2200", + "ap1300", + "ap1300lte", + "x1200", + "x750", + "x300b", + "xe300", + "ar750s", + "ar750", + "ar300m", + "n300", +] + + +@pytest.fixture(scope="session") +def event_loop(): + """Create a new event loop for each test session.""" + policy = asyncio.get_event_loop_policy() + loop = policy.new_event_loop() + yield loop + loop.close() + + +@pytest.mark.asyncio +async def test_router_reachable() -> None: + """Test if the router is reachable.""" + response = await router.router_reachable() + assert response + print(response) + + +@pytest.mark.asyncio +async def test_login() -> None: + """Test logging into the router.""" + with open("router_pwd", "r", encoding="utf-8") as file: + pwd = str(file.read()) + assert not router.logged_in + await router.login("root", pwd) + assert router.logged_in + print(router.sid) + + +@pytest.mark.asyncio +async def test_router_info() -> None: + """Test retrieving router information.""" + response = await router.router_info() + assert "model" in response + assert "firmware_version" in response + assert "mac" in response + print(response) + + +@pytest.mark.asyncio +async def test_router_get_status() -> None: + """Test retrieving router status.""" + response = await router.router_get_status() + assert "service" in response + assert "network" in response + assert "system" in response + assert "wifi" in response + system = response.get("system") + assert "uptime" in system + assert "load_average" in system + print(response) + + +@pytest.mark.asyncio +async def test_router_get_load() -> None: + """Test retrieving router load information.""" + response = await router.router_get_load() + assert "load_average" in response + assert "memory_free" in response + assert "memory_total" in response + print(response) + + +@pytest.mark.asyncio +async def test_router_mac() -> None: + """Test retrieving the router's MAC address.""" + response = await router.router_mac() + assert "factory_mac" in response + print(response) + + +@pytest.mark.asyncio +async def test_connected_clients() -> None: + """Test retrieving connected clients.""" + clients = await router.connected_clients() + print(len(clients)) + assert len(clients) > 0 + + +@pytest.mark.asyncio +async def test_wifi_ifaces_get() -> None: + """Test retrieving WiFi interfaces.""" + wifi_ifaces = await router.wifi_ifaces_get() + print(wifi_ifaces) + for iface in wifi_ifaces.values(): + assert "enabled" in iface + assert "ssid" in iface + assert "name" in iface + assert "key" in iface + + +@pytest.mark.asyncio +async def test_wifi_ifaces_set_enabled() -> None: + """Test enabling/disabling a WiFi interface.""" + + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + wifi_ifaces = await router.wifi_ifaces_get() + iface = next(iter(wifi_ifaces.values())) + iface_enabled = iface.get("enabled") + + response = await router.wifi_iface_set_enabled(iface.get("name"), not iface_enabled) + print(response) + await asyncio.sleep(1) + + wifi_ifaces2 = await router.wifi_ifaces_get() + iface_enabled_after = wifi_ifaces2.get(iface.get("name")).get("enabled") + assert iface_enabled_after != iface_enabled + + +@pytest.mark.asyncio +async def test_connected_to_internet() -> None: + """Test checking if the router is connected to the internet.""" + response = await router.connected_to_internet() + print(response) + assert response["detected"] in [0, 1, 2, 3] + assert "ip" in response + + +@pytest.mark.asyncio +async def test_ping() -> None: + """Test pinging a host.""" + response = await router.ping("google.com") + assert response + print(response) + response = await router.ping("8.8.8.8") + assert response + response = await router.ping("0.0.0.1") + assert not response + + +@pytest.mark.asyncio +async def test_wireguard_client_list() -> None: + """Test retrieving the list of WireGuard clients.""" + response = await router.wireguard_client_list() + print(response) + # assert(response['enable'] in [True,False]) + + +@pytest.mark.asyncio +async def test_wireguard_client_state() -> None: + """Test retrieving the state of the WireGuard client.""" + # We need to get the proper firmware version for this + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + parsed_version = Version.parse(firmware_version) + response = await router.wireguard_client_state(firmware_version) + print(response) + first_status = response[0] + # In newer version, status only exists when enabled is True + # In older versions, status is always present + if parsed_version >= NEW_VPN_CLIENT_VERSION: + assert first_status["enabled"] in [True, False] + else: + assert first_status["status"] in [0, 1, 2] + +@pytest.mark.asyncio +async def test_wireguard_start() -> None: + """Test starting the WireGuard client.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + status_list = await router.wireguard_client_state(firmware_version) + if status_list is None or len(status_list) == 0: + pytest.skip("No WireGuard client configured, skipping test.") + return + + first_status = status_list[0] + group_id = first_status["group_id"] + peer_id = first_status["peer_id"] + tunnel_id = first_status["tunnel_id"] + + result = await router.wireguard_client_start(group_id, peer_id, tunnel_id, firmware_version) + print("RESULT: ", result) + assert result["tunnel_id"] == tunnel_id + + # Wait for the client to connect or timeout with 10 seconds + for i in range(10): + status_list = await router.wireguard_client_state(firmware_version) + first_status = status_list[0] + if "status" in first_status and first_status["status"] == 1 and "enabled" in first_status and first_status["enabled"]: + break + await asyncio.sleep(1) + + if i == 9: + pytest.fail("WireGuard client took too long to connect.") + +@pytest.mark.asyncio +async def test_wireguard_stop() -> None: + """Test stopping the WireGuard client.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + status_list = await router.wireguard_client_state(firmware_version) + if status_list is None or len(status_list) == 0: + pytest.skip("No WireGuard client configured, skipping test.") + return + + first_status = status_list[0] + tunnel_id = first_status["tunnel_id"] + + result = await router.wireguard_client_stop(tunnel_id, firmware_version) + print("RESULT: ", result) + assert result["tunnel_id"] == tunnel_id + + parsed_version = Version.parse(firmware_version) + + # Wait for the client to disconnect or timeout with 10 seconds + for i in range(10): + status_list = await router.wireguard_client_state(firmware_version) + first_status = status_list[0] + # In newer version, status only exists when enabled is True + # In older versions, status is always present + if parsed_version >= NEW_VPN_CLIENT_VERSION: + if "enabled" in first_status and not first_status["enabled"]: + break + else: + if "status" in first_status and first_status["status"] == 0: + break + + await asyncio.sleep(1) + + if i == 9: + pytest.fail("WireGuard client took too long to disconnect.") + + +@pytest.mark.asyncio +async def test_tailscale_status() -> None: + """Test retrieving the Tailscale status.""" + response = await router._tailscale_status() # pylint: disable=protected-access + print(response) + assert dict(response).get("status", 0) in [1, 2, 3, 4] or response == [] + + +@pytest.mark.asyncio +async def test_tailscale_connection() -> None: + """Test retrieving the Tailscale connection state.""" + response = await router.tailscale_connection_state() + print(response) + assert response in [TailscaleConnection.DISCONNECTED, TailscaleConnection.CONNECTED] + + +@pytest.mark.asyncio +async def test_tailscale_configured() -> None: + """Test checking if Tailscale is configured.""" + response = await router.tailscale_configured() + print("Tailscale configured:", response) + assert response in [True, False] + + +@pytest.mark.asyncio +async def test_tailscale_get_config() -> None: + """Test retrieving the Tailscale configuration.""" + response = await router._tailscale_get_config() # pylint: disable=protected-access + print(response["enabled"]) + assert response["enabled"] in [True, False] + + +@pytest.mark.asyncio +async def test_tailscale_start() -> None: + """Test starting Tailscale.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + result = await router.tailscale_start() + print(result) + assert result in [True, False] + + +@pytest.mark.asyncio +async def test_tailscale_stop() -> None: + """Test stopping Tailscale.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + result = await router.tailscale_stop() + print(result) + assert result in [True, False] + + +@pytest.mark.asyncio +async def test_router_reboot() -> None: + """Test rebooting the router.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + response = await router.router_reboot() + print(response) + print("waiting `15s` for router to shutdown") + await asyncio.sleep(15) + while not await router.router_reachable(): + print("waiting for router to wake") + await asyncio.sleep(1) + with pytest.raises(NonZeroResponse): + await router.router_info() From bc922c4803eec7bb8c69578863a3c74c1fe366d0 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 15:35:08 +0200 Subject: [PATCH 10/21] Revert "Add wireguard tests" This reverts commit f5226d401f52c43606b6c4fbbf3a4017ad1abbc3. --- tests/test_glinet.py | 602 +++++++++++++++++++------------------------ 1 file changed, 258 insertions(+), 344 deletions(-) diff --git a/tests/test_glinet.py b/tests/test_glinet.py index 0b41bcf..db75fb2 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -1,344 +1,258 @@ -"""Tests for the GLinet router API using gli4py, must be run against a GLinet router.""" - -import asyncio -import pytest -from gli4py.enums import TailscaleConnection -from gli4py.error_handling import NonZeroResponse -from gli4py.glinet import GLinet, NEW_VPN_CLIENT_VERSION -from gli4py.version import Version - -router = GLinet(base_url="http://192.168.0.1/rpc") -PERFORM_DISTRUPTIVE_TESTS = False - -models = [ - "mt1300", - "x3000", - "mt2500", - "mt2500a", - "axt1800", - "a1300", - "ax1800", - "sft1200", - "e750", - "mv100", - "mv1000w", - "s10", - "s200", - "s1300", - "sf1200", - "b1300", - "b2200", - "ap1300", - "ap1300lte", - "x1200", - "x750", - "x300b", - "xe300", - "ar750s", - "ar750", - "ar300m", - "n300", -] - - -@pytest.fixture(scope="session") -def event_loop(): - """Create a new event loop for each test session.""" - policy = asyncio.get_event_loop_policy() - loop = policy.new_event_loop() - yield loop - loop.close() - - -@pytest.mark.asyncio -async def test_router_reachable() -> None: - """Test if the router is reachable.""" - response = await router.router_reachable() - assert response - print(response) - - -@pytest.mark.asyncio -async def test_login() -> None: - """Test logging into the router.""" - with open("router_pwd", "r", encoding="utf-8") as file: - pwd = str(file.read()) - assert not router.logged_in - await router.login("root", pwd) - assert router.logged_in - print(router.sid) - - -@pytest.mark.asyncio -async def test_router_info() -> None: - """Test retrieving router information.""" - response = await router.router_info() - assert "model" in response - assert "firmware_version" in response - assert "mac" in response - print(response) - - -@pytest.mark.asyncio -async def test_router_get_status() -> None: - """Test retrieving router status.""" - response = await router.router_get_status() - assert "service" in response - assert "network" in response - assert "system" in response - assert "wifi" in response - system = response.get("system") - assert "uptime" in system - assert "load_average" in system - print(response) - - -@pytest.mark.asyncio -async def test_router_get_load() -> None: - """Test retrieving router load information.""" - response = await router.router_get_load() - assert "load_average" in response - assert "memory_free" in response - assert "memory_total" in response - print(response) - - -@pytest.mark.asyncio -async def test_router_mac() -> None: - """Test retrieving the router's MAC address.""" - response = await router.router_mac() - assert "factory_mac" in response - print(response) - - -@pytest.mark.asyncio -async def test_connected_clients() -> None: - """Test retrieving connected clients.""" - clients = await router.connected_clients() - print(len(clients)) - assert len(clients) > 0 - - -@pytest.mark.asyncio -async def test_wifi_ifaces_get() -> None: - """Test retrieving WiFi interfaces.""" - wifi_ifaces = await router.wifi_ifaces_get() - print(wifi_ifaces) - for iface in wifi_ifaces.values(): - assert "enabled" in iface - assert "ssid" in iface - assert "name" in iface - assert "key" in iface - - -@pytest.mark.asyncio -async def test_wifi_ifaces_set_enabled() -> None: - """Test enabling/disabling a WiFi interface.""" - - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - wifi_ifaces = await router.wifi_ifaces_get() - iface = next(iter(wifi_ifaces.values())) - iface_enabled = iface.get("enabled") - - response = await router.wifi_iface_set_enabled(iface.get("name"), not iface_enabled) - print(response) - await asyncio.sleep(1) - - wifi_ifaces2 = await router.wifi_ifaces_get() - iface_enabled_after = wifi_ifaces2.get(iface.get("name")).get("enabled") - assert iface_enabled_after != iface_enabled - - -@pytest.mark.asyncio -async def test_connected_to_internet() -> None: - """Test checking if the router is connected to the internet.""" - response = await router.connected_to_internet() - print(response) - assert response["detected"] in [0, 1, 2, 3] - assert "ip" in response - - -@pytest.mark.asyncio -async def test_ping() -> None: - """Test pinging a host.""" - response = await router.ping("google.com") - assert response - print(response) - response = await router.ping("8.8.8.8") - assert response - response = await router.ping("0.0.0.1") - assert not response - - -@pytest.mark.asyncio -async def test_wireguard_client_list() -> None: - """Test retrieving the list of WireGuard clients.""" - response = await router.wireguard_client_list() - print(response) - # assert(response['enable'] in [True,False]) - - -@pytest.mark.asyncio -async def test_wireguard_client_state() -> None: - """Test retrieving the state of the WireGuard client.""" - # We need to get the proper firmware version for this - info_response = await router.router_info() - firmware_version = info_response["firmware_version"] - parsed_version = Version.parse(firmware_version) - response = await router.wireguard_client_state(firmware_version) - print(response) - first_status = response[0] - # In newer version, status only exists when enabled is True - # In older versions, status is always present - if parsed_version >= NEW_VPN_CLIENT_VERSION: - assert first_status["enabled"] in [True, False] - else: - assert first_status["status"] in [0, 1, 2] - -@pytest.mark.asyncio -async def test_wireguard_start() -> None: - """Test starting the WireGuard client.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - - info_response = await router.router_info() - firmware_version = info_response["firmware_version"] - status_list = await router.wireguard_client_state(firmware_version) - if status_list is None or len(status_list) == 0: - pytest.skip("No WireGuard client configured, skipping test.") - return - - first_status = status_list[0] - group_id = first_status["group_id"] - peer_id = first_status["peer_id"] - tunnel_id = first_status["tunnel_id"] - - result = await router.wireguard_client_start(group_id, peer_id, tunnel_id, firmware_version) - print("RESULT: ", result) - assert result["tunnel_id"] == tunnel_id - - # Wait for the client to connect or timeout with 10 seconds - for i in range(10): - status_list = await router.wireguard_client_state(firmware_version) - first_status = status_list[0] - if "status" in first_status and first_status["status"] == 1 and "enabled" in first_status and first_status["enabled"]: - break - await asyncio.sleep(1) - - if i == 9: - pytest.fail("WireGuard client took too long to connect.") - -@pytest.mark.asyncio -async def test_wireguard_stop() -> None: - """Test stopping the WireGuard client.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - - info_response = await router.router_info() - firmware_version = info_response["firmware_version"] - status_list = await router.wireguard_client_state(firmware_version) - if status_list is None or len(status_list) == 0: - pytest.skip("No WireGuard client configured, skipping test.") - return - - first_status = status_list[0] - tunnel_id = first_status["tunnel_id"] - - result = await router.wireguard_client_stop(tunnel_id, firmware_version) - print("RESULT: ", result) - assert result["tunnel_id"] == tunnel_id - - parsed_version = Version.parse(firmware_version) - - # Wait for the client to disconnect or timeout with 10 seconds - for i in range(10): - status_list = await router.wireguard_client_state(firmware_version) - first_status = status_list[0] - # In newer version, status only exists when enabled is True - # In older versions, status is always present - if parsed_version >= NEW_VPN_CLIENT_VERSION: - if "enabled" in first_status and not first_status["enabled"]: - break - else: - if "status" in first_status and first_status["status"] == 0: - break - - await asyncio.sleep(1) - - if i == 9: - pytest.fail("WireGuard client took too long to disconnect.") - - -@pytest.mark.asyncio -async def test_tailscale_status() -> None: - """Test retrieving the Tailscale status.""" - response = await router._tailscale_status() # pylint: disable=protected-access - print(response) - assert dict(response).get("status", 0) in [1, 2, 3, 4] or response == [] - - -@pytest.mark.asyncio -async def test_tailscale_connection() -> None: - """Test retrieving the Tailscale connection state.""" - response = await router.tailscale_connection_state() - print(response) - assert response in [TailscaleConnection.DISCONNECTED, TailscaleConnection.CONNECTED] - - -@pytest.mark.asyncio -async def test_tailscale_configured() -> None: - """Test checking if Tailscale is configured.""" - response = await router.tailscale_configured() - print("Tailscale configured:", response) - assert response in [True, False] - - -@pytest.mark.asyncio -async def test_tailscale_get_config() -> None: - """Test retrieving the Tailscale configuration.""" - response = await router._tailscale_get_config() # pylint: disable=protected-access - print(response["enabled"]) - assert response["enabled"] in [True, False] - - -@pytest.mark.asyncio -async def test_tailscale_start() -> None: - """Test starting Tailscale.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - result = await router.tailscale_start() - print(result) - assert result in [True, False] - - -@pytest.mark.asyncio -async def test_tailscale_stop() -> None: - """Test stopping Tailscale.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - result = await router.tailscale_stop() - print(result) - assert result in [True, False] - - -@pytest.mark.asyncio -async def test_router_reboot() -> None: - """Test rebooting the router.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - response = await router.router_reboot() - print(response) - print("waiting `15s` for router to shutdown") - await asyncio.sleep(15) - while not await router.router_reachable(): - print("waiting for router to wake") - await asyncio.sleep(1) - with pytest.raises(NonZeroResponse): - await router.router_info() +"""Tests for the GLinet router API using gli4py, must be run against a GLinet router.""" + +import asyncio +import pytest +from gli4py.enums import TailscaleConnection +from gli4py.error_handling import NonZeroResponse +from gli4py.glinet import GLinet + +router = GLinet(base_url="http://192.168.0.1/rpc") +PERFORM_DISTRUPTIVE_TESTS = False + +models = [ + "mt1300", + "x3000", + "mt2500", + "mt2500a", + "axt1800", + "a1300", + "ax1800", + "sft1200", + "e750", + "mv100", + "mv1000w", + "s10", + "s200", + "s1300", + "sf1200", + "b1300", + "b2200", + "ap1300", + "ap1300lte", + "x1200", + "x750", + "x300b", + "xe300", + "ar750s", + "ar750", + "ar300m", + "n300", +] + + +@pytest.fixture(scope="session") +def event_loop(): + """Create a new event loop for each test session.""" + policy = asyncio.get_event_loop_policy() + loop = policy.new_event_loop() + yield loop + loop.close() + + +@pytest.mark.asyncio +async def test_router_reachable() -> None: + """Test if the router is reachable.""" + response = await router.router_reachable() + assert response + print(response) + + +@pytest.mark.asyncio +async def test_login() -> None: + """Test logging into the router.""" + with open("router_pwd", "r", encoding="utf-8") as file: + pwd = str(file.read()) + assert not router.logged_in + await router.login("root", pwd) + assert router.logged_in + print(router.sid) + + +@pytest.mark.asyncio +async def test_router_info() -> None: + """Test retrieving router information.""" + response = await router.router_info() + assert "model" in response + assert "firmware_version" in response + assert "mac" in response + print(response) + + +@pytest.mark.asyncio +async def test_router_get_status() -> None: + """Test retrieving router status.""" + response = await router.router_get_status() + assert "service" in response + assert "network" in response + assert "system" in response + assert "wifi" in response + system = response.get("system") + assert "uptime" in system + assert "load_average" in system + print(response) + + +@pytest.mark.asyncio +async def test_router_get_load() -> None: + """Test retrieving router load information.""" + response = await router.router_get_load() + assert "load_average" in response + assert "memory_free" in response + assert "memory_total" in response + print(response) + + +@pytest.mark.asyncio +async def test_router_mac() -> None: + """Test retrieving the router's MAC address.""" + response = await router.router_mac() + assert "factory_mac" in response + print(response) + + +@pytest.mark.asyncio +async def test_connected_clients() -> None: + """Test retrieving connected clients.""" + clients = await router.connected_clients() + print(len(clients)) + assert len(clients) > 0 + + +@pytest.mark.asyncio +async def test_wifi_ifaces_get() -> None: + """Test retrieving WiFi interfaces.""" + wifi_ifaces = await router.wifi_ifaces_get() + print(wifi_ifaces) + for iface in wifi_ifaces.values(): + assert "enabled" in iface + assert "ssid" in iface + assert "name" in iface + assert "key" in iface + + +@pytest.mark.asyncio +async def test_wifi_ifaces_set_enabled() -> None: + """Test enabling/disabling a WiFi interface.""" + + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + wifi_ifaces = await router.wifi_ifaces_get() + iface = next(iter(wifi_ifaces.values())) + iface_enabled = iface.get("enabled") + + response = await router.wifi_iface_set_enabled(iface.get("name"), not iface_enabled) + print(response) + await asyncio.sleep(1) + + wifi_ifaces2 = await router.wifi_ifaces_get() + iface_enabled_after = wifi_ifaces2.get(iface.get("name")).get("enabled") + assert iface_enabled_after != iface_enabled + + +@pytest.mark.asyncio +async def test_connected_to_internet() -> None: + """Test checking if the router is connected to the internet.""" + response = await router.connected_to_internet() + print(response) + assert response["detected"] in [0, 1, 2, 3] + assert "ip" in response + + +@pytest.mark.asyncio +async def test_ping() -> None: + """Test pinging a host.""" + response = await router.ping("google.com") + assert response + print(response) + response = await router.ping("8.8.8.8") + assert response + response = await router.ping("0.0.0.1") + assert not response + + +@pytest.mark.asyncio +async def test_wireguard_client_list() -> None: + """Test retrieving the list of WireGuard clients.""" + response = await router.wireguard_client_list() + print(response) + # assert(response['enable'] in [True,False]) + + +@pytest.mark.asyncio +async def test_wireguard_client_state() -> None: + """Test retrieving the state of the WireGuard client.""" + response = await router.wireguard_client_state() + print(response) + assert response["status"] in [0, 1, 2] + + +@pytest.mark.asyncio +async def test_tailscale_status() -> None: + """Test retrieving the Tailscale status.""" + response = await router._tailscale_status() # pylint: disable=protected-access + print(response) + assert dict(response).get("status", 0) in [1, 2, 3, 4] or response == [] + + +@pytest.mark.asyncio +async def test_tailscale_connection() -> None: + """Test retrieving the Tailscale connection state.""" + response = await router.tailscale_connection_state() + print(response) + assert response in [TailscaleConnection.DISCONNECTED, TailscaleConnection.CONNECTED] + + +@pytest.mark.asyncio +async def test_tailscale_configured() -> None: + """Test checking if Tailscale is configured.""" + response = await router.tailscale_configured() + print("Tailscale configured:", response) + assert response in [True, False] + + +@pytest.mark.asyncio +async def test_tailscale_get_config() -> None: + """Test retrieving the Tailscale configuration.""" + response = await router._tailscale_get_config() # pylint: disable=protected-access + print(response["enabled"]) + assert response["enabled"] in [True, False] + + +@pytest.mark.asyncio +async def test_tailscale_start() -> None: + """Test starting Tailscale.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + result = await router.tailscale_start() + print(result) + assert result in [True, False] + + +@pytest.mark.asyncio +async def test_tailscale_stop() -> None: + """Test stopping Tailscale.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + result = await router.tailscale_stop() + print(result) + assert result in [True, False] + + +@pytest.mark.asyncio +async def test_router_reboot() -> None: + """Test rebooting the router.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + response = await router.router_reboot() + print(response) + print("waiting `15s` for router to shutdown") + await asyncio.sleep(15) + while not await router.router_reachable(): + print("waiting for router to wake") + await asyncio.sleep(1) + with pytest.raises(NonZeroResponse): + await router.router_info() From 045363d6a6931fdb255e6e99bd55eb5776aad3f8 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 15:36:32 +0200 Subject: [PATCH 11/21] Hopefully fix test file git --- tests/test_glinet.py | 91 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 88 insertions(+), 3 deletions(-) diff --git a/tests/test_glinet.py b/tests/test_glinet.py index db75fb2..878a766 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -4,7 +4,7 @@ import pytest from gli4py.enums import TailscaleConnection from gli4py.error_handling import NonZeroResponse -from gli4py.glinet import GLinet +from gli4py.glinet import GLinet, NEW_VPN_CLIENT_VERSION router = GLinet(base_url="http://192.168.0.1/rpc") PERFORM_DISTRUPTIVE_TESTS = False @@ -182,9 +182,94 @@ async def test_wireguard_client_list() -> None: @pytest.mark.asyncio async def test_wireguard_client_state() -> None: """Test retrieving the state of the WireGuard client.""" - response = await router.wireguard_client_state() + # We need to get the proper firmware version for this + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + parsed_version = Version.parse(firmware_version) + response = await router.wireguard_client_state(firmware_version) print(response) - assert response["status"] in [0, 1, 2] + first_status = response[0] + # In newer version, status only exists when enabled is True + # In older versions, status is always present + if parsed_version >= NEW_VPN_CLIENT_VERSION: + assert first_status["enabled"] in [True, False] + else: + assert first_status["status"] in [0, 1, 2] + +@pytest.mark.asyncio +async def test_wireguard_start() -> None: + """Test starting the WireGuard client.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + status_list = await router.wireguard_client_state(firmware_version) + if status_list is None or len(status_list) == 0: + pytest.skip("No WireGuard client configured, skipping test.") + return + + first_status = status_list[0] + group_id = first_status["group_id"] + peer_id = first_status["peer_id"] + tunnel_id = first_status["tunnel_id"] + + result = await router.wireguard_client_start(group_id, peer_id, tunnel_id, firmware_version) + print("RESULT: ", result) + assert result["tunnel_id"] == tunnel_id + + # Wait for the client to connect or timeout with 10 seconds + for i in range(10): + status_list = await router.wireguard_client_state(firmware_version) + first_status = status_list[0] + if "status" in first_status and first_status["status"] == 1 and "enabled" in first_status and first_status["enabled"]: + break + await asyncio.sleep(1) + + if i == 9: + pytest.fail("WireGuard client took too long to connect.") + +@pytest.mark.asyncio +async def test_wireguard_stop() -> None: + """Test stopping the WireGuard client.""" + assert ( + PERFORM_DISTRUPTIVE_TESTS + ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + status_list = await router.wireguard_client_state(firmware_version) + if status_list is None or len(status_list) == 0: + pytest.skip("No WireGuard client configured, skipping test.") + return + + first_status = status_list[0] + tunnel_id = first_status["tunnel_id"] + + result = await router.wireguard_client_stop(tunnel_id, firmware_version) + print("RESULT: ", result) + assert result["tunnel_id"] == tunnel_id + + parsed_version = Version.parse(firmware_version) + + # Wait for the client to disconnect or timeout with 10 seconds + for i in range(10): + status_list = await router.wireguard_client_state(firmware_version) + first_status = status_list[0] + # In newer version, status only exists when enabled is True + # In older versions, status is always present + if parsed_version >= NEW_VPN_CLIENT_VERSION: + if "enabled" in first_status and not first_status["enabled"]: + break + else: + if "status" in first_status and first_status["status"] == 0: + break + + await asyncio.sleep(1) + + if i == 9: + pytest.fail("WireGuard client took too long to disconnect.") @pytest.mark.asyncio From 7c96d5de9f9bb70aee48200397f2e49f58a5615f Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 15:46:19 +0200 Subject: [PATCH 12/21] Return response array directly --- gli4py/glinet.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 0f7d82a..d7e996e 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -328,7 +328,7 @@ async def wireguard_client_state(self, version_string: str) -> list: # If the version is less than 4.8 we need to adjust the response to match the new format # The old format does not return an array, but just a single object. # We will wrap it in an array to match the new format. - response = {"status_list": [response]} + return [response] return response.get("status_list", []) From edb9644696e3ea2a4a53e4bb6d7eeb3c2ef03809 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 21:10:26 +0200 Subject: [PATCH 13/21] Use existing semver library instead of using own version --- gli4py/glinet.py | 4 +- gli4py/version.py | 108 ----------------------- poetry.lock | 58 ++++++++++--- pyproject.toml | 1 + tests/test_glinet.py | 2 + tests/test_version.py | 198 ------------------------------------------ 6 files changed, 53 insertions(+), 318 deletions(-) delete mode 100644 gli4py/version.py delete mode 100644 tests/test_version.py diff --git a/gli4py/glinet.py b/gli4py/glinet.py index d7e996e..1e2af51 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -2,14 +2,14 @@ import asyncio import hashlib +import semver from typing import Any, Optional from requests import Response, exceptions from uplink import Consumer, json, post, response_handler, AiohttpClient, timeout, Body from passlib.hash import md5_crypt, sha256_crypt, sha512_crypt +from semver import Version from gli4py.enums import TailscaleConnection -from gli4py.version import Version - from .error_handling import APIClientError, AuthenticationError, raise_for_status # , timeout_error diff --git a/gli4py/version.py b/gli4py/version.py deleted file mode 100644 index 3fcfb46..0000000 --- a/gli4py/version.py +++ /dev/null @@ -1,108 +0,0 @@ -"""Version utility class for handling semantic versioning.""" - -import re - - -class Version: - """A class to represent and parse semantic version numbers. - - Attributes: - major (int): The major version number - minor (int): The minor version number - patch (int): The patch version number - build (int): The build version number - """ - - def __init__(self, major: int = 0, minor: int = 0, patch: int = 0, build: int = 0): - """Initialize a Version instance. - - Args: - major (int): The major version number (default: 0) - minor (int): The minor version number (default: 0) - patch (int): The patch version number (default: 0) - build (int): The build version number (default: 0) - """ - self.major = int(major) - self.minor = int(minor) - self.patch = int(patch) - self.build = int(build) - - @classmethod - def parse(cls, version_string: str) -> 'Version': - """Parse a version string into a Version object. - - Args: - version_string (str): A version string in the format "major.minor.patch" or "major.minor.patch.build" - (e.g., "1.2.3", "12.34.56.78") - - Returns: - Version: A Version instance with the parsed values - - Raises: - ValueError: If the version string format is invalid - """ - if not isinstance(version_string, str): - raise ValueError("Version string must be a string") - - # Remove any leading 'v' if present (e.g., "v1.2.3" -> "1.2.3") - version_string = version_string.lstrip('v') - - # Regular expression to match 3-part or 4-part version pattern - pattern_3_part = r'^(\d+)\.(\d+)\.(\d+)$' - pattern_4_part = r'^(\d+)\.(\d+)\.(\d+)\.(\d+)$' - - match_3 = re.match(pattern_3_part, version_string) - match_4 = re.match(pattern_4_part, version_string) - - if match_4: - # 4-part version: major.minor.patch.build - major, minor, patch, build = match_4.groups() - return cls(int(major), int(minor), int(patch), int(build)) - elif match_3: - # 3-part version: major.minor.patch (build defaults to 0) - major, minor, patch = match_3.groups() - return cls(int(major), int(minor), int(patch), 0) - else: - raise ValueError( - f"Invalid version string format: '{version_string}'. Expected format: 'major.minor.patch' or 'major.minor.patch.build'") - - def __str__(self) -> str: - """Return the string representation of the version.""" - if self.build == 0: - return f"{self.major}.{self.minor}.{self.patch}" - else: - return f"{self.major}.{self.minor}.{self.patch}.{self.build}" - - def __repr__(self) -> str: - """Return the detailed string representation of the version.""" - return f"Version(major={self.major}, minor={self.minor}, patch={self.patch}, build={self.build})" - - def __eq__(self, other) -> bool: - """Check if two versions are equal.""" - if not isinstance(other, Version): - return False - return (self.major, self.minor, self.patch, self.build) == (other.major, other.minor, other.patch, other.build) - - def __lt__(self, other) -> bool: - """Check if this version is less than another version.""" - if not isinstance(other, Version): - return NotImplemented - return (self.major, self.minor, self.patch, self.build) < (other.major, other.minor, other.patch, other.build) - - def __le__(self, other) -> bool: - """Check if this version is less than or equal to another version.""" - return self == other or self < other - - def __gt__(self, other) -> bool: - """Check if this version is greater than another version.""" - if not isinstance(other, Version): - return NotImplemented - return (self.major, self.minor, self.patch, self.build) > (other.major, other.minor, other.patch, other.build) - - def __ge__(self, other) -> bool: - """Check if this version is greater than or equal to another version.""" - return self == other or self > other - - def to_tuple(self) -> tuple[int, int, int, int]: - """Return the version as a tuple (major, minor, patch, build).""" - return (self.major, self.minor, self.patch, self.build) diff --git a/poetry.lock b/poetry.lock index 232c436..9cae551 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand. [[package]] name = "aiohappyeyeballs" @@ -6,6 +6,7 @@ version = "2.6.1" description = "Happy Eyeballs for asyncio" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "aiohappyeyeballs-2.6.1-py3-none-any.whl", hash = "sha256:f349ba8f4b75cb25c99c5c2d84e997e485204d2902a9597802b0371f09331fb8"}, {file = "aiohappyeyeballs-2.6.1.tar.gz", hash = "sha256:c3f9d0113123803ccadfdf3f0faa505bc78e6a72d1cc4806cbd719826e943558"}, @@ -17,6 +18,7 @@ version = "3.12.13" description = "Async http client/server framework (asyncio)" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "aiohttp-3.12.13-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:5421af8f22a98f640261ee48aae3a37f0c41371e99412d55eaf2f8a46d5dad29"}, {file = "aiohttp-3.12.13-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:0fcda86f6cb318ba36ed8f1396a6a4a3fd8f856f84d426584392083d10da4de0"}, @@ -116,7 +118,7 @@ propcache = ">=0.2.0" yarl = ">=1.17.0,<2.0" [package.extras] -speedups = ["Brotli", "aiodns (>=3.3.0)", "brotlicffi"] +speedups = ["Brotli ; platform_python_implementation == \"CPython\"", "aiodns (>=3.3.0)", "brotlicffi ; platform_python_implementation != \"CPython\""] [[package]] name = "aiosignal" @@ -124,6 +126,7 @@ version = "1.3.2" description = "aiosignal: a list of registered asynchronous callbacks" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "aiosignal-1.3.2-py2.py3-none-any.whl", hash = "sha256:45cde58e409a301715980c2b01d0c28bdde3770d8290b5eb2173759d9acb31a5"}, {file = "aiosignal-1.3.2.tar.gz", hash = "sha256:a8c255c66fafb1e499c9351d0bf32ff2d8a0321595ebac3b93713656d2436f54"}, @@ -138,6 +141,7 @@ version = "3.4.3" description = "reference implementation of PEP 3156" optional = false python-versions = "*" +groups = ["dev"] files = [ {file = "asyncio-3.4.3-cp33-none-win32.whl", hash = "sha256:b62c9157d36187eca799c378e572c969f0da87cd5fc42ca372d92cdb06e7e1de"}, {file = "asyncio-3.4.3-cp33-none-win_amd64.whl", hash = "sha256:c46a87b48213d7464f22d9a497b9eef8c1928b68320a2fa94240f969f6fec08c"}, @@ -151,18 +155,19 @@ version = "25.3.0" description = "Classes Without Boilerplate" optional = false python-versions = ">=3.8" +groups = ["dev"] files = [ {file = "attrs-25.3.0-py3-none-any.whl", hash = "sha256:427318ce031701fea540783410126f03899a97ffc6f61596ad581ac2e40e3bc3"}, {file = "attrs-25.3.0.tar.gz", hash = "sha256:75d7cefc7fb576747b2c81b4442d4d4a1ce0900973527c011d1030fd3bf4af1b"}, ] [package.extras] -benchmark = ["cloudpickle", "hypothesis", "mypy (>=1.11.1)", "pympler", "pytest (>=4.3.0)", "pytest-codspeed", "pytest-mypy-plugins", "pytest-xdist[psutil]"] -cov = ["cloudpickle", "coverage[toml] (>=5.3)", "hypothesis", "mypy (>=1.11.1)", "pympler", "pytest (>=4.3.0)", "pytest-mypy-plugins", "pytest-xdist[psutil]"] -dev = ["cloudpickle", "hypothesis", "mypy (>=1.11.1)", "pre-commit-uv", "pympler", "pytest (>=4.3.0)", "pytest-mypy-plugins", "pytest-xdist[psutil]"] +benchmark = ["cloudpickle ; platform_python_implementation == \"CPython\"", "hypothesis", "mypy (>=1.11.1) ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pympler", "pytest (>=4.3.0)", "pytest-codspeed", "pytest-mypy-plugins ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pytest-xdist[psutil]"] +cov = ["cloudpickle ; platform_python_implementation == \"CPython\"", "coverage[toml] (>=5.3)", "hypothesis", "mypy (>=1.11.1) ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pympler", "pytest (>=4.3.0)", "pytest-mypy-plugins ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pytest-xdist[psutil]"] +dev = ["cloudpickle ; platform_python_implementation == \"CPython\"", "hypothesis", "mypy (>=1.11.1) ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pre-commit-uv", "pympler", "pytest (>=4.3.0)", "pytest-mypy-plugins ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pytest-xdist[psutil]"] docs = ["cogapp", "furo", "myst-parser", "sphinx", "sphinx-notfound-page", "sphinxcontrib-towncrier", "towncrier"] -tests = ["cloudpickle", "hypothesis", "mypy (>=1.11.1)", "pympler", "pytest (>=4.3.0)", "pytest-mypy-plugins", "pytest-xdist[psutil]"] -tests-mypy = ["mypy (>=1.11.1)", "pytest-mypy-plugins"] +tests = ["cloudpickle ; platform_python_implementation == \"CPython\"", "hypothesis", "mypy (>=1.11.1) ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pympler", "pytest (>=4.3.0)", "pytest-mypy-plugins ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pytest-xdist[psutil]"] +tests-mypy = ["mypy (>=1.11.1) ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\"", "pytest-mypy-plugins ; platform_python_implementation == \"CPython\" and python_version >= \"3.10\""] [[package]] name = "certifi" @@ -170,6 +175,7 @@ version = "2025.6.15" description = "Python package for providing Mozilla's CA Bundle." optional = false python-versions = ">=3.7" +groups = ["main", "dev"] files = [ {file = "certifi-2025.6.15-py3-none-any.whl", hash = "sha256:2e0c7ce7cb5d8f8634ca55d2ba7e6ec2689a2fd6537d8dec1296a477a4910057"}, {file = "certifi-2025.6.15.tar.gz", hash = "sha256:d747aa5a8b9bbbb1bb8c22bb13e22bd1f18e9796defa16bab421f7f7a317323b"}, @@ -181,6 +187,7 @@ version = "3.4.2" description = "The Real First Universal Charset Detector. Open, modern and actively maintained alternative to Chardet." optional = false python-versions = ">=3.7" +groups = ["main", "dev"] files = [ {file = "charset_normalizer-3.4.2-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:7c48ed483eb946e6c04ccbe02c6b4d1d48e51944b6db70f697e089c193404941"}, {file = "charset_normalizer-3.4.2-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b2d318c11350e10662026ad0eb71bb51c7812fc8590825304ae0bdd4ac283acd"}, @@ -282,6 +289,8 @@ version = "0.4.6" description = "Cross-platform colored terminal text." optional = false python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7" +groups = ["dev"] +markers = "sys_platform == \"win32\"" files = [ {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"}, {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, @@ -293,6 +302,7 @@ version = "1.7.0" description = "A list-like structure which implements collections.abc.MutableSequence" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "frozenlist-1.7.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:cc4df77d638aa2ed703b878dd093725b72a824c3c546c076e8fdf276f78ee84a"}, {file = "frozenlist-1.7.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:716a9973a2cc963160394f701964fe25012600f3d311f60c790400b00e568b61"}, @@ -406,6 +416,7 @@ version = "3.10" description = "Internationalized Domain Names in Applications (IDNA)" optional = false python-versions = ">=3.6" +groups = ["main", "dev"] files = [ {file = "idna-3.10-py3-none-any.whl", hash = "sha256:946d195a0d259cbba61165e88e65941f16e9b36ea6ddb97f00452bae8b1287d3"}, {file = "idna-3.10.tar.gz", hash = "sha256:12f65c9b470abda6dc35cf8e63cc574b1c52b11df2c86030af0ac09b01b13ea9"}, @@ -420,6 +431,7 @@ version = "2.1.0" description = "brain-dead simple config-ini parsing" optional = false python-versions = ">=3.8" +groups = ["dev"] files = [ {file = "iniconfig-2.1.0-py3-none-any.whl", hash = "sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760"}, {file = "iniconfig-2.1.0.tar.gz", hash = "sha256:3abbd2e30b36733fee78f9c7f7308f2d0050e88f0087fd25c2645f63c773e1c7"}, @@ -431,6 +443,7 @@ version = "0.1.0" description = "A tool for converting JSON documents to another JSON document format." optional = false python-versions = ">=3.6" +groups = ["dev"] files = [ {file = "json2json-0.1.0-py3-none-any.whl", hash = "sha256:577be5bd0e81097400f83afeaed86a65daf4b1403eb9095caea940be9ce4c141"}, {file = "json2json-0.1.0.tar.gz", hash = "sha256:39c9a51dc7807c7ffed2e392e0bd9ef259d82ab28976b5ab39825b1c748c87dd"}, @@ -442,6 +455,7 @@ version = "6.5.0" description = "multidict implementation" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "multidict-6.5.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:2e118a202904623b1d2606d1c8614e14c9444b59d64454b0c355044058066469"}, {file = "multidict-6.5.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:a42995bdcaff4e22cb1280ae7752c3ed3fbb398090c6991a2797a4a0e5ed16a9"}, @@ -561,6 +575,7 @@ version = "25.0" description = "Core utilities for Python packages" optional = false python-versions = ">=3.8" +groups = ["dev"] files = [ {file = "packaging-25.0-py3-none-any.whl", hash = "sha256:29572ef2b1f17581046b3a2227d5c611fb25ec70ca1ba8554b24b0e69331a484"}, {file = "packaging-25.0.tar.gz", hash = "sha256:d443872c98d677bf60f6a1f2f8c1cb748e8fe762d2bf9d3148b5599295b0fc4f"}, @@ -572,6 +587,7 @@ version = "1.7.4" description = "comprehensive password hashing framework supporting over 30 schemes" optional = false python-versions = "*" +groups = ["main"] files = [ {file = "passlib-1.7.4-py2.py3-none-any.whl", hash = "sha256:aa6bca462b8d8bda89c70b382f0c298a20b5560af6cbfa2dce410c0a2fb669f1"}, {file = "passlib-1.7.4.tar.gz", hash = "sha256:defd50f72b65c5402ab2c573830a6978e5f202ad0d984793c8dde2c4152ebe04"}, @@ -589,6 +605,7 @@ version = "1.6.0" description = "plugin and hook calling mechanisms for python" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746"}, {file = "pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3"}, @@ -604,6 +621,7 @@ version = "0.3.2" description = "Accelerated property cache" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "propcache-0.3.2-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:22d9962a358aedbb7a2e36187ff273adeaab9743373a272976d2e348d08c7770"}, {file = "propcache-0.3.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:0d0fda578d1dc3f77b6b5a5dce3b9ad69a8250a891760a548df850a5e8da87f3"}, @@ -711,6 +729,7 @@ version = "7.4.4" description = "pytest: simple powerful testing with Python" optional = false python-versions = ">=3.7" +groups = ["dev"] files = [ {file = "pytest-7.4.4-py3-none-any.whl", hash = "sha256:b090cdf5ed60bf4c45261be03239c2c1c22df034fbffe691abe93cd80cea01d8"}, {file = "pytest-7.4.4.tar.gz", hash = "sha256:2cf0005922c6ace4a3e2ec8b4080eb0d9753fdc93107415332f50ce9e7994280"}, @@ -731,6 +750,7 @@ version = "0.21.2" description = "Pytest support for asyncio" optional = false python-versions = ">=3.7" +groups = ["dev"] files = [ {file = "pytest_asyncio-0.21.2-py3-none-any.whl", hash = "sha256:ab664c88bb7998f711d8039cacd4884da6430886ae8bbd4eded552ed2004f16b"}, {file = "pytest_asyncio-0.21.2.tar.gz", hash = "sha256:d67738fc232b94b326b9d060750beb16e0074210b98dd8b58a5239fa2a154f45"}, @@ -749,6 +769,7 @@ version = "2.32.4" description = "Python HTTP for Humans." optional = false python-versions = ">=3.8" +groups = ["main", "dev"] files = [ {file = "requests-2.32.4-py3-none-any.whl", hash = "sha256:27babd3cda2a6d50b30443204ee89830707d396671944c998b5975b031ac2b2c"}, {file = "requests-2.32.4.tar.gz", hash = "sha256:27d0316682c8a29834d3264820024b62a36942083d52caf2f14c0591336d3422"}, @@ -764,12 +785,25 @@ urllib3 = ">=1.21.1,<3" socks = ["PySocks (>=1.5.6,!=1.5.7)"] use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"] +[[package]] +name = "semver" +version = "3.0.4" +description = "Python helper for Semantic Versioning (https://semver.org)" +optional = false +python-versions = ">=3.7" +groups = ["dev"] +files = [ + {file = "semver-3.0.4-py3-none-any.whl", hash = "sha256:9c824d87ba7f7ab4a1890799cec8596f15c1241cb473404ea1cb0c55e4b04746"}, + {file = "semver-3.0.4.tar.gz", hash = "sha256:afc7d8c584a5ed0a11033af086e8af226a9c0b206f313e0301f8dd7b6b589602"}, +] + [[package]] name = "six" version = "1.17.0" description = "Python 2 and 3 compatibility utilities" optional = false python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,>=2.7" +groups = ["main", "dev"] files = [ {file = "six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274"}, {file = "six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81"}, @@ -781,6 +815,7 @@ version = "0.10.0" description = "A Declarative HTTP Client for Python." optional = false python-versions = ">=3.10" +groups = ["main", "dev"] files = [ {file = "uplink-0.10.0-py3-none-any.whl", hash = "sha256:03212163f8a83a608480ec15122884988eb82cc7a2368b9072d9af8ede2246d9"}, {file = "uplink-0.10.0.tar.gz", hash = "sha256:a3b76b1cac5394126a72698d72b209bb80c8a94bad091870e463919979e4ab63"}, @@ -803,6 +838,7 @@ version = "4.2.0" description = "Implementation of RFC 6570 URI Templates" optional = false python-versions = ">=3.9" +groups = ["main", "dev"] files = [ {file = "uritemplate-4.2.0-py3-none-any.whl", hash = "sha256:962201ba1c4edcab02e60f9a0d3821e82dfc5d2d6662a21abd533879bdb8a686"}, {file = "uritemplate-4.2.0.tar.gz", hash = "sha256:480c2ed180878955863323eea31b0ede668795de182617fef9c6ca09e6ec9d0e"}, @@ -814,13 +850,14 @@ version = "2.5.0" description = "HTTP library with thread-safe connection pooling, file post, and more." optional = false python-versions = ">=3.9" +groups = ["main", "dev"] files = [ {file = "urllib3-2.5.0-py3-none-any.whl", hash = "sha256:e6b01673c0fa6a13e374b50871808eb3bf7046c4b125b216f6bf1cc604cff0dc"}, {file = "urllib3-2.5.0.tar.gz", hash = "sha256:3fc47733c7e419d4bc3f6b3dc2b4f890bb743906a30d56ba4a5bfa4bbff92760"}, ] [package.extras] -brotli = ["brotli (>=1.0.9)", "brotlicffi (>=0.8.0)"] +brotli = ["brotli (>=1.0.9) ; platform_python_implementation == \"CPython\"", "brotlicffi (>=0.8.0) ; platform_python_implementation != \"CPython\""] h2 = ["h2 (>=4,<5)"] socks = ["pysocks (>=1.5.6,!=1.5.7,<2.0)"] zstd = ["zstandard (>=0.18.0)"] @@ -831,6 +868,7 @@ version = "1.20.1" description = "Yet another URL library" optional = false python-versions = ">=3.9" +groups = ["dev"] files = [ {file = "yarl-1.20.1-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:6032e6da6abd41e4acda34d75a816012717000fa6839f37124a47fcefc49bec4"}, {file = "yarl-1.20.1-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:2c7b34d804b8cf9b214f05015c4fee2ebe7ed05cf581e7192c06555c71f4446a"}, @@ -944,6 +982,6 @@ multidict = ">=4.0" propcache = ">=0.2.1" [metadata] -lock-version = "2.0" +lock-version = "2.1" python-versions = "^3.11" -content-hash = "ea7eac93773d1b2474e4b09e0c1144bef028a922f7cbde111f981e40a2c09841" +content-hash = "4993b19b604fe917e0c6455864c6066897e799a51e4fdacdc3cb5b161985dedc" diff --git a/pyproject.toml b/pyproject.toml index 5b405ad..f46fa0e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ asyncio = "^3.4.3" json2json = "^0.1.0" pytest = "^7.2.2" pytest-asyncio = "^0.21.0" +semver = "^3.0.0" [build-system] requires = ["poetry-core"] diff --git a/tests/test_glinet.py b/tests/test_glinet.py index 878a766..eaf8e41 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -2,6 +2,8 @@ import asyncio import pytest +import semver +from semver import Version from gli4py.enums import TailscaleConnection from gli4py.error_handling import NonZeroResponse from gli4py.glinet import GLinet, NEW_VPN_CLIENT_VERSION diff --git a/tests/test_version.py b/tests/test_version.py deleted file mode 100644 index 8726e4c..0000000 --- a/tests/test_version.py +++ /dev/null @@ -1,198 +0,0 @@ -"""Tests for the Version class.""" - -import pytest -from gli4py.version import Version - - -class TestVersion: - """Test cases for the Version class.""" - - def test_init_default(self): - """Test Version initialization with default values.""" - version = Version() - assert version.major == 0 - assert version.minor == 0 - assert version.patch == 0 - assert version.build == 0 - - def test_init_with_values(self): - """Test Version initialization with specific values.""" - version = Version(1, 2, 3, 4) - assert version.major == 1 - assert version.minor == 2 - assert version.patch == 3 - assert version.build == 4 - - def test_init_with_three_values(self): - """Test Version initialization with three values (build defaults to 0).""" - version = Version(1, 2, 3) - assert version.major == 1 - assert version.minor == 2 - assert version.patch == 3 - assert version.build == 0 - - def test_parse_valid_version_3_part(self): - """Test parsing valid 3-part version strings.""" - version = Version.parse("1.2.3") - assert version.major == 1 - assert version.minor == 2 - assert version.patch == 3 - assert version.build == 0 - - def test_parse_valid_version_4_part(self): - """Test parsing valid 4-part version strings.""" - version = Version.parse("12.34.56.78") - assert version.major == 12 - assert version.minor == 34 - assert version.patch == 56 - assert version.build == 78 - - def test_parse_with_v_prefix_3_part(self): - """Test parsing 3-part version strings with 'v' prefix.""" - version = Version.parse("v2.5.1") - assert version.major == 2 - assert version.minor == 5 - assert version.patch == 1 - assert version.build == 0 - - def test_parse_with_v_prefix_4_part(self): - """Test parsing 4-part version strings with 'v' prefix.""" - version = Version.parse("v2.5.1.9") - assert version.major == 2 - assert version.minor == 5 - assert version.patch == 1 - assert version.build == 9 - - def test_parse_zero_values_3_part(self): - """Test parsing 3-part version strings with zero values.""" - version = Version.parse("0.0.0") - assert version.major == 0 - assert version.minor == 0 - assert version.patch == 0 - assert version.build == 0 - - def test_parse_zero_values_4_part(self): - """Test parsing 4-part version strings with zero values.""" - version = Version.parse("0.0.0.0") - assert version.major == 0 - assert version.minor == 0 - assert version.patch == 0 - assert version.build == 0 - - def test_parse_large_numbers_3_part(self): - """Test parsing 3-part version strings with large numbers.""" - version = Version.parse("123.456.789") - assert version.major == 123 - assert version.minor == 456 - assert version.patch == 789 - assert version.build == 0 - - def test_parse_large_numbers_4_part(self): - """Test parsing 4-part version strings with large numbers.""" - version = Version.parse("123.456.789.101112") - assert version.major == 123 - assert version.minor == 456 - assert version.patch == 789 - assert version.build == 101112 - - def test_parse_invalid_format(self): - """Test parsing invalid version string formats.""" - with pytest.raises(ValueError, match="Invalid version string format"): - Version.parse("1.2") - - with pytest.raises(ValueError, match="Invalid version string format"): - Version.parse("1.2.3.4.5") - - with pytest.raises(ValueError, match="Invalid version string format"): - Version.parse("1.2.a") - - with pytest.raises(ValueError, match="Invalid version string format"): - Version.parse("a.b.c") - - with pytest.raises(ValueError, match="Invalid version string format"): - Version.parse("1-2-3") - - with pytest.raises(ValueError, match="Invalid version string format"): - Version.parse("") - - def test_parse_non_string(self): - """Test parsing non-string inputs.""" - with pytest.raises(ValueError, match="Version string must be a string"): - Version.parse(123) - - with pytest.raises(ValueError, match="Version string must be a string"): - Version.parse(None) - - def test_str_representation_3_part(self): - """Test string representation of Version with build=0.""" - version = Version(1, 2, 3, 0) - assert str(version) == "1.2.3" - - def test_str_representation_4_part(self): - """Test string representation of Version with build>0.""" - version = Version(1, 2, 3, 4) - assert str(version) == "1.2.3.4" - - def test_repr_representation(self): - """Test repr representation of Version.""" - version = Version(1, 2, 3, 4) - assert repr(version) == "Version(major=1, minor=2, patch=3, build=4)" - - def test_equality(self): - """Test version equality comparison.""" - version1 = Version(1, 2, 3, 0) - version2 = Version(1, 2, 3, 0) - version3 = Version(1, 2, 4, 0) - version4 = Version(1, 2, 3, 1) - - assert version1 == version2 - assert version1 != version3 - assert version1 != version4 - assert version1 != "1.2.3" # Different type - - def test_comparison_operators(self): - """Test version comparison operators.""" - v1_0_0_0 = Version(1, 0, 0, 0) - v1_2_3_0 = Version(1, 2, 3, 0) - v1_2_3_1 = Version(1, 2, 3, 1) - v1_2_4_0 = Version(1, 2, 4, 0) - v2_0_0_0 = Version(2, 0, 0, 0) - - # Less than - assert v1_0_0_0 < v1_2_3_0 - assert v1_2_3_0 < v1_2_3_1 - assert v1_2_3_1 < v1_2_4_0 - assert v1_2_4_0 < v2_0_0_0 - - # Less than or equal - assert v1_0_0_0 <= v1_2_3_0 - assert v1_2_3_0 <= Version(1, 2, 3, 0) # Equal case - - # Greater than - assert v2_0_0_0 > v1_2_4_0 - assert v1_2_4_0 > v1_2_3_1 - assert v1_2_3_1 > v1_2_3_0 - assert v1_2_3_0 > v1_0_0_0 - - # Greater than or equal - assert v2_0_0_0 >= v1_2_4_0 - assert v1_2_3_0 >= Version(1, 2, 3, 0) # Equal case - - def test_to_tuple(self): - """Test conversion to tuple.""" - version = Version(1, 2, 3, 4) - assert version.to_tuple() == (1, 2, 3, 4) - - def test_parse_and_str_roundtrip_3_part(self): - """Test that parsing a 3-part version and converting back to string works.""" - original = "1.2.3" - version = Version.parse(original) - result = str(version) - assert result == original - - def test_parse_and_str_roundtrip_4_part(self): - """Test that parsing a 4-part version and converting back to string works.""" - original = "1.2.3.4" - version = Version.parse(original) - result = str(version) - assert result == original From 42b860aaff4209fe6aa6ea412ace9677115588ae Mon Sep 17 00:00:00 2001 From: Hertzole Date: Wed, 24 Sep 2025 21:30:18 +0200 Subject: [PATCH 14/21] Keep track of firmware version self --- gli4py/__init__.py | 1 - gli4py/glinet.py | 41 +++++++++++++++++++++++++++-------------- tests/test_glinet.py | 17 +++++++---------- 3 files changed, 34 insertions(+), 25 deletions(-) diff --git a/gli4py/__init__.py b/gli4py/__init__.py index 74ba59f..6b8cd1d 100644 --- a/gli4py/__init__.py +++ b/gli4py/__init__.py @@ -1,6 +1,5 @@ """gli4py - A Python library for GL.iNet routers""" from .glinet import GLinet -from .version import Version if __name__ == "__main__": pass diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 1e2af51..e33f8df 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -22,6 +22,8 @@ class GLinet(Consumer): """A Python Client for the GL-inet API.""" + _firmware_version: Optional[Version] = None + def __init__(self, sid: Optional[str] = None, **kwargs): self.sid: str = sid self._logged_in = self.sid is not None @@ -139,10 +141,19 @@ async def login(self, username: str, password: str) -> None: async def router_info(self) -> dict: """Retrieves information about the router, requires authentication.""" - return await self._request( + response = await self._request( self.gen_sid_payload("call", ["system", "get_info"], self.sid) ) + # Sanity check for firmware version + if "firmware_version" in response: + self._firmware_version = Version.parse(response["firmware_version"]) + else: + # No firmware version found, error + raise ValueError("No firmware version found in router info") + + return response + async def router_get_status(self) -> dict[str, list[dict[str, Any]]]: """Retrieves the status of the router, requires authentication.""" response: dict[str, list[dict[str, Any]]] = await self._request( @@ -312,19 +323,21 @@ async def wireguard_client_list(self) -> dict: ) return configs - async def wireguard_client_state(self, version_string: str) -> list: + async def wireguard_client_state(self) -> list: """ {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} """ - parsed_version = Version.parse(version_string) + if self._firmware_version is None: + await self.router_info() + # If version is 4.8 or greater use vpn-client otherwise use wg-client - target_call = "vpn-client" if parsed_version >= NEW_VPN_CLIENT_VERSION else "wg-client" + target_call = "vpn-client" if self._firmware_version >= NEW_VPN_CLIENT_VERSION else "wg-client" response = await self._request( self.gen_sid_payload("call", [target_call, "get_status"], self.sid) ) - if parsed_version < NEW_VPN_CLIENT_VERSION: + if self._firmware_version < NEW_VPN_CLIENT_VERSION: # If the version is less than 4.8 we need to adjust the response to match the new format # The old format does not return an array, but just a single object. # We will wrap it in an array to match the new format. @@ -332,22 +345,22 @@ async def wireguard_client_state(self, version_string: str) -> list: return response.get("status_list", []) - async def wireguard_client_start(self, group_id: int, peer_id: int, tunnel_id: int, version_string: str) -> dict: + async def wireguard_client_start(self, group_id: int, peer_id: int, tunnel_id: int) -> dict: """Starts a WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(group_id, peer_id, tunnel_id, True, version_string) + return await self._wireguard_set_client_enabled(group_id, peer_id, tunnel_id, True) - async def wireguard_client_stop(self, tunnel_id: int, version_string: str) -> dict: + async def wireguard_client_stop(self, tunnel_id: int) -> dict: """Stops the WireGuard client with the specified tunnel ID.""" # Pass -1 for group_id and peer_id as they are not needed to stop the client - return await self._wireguard_set_client_enabled(-1, -1, tunnel_id, False, version_string) + return await self._wireguard_set_client_enabled(-1, -1, tunnel_id, False) - async def _wireguard_set_client_enabled( - self, group_id: int, peer_id: int, tunnel_id: int, enabled: bool, version_string: str - ) -> dict: + async def _wireguard_set_client_enabled(self, group_id: int, peer_id: int, tunnel_id: int, enabled: bool) -> dict: """Sets the WireGuard client enabled state.""" - parsed_version = Version.parse(version_string) + if self._firmware_version is None: + await self.router_info() + # If version is 4.8 or greater use vpn-client otherwise use wg-client - if parsed_version >= NEW_VPN_CLIENT_VERSION: + if self._firmware_version >= NEW_VPN_CLIENT_VERSION: return await self._request( self.gen_sid_payload( "call", diff --git a/tests/test_glinet.py b/tests/test_glinet.py index eaf8e41..27c612d 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -2,7 +2,6 @@ import asyncio import pytest -import semver from semver import Version from gli4py.enums import TailscaleConnection from gli4py.error_handling import NonZeroResponse @@ -188,7 +187,7 @@ async def test_wireguard_client_state() -> None: info_response = await router.router_info() firmware_version = info_response["firmware_version"] parsed_version = Version.parse(firmware_version) - response = await router.wireguard_client_state(firmware_version) + response = await router.wireguard_client_state() print(response) first_status = response[0] # In newer version, status only exists when enabled is True @@ -205,9 +204,7 @@ async def test_wireguard_start() -> None: PERFORM_DISTRUPTIVE_TESTS ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - info_response = await router.router_info() - firmware_version = info_response["firmware_version"] - status_list = await router.wireguard_client_state(firmware_version) + status_list = await router.wireguard_client_state() if status_list is None or len(status_list) == 0: pytest.skip("No WireGuard client configured, skipping test.") return @@ -217,13 +214,13 @@ async def test_wireguard_start() -> None: peer_id = first_status["peer_id"] tunnel_id = first_status["tunnel_id"] - result = await router.wireguard_client_start(group_id, peer_id, tunnel_id, firmware_version) + result = await router.wireguard_client_start(group_id, peer_id, tunnel_id) print("RESULT: ", result) assert result["tunnel_id"] == tunnel_id # Wait for the client to connect or timeout with 10 seconds for i in range(10): - status_list = await router.wireguard_client_state(firmware_version) + status_list = await router.wireguard_client_state() first_status = status_list[0] if "status" in first_status and first_status["status"] == 1 and "enabled" in first_status and first_status["enabled"]: break @@ -241,7 +238,7 @@ async def test_wireguard_stop() -> None: info_response = await router.router_info() firmware_version = info_response["firmware_version"] - status_list = await router.wireguard_client_state(firmware_version) + status_list = await router.wireguard_client_state() if status_list is None or len(status_list) == 0: pytest.skip("No WireGuard client configured, skipping test.") return @@ -249,7 +246,7 @@ async def test_wireguard_stop() -> None: first_status = status_list[0] tunnel_id = first_status["tunnel_id"] - result = await router.wireguard_client_stop(tunnel_id, firmware_version) + result = await router.wireguard_client_stop(tunnel_id) print("RESULT: ", result) assert result["tunnel_id"] == tunnel_id @@ -257,7 +254,7 @@ async def test_wireguard_stop() -> None: # Wait for the client to disconnect or timeout with 10 seconds for i in range(10): - status_list = await router.wireguard_client_state(firmware_version) + status_list = await router.wireguard_client_state() first_status = status_list[0] # In newer version, status only exists when enabled is True # In older versions, status is always present From d201eaca2aee7be03e77c76a3ac307ef39665b8e Mon Sep 17 00:00:00 2001 From: Hertzole Date: Fri, 26 Sep 2025 02:59:28 +0200 Subject: [PATCH 15/21] Fix import --- gli4py/glinet.py | 1 - 1 file changed, 1 deletion(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index e33f8df..2da145b 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -2,7 +2,6 @@ import asyncio import hashlib -import semver from typing import Any, Optional from requests import Response, exceptions from uplink import Consumer, json, post, response_handler, AiohttpClient, timeout, Body From af1884b2411153b574f90474edc4d89335f2058a Mon Sep 17 00:00:00 2001 From: Hertzole Date: Fri, 26 Sep 2025 02:59:39 +0200 Subject: [PATCH 16/21] Fix version type --- gli4py/glinet.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 2da145b..1c26dac 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -21,7 +21,7 @@ class GLinet(Consumer): """A Python Client for the GL-inet API.""" - _firmware_version: Optional[Version] = None + _firmware_version: Version | None = None def __init__(self, sid: Optional[str] = None, **kwargs): self.sid: str = sid From 1ddaa445a26bb70297e13c54b567175b69b46250 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Fri, 26 Sep 2025 03:00:02 +0200 Subject: [PATCH 17/21] Fix linting errors --- gli4py/glinet.py | 33 +++++++++++++++++---------------- 1 file changed, 17 insertions(+), 16 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 1c26dac..9a56628 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -328,7 +328,7 @@ async def wireguard_client_state(self) -> list: """ if self._firmware_version is None: await self.router_info() - + # If version is 4.8 or greater use vpn-client otherwise use wg-client target_call = "vpn-client" if self._firmware_version >= NEW_VPN_CLIENT_VERSION else "wg-client" @@ -357,7 +357,7 @@ async def _wireguard_set_client_enabled(self, group_id: int, peer_id: int, tunne """Sets the WireGuard client enabled state.""" if self._firmware_version is None: await self.router_info() - + # If version is 4.8 or greater use vpn-client otherwise use wg-client if self._firmware_version >= NEW_VPN_CLIENT_VERSION: return await self._request( @@ -370,22 +370,23 @@ async def _wireguard_set_client_enabled(self, group_id: int, peer_id: int, tunne ], self.sid, ) - ) - else: - if enabled: - return await self._request( - self.gen_sid_payload( - "call", - ["wg-client", "start", {"group_id": group_id, "peer_id": peer_id}], - self.sid, - ) - ) - else: - return await self._request( - self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) + ) + + # Not version 4.8 or greater so use wg-client + # If enabled, call the start method with group_id and peer_id + if enabled: + return await self._request( + self.gen_sid_payload( + "call", + ["wg-client", "start", {"group_id": group_id, "peer_id": peer_id}], + self.sid, ) + ) - + # Not enabled, call the stop method + return await self._request( + self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) + ) async def _tailscale_get_config(self) -> dict | bool: """ From 00d3ddf54a416ef988274ae59296ae3142987fb1 Mon Sep 17 00:00:00 2001 From: Hertzole Date: Fri, 26 Sep 2025 03:04:37 +0200 Subject: [PATCH 18/21] Fixed even more trailing whitespaces --- tests/test_glinet.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_glinet.py b/tests/test_glinet.py index 27c612d..9432101 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -208,8 +208,8 @@ async def test_wireguard_start() -> None: if status_list is None or len(status_list) == 0: pytest.skip("No WireGuard client configured, skipping test.") return - - first_status = status_list[0] + + first_status = status_list[0] group_id = first_status["group_id"] peer_id = first_status["peer_id"] tunnel_id = first_status["tunnel_id"] @@ -242,8 +242,8 @@ async def test_wireguard_stop() -> None: if status_list is None or len(status_list) == 0: pytest.skip("No WireGuard client configured, skipping test.") return - - first_status = status_list[0] + + first_status = status_list[0] tunnel_id = first_status["tunnel_id"] result = await router.wireguard_client_stop(tunnel_id) From 85d59c2d47b63d47fbaf58ee8289836011372b8e Mon Sep 17 00:00:00 2001 From: HarvsG <11440490+HarvsG@users.noreply.github.com> Date: Fri, 26 Sep 2025 10:35:43 +0000 Subject: [PATCH 19/21] make semver non-dev, linting --- poetry.lock | 4 +- pyproject.toml | 4 +- tests/test_glinet.py | 209 ++++++++++++++++++++++--------------------- 3 files changed, 112 insertions(+), 105 deletions(-) diff --git a/poetry.lock b/poetry.lock index 9cae551..87cac45 100644 --- a/poetry.lock +++ b/poetry.lock @@ -791,7 +791,7 @@ version = "3.0.4" description = "Python helper for Semantic Versioning (https://semver.org)" optional = false python-versions = ">=3.7" -groups = ["dev"] +groups = ["main"] files = [ {file = "semver-3.0.4-py3-none-any.whl", hash = "sha256:9c824d87ba7f7ab4a1890799cec8596f15c1241cb473404ea1cb0c55e4b04746"}, {file = "semver-3.0.4.tar.gz", hash = "sha256:afc7d8c584a5ed0a11033af086e8af226a9c0b206f313e0301f8dd7b6b589602"}, @@ -984,4 +984,4 @@ propcache = ">=0.2.1" [metadata] lock-version = "2.1" python-versions = "^3.11" -content-hash = "4993b19b604fe917e0c6455864c6066897e799a51e4fdacdc3cb5b161985dedc" +content-hash = "bb1133ea615d1424ceb3e7458978dc9a837de6029ccb643b6d09787de0936555" diff --git a/pyproject.toml b/pyproject.toml index f46fa0e..ada8f2a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "gli4py" -version = "0.0.13" +version = "0.0.14" description = "A python 3 API wrapper for GL-inet routers for consumption by Home Assistant" authors = ["HarvsG "] license = "GNU GENERAL PUBLIC LICENSE" @@ -11,6 +11,7 @@ packages = [{include = "gli4py"}] python = "^3.11" uplink = "0.10.0" passlib = "^1.7.4" +semver = "^3.0.0" [tool.poetry.group.dev.dependencies] uplink = "^0.10.0" @@ -20,7 +21,6 @@ asyncio = "^3.4.3" json2json = "^0.1.0" pytest = "^7.2.2" pytest-asyncio = "^0.21.0" -semver = "^3.0.0" [build-system] requires = ["poetry-core"] diff --git a/tests/test_glinet.py b/tests/test_glinet.py index 9432101..992b14e 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -2,10 +2,10 @@ import asyncio import pytest -from semver import Version +from semver import Version from gli4py.enums import TailscaleConnection from gli4py.error_handling import NonZeroResponse -from gli4py.glinet import GLinet, NEW_VPN_CLIENT_VERSION +from gli4py.glinet import GLinet, NEW_VPN_CLIENT_VERSION router = GLinet(base_url="http://192.168.0.1/rpc") PERFORM_DISTRUPTIVE_TESTS = False @@ -135,9 +135,9 @@ async def test_wifi_ifaces_get() -> None: async def test_wifi_ifaces_set_enabled() -> None: """Test enabling/disabling a WiFi interface.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + assert PERFORM_DISTRUPTIVE_TESTS, ( + "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + ) wifi_ifaces = await router.wifi_ifaces_get() iface = next(iter(wifi_ifaces.values())) iface_enabled = iface.get("enabled") @@ -183,98 +183,105 @@ async def test_wireguard_client_list() -> None: @pytest.mark.asyncio async def test_wireguard_client_state() -> None: """Test retrieving the state of the WireGuard client.""" - # We need to get the proper firmware version for this - info_response = await router.router_info() - firmware_version = info_response["firmware_version"] - parsed_version = Version.parse(firmware_version) - response = await router.wireguard_client_state() + # We need to get the proper firmware version for this + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + parsed_version = Version.parse(firmware_version) + response = await router.wireguard_client_state() print(response) - first_status = response[0] - # In newer version, status only exists when enabled is True - # In older versions, status is always present - if parsed_version >= NEW_VPN_CLIENT_VERSION: - assert first_status["enabled"] in [True, False] - else: - assert first_status["status"] in [0, 1, 2] - -@pytest.mark.asyncio -async def test_wireguard_start() -> None: - """Test starting the WireGuard client.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - - status_list = await router.wireguard_client_state() - if status_list is None or len(status_list) == 0: - pytest.skip("No WireGuard client configured, skipping test.") - return - - first_status = status_list[0] - group_id = first_status["group_id"] - peer_id = first_status["peer_id"] - tunnel_id = first_status["tunnel_id"] - - result = await router.wireguard_client_start(group_id, peer_id, tunnel_id) - print("RESULT: ", result) - assert result["tunnel_id"] == tunnel_id - - # Wait for the client to connect or timeout with 10 seconds - for i in range(10): - status_list = await router.wireguard_client_state() - first_status = status_list[0] - if "status" in first_status and first_status["status"] == 1 and "enabled" in first_status and first_status["enabled"]: - break - await asyncio.sleep(1) - - if i == 9: - pytest.fail("WireGuard client took too long to connect.") - -@pytest.mark.asyncio -async def test_wireguard_stop() -> None: - """Test stopping the WireGuard client.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." - - info_response = await router.router_info() - firmware_version = info_response["firmware_version"] - status_list = await router.wireguard_client_state() - if status_list is None or len(status_list) == 0: - pytest.skip("No WireGuard client configured, skipping test.") - return - - first_status = status_list[0] - tunnel_id = first_status["tunnel_id"] - - result = await router.wireguard_client_stop(tunnel_id) - print("RESULT: ", result) - assert result["tunnel_id"] == tunnel_id - - parsed_version = Version.parse(firmware_version) - - # Wait for the client to disconnect or timeout with 10 seconds - for i in range(10): - status_list = await router.wireguard_client_state() - first_status = status_list[0] - # In newer version, status only exists when enabled is True - # In older versions, status is always present - if parsed_version >= NEW_VPN_CLIENT_VERSION: - if "enabled" in first_status and not first_status["enabled"]: - break - else: - if "status" in first_status and first_status["status"] == 0: - break - - await asyncio.sleep(1) - - if i == 9: - pytest.fail("WireGuard client took too long to disconnect.") + first_status = response[0] + # In newer version, status only exists when enabled is True + # In older versions, status is always present + if parsed_version >= NEW_VPN_CLIENT_VERSION: + assert first_status["enabled"] in [True, False] + else: + assert first_status["status"] in [0, 1, 2] + + +@pytest.mark.asyncio +async def test_wireguard_start() -> None: + """Test starting the WireGuard client.""" + assert PERFORM_DISTRUPTIVE_TESTS, ( + "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + ) + + status_list = await router.wireguard_client_state() + if status_list is None or len(status_list) == 0: + pytest.skip("No WireGuard client configured, skipping test.") + return + + first_status = status_list[0] + group_id = first_status["group_id"] + peer_id = first_status["peer_id"] + tunnel_id = first_status["tunnel_id"] + + result = await router.wireguard_client_start(group_id, peer_id, tunnel_id) + print("RESULT: ", result) + assert result["tunnel_id"] == tunnel_id + + # Wait for the client to connect or timeout with 10 seconds + for i in range(10): + status_list = await router.wireguard_client_state() + first_status = status_list[0] + if ( + "status" in first_status + and first_status["status"] == 1 + and "enabled" in first_status + and first_status["enabled"] + ): + break + await asyncio.sleep(1) + + if i == 9: + pytest.fail("WireGuard client took too long to connect.") + + +@pytest.mark.asyncio +async def test_wireguard_stop() -> None: + """Test stopping the WireGuard client.""" + assert PERFORM_DISTRUPTIVE_TESTS, ( + "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + ) + + info_response = await router.router_info() + firmware_version = info_response["firmware_version"] + status_list = await router.wireguard_client_state() + if status_list is None or len(status_list) == 0: + pytest.skip("No WireGuard client configured, skipping test.") + return + + first_status = status_list[0] + tunnel_id = first_status["tunnel_id"] + + result = await router.wireguard_client_stop(tunnel_id) + print("RESULT: ", result) + assert result["tunnel_id"] == tunnel_id + + parsed_version = Version.parse(firmware_version) + + # Wait for the client to disconnect or timeout with 10 seconds + for i in range(10): + status_list = await router.wireguard_client_state() + first_status = status_list[0] + # In newer version, status only exists when enabled is True + # In older versions, status is always present + if parsed_version >= NEW_VPN_CLIENT_VERSION: + if "enabled" in first_status and not first_status["enabled"]: + break + else: + if "status" in first_status and first_status["status"] == 0: + break + + await asyncio.sleep(1) + + if i == 9: + pytest.fail("WireGuard client took too long to disconnect.") @pytest.mark.asyncio async def test_tailscale_status() -> None: """Test retrieving the Tailscale status.""" - response = await router._tailscale_status() # pylint: disable=protected-access + response = await router._tailscale_status() # pylint: disable=protected-access print(response) assert dict(response).get("status", 0) in [1, 2, 3, 4] or response == [] @@ -298,7 +305,7 @@ async def test_tailscale_configured() -> None: @pytest.mark.asyncio async def test_tailscale_get_config() -> None: """Test retrieving the Tailscale configuration.""" - response = await router._tailscale_get_config() # pylint: disable=protected-access + response = await router._tailscale_get_config() # pylint: disable=protected-access print(response["enabled"]) assert response["enabled"] in [True, False] @@ -306,9 +313,9 @@ async def test_tailscale_get_config() -> None: @pytest.mark.asyncio async def test_tailscale_start() -> None: """Test starting Tailscale.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + assert PERFORM_DISTRUPTIVE_TESTS, ( + "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + ) result = await router.tailscale_start() print(result) assert result in [True, False] @@ -317,9 +324,9 @@ async def test_tailscale_start() -> None: @pytest.mark.asyncio async def test_tailscale_stop() -> None: """Test stopping Tailscale.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + assert PERFORM_DISTRUPTIVE_TESTS, ( + "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + ) result = await router.tailscale_stop() print(result) assert result in [True, False] @@ -328,9 +335,9 @@ async def test_tailscale_stop() -> None: @pytest.mark.asyncio async def test_router_reboot() -> None: """Test rebooting the router.""" - assert ( - PERFORM_DISTRUPTIVE_TESTS - ), "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + assert PERFORM_DISTRUPTIVE_TESTS, ( + "Disruptive tests are disabled, set PERFORM_DISTRUPTIVE_TESTS to True to run this test." + ) response = await router.router_reboot() print(response) print("waiting `15s` for router to shutdown") From 1867f262141c20709ed8edb4e4acf59811696503 Mon Sep 17 00:00:00 2001 From: HarvsG <11440490+HarvsG@users.noreply.github.com> Date: Fri, 26 Sep 2025 13:43:27 +0000 Subject: [PATCH 20/21] improve typing and unify wireguard calls --- gli4py/glinet.py | 167 +++++++++++++++++++++++++---------------------- 1 file changed, 90 insertions(+), 77 deletions(-) diff --git a/gli4py/glinet.py b/gli4py/glinet.py index 9a56628..7024508 100644 --- a/gli4py/glinet.py +++ b/gli4py/glinet.py @@ -6,23 +6,21 @@ from requests import Response, exceptions from uplink import Consumer, json, post, response_handler, AiohttpClient, timeout, Body from passlib.hash import md5_crypt, sha256_crypt, sha512_crypt -from semver import Version +from semver import Version from gli4py.enums import TailscaleConnection from .error_handling import APIClientError, AuthenticationError, raise_for_status # , timeout_error # typical base url http://192.168.8.1/rpc -NEW_VPN_CLIENT_VERSION = Version(4, 8, 0, 0) +NEW_VPN_CLIENT_VERSION = Version(4, 8, 0, 0) -@response_handler(raise_for_status) -@json class GLinet(Consumer): """A Python Client for the GL-inet API.""" - _firmware_version: Version | None = None - + _firmware_version: Version | None = None + def __init__(self, sid: Optional[str] = None, **kwargs): self.sid: str = sid self._logged_in = self.sid is not None @@ -55,11 +53,15 @@ def gen_no_auth_payload(method: str, params: dict) -> dict: } return payload + @response_handler(raise_for_status) + @json @post("") @timeout(2) async def _request(self, data: Body) -> Response: """Base method to make a request to the GL-inet API.""" + @response_handler(raise_for_status) + @json @post("") @timeout(5) async def _request_long_timeout(self, data: Body) -> Response: @@ -140,19 +142,19 @@ async def login(self, username: str, password: str) -> None: async def router_info(self) -> dict: """Retrieves information about the router, requires authentication.""" - response = await self._request( + response = await self._request( self.gen_sid_payload("call", ["system", "get_info"], self.sid) ) - # Sanity check for firmware version - if "firmware_version" in response: - self._firmware_version = Version.parse(response["firmware_version"]) - else: - # No firmware version found, error - raise ValueError("No firmware version found in router info") - - return response - + # Sanity check for firmware version + if "firmware_version" in response: + self._firmware_version = Version.parse(response["firmware_version"]) + else: + # No firmware version found, error + raise ValueError("No firmware version found in router info") + + return response + async def router_get_status(self) -> dict[str, list[dict[str, Any]]]: """Retrieves the status of the router, requires authentication.""" response: dict[str, list[dict[str, Any]]] = await self._request( @@ -303,7 +305,7 @@ async def wifi_iface_set_enabled(self, iface_name: str, enabled: bool) -> dict: # VPN information - async def wireguard_client_list(self) -> dict: + async def wireguard_client_list(self) -> list[dict[str, any]]: """Gets the list of WireGuard clients.""" response: dict = await self._request( self.gen_sid_payload("call", ["wg-client", "get_all_config_list"], self.sid) @@ -322,70 +324,81 @@ async def wireguard_client_list(self) -> dict: ) return configs - async def wireguard_client_state(self) -> list: + async def wireguard_client_state(self) -> list[dict[str, Any]]: """ - {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} + Firmware 4.8 and greater returns a list of status objects + {"status_list": [{"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""}]} + Firmware less than 4.8 returns a single status object of the most recently started client + {"rx_bytes":0,"ipv6":"","tx_bytes":0,"domain":"vpn.example.com","group_id":7707,"port":51820,"name":"TheOracle","peer_id":1341,"enabled":true,"proxy":True,"log":"","ipv4":""} """ - if self._firmware_version is None: - await self.router_info() - - # If version is 4.8 or greater use vpn-client otherwise use wg-client - target_call = "vpn-client" if self._firmware_version >= NEW_VPN_CLIENT_VERSION else "wg-client" - - response = await self._request( - self.gen_sid_payload("call", [target_call, "get_status"], self.sid) + if self._firmware_version is None: + await self.router_info() + + # If version is 4.8 or greater use vpn-client otherwise use wg-client + target_call = "vpn-client" if self._firmware_version >= NEW_VPN_CLIENT_VERSION else "wg-client" + + response = await self._request( + self.gen_sid_payload("call", [target_call, "get_status"], self.sid) ) - - if self._firmware_version < NEW_VPN_CLIENT_VERSION: - # If the version is less than 4.8 we need to adjust the response to match the new format - # The old format does not return an array, but just a single object. - # We will wrap it in an array to match the new format. - return [response] - - return response.get("status_list", []) - - async def wireguard_client_start(self, group_id: int, peer_id: int, tunnel_id: int) -> dict: - """Starts a WireGuard client with the specified tunnel ID.""" - return await self._wireguard_set_client_enabled(group_id, peer_id, tunnel_id, True) - - async def wireguard_client_stop(self, tunnel_id: int) -> dict: - """Stops the WireGuard client with the specified tunnel ID.""" - # Pass -1 for group_id and peer_id as they are not needed to stop the client - return await self._wireguard_set_client_enabled(-1, -1, tunnel_id, False) - - async def _wireguard_set_client_enabled(self, group_id: int, peer_id: int, tunnel_id: int, enabled: bool) -> dict: - """Sets the WireGuard client enabled state.""" - if self._firmware_version is None: - await self.router_info() - - # If version is 4.8 or greater use vpn-client otherwise use wg-client - if self._firmware_version >= NEW_VPN_CLIENT_VERSION: - return await self._request( + + if self._firmware_version < NEW_VPN_CLIENT_VERSION: + # If the version is less than 4.8 we need to adjust the response to match the new format + # The old format does not return an array, but just a single object. + # We will wrap it in an array to match the new format. + return [response] + + return response.get("status_list", []) + + async def wireguard_client_start( + self, group_id: int, peer_or_tunnel_id: int + ) -> dict: + """Starts a WireGuard client with the specified tunnel ID.""" + return await self._wireguard_set_client_enabled( + group_id, peer_or_tunnel_id, True + ) + + async def wireguard_client_stop(self, peer_or_tunnel_id: int) -> dict: + """Stops the WireGuard client with the specified tunnel ID.""" + # Pass -1 for group_id and peer_id as they are not needed to stop the client + return await self._wireguard_set_client_enabled(-1, peer_or_tunnel_id, False) + + async def _wireguard_set_client_enabled( + self, group_id: int, peer_or_tunnel_id: int, enabled: bool + ) -> dict: + """Sets the WireGuard client enabled state.""" + if self._firmware_version is None: + await self.router_info() + + # If version is 4.8 or greater use vpn-client otherwise use wg-client + if self._firmware_version >= NEW_VPN_CLIENT_VERSION: + tunnel_id = peer_or_tunnel_id + return await self._request( self.gen_sid_payload( - "call", - [ - "vpn-client", - "set_tunnel", - {"enabled": enabled, "tunnel_id": tunnel_id}, - ], - self.sid, - ) - ) - - # Not version 4.8 or greater so use wg-client - # If enabled, call the start method with group_id and peer_id - if enabled: - return await self._request( - self.gen_sid_payload( - "call", - ["wg-client", "start", {"group_id": group_id, "peer_id": peer_id}], - self.sid, - ) - ) - - # Not enabled, call the stop method - return await self._request( - self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) + "call", + [ + "vpn-client", + "set_tunnel", + {"enabled": enabled, "tunnel_id": tunnel_id}, + ], + self.sid, + ) + ) + + # Not version 4.8 or greater so use wg-client + # If enabled, call the start method with group_id and peer_id + peer_id = peer_or_tunnel_id + if enabled: + return await self._request( + self.gen_sid_payload( + "call", + ["wg-client", "start", {"group_id": group_id, "peer_id": peer_id}], + self.sid, + ) + ) + + # Not enabled, call the stop method + return await self._request( + self.gen_sid_payload("call", ["wg-client", "stop"], self.sid) ) async def _tailscale_get_config(self) -> dict | bool: From bd08ceb45205432026b8a126689bc39fba3cd5ab Mon Sep 17 00:00:00 2001 From: HarvsG <11440490+HarvsG@users.noreply.github.com> Date: Fri, 26 Sep 2025 14:11:51 +0000 Subject: [PATCH 21/21] make tunnel ID optional --- tests/test_glinet.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_glinet.py b/tests/test_glinet.py index 992b14e..3c3cb35 100644 --- a/tests/test_glinet.py +++ b/tests/test_glinet.py @@ -213,9 +213,9 @@ async def test_wireguard_start() -> None: first_status = status_list[0] group_id = first_status["group_id"] peer_id = first_status["peer_id"] - tunnel_id = first_status["tunnel_id"] + tunnel_id = first_status.get("tunnel_id") - result = await router.wireguard_client_start(group_id, peer_id, tunnel_id) + result = await router.wireguard_client_start(group_id, tunnel_id or peer_id) print("RESULT: ", result) assert result["tunnel_id"] == tunnel_id