Skip to content

Commit

Permalink
stronger type hints
Browse files Browse the repository at this point in the history
  • Loading branch information
totaam committed Mar 9, 2025
1 parent 80b7b47 commit c78ee53
Show file tree
Hide file tree
Showing 3 changed files with 16 additions and 14 deletions.
14 changes: 8 additions & 6 deletions xpra/platform/win32/proxy_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

import os
from time import sleep
from typing import Any
from collections.abc import Sequence

from xpra.server.proxy.server import ProxyServer as _ProxyServer, get_proxy_env
Expand Down Expand Up @@ -59,12 +60,13 @@ def exec_command(username: str, password: str, args: Sequence[str], exe: str, cw

class ProxyServer(_ProxyServer):

def start_new_session(self, username, password, uid, gid, new_session_dict=None, displays=()):
log("start_new_session%s", (username, "..", uid, gid, new_session_dict, displays))
return self.start_win32_shadow(username, password, new_session_dict)
def start_new_session(self, username: str, password: str, uid: int, gid: int,
sess_options: dict, displays=()) -> tuple[Any, str, str]:
log("start_new_session%s", (username, "..", uid, gid, sess_options, displays))
return self.start_win32_shadow(username, password, sess_options)

def start_win32_shadow(self, username, password, new_session_dict):
log("start_win32_shadow%s", (username, "..", new_session_dict))
def start_win32_shadow(self, username: str, password: str, sess_options: dict) -> tuple[Any, str, str]:
log("start_win32_shadow%s", (username, "..", sess_options))
# pylint: disable=import-outside-toplevel
from xpra.platform.win32.wtsapi import find_session
session_info = find_session(username)
Expand Down Expand Up @@ -107,7 +109,7 @@ def start_win32_shadow(self, username, password, new_session_dict):
# "--tray=no",
]
# unless explicitly stated otherwise, exit with client:
if new_session_dict.get("exit-with-client", None) is not False:
if sess_options.get("exit-with-client", None) is not False:
cmd.append("--exit-with-client=yes")
from xpra.log import debug_enabled_categories
if debug_enabled_categories:
Expand Down
6 changes: 3 additions & 3 deletions xpra/platform/win32/wtsapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ class WTS_CLIENT_DISPLAY(Structure):
WTSReset = 7
WTSDown = 8
WTSInit = 9
CONNECT_STATE = {
CONNECT_STATE: dict[int, str] = {
WTSActive: "Active",
WTSConnected: "Connected",
WTSConnectQuery: "ConnectQuery",
Expand Down Expand Up @@ -96,7 +96,7 @@ class WTS_CLIENT_DISPLAY(Structure):
WTSValidationInfo = 27
WTSSessionAddressV4 = 28
WTSIsRemoteSession = 29
WTS_INFO_CLASS = {
WTS_INFO_CLASS: dict[int, str] = {
WTSInitialProgram: "InitialProgram",
WTSApplicationName: "ApplicationName",
WTSWorkingDirectory: "WorkingDirectory",
Expand Down Expand Up @@ -230,7 +230,7 @@ def get_sessions() -> dict[int, dict[str, Any]]:
return sessions


def find_session(username, with_display=True) -> dict:
def find_session(username: str, with_display=True) -> dict:
if username:
for sid, info in get_sessions().items():
if with_display and not info.get("Display"):
Expand Down
10 changes: 5 additions & 5 deletions xpra/server/proxy/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,7 @@ def nosession(*extras) -> None:
proc = None
socket_path = None
display = None
sns = typedict(c.dictget("start-new-session", {}))
sns = c.dictget("start-new-session", {})
authlog("proxy_session: displays=%s, start_sessions=%s, start-new-session=%s",
displays, self._start_sessions, sns)
if not displays or sns:
Expand Down Expand Up @@ -473,7 +473,7 @@ def nosession(*extras) -> None:
if socket_path:
hello["socket-path"] = socket_path
# echo mode if present:
mode = sns.strget("mode")
mode = str(sns.get("mode", ""))
if mode:
hello["mode"] = mode
client_proto.send_now(("hello", hello))
Expand Down Expand Up @@ -603,9 +603,9 @@ def unexpected_packet(packet) -> None:
start_thread(start_proxy_process, f"start_proxy({client_proto})")

def start_new_session(self, username: str, _password, uid: int, gid: int,
new_session_dict=None, displays=()) -> tuple[Any, str, str]:
log("start_new_session%s", (username, "..", uid, gid, new_session_dict, displays))
sns = typedict(new_session_dict or {})
sess_options: dict, displays=()) -> tuple[Any, str, str]:
log("start_new_session%s", (username, "..", uid, gid, sess_options, displays))
sns = typedict(sess_options or {})
mode = sns.strget("mode", "start")
mode = MODE_ALIAS.get(mode, mode)
if mode not in ("seamless", "desktop", "shadow", "monitor", "expand"):
Expand Down

0 comments on commit c78ee53

Please # to comment.