NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-09-30 07:40:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2928 from markparonyan/mypy-utils

Browse Source 
fix(typing): resolve mypy errors in gns3server.utils
This commit is contained in:
Jeremy Grossmann 2026-09-29 13:42:11 +02:00 committed by GitHub
commit 5f9fd6e595
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 32 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
gns3server/utils/http_client.py
View File

@ -16,6 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import aiohttp
from pydantic import SecretStr
import socket
import logging
@ -28,7 +29,7 @@ class HTTPClient:
HTTP client for request to computes and external services.
"""
_aiohttp_client: aiohttp.ClientSession = None
_aiohttp_client: aiohttp.ClientSession | None = None
@classmethod
def get_client(cls, ssl_context=None) -> aiohttp.ClientSession:
@ -44,15 +45,21 @@ class HTTPClient:
cls._aiohttp_client = None
@classmethod
def request(cls, method: str, url: str, user: str = None, password: str = None, ssl_context=None, **kwargs):
def request(
cls,
method: str,
url: str,
user: str | None = None,
password: SecretStr | None = None,
ssl_context=None,
**kwargs,
):
client = cls.get_client(ssl_context=ssl_context)
basic_auth = None
if user:
if not password:
password = ""
try:
basic_auth = aiohttp.BasicAuth(user, password.get_secret_value(), "utf-8")
basic_auth = aiohttp.BasicAuth(user, password.get_secret_value() if password else "", "utf-8")
except ValueError as e:
log.error(f"Basic authentication set-up error: {e}")

47
gns3server/utils/websocket_to_websocket.py
View File

@ -25,7 +25,6 @@ Similar pattern to VNC console implementation in base_node.py:
import asyncio
import logging
import sys
from typing import Optional
import aiohttp
@ -39,7 +38,7 @@ log = logging.getLogger(__name__)
async def websocket_proxy(
client_ws: WebSocket,
target_url: str,
requested_protocols: list = None,
requested_protocols: Optional[list] = None,
buffer_size: int = 65536,
timeout: Optional[float] = None,
) -> None:
@ -59,7 +58,7 @@ async def websocket_proxy(
Raises:
aiohttp.ClientError: If connection to target fails
"""
client_info = f"{client_ws.client.host}:{client_ws.client.port}"
client_info = f"{client_ws.client.host}:{client_ws.client.port}" if client_ws.client else "unknown"
async def forward_client_to_target(target_ws):
"""Client → Target: Forward binary WebSocket data."""
@ -101,25 +100,20 @@ async def websocket_proxy(
subprotocols = requested_protocols or ["binary"]
log.info(f"Client requested subprotocols: {subprotocols}")
timeout_config = {}
if timeout:
timeout_config = {"timeout": aiohttp.ClientTimeout(total=timeout)}
session_timeout = aiohttp.ClientTimeout(total=timeout) if timeout else None
async with aiohttp.ClientSession() as session:
async with session.ws_connect(target_url, protocols=subprotocols, **timeout_config) as target_ws:
async with aiohttp.ClientSession(timeout=session_timeout) as session:
async with session.ws_connect(target_url, protocols=subprotocols) as target_ws:
negotiated_protocol = target_ws.protocol
log.info(f"Target WebSocket negotiated protocol: {negotiated_protocol}")
log.info(f"WebSocket proxy established: {client_info} → {target_url}")
# Run both forwarding tasks in parallel
# Similar pattern to base_node.py VNC implementation
if sys.version_info >= (3, 11, 0):
aws = [
asyncio.create_task(forward_client_to_target(target_ws)),
asyncio.create_task(forward_target_to_client(target_ws)),
]
else:
aws = [forward_client_to_target(target_ws), forward_target_to_client(target_ws)]
aws = [
asyncio.create_task(forward_client_to_target(target_ws)),
asyncio.create_task(forward_target_to_client(target_ws)),
]
try:
done, pending = await asyncio.wait(aws, return_when=asyncio.FIRST_COMPLETED)
@ -146,7 +140,7 @@ async def websocket_proxy(
async def websocket_proxy_with_manual_accept(
client_ws: StarletteWebSocket,
target_url: str,
requested_protocols: list = None,
requested_protocols: Optional[list] = None,
buffer_size: int = 65536,
timeout: Optional[float] = None,
) -> None:
@ -167,7 +161,7 @@ async def websocket_proxy_with_manual_accept(
Raises:
aiohttp.ClientError: If connection to target fails
"""
client_info = f"{client_ws.client.host}:{client_ws.client.port}" if hasattr(client_ws, "client") else "unknown"
client_info = f"{client_ws.client.host}:{client_ws.client.port}" if client_ws.client else "unknown"
async def forward_client_to_target(target_ws):
"""Client → Target: Forward binary WebSocket data."""
@ -213,13 +207,11 @@ async def websocket_proxy_with_manual_accept(
subprotocols = requested_protocols or ["binary"]
log.info(f"Client requested subprotocols: {subprotocols}")
timeout_config = {}
if timeout:
timeout_config = {"timeout": aiohttp.ClientTimeout(total=timeout)}
session_timeout = aiohttp.ClientTimeout(total=timeout) if timeout else None
async with aiohttp.ClientSession() as session:
async with aiohttp.ClientSession(timeout=session_timeout) as session:
log.info(f"About to ws_connect to {target_url}")
ws_conn = session.ws_connect(target_url, protocols=subprotocols, **timeout_config)
ws_conn = session.ws_connect(target_url, protocols=subprotocols)
log.info("ws_connect coroutine created, about to enter")
async with ws_conn as target_ws:
negotiated_protocol = target_ws.protocol
@ -235,13 +227,10 @@ async def websocket_proxy_with_manual_accept(
log.info(f"WebSocket proxy established: {client_info} → {target_url}")
# Run both forwarding tasks in parallel
if sys.version_info >= (3, 11, 0):
aws = [
asyncio.create_task(forward_client_to_target(target_ws)),
asyncio.create_task(forward_target_to_client(target_ws)),
]
else:
aws = [forward_client_to_target(target_ws), forward_target_to_client(target_ws)]
aws = [
asyncio.create_task(forward_client_to_target(target_ws)),
asyncio.create_task(forward_target_to_client(target_ws)),
]
log.info(f"About to call asyncio.wait with {len(aws)} tasks")
try:

3
gns3server/utils/zipfile_zstd/_patcher.py
View File

@ -1,8 +1,9 @@
import functools
from typing import Any
class patch:
originals = {}
originals: dict[str, Any] = {}
def __init__(self, host, name):
self.host = host

3
pyproject.toml
View File

@ -313,8 +313,5 @@ module = [
"gns3server.db.repositories.users", # 12
"gns3server.db.tasks", # 12
"gns3server.schemas.controller.computes", # 6
"gns3server.utils.http_client", # 4
"gns3server.utils.websocket_to_websocket", # 12
"gns3server.utils.zipfile_zstd._patcher", # 1
]
ignore_errors = true