#3930 don't import from 'typing' if we don't have to

tests/unittests/unit/server/auth_test.py

@@ -13,7 +13,7 @@
 import uuid
 import hmac
 from time import monotonic
-from typing import Tuple, Callable
+from typing import Callable
 
 from xpra.os_util import (
     strtobytes, bytestostr,
@@ -279,7 +279,7 @@ def genfiledata(_a):
             a.load_password_file()
 
     def test_multifile(self):
-        def genfiledata(a) -> Tuple[str,str]:
+        def genfiledata(a) -> tuple[str,str]:
             password = uuid.uuid4().hex
             lines = [
                 "#comment",
@@ -292,7 +292,7 @@ def genfiledata(a) -> Tuple[str,str]:
                 ]
             return password, "\n".join(lines)
         self._test_file_auth("multifile", genfiledata, 1)
-        def nodata(_a) -> Tuple[str,str]:
+        def nodata(_a) -> tuple[str,str]:
             return "abc", ""
         try:
             self._test_file_auth("multifile", nodata, 1)

tests/unittests/unit/server/server_sockets_test.py

@@ -10,7 +10,6 @@
 import tempfile
 from time import monotonic
 from subprocess import Popen
-from typing import List
 
 from xpra.util import repr_ellipsized, envint
 from xpra.os_util import load_binary_file, pollwait, OSX, POSIX
@@ -248,7 +247,7 @@ def test_ssl(self):
                     openssl = self.run_command(openssl_verify_command, stdin=devnull, shell=True)
                     r = pollwait(openssl, 10)
                     assert r==0, "openssl certificate verification failed, returned %s" % r
-                errors : List[str] = []
+                errors : list[str] = []
                 def tc(mode:str, port:int):
                     uri = f"{mode}://foo:[email protected]:{port}/"
                     stype = ports_proto.get(port, "").rjust(5)

tests/unittests/unit/server_test_util.py

@@ -8,7 +8,6 @@
 import time
 import tempfile
 import subprocess
-from typing import List, Dict
 
 from unit.process_test_util import ProcessTestUtil
 from xpra.util import envint
@@ -48,7 +47,7 @@ def setUpClass(cls):
         ProcessTestUtil.setUpClass()
         tmpdir = tempfile.gettempdir()
         cls.dotxpra = DotXpra(tmpdir, [tmpdir])
-        cls.default_xpra_args : List[str] = ["--speaker=no", "--microphone=no"]
+        cls.default_xpra_args : list[str] = ["--speaker=no", "--microphone=no"]
         if not WIN32:
             cls.default_xpra_args += ["--systemd-run=no", "--pulseaudio=no"]
             for x in cls.dotxpra._sockdirs:
@@ -89,7 +88,7 @@ def run_xpra(self, xpra_args, **kwargs):
 
 
     @classmethod
-    def get_xpra_cmd(cls) -> List[str]:
+    def get_xpra_cmd(cls) -> list[str]:
         return ProcessTestUtil.get_xpra_cmd() + cls.default_xpra_args
 
 
@@ -100,7 +99,7 @@ def run_server(self, *args):
         return server
 
     def check_fast_start_server(self, display:str, *args):
-        defaults : Dict[str,str] = dict((k, "no") for k in (
+        defaults : dict[str,str] = dict((k, "no") for k in (
             "av-sync", "remote-logging",
             "windows",
             "mdns",
@@ -135,7 +134,7 @@ def check_server(self, subcommand:str, display:str, *args):
             self.show_proc_error(server_proc, "server failed to start")
         if display:
             #wait until the socket shows up:
-            live : List[str] = []
+            live : list[str] = []
             for _ in range(20):
                 live = self.dotxpra.displays()
                 if display in live:

tests/unittests/unit/util_test.py

@@ -56,7 +56,7 @@ def test_MutableInteger(self):
         self._test_IntegerClass(MutableInteger)
 
 
-class TestTypeDict(unittest.TestCase):
+class TestTypedict(unittest.TestCase):
 
     def test_typedict(self):
         d = typedict({

tests/xpra/test_apps/screensaver_listener.py

@@ -17,7 +17,7 @@
 
 def main():
     from xpra.platform import program_context
-    with program_context("ScreenSaver-Listener", "ScreenSaver Listener"):
+    with program_context("ScreenSaver-listener", "ScreenSaver listener"):
         dbus_session = dbus.SessionBus()
         def active_changed(active):
             print("screensaver active status changed: %s" % (active, ))

xpra/audio/audio_pipeline.py

@@ -7,7 +7,7 @@
 #pylint: disable=wrong-import-position
 
 import os
-from typing import Dict, Any
+from typing import Any
 
 from xpra.util import AtomicInteger
 from xpra.os_util import register_SIGUSR_signals
@@ -43,7 +43,7 @@ def __init__(self, codec):
         self.byte_count = 0
         self.emit_info_timer = 0
         self.volume = None
-        self.info : Dict[str,Any] = {
+        self.info : dict[str,Any] = {
                      "codec"        : self.codec,
                      "state"        : self.state,
                      }

xpra/audio/common.py

@@ -4,8 +4,6 @@
 # Xpra is released under the terms of the GNU GPL v2, or, at your option, any
 # later version. See the file COPYING for details.
 
-from typing import Tuple
-
 from xpra.util import csv, engs
 from xpra.log import Logger
 log = Logger("audio")
@@ -46,7 +44,7 @@
 
 
 #used for parsing codec names specified on the command line:
-def audio_option_or_all(name:str, options, all_values:Tuple[str,...]) -> Tuple[str,...]:
+def audio_option_or_all(name:str, options, all_values:tuple[str,...]) -> tuple[str,...]:
     log("audio_option_or_all%s", (name, options, all_values))
     if not options:
         v = list(all_values)              #not specified on command line: use default

xpra/audio/gstreamer_util.py

@@ -6,7 +6,7 @@
 
 import sys
 import os
-from typing import Dict, Tuple, List, Any, Callable
+from typing import Any, Callable
 
 from xpra.gst_common import (
     has_plugins, get_all_plugin_names,
@@ -92,7 +92,7 @@ def force_enabled(codec_name):
 #format: encoder, container-formatter, decoder, container-parser, stream-compressor
 #we keep multiple options here for the same encoding
 #and will populate the ones that are actually available into the "CODECS" dict
-CODEC_OPTIONS : Tuple[Tuple[str,str,str,str,str,str],...] = (
+CODEC_OPTIONS : tuple[tuple[str,str,str,str,str,str],...] = (
         (VORBIS_MKA , "vorbisenc",      "matroskamux",  "vorbisdec",                    "matroskademux",    ""),
         (VORBIS_MKA , "vorbisenc",      "webmmux",      "vorbisdec",                    "matroskademux",    ""),
         #those two used to fail silently (older versions of gstreamer?)
@@ -113,7 +113,7 @@ def force_enabled(codec_name):
         (AAC_MPEG4  , "voaacenc",       "mp4mux",       "faad",                         "qtdemux",          ""),
             )
 
-MUX_OPTIONS : Tuple[Tuple[str,str,str], ...] = (
+MUX_OPTIONS : tuple[tuple[str,str,str], ...] = (
                (OGG,    "oggmux",   "oggdemux"),
                (MKA,    "webmmux",  "matroskademux"),
                (MKA,    "matroskamux",  "matroskademux"),
@@ -143,7 +143,7 @@ def force_enabled(codec_name):
 
 #options we use to tune for low latency:
 OGG_DELAY = 20*MS_TO_NS
-ENCODER_DEFAULT_OPTIONS_COMMON : Dict[str,Dict[str,Any]] = {
+ENCODER_DEFAULT_OPTIONS_COMMON : dict[str,dict[str,Any]] = {
             "lamemp3enc"    : {
                                "encoding-engine-quality" : 0,
                                },   #"fast"
@@ -160,15 +160,15 @@ def force_enabled(codec_name):
                                },
             #"vorbisenc"     : {"perfect-timestamp" : 1},
                            }
-ENCODER_DEFAULT_OPTIONS : Dict[str,Dict[str,Any]] = {
+ENCODER_DEFAULT_OPTIONS : dict[str,dict[str,Any]] = {
     "opusenc"       : {
         #only available with 1.6 onwards?
         "bitrate-type"   : 1,      #vbr
         "complexity"     : 0
         },
     }
 #we may want to review this if/when we implement UDP transport:
-MUXER_DEFAULT_OPTIONS : Dict[str,Dict[str,Any]] = {
+MUXER_DEFAULT_OPTIONS : dict[str,dict[str,Any]] = {
             "oggmux"        : {
                                "max-delay"          : OGG_DELAY,
                                "max-page-delay"     : OGG_DELAY,
@@ -192,7 +192,7 @@ def force_enabled(codec_name):
 
 #based on the encoder options above:
 RECORD_PIPELINE_LATENCY = 25
-ENCODER_LATENCY : Dict[str,int] = {
+ENCODER_LATENCY : dict[str,int] = {
         VORBIS      : 0,
         VORBIS_OGG  : 0,
         VORBIS_MKA  : 0,
@@ -216,16 +216,16 @@ def force_enabled(codec_name):
     )
 
 
-def get_encoder_default_options(encoder:str) -> Dict[str,Any]:
+def get_encoder_default_options(encoder:str) -> dict[str,Any]:
     #strip the muxer:
     enc = encoder.split("+")[0]
     options = ENCODER_DEFAULT_OPTIONS_COMMON.get(enc, {}).copy()
     options.update(ENCODER_DEFAULT_OPTIONS.get(enc, {}))
     return options
 
 
-CODECS : Dict[str,bool] = {}
-CODEC_DEFS = Dict[str, Tuple[Any,Any,Any]]
+CODECS : dict[str,bool] = {}
+CODEC_DEFS = dict[str, tuple[Any,Any,Any]]
 ENCODERS : CODEC_DEFS = {}       #(encoder, payloader, stream-compressor)
 DECODERS : CODEC_DEFS = {}       #(decoder, depayloader, stream-compressor)
 
@@ -316,24 +316,24 @@ def has_stream_compressor(stream_compressor:str) -> bool:
         return False
     return True
 
-def get_muxers() -> List[str]:
+def get_muxers() -> list[str]:
     muxers = []
     for name,muxer,_ in MUX_OPTIONS:
         if has_plugins(muxer):
             muxers.append(name)
     return muxers
 
-def get_demuxers() -> List[str]:
+def get_demuxers() -> list[str]:
     demuxers = []
     for name,_,demuxer in MUX_OPTIONS:
         if has_plugins(demuxer):
             demuxers.append(name)
     return demuxers
 
-def get_stream_compressors() -> List[str]:
+def get_stream_compressors() -> list[str]:
     return [x for x in ("lz4", ) if has_stream_compressor(x)]
 
-def get_encoder_elements(name) -> Tuple[str,str,str]:
+def get_encoder_elements(name) -> tuple[str,str,str]:
     encoders = get_encoders()
     enc = encoders.get(name)
     if enc is None:
@@ -347,7 +347,7 @@ def get_encoder_elements(name) -> Tuple[str,str,str]:
         raise RuntimeError(f"formatter {formatter} not found")
     return encoder, formatter, stream_compressor
 
-def get_decoder_elements(name) -> Tuple[str,str,str]:
+def get_decoder_elements(name) -> tuple[str,str,str]:
     decoders = get_decoders()
     dec = decoders.get(name)
     if dec is None:
@@ -380,14 +380,14 @@ def has_decoder(name) -> bool:
 def has_codec(name) -> bool:
     return has_encoder(name) and has_decoder(name)
 
-def can_encode() -> List[str]:
+def can_encode() -> list[str]:
     return [x for x in CODEC_ORDER if has_encoder(x)]
 
-def can_decode() -> List[str]:
+def can_decode() -> list[str]:
     return [x for x in CODEC_ORDER if has_decoder(x)]
 
 
-def get_source_plugins() -> List[str]:
+def get_source_plugins() -> list[str]:
     sources = []
     if POSIX and not OSX:
         try:
@@ -436,7 +436,7 @@ def get_default_source() -> str:
             return source
     return ""
 
-def get_sink_plugins() -> List[str]:
+def get_sink_plugins() -> list[str]:
     SINKS = []
     if OSX:
         SINKS.append("osxaudiosink")
@@ -480,12 +480,12 @@ def get_default_sink_plugin() -> str:
     return ""
 
 
-def get_test_defaults(*_args) -> Dict[str,Any]:
+def get_test_defaults(*_args) -> dict[str,Any]:
     return  {"wave" : 2, "freq" : 110, "volume" : 0.4}
 
 WARNED_MULTIPLE_DEVICES = False
 def get_pulse_defaults(device_name_match=None, want_monitor_device=True,
-                       input_or_output=None, remote=None, env_device_name=None) -> Dict[str,Any]:
+                       input_or_output=None, remote=None, env_device_name=None) -> dict[str,Any]:
     try:
         device = get_pulse_device(device_name_match, want_monitor_device, input_or_output, remote, env_device_name)
     except Exception as e:
@@ -648,16 +648,16 @@ def get_pulse_device(device_name_match=None, want_monitor_device=True,
     log.info(" '%s'", bytestostr(device_name))
     return device
 
-def get_pulse_source_defaults(device_name_match=None, want_monitor_device=True, remote=None) -> Dict[str,Any]:
+def get_pulse_source_defaults(device_name_match=None, want_monitor_device=True, remote=None) -> dict[str,Any]:
     return get_pulse_defaults(device_name_match, want_monitor_device,
                               input_or_output=not want_monitor_device, remote=remote,
                               env_device_name=XPRA_PULSE_SOURCE_DEVICE_NAME)
 
-def get_pulse_sink_defaults() -> Dict[str,Any]:
+def get_pulse_sink_defaults() -> dict[str,Any]:
     return get_pulse_defaults(want_monitor_device=False, input_or_output=False,
                               env_device_name=XPRA_PULSE_SINK_DEVICE_NAME)
 
-def get_directsound_source_defaults(device_name_match=None, want_monitor_device=True, remote=None) -> Dict[str,Any]:
+def get_directsound_source_defaults(device_name_match=None, want_monitor_device=True, remote=None) -> dict[str,Any]:
     try:
         from xpra.platform.win32.directsound import get_devices, get_capture_devices
         if not want_monitor_device:
@@ -706,11 +706,11 @@ def get_directsound_source_defaults(device_name_match=None, want_monitor_device=
 #a list of functions to call to get the plugin options
 #at runtime (so we can perform runtime checks on remote data,
 # to avoid audio loops for example)
-DEFAULT_SRC_PLUGIN_OPTIONS : Dict[str,Callable] = {
+DEFAULT_SRC_PLUGIN_OPTIONS : dict[str,Callable] = {
     "test"                  : get_test_defaults,
     "direct"                : get_directsound_source_defaults,
     }
-DEFAULT_SINK_PLUGIN_OPTIONS : Dict[str,Any] = {}
+DEFAULT_SINK_PLUGIN_OPTIONS : dict[str,Any] = {}
 if POSIX and not OSX:
     DEFAULT_SINK_PLUGIN_OPTIONS["pulse"] = get_pulse_sink_defaults
     DEFAULT_SRC_PLUGIN_OPTIONS["pulse"] = get_pulse_source_defaults
@@ -779,7 +779,7 @@ def parse_audio_source(all_plugins, audio_source_plugin, device, want_monitor_de
     return gst_audio_source_plugin, options
 
 
-def loop_warning_messages(mode="speaker") -> List[str]:
+def loop_warning_messages(mode="speaker") -> list[str]:
     return [
         "Cannot start %s forwarding:" % mode,
         "client and server environment are identical,",

xpra/audio/pulseaudio/pulseaudio_none_util.py

@@ -5,7 +5,7 @@
 # later version. See the file COPYING for details.
 
 import sys
-from typing import Dict, Any
+from typing import Any
 
 from xpra.log import Logger
 log = Logger("audio")
@@ -36,10 +36,10 @@ def get_pulse_cookie_hash() -> str:
 def get_pactl_server() -> str:
     return ""
 
-def get_pa_device_options(*_args) -> Dict[str,Any]:
+def get_pa_device_options(*_args) -> dict[str,Any]:
     return {}
 
-def get_info() -> Dict[str,Any]:
+def get_info() -> dict[str,Any]:
     return {
             "pulseaudio.wrapper": "none",
             "pulseaudio.found"  : has_pa(),