diff --git a/CHANGES b/CHANGES index 71c828ceb66..35ed7914e3e 100644 --- a/CHANGES +++ b/CHANGES @@ -19,6 +19,10 @@ $ pipx install --suffix=@next 'tmuxp' --pip-args '\--pre' --force +### Breaking changes + +- Type annotations: Add strict mypy typings (#796) + ## tmuxp 1.22.1 (2022-12-27) _Maintenance only, no bug fixes or features_ diff --git a/src/tmuxp/cli/convert.py b/src/tmuxp/cli/convert.py index 6236b58551a..d10e5bd4359 100644 --- a/src/tmuxp/cli/convert.py +++ b/src/tmuxp/cli/convert.py @@ -8,6 +8,11 @@ from .utils import prompt_yes_no +if t.TYPE_CHECKING: + from typing_extensions import Literal + + AllowedFileTypes = Literal["json", "yaml"] + def create_convert_subparser( parser: argparse.ArgumentParser, @@ -50,6 +55,7 @@ def command_convert( _, ext = os.path.splitext(workspace_file) ext = ext.lower() + to_filetype: "AllowedFileTypes" if ext == ".json": to_filetype = "yaml" elif ext in [".yaml", ".yml"]: @@ -60,8 +66,11 @@ def command_convert( configparser = ConfigReader.from_file(workspace_file) newfile = workspace_file.parent / (str(workspace_file.stem) + f".{to_filetype}") - export_kwargs = {"default_flow_style": False} if to_filetype == "yaml" else {} - new_workspace = configparser.dump(format=to_filetype, indent=2, **export_kwargs) + new_workspace = configparser.dump( + format=to_filetype, + indent=2, + **{"default_flow_style": False} if to_filetype == "yaml" else {}, + ) if not answer_yes: if prompt_yes_no(f"Convert to <{workspace_file}> to {to_filetype}?"): diff --git a/src/tmuxp/cli/load.py b/src/tmuxp/cli/load.py index 00dde4e0b1c..41fbf4a869a 100644 --- a/src/tmuxp/cli/load.py +++ b/src/tmuxp/cli/load.py @@ -107,7 +107,7 @@ def set_layout_hook(session: Session, hook_name: str) -> None: session.cmd(*cmd) -def load_plugins(sconf: t.Any) -> t.List[t.Any]: +def load_plugins(sconf: t.Dict[str, t.Any]) -> t.List[t.Any]: """ Load and return plugins in workspace """ @@ -158,6 +158,7 @@ def _reattach(builder: WorkspaceBuilder): If not, ``tmux attach-session`` loads the client to the target session. """ + assert builder.session is not None for plugin in builder.plugins: plugin.reattach(builder.session) proc = builder.session.cmd("display-message", "-p", "'#S'") @@ -181,6 +182,7 @@ def _load_attached(builder: WorkspaceBuilder, detached: bool) -> None: detached : bool """ builder.build() + assert builder.session is not None if "TMUX" in os.environ: # tmuxp ran from inside tmux # unset TMUX, save it, e.g. '/tmp/tmux-1000/default,30668,0' @@ -214,6 +216,8 @@ def _load_detached(builder: WorkspaceBuilder) -> None: """ builder.build() + assert builder.session is not None + if has_gte_version("2.6"): # prepare for both cases set_layout_hook(builder.session, "client-attached") set_layout_hook(builder.session, "client-session-changed") @@ -231,6 +235,7 @@ def _load_append_windows_to_current_session(builder: WorkspaceBuilder) -> None: """ current_attached_session = builder.find_current_attached_session() builder.build(current_attached_session, append=True) + assert builder.session is not None if has_gte_version("2.6"): # prepare for both cases set_layout_hook(builder.session, "client-attached") set_layout_hook(builder.session, "client-session-changed") @@ -244,6 +249,7 @@ def _setup_plugins(builder: WorkspaceBuilder) -> Session: ---------- builder: :class:`workspace.builder.WorkspaceBuilder` """ + assert builder.session is not None for plugin in builder.plugins: plugin.before_script(builder.session) @@ -458,8 +464,9 @@ def load_workspace( ) if choice == "k": - builder.session.kill_session() - tmuxp_echo("Session killed.") + if builder.session is not None: + builder.session.kill_session() + tmuxp_echo("Session killed.") elif choice == "a": _reattach(builder) else: diff --git a/src/tmuxp/config_reader.py b/src/tmuxp/config_reader.py index 4b64d127b59..6f3683c6dbb 100644 --- a/src/tmuxp/config_reader.py +++ b/src/tmuxp/config_reader.py @@ -22,11 +22,11 @@ class ConfigReader: '{\n "session_name": "my session"\n}' """ - def __init__(self, content: "RawConfigData"): + def __init__(self, content: "RawConfigData") -> None: self.content = content @staticmethod - def _load(format: "FormatLiteral", content: str): + def _load(format: "FormatLiteral", content: str) -> t.Dict[str, t.Any]: """Load raw config data and directly return it. >>> ConfigReader._load("json", '{ "session_name": "my session" }') @@ -46,7 +46,7 @@ def _load(format: "FormatLiteral", content: str): raise NotImplementedError(f"{format} not supported in configuration") @classmethod - def load(cls, format: "FormatLiteral", content: str): + def load(cls, format: "FormatLiteral", content: str) -> "ConfigReader": """Load raw config data into a ConfigReader instance (to dump later). >>> cfg = ConfigReader.load("json", '{ "session_name": "my session" }') @@ -69,7 +69,7 @@ def load(cls, format: "FormatLiteral", content: str): ) @classmethod - def _from_file(cls, path: pathlib.Path): + def _from_file(cls, path: pathlib.Path) -> t.Dict[str, t.Any]: r"""Load data from file path directly to dictionary. **YAML file** @@ -114,7 +114,7 @@ def _from_file(cls, path: pathlib.Path): ) @classmethod - def from_file(cls, path: pathlib.Path): + def from_file(cls, path: pathlib.Path) -> "ConfigReader": r"""Load data from file path **YAML file** diff --git a/src/tmuxp/exc.py b/src/tmuxp/exc.py index 3ce3c73ee6b..e8801e19f0c 100644 --- a/src/tmuxp/exc.py +++ b/src/tmuxp/exc.py @@ -4,6 +4,8 @@ ~~~~~~~~~ """ +import typing as t + from ._compat import implements_to_string @@ -28,7 +30,7 @@ class TmuxpPluginException(TmuxpException): class BeforeLoadScriptNotExists(OSError): - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) self.strerror = "before_script file '%s' doesn't exist." % self.strerror @@ -41,7 +43,9 @@ class BeforeLoadScriptError(Exception): :meth:`tmuxp.util.run_before_script`. """ - def __init__(self, returncode, cmd, output=None): + def __init__( + self, returncode: int, cmd: str, output: t.Optional[str] = None + ) -> None: self.returncode = returncode self.cmd = cmd self.output = output diff --git a/src/tmuxp/log.py b/src/tmuxp/log.py index 4f54a5fa588..32e877f2baf 100644 --- a/src/tmuxp/log.py +++ b/src/tmuxp/log.py @@ -29,7 +29,9 @@ } -def setup_logger(logger=None, level="INFO"): +def setup_logger( + logger: t.Optional[logging.Logger] = None, level: str = "INFO" +) -> None: """ Setup logging for CLI use. @@ -49,80 +51,82 @@ def setup_logger(logger=None, level="INFO"): def set_style( - message, stylized, style_before=None, style_after=None, prefix="", suffix="" -): + message: str, + stylized: bool, + style_before: str = "", + style_after: str = "", + prefix: str = "", + suffix: str = "", +) -> str: if stylized: return prefix + style_before + message + style_after + suffix return prefix + message + suffix -def default_log_template( - self: t.Type[logging.Formatter], - record: logging.LogRecord, - stylized: t.Optional[bool] = False, - **kwargs: t.Any, -) -> str: - """ - Return the prefix for the log message. Template for Formatter. - - Parameters - ---------- - :py:class:`logging.LogRecord` : - object. this is passed in from inside the - :py:meth:`logging.Formatter.format` record. - - Returns - ------- - str - template for logger message - """ - - reset = Style.RESET_ALL - levelname = set_style( - "(%(levelname)s)", - stylized, - style_before=(LEVEL_COLORS.get(record.levelname, "") + Style.BRIGHT), - style_after=Style.RESET_ALL, - suffix=" ", - ) - asctime = set_style( - "%(asctime)s", - stylized, - style_before=(Fore.BLACK + Style.DIM + Style.BRIGHT), - style_after=(Fore.RESET + Style.RESET_ALL), - prefix="[", - suffix="]", - ) - name = set_style( - "%(name)s", - stylized, - style_before=(Fore.WHITE + Style.DIM + Style.BRIGHT), - style_after=(Fore.RESET + Style.RESET_ALL), - prefix=" ", - suffix=" ", - ) - - if stylized: - return reset + levelname + asctime + name + reset - - return levelname + asctime + name - - class LogFormatter(logging.Formatter): - template = default_log_template - - def __init__(self, color=True, *args, **kwargs): + def template( + self: logging.Formatter, + record: logging.LogRecord, + stylized: bool = False, + **kwargs: t.Any, + ) -> str: + """ + Return the prefix for the log message. Template for Formatter. + + Parameters + ---------- + :py:class:`logging.LogRecord` : + object. this is passed in from inside the + :py:meth:`logging.Formatter.format` record. + + Returns + ------- + str + template for logger message + """ + reset = Style.RESET_ALL + levelname = set_style( + "(%(levelname)s)", + stylized, + style_before=(LEVEL_COLORS.get(record.levelname, "") + Style.BRIGHT), + style_after=Style.RESET_ALL, + suffix=" ", + ) + asctime = set_style( + "%(asctime)s", + stylized, + style_before=(Fore.BLACK + Style.DIM + Style.BRIGHT), + style_after=(Fore.RESET + Style.RESET_ALL), + prefix="[", + suffix="]", + ) + name = set_style( + "%(name)s", + stylized, + style_before=(Fore.WHITE + Style.DIM + Style.BRIGHT), + style_after=(Fore.RESET + Style.RESET_ALL), + prefix=" ", + suffix=" ", + ) + + if stylized: + return reset + levelname + asctime + name + reset + + return levelname + asctime + name + + def __init__(self, color: bool = True, *args, **kwargs) -> None: logging.Formatter.__init__(self, *args, **kwargs) - def format(self, record): + def format(self, record: logging.LogRecord) -> str: try: record.message = record.getMessage() except Exception as e: record.message = f"Bad message ({e!r}): {record.__dict__!r}" date_format = "%H:%m:%S" - record.asctime = time.strftime(date_format, self.converter(record.created)) + formatting = self.converter(record.created) # type:ignore + record.asctime = time.strftime(date_format, formatting) prefix = self.template(record) % record.__dict__ diff --git a/src/tmuxp/plugin.py b/src/tmuxp/plugin.py index d6af25fe3ae..8417d28f6c1 100644 --- a/src/tmuxp/plugin.py +++ b/src/tmuxp/plugin.py @@ -1,3 +1,5 @@ +import typing as t + import libtmux from libtmux._compat import LegacyVersion as Version from libtmux.common import get_version @@ -24,20 +26,35 @@ TMUXP_MAX_VERSION = None +if t.TYPE_CHECKING: + from typing_extensions import TypedDict + + class VersionConstraints(TypedDict): + version: t.Union[Version, str] + vmin: str + vmax: t.Optional[str] + incompatible: t.List[t.Union[t.Any, str]] + + class TmuxpPluginVersionConstraints(TypedDict): + tmux: VersionConstraints + tmuxp: VersionConstraints + libtmux: VersionConstraints + + class TmuxpPlugin: def __init__( self, - plugin_name="tmuxp-plugin", - tmux_min_version=TMUX_MIN_VERSION, - tmux_max_version=TMUX_MAX_VERSION, - tmux_version_incompatible=None, - libtmux_min_version=LIBTMUX_MIN_VERSION, - libtmux_max_version=LIBTMUX_MAX_VERSION, - libtmux_version_incompatible=None, - tmuxp_min_version=TMUXP_MIN_VERSION, - tmuxp_max_version=TMUXP_MAX_VERSION, - tmuxp_version_incompatible=None, - ): + plugin_name: str = "tmuxp-plugin", + tmux_min_version: str = TMUX_MIN_VERSION, + tmux_max_version: t.Optional[str] = TMUX_MAX_VERSION, + tmux_version_incompatible: t.Optional[t.List[str]] = None, + libtmux_min_version: str = LIBTMUX_MIN_VERSION, + libtmux_max_version: t.Optional[str] = LIBTMUX_MAX_VERSION, + libtmux_version_incompatible: t.Optional[t.List[str]] = None, + tmuxp_min_version: str = TMUXP_MIN_VERSION, + tmuxp_max_version: t.Optional[str] = TMUXP_MAX_VERSION, + tmuxp_version_incompatible: t.Optional[t.List[str]] = None, + ) -> None: """ Initialize plugin. @@ -82,10 +99,10 @@ def __init__( # Dependency versions self.tmux_version = get_version() - self.libtmux_version = libtmux.__version__ + self.libtmux_version = libtmux.__about__.__version__ self.tmuxp_version = Version(__version__) - self.version_constraints = { + self.version_constraints: "TmuxpPluginVersionConstraints" = { "tmux": { "version": self.tmux_version, "vmin": tmux_min_version, @@ -114,11 +131,12 @@ def __init__( self._version_check() - def _version_check(self): + def _version_check(self) -> None: """ Check all dependency versions for compatibility. """ for dep, constraints in self.version_constraints.items(): + assert isinstance(constraints, dict) try: assert self._pass_version_check(**constraints) except AssertionError: @@ -130,7 +148,13 @@ def _version_check(self): ) ) - def _pass_version_check(self, version, vmin, vmax, incompatible): + def _pass_version_check( + self, + version: t.Union[str, Version], + vmin: str, + vmax: t.Optional[str], + incompatible: t.List[t.Union[t.Any, str]], + ) -> bool: """ Provide affirmative if version compatibility is correct. """ diff --git a/src/tmuxp/shell.py b/src/tmuxp/shell.py index 0113eb48e12..44aa6fe55e1 100644 --- a/src/tmuxp/shell.py +++ b/src/tmuxp/shell.py @@ -6,11 +6,19 @@ """ import logging import os +import typing as t logger = logging.getLogger(__name__) +if t.TYPE_CHECKING: + from typing_extensions import Literal, TypeAlias -def has_ipython(): + CLIShellLiteral: TypeAlias = Literal[ + "best", "pdb", "code", "ptipython", "ptpython", "ipython", "bpython" + ] + + +def has_ipython() -> bool: try: from IPython import start_ipython # NOQA F841 except ImportError: @@ -22,7 +30,7 @@ def has_ipython(): return True -def has_ptpython(): +def has_ptpython() -> bool: try: from ptpython.repl import embed, run_config # NOQA F841 except ImportError: @@ -34,7 +42,7 @@ def has_ptpython(): return True -def has_ptipython(): +def has_ptipython() -> bool: try: from ptpython.ipython import embed # NOQA F841 from ptpython.repl import run_config # NOQA F841 @@ -48,7 +56,7 @@ def has_ptipython(): return True -def has_bpython(): +def has_bpython() -> bool: try: from bpython import embed # NOQA F841 except ImportError: @@ -56,7 +64,7 @@ def has_bpython(): return True -def detect_best_shell(): +def detect_best_shell() -> "CLIShellLiteral": if has_ptipython(): return "ptipython" elif has_ptpython(): @@ -220,7 +228,12 @@ def launch_code(): return launch_code -def launch(shell="best", use_pythonrc=False, use_vi_mode=False, **kwargs): +def launch( + shell: t.Optional["CLIShellLiteral"] = "best", + use_pythonrc: bool = False, + use_vi_mode: bool = False, + **kwargs +) -> None: # Also allowing passing shell='code' to force using code.interact imported_objects = get_launch_args(**kwargs) diff --git a/src/tmuxp/util.py b/src/tmuxp/util.py index b70a61ea008..90acacf05a7 100644 --- a/src/tmuxp/util.py +++ b/src/tmuxp/util.py @@ -6,20 +6,30 @@ """ import logging import os +import pathlib import shlex import subprocess import sys +import typing as t from libtmux._compat import console_to_str from . import exc +if t.TYPE_CHECKING: + from libtmux.pane import Pane + from libtmux.server import Server + from libtmux.session import Session + from libtmux.window import Window + logger = logging.getLogger(__name__) PY2 = sys.version_info[0] == 2 -def run_before_script(script_file, cwd=None): +def run_before_script( + script_file: t.Union[str, pathlib.Path], cwd: t.Optional[pathlib.Path] = None +) -> int: """Function to wrap try/except for subprocess.check_call().""" try: proc = subprocess.Popen( @@ -28,18 +38,19 @@ def run_before_script(script_file, cwd=None): stdout=subprocess.PIPE, cwd=cwd, ) - for line in iter(proc.stdout.readline, b""): - sys.stdout.write(console_to_str(line)) + if proc.stdout is not None: + for line in iter(proc.stdout.readline, b""): + sys.stdout.write(console_to_str(line)) proc.wait() - if proc.returncode: + if proc.returncode and proc.stderr is not None: stderr = proc.stderr.read() proc.stderr.close() - stderr = console_to_str(stderr).split("\n") - stderr = "\n".join(list(filter(None, stderr))) # filter empty + stderr_strlist = console_to_str(stderr).split("\n") + stderr_str = "\n".join(list(filter(None, stderr_strlist))) # filter empty raise exc.BeforeLoadScriptError( - proc.returncode, os.path.abspath(script_file), stderr + proc.returncode, os.path.abspath(script_file), stderr_str ) return proc.returncode @@ -50,14 +61,12 @@ def run_before_script(script_file, cwd=None): raise e -def oh_my_zsh_auto_title(): +def oh_my_zsh_auto_title() -> None: """Give warning and offer to fix ``DISABLE_AUTO_TITLE``. - see: https://github.com/robbyrussell/oh-my-zsh/pull/257 - + See: https://github.com/robbyrussell/oh-my-zsh/pull/257 """ - - if "SHELL" in os.environ and "zsh" in os.environ.get("SHELL"): + if "SHELL" in os.environ and "zsh" in os.environ.get("SHELL", ""): if os.path.exists(os.path.expanduser("~/.oh-my-zsh")): # oh-my-zsh exists if ( @@ -74,16 +83,21 @@ def oh_my_zsh_auto_title(): ) -def get_current_pane(server): +def get_current_pane(server: "Server") -> t.Optional["Pane"]: """Return Pane if one found in env""" if os.getenv("TMUX_PANE") is not None: try: return [p for p in server.panes if p.pane_id == os.getenv("TMUX_PANE")][0] except IndexError: pass + return None -def get_session(server, session_name=None, current_pane=None): +def get_session( + server: "Server", + session_name: t.Optional[str] = None, + current_pane: t.Optional["Pane"] = None, +) -> "Session": try: if session_name: session = server.sessions.get(session_name=session_name) @@ -107,7 +121,11 @@ def get_session(server, session_name=None, current_pane=None): return session -def get_window(session, window_name=None, current_pane=None): +def get_window( + session: "Session", + window_name: t.Optional[str] = None, + current_pane: t.Optional["Pane"] = None, +) -> "Window": try: if window_name: window = session.windows.get(window_name=window_name) @@ -129,7 +147,8 @@ def get_window(session, window_name=None, current_pane=None): return window -def get_pane(window, current_pane=None): +def get_pane(window: "Window", current_pane: t.Optional["Pane"] = None) -> "Pane": + pane = None try: if current_pane is not None: pane = window.panes.get(pane_id=current_pane.pane_id) # NOQA: F841 @@ -137,7 +156,6 @@ def get_pane(window, current_pane=None): pane = window.attached_pane # NOQA: F841 except exc.TmuxpException as e: print(e) - return if pane is None: if current_pane: diff --git a/src/tmuxp/workspace/builder.py b/src/tmuxp/workspace/builder.py index 2fa68581d40..619e67f86d0 100644 --- a/src/tmuxp/workspace/builder.py +++ b/src/tmuxp/workspace/builder.py @@ -6,7 +6,9 @@ """ import logging import time +import typing as t +from libtmux._internal.query_list import ObjectDoesNotExist from libtmux.common import has_lt_version from libtmux.exc import TmuxSessionExists from libtmux.pane import Pane @@ -134,7 +136,15 @@ class WorkspaceBuilder: a session inside tmux (when `$TMUX` is in the env variables). """ - def __init__(self, sconf, plugins=[], server=None): + server: t.Optional["Server"] + session: t.Optional["Session"] + + def __init__( + self, + sconf: t.Dict[str, t.Any], + plugins: t.List[t.Any] = [], + server: t.Optional[Server] = None, + ) -> None: """ Initialize workspace loading. @@ -162,25 +172,26 @@ def __init__(self, sconf, plugins=[], server=None): if isinstance(server, Server): self.server = server - else: - self.server = None self.sconf = sconf self.plugins = plugins - def session_exists(self, session_name=None): + def session_exists(self, session_name: str) -> bool: + assert session_name is not None + assert isinstance(session_name, str) + assert self.server is not None exists = self.server.has_session(session_name) if not exists: return exists try: - self.session = self.server.sessions.filter(session_name=session_name)[0] - except IndexError: + self.server.sessions.get(session_name=session_name) + except ObjectDoesNotExist: return False return True - def build(self, session=None, append=False): + def build(self, session: t.Optional[Session] = None, append: bool = False) -> None: """ Build tmux workspace in session. @@ -207,15 +218,15 @@ def build(self, session=None, append=False): if self.server.has_session(self.sconf["session_name"]): try: - self.session = self.server.sessions.filter( + self.session = self.server.sessions.get( session_name=self.sconf["session_name"] - )[0] + ) raise TmuxSessionExists( "Session name %s is already running." % self.sconf["session_name"] ) - except IndexError: + except ObjectDoesNotExist: pass else: new_session_kwargs = {} @@ -227,13 +238,19 @@ def build(self, session=None, append=False): session_name=self.sconf["session_name"], **new_session_kwargs, ) + assert session is not None assert self.sconf["session_name"] == session.name assert len(self.sconf["session_name"]) > 0 - self.session = session - self.server = session.server + assert session is not None + assert session.name is not None + + self.session: "Session" = session + assert session.server is not None + + self.server: "Server" = session.server self.server.sessions assert self.server.has_session(session.name) assert session.id @@ -298,7 +315,9 @@ def build(self, session=None, append=False): if focus: focus.select_window() - def iter_create_windows(self, session, append=False): + def iter_create_windows( + self, session: Session, append: bool = False + ) -> t.Iterator[t.Any]: """ Return :class:`libtmux.Window` iterating through session config dict. @@ -331,7 +350,7 @@ def iter_create_windows(self, session, append=False): w1 = None if is_first_window_pass: # if first window, use window 1 w1 = session.attached_window - w1.move_window(99) + w1.move_window("99") if "start_directory" in wconf: sd = wconf["start_directory"] @@ -395,7 +414,9 @@ def iter_create_windows(self, session, append=False): yield w, wconf - def iter_create_panes(self, w, wconf): + def iter_create_panes( + self, w: Window, wconf: t.Dict[str, t.Any] + ) -> t.Iterator[t.Any]: """ Return :class:`libtmux.Pane` iterating through window config dict. @@ -416,7 +437,9 @@ def iter_create_panes(self, w, wconf): """ assert isinstance(w, Window) - pane_base_index = int(w.show_window_option("pane-base-index", g=True)) + pane_base_index_str = w.show_window_option("pane-base-index", g=True) + assert pane_base_index_str is not None + pane_base_index = int(pane_base_index_str) p = None @@ -453,6 +476,8 @@ def get_pane_shell(): ) environment = None + assert p is not None + p = w.split_window( attach=True, start_directory=get_pane_start_directory(), @@ -494,7 +519,7 @@ def get_pane_shell(): yield p, pconf - def config_after_window(self, w, wconf): + def config_after_window(self, w: Window, wconf: t.Dict[str, t.Any]) -> None: """Actions to apply to window after window and pane finished. When building a tmux session, sometimes its easier to postpone things @@ -512,10 +537,12 @@ def config_after_window(self, w, wconf): for key, val in wconf["options_after"].items(): w.set_window_option(key, val) - def find_current_attached_session(self): + def find_current_attached_session(self) -> Session: + assert self.server is not None + current_active_pane = get_current_pane(self.server) - if not current_active_pane: + if current_active_pane is None: raise exc.TmuxpException("No session active.") return next( @@ -524,8 +551,7 @@ def find_current_attached_session(self): for s in self.server.sessions if s.session_id == current_active_pane.session_id ), - None, ) - def first_window_pass(self, i, session, append): + def first_window_pass(self, i: int, session: Session, append: bool) -> bool: return len(session.windows) == 1 and i == 1 and not append diff --git a/src/tmuxp/workspace/finders.py b/src/tmuxp/workspace/finders.py index 85e2ced513b..2cb0360af9b 100644 --- a/src/tmuxp/workspace/finders.py +++ b/src/tmuxp/workspace/finders.py @@ -1,5 +1,6 @@ import logging import os +import pathlib import typing as t from colorama import Fore @@ -10,8 +11,20 @@ logger = logging.getLogger(__name__) +if t.TYPE_CHECKING: + from typing_extensions import Literal, TypeAlias -def is_workspace_file(filename, extensions=[".yml", ".yaml", ".json"]): + ValidExtensions: TypeAlias = Literal[".yml", ".yaml", ".json"] + + +def is_workspace_file( + filename: str, + extensions: t.Union["ValidExtensions", t.List["ValidExtensions"]] = [ + ".yml", + ".yaml", + ".json", + ], +) -> bool: """ Return True if file has a valid workspace file type. @@ -31,8 +44,10 @@ def is_workspace_file(filename, extensions=[".yml", ".yaml", ".json"]): def in_dir( - workspace_dir=os.path.expanduser("~/.tmuxp"), extensions=[".yml", ".yaml", ".json"] -): + workspace_dir: t.Union[pathlib.Path, str] = os.path.expanduser("~/.tmuxp"), + extensions: t.List["ValidExtensions"] = [".yml", ".yaml", ".json"], +) -> t.List[str]: + """ Return a list of workspace_files in ``workspace_dir``. @@ -56,7 +71,7 @@ def in_dir( return workspace_files -def in_cwd(): +def in_cwd() -> t.List[str]: """ Return list of workspace_files in current working directory. diff --git a/src/tmuxp/workspace/importers.py b/src/tmuxp/workspace/importers.py index ff05e497a22..b5eff6606ba 100644 --- a/src/tmuxp/workspace/importers.py +++ b/src/tmuxp/workspace/importers.py @@ -1,4 +1,7 @@ -def import_tmuxinator(workspace_dict): +import typing as t + + +def import_tmuxinator(workspace_dict: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]: """Return tmuxp workspace from a `tmuxinator`_ yaml workspace. .. _tmuxinator: https://github.com/aziz/tmuxinator @@ -96,7 +99,7 @@ def import_tmuxinator(workspace_dict): return tmuxp_workspace -def import_teamocil(workspace_dict): +def import_teamocil(workspace_dict: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]: """Return tmuxp workspace from a `teamocil`_ yaml workspace. .. _teamocil: https://github.com/remiprev/teamocil diff --git a/src/tmuxp/workspace/validation.py b/src/tmuxp/workspace/validation.py index f1d99ec60f6..bbe1ede7f9e 100644 --- a/src/tmuxp/workspace/validation.py +++ b/src/tmuxp/workspace/validation.py @@ -1,7 +1,9 @@ +import typing as t + from .. import exc -def validate_schema(workspace_dict): +def validate_schema(workspace_dict: t.Any) -> bool: """ Return True if workspace schema is correct. diff --git a/tests/fixtures/pluginsystem/partials/all_pass.py b/tests/fixtures/pluginsystem/partials/all_pass.py index 72c4116a427..8fea533d2f1 100644 --- a/tests/fixtures/pluginsystem/partials/all_pass.py +++ b/tests/fixtures/pluginsystem/partials/all_pass.py @@ -2,7 +2,7 @@ class AllVersionPassPlugin(MyTestTmuxpPlugin): - def __init__(self): + def __init__(self) -> None: config = { "plugin_name": "tmuxp-plugin-my-tmuxp-plugin", "tmux_min_version": "1.8", diff --git a/tests/fixtures/pluginsystem/partials/test_plugin_helpers.py b/tests/fixtures/pluginsystem/partials/test_plugin_helpers.py index 1ca4ec5fa7e..1f7bcdb7ee7 100644 --- a/tests/fixtures/pluginsystem/partials/test_plugin_helpers.py +++ b/tests/fixtures/pluginsystem/partials/test_plugin_helpers.py @@ -1,8 +1,11 @@ +from typing import Iterable, Mapping, Union + from tmuxp.plugin import TmuxpPlugin class MyTestTmuxpPlugin(TmuxpPlugin): - def __init__(self, config): + def __init__(self, config: Mapping[str, Union[str, Iterable[str]]]) -> None: + assert isinstance(config, dict) tmux_version = config.pop("tmux_version", None) libtmux_version = config.pop("libtmux_version", None) tmuxp_version = config.pop("tmuxp_version", None) diff --git a/tests/fixtures/utils.py b/tests/fixtures/utils.py index 7b296eb3574..d6eb9cbbac4 100644 --- a/tests/fixtures/utils.py +++ b/tests/fixtures/utils.py @@ -1,13 +1,21 @@ import pathlib +import typing as t from ..constants import FIXTURE_PATH -def get_workspace_file(_file): # return fixture data, relative to __file__ +def get_workspace_file( + _file: t.Union[str, pathlib.Path], +) -> pathlib.Path: + """Return fixture data, relative to __file__""" + if isinstance(_file, str): + _file = pathlib.Path(_file) return FIXTURE_PATH / _file -def read_workspace_file(_file): # return fixture data, relative to __file__ +def read_workspace_file(_file: t.Union[pathlib.Path, str]) -> str: + """Return fixture data, relative to __file__""" + return open(get_workspace_file(_file)).read() diff --git a/tests/fixtures/workspace/shell_command_before.py b/tests/fixtures/workspace/shell_command_before.py index f1b9d601445..6a7f0d6ab32 100644 --- a/tests/fixtures/workspace/shell_command_before.py +++ b/tests/fixtures/workspace/shell_command_before.py @@ -1,4 +1,5 @@ import os +from typing import Any, Dict config_unexpanded = { # shell_command_before is string in some areas "session_name": "sample workspace", @@ -37,7 +38,7 @@ } -def config_expanded(): +def config_expanded() -> Dict[str, Any]: return { # shell_command_before is string in some areas "session_name": "sample workspace", "start_directory": "/", @@ -89,7 +90,7 @@ def config_expanded(): } -def config_after(): +def config_after() -> Dict[str, Any]: return { # shell_command_before is string in some areas "session_name": "sample workspace", "start_directory": "/", diff --git a/tests/test_cli.py b/tests/test_cli.py index ff8ce8cdefa..eb65417b854 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1273,6 +1273,7 @@ def test_reattach_plugins( except libtmux.exc.LibTmuxException: pass + assert builder.session is not None proc = builder.session.cmd("display-message", "-p", "'#S'") assert proc.stdout[0] == "'plugin_test_r'" diff --git a/tests/test_plugin.py b/tests/test_plugin.py index adb47044cfc..2151d7ce950 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -22,63 +22,63 @@ @pytest.fixture(autouse=True) -def autopatch_sitedir(monkeypatch_plugin_test_packages): +def autopatch_sitedir(monkeypatch_plugin_test_packages: None) -> None: pass -def test_all_pass(): +def test_all_pass() -> None: AllVersionPassPlugin() -def test_tmux_version_fail_min(): +def test_tmux_version_fail_min() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: TmuxVersionFailMinPlugin() assert "tmux-min-version-fail" in str(exc_info.value) -def test_tmux_version_fail_max(): +def test_tmux_version_fail_max() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: TmuxVersionFailMaxPlugin() assert "tmux-max-version-fail" in str(exc_info.value) -def test_tmux_version_fail_incompatible(): +def test_tmux_version_fail_incompatible() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: TmuxVersionFailIncompatiblePlugin() assert "tmux-incompatible-version-fail" in str(exc_info.value) -def test_tmuxp_version_fail_min(): +def test_tmuxp_version_fail_min() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: TmuxpVersionFailMinPlugin() assert "tmuxp-min-version-fail" in str(exc_info.value) -def test_tmuxp_version_fail_max(): +def test_tmuxp_version_fail_max() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: TmuxpVersionFailMaxPlugin() assert "tmuxp-max-version-fail" in str(exc_info.value) -def test_tmuxp_version_fail_incompatible(): +def test_tmuxp_version_fail_incompatible() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: TmuxpVersionFailIncompatiblePlugin() assert "tmuxp-incompatible-version-fail" in str(exc_info.value) -def test_libtmux_version_fail_min(): +def test_libtmux_version_fail_min() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: LibtmuxVersionFailMinPlugin() assert "libtmux-min-version-fail" in str(exc_info.value) -def test_libtmux_version_fail_max(): +def test_libtmux_version_fail_max() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: LibtmuxVersionFailMaxPlugin() assert "libtmux-max-version-fail" in str(exc_info.value) -def test_libtmux_version_fail_incompatible(): +def test_libtmux_version_fail_incompatible() -> None: with pytest.raises(TmuxpPluginException, match=r"Incompatible.*") as exc_info: LibtmuxVersionFailIncompatiblePlugin() assert "libtmux-incompatible-version-fail" in str(exc_info.value) diff --git a/tests/test_shell.py b/tests/test_shell.py index ca6a97b2652..2f87bddba9b 100644 --- a/tests/test_shell.py +++ b/tests/test_shell.py @@ -1,12 +1,12 @@ from tmuxp import shell -def test_detect_best_shell(): +def test_detect_best_shell() -> None: result = shell.detect_best_shell() assert isinstance(result, str) -def test_shell_detect(): +def test_shell_detect() -> None: assert isinstance(shell.has_bpython(), bool) assert isinstance(shell.has_ipython(), bool) assert isinstance(shell.has_ptpython(), bool) diff --git a/tests/tests/test_helpers.py b/tests/tests/test_helpers.py index 6418cce5795..5e93ea45ebb 100644 --- a/tests/tests/test_helpers.py +++ b/tests/tests/test_helpers.py @@ -1,10 +1,11 @@ """Tests for .'s helper and utility functions.""" import pytest +from libtmux.server import Server from libtmux.test import get_test_session_name, temp_session -def test_kills_session(server): +def test_kills_session(server: Server) -> None: server = server session_name = get_test_session_name(server=server) @@ -16,7 +17,7 @@ def test_kills_session(server): @pytest.mark.flaky(reruns=5) -def test_if_session_killed_before(server): +def test_if_session_killed_before(server: Server) -> None: """Handles situation where session already closed within context""" server = server diff --git a/tests/workspace/conftest.py b/tests/workspace/conftest.py index f870c845ec8..932055675af 100644 --- a/tests/workspace/conftest.py +++ b/tests/workspace/conftest.py @@ -6,7 +6,7 @@ @pytest.fixture -def config_fixture(): +def config_fixture() -> WorkspaceTestData: """Deferred import of tmuxp.tests.fixtures.* pytest setup (conftest.py) patches os.environ["HOME"], delay execution of diff --git a/tests/workspace/test_builder.py b/tests/workspace/test_builder.py index 4c77dfc27f7..44e577f5c76 100644 --- a/tests/workspace/test_builder.py +++ b/tests/workspace/test_builder.py @@ -9,6 +9,7 @@ import libtmux from libtmux.common import has_gte_version, has_lt_version +from libtmux.pane import Pane from libtmux.session import Session from libtmux.test import retry_until, temp_session from libtmux.window import Window @@ -80,9 +81,11 @@ def test_focus_pane_index(session): assert session.attached_window.name == "focused window" - pane_base_index = int( - session.attached_window.show_window_option("pane-base-index", g=True) + _pane_base_index = session.attached_window.show_window_option( + "pane-base-index", g=True ) + assert isinstance(_pane_base_index, int) + pane_base_index = int(_pane_base_index) if not pane_base_index: pane_base_index = 0 @@ -104,13 +107,15 @@ def test_focus_pane_index(session): pane_path = "/usr" p = None - def f(): + def f_check() -> bool: nonlocal p p = w.attached_pane + assert p is not None return p.pane_current_path == pane_path - assert retry_until(f) + assert retry_until(f_check) + assert p is not None assert p.pane_current_path == pane_path proc = session.cmd("show-option", "-gv", "base-index") @@ -122,13 +127,17 @@ def f(): p = None pane_path = "/" - def f(): + def f_check_again(): nonlocal p p = window3.attached_pane + assert p is not None return p.pane_current_path == pane_path - assert retry_until(f) + assert retry_until(f_check_again) + assert p is not None + assert p.pane_current_path is not None + assert isinstance(p.pane_current_path, str) assert p.pane_current_path == pane_path @@ -201,8 +210,13 @@ def test_session_options(session): builder = WorkspaceBuilder(sconf=workspace) builder.build(session=session) - assert "/bin/sh" in session.show_option("default-shell") - assert "/bin/sh" in session.show_option("default-command") + _default_shell = session.show_option("default-shell") + assert isinstance(_default_shell, str) + assert "/bin/sh" in _default_shell + + _default_command = session.show_option("default-command") + assert isinstance(_default_command, str) + assert "/bin/sh" in _default_command def test_global_options(session): @@ -214,7 +228,9 @@ def test_global_options(session): builder = WorkspaceBuilder(sconf=workspace) builder.build(session=session) - assert "top" in session.show_option("status-position", _global=True) + _status_position = session.show_option("status-position", _global=True) + assert isinstance(_status_position, str) + assert "top" in _status_position assert 493 == session.show_option("repeat-time", _global=True) @@ -234,7 +250,9 @@ def test_global_session_env_options(session, monkeypatch): builder = WorkspaceBuilder(sconf=workspace) builder.build(session=session) - assert visual_silence in session.show_option("visual-silence", _global=True) + _visual_silence = session.show_option("visual-silence", _global=True) + assert isinstance(_visual_silence, str) + assert visual_silence in _visual_silence assert repeat_time == session.show_option("repeat-time") assert main_pane_height == session.attached_window.show_window_option( "main-pane-height" @@ -434,6 +452,7 @@ def test_automatic_rename_option( builder = WorkspaceBuilder(sconf=workspace, server=server) builder.build() + assert builder.session is not None session: Session = builder.session w: Window = session.windows[0] assert len(session.windows) == 1 @@ -473,15 +492,19 @@ def test_blank_pane_count(session): assert session == builder.session window1 = session.windows.get(window_name="Blank pane test") + assert window1 is not None assert len(window1.panes) == 3 window2 = session.windows.get(window_name="More blank panes") + assert window2 is not None assert len(window2.panes) == 3 window3 = session.windows.get(window_name="Empty string (return)") + assert window3 is not None assert len(window3.panes) == 3 window4 = session.windows.get(window_name="Blank with options") + assert window4 is not None assert len(window4.panes) == 2 @@ -519,12 +542,12 @@ def test_start_directory_relative(session, tmp_path: pathlib.Path): """Same as above test, but with relative start directory, mimicking loading it from a location of project file. Like:: - $ tmuxp load ~/workspace/myproject/.tmuxp.yaml + $ tmuxp load ~/workspace/myproject/.tmuxp.yaml instead of:: - $ cd ~/workspace/myproject/.tmuxp.yaml - $ tmuxp load . + $ cd ~/workspace/myproject/.tmuxp.yaml + $ tmuxp load . """ yaml_workspace = test_utils.read_workspace_file( @@ -1068,7 +1091,9 @@ def test_load_workspace_enter( builder.build() session = builder.session + assert isinstance(session, Session) pane = session.attached_pane + assert isinstance(pane, Pane) def fn(): captured_pane = "\n".join(pane.capture_pane()) @@ -1186,18 +1211,27 @@ def test_load_workspace_sleep( workspace = loader.trickle(workspace) builder = WorkspaceBuilder(sconf=workspace, server=server) - t = time.process_time() + start_time = time.process_time() builder.build() time.sleep(0.5) session = builder.session + assert isinstance(builder.session, Session) + assert session is not None pane = session.attached_pane + assert isinstance(pane, Pane) + + assert pane is not None - while (time.process_time() - t) * 1000 < sleep: + assert not isinstance(pane.capture_pane, str) + assert callable(pane.capture_pane) + + while (time.process_time() - start_time) * 1000 < sleep: captured_pane = "\n".join(pane.capture_pane()) assert output not in captured_pane time.sleep(0.1) + captured_pane = "\n".join(pane.capture_pane()) assert output in captured_pane @@ -1340,6 +1374,10 @@ def test_issue_800_default_size_many_windows( with pytest.raises(Exception): builder.build() + assert builder is not None + assert builder.session is not None + assert isinstance(builder.session, Session) + assert callable(builder.session.kill_session) builder.session.kill_session() with pytest.raises(libtmux.exc.LibTmuxException, match="no space for new pane"): diff --git a/tests/workspace/test_config.py b/tests/workspace/test_config.py index bba52886fa4..9fe380eba61 100644 --- a/tests/workspace/test_config.py +++ b/tests/workspace/test_config.py @@ -1,8 +1,7 @@ """Test for tmuxp configuration import, inlining, expanding and export.""" import os import pathlib -import typing -from typing import Union +import typing as t import pytest @@ -12,17 +11,17 @@ from ..constants import EXAMPLE_PATH -if typing.TYPE_CHECKING: +if t.TYPE_CHECKING: from ..fixtures.structures import WorkspaceTestData -def load_yaml(path: Union[str, pathlib.Path]) -> str: +def load_yaml(path: t.Union[str, pathlib.Path]) -> t.Dict[str, t.Any]: return ConfigReader._from_file( pathlib.Path(path) if isinstance(path, str) else path ) -def load_workspace(path: Union[str, pathlib.Path]) -> str: +def load_workspace(path: t.Union[str, pathlib.Path]) -> t.Dict[str, t.Any]: return ConfigReader._from_file( pathlib.Path(path) if isinstance(path, str) else path ) @@ -170,7 +169,7 @@ def test_shell_command_before(config_fixture: "WorkspaceTestData"): assert test_workspace == config_fixture.shell_command_before.config_after() -def test_in_session_scope(config_fixture: "WorkspaceTestData"): +def test_in_session_scope(config_fixture: "WorkspaceTestData") -> None: sconfig = ConfigReader._load( format="yaml", content=config_fixture.shell_command_before_session.before ) @@ -183,7 +182,7 @@ def test_in_session_scope(config_fixture: "WorkspaceTestData"): ) -def test_trickle_relative_start_directory(config_fixture: "WorkspaceTestData"): +def test_trickle_relative_start_directory(config_fixture: "WorkspaceTestData") -> None: test_workspace = loader.trickle(config_fixture.trickle.before) assert test_workspace == config_fixture.trickle.expected @@ -206,7 +205,7 @@ def test_trickle_window_with_no_pane_workspace(): } -def test_expands_blank_panes(config_fixture: "WorkspaceTestData"): +def test_expands_blank_panes(config_fixture: "WorkspaceTestData") -> None: """Expand blank config into full form. Handle ``NoneType`` and 'blank':: @@ -256,7 +255,7 @@ def test_no_session_name(): with pytest.raises(exc.WorkspaceError) as excinfo: validation.validate_schema(sconfig) - assert excinfo.matches(r'requires "session_name"') + assert excinfo.match(r'requires "session_name"') def test_no_windows(): @@ -290,10 +289,10 @@ def test_no_window_name(): with pytest.raises(exc.WorkspaceError) as excinfo: validation.validate_schema(sconfig) - assert excinfo.matches('missing "window_name"') + assert excinfo.match('missing "window_name"') -def test_replaces_env_variables(monkeypatch): +def test_replaces_env_variables(monkeypatch: pytest.MonkeyPatch) -> None: env_key = "TESTHEY92" env_val = "HEYO1" yaml_workspace = """ @@ -352,4 +351,4 @@ def test_plugins(): with pytest.raises(exc.WorkspaceError) as excinfo: validation.validate_schema(sconfig) - assert excinfo.matches("only supports list type") + assert excinfo.match("only supports list type") diff --git a/tests/workspace/test_import_teamocil.py b/tests/workspace/test_import_teamocil.py index 39da8d83da4..9000ac225a8 100644 --- a/tests/workspace/test_import_teamocil.py +++ b/tests/workspace/test_import_teamocil.py @@ -1,4 +1,6 @@ """Test for tmuxp teamocil configuration.""" +import typing as t + import pytest from tmuxp import config_reader @@ -32,7 +34,12 @@ ), ], ) -def test_config_to_dict(teamocil_yaml, teamocil_dict, tmuxp_dict): +def test_config_to_dict( + teamocil_yaml: str, + teamocil_dict: t.Dict[str, t.Any], + tmuxp_dict: t.Dict[str, t.Any], +) -> None: + yaml_to_dict = config_reader.ConfigReader._load( format="yaml", content=teamocil_yaml ) @@ -44,14 +51,17 @@ def test_config_to_dict(teamocil_yaml, teamocil_dict, tmuxp_dict): @pytest.fixture(scope="module") -def multisession_config(): +def multisession_config() -> t.Dict[ + str, + t.Dict[str, t.Any], +]: """Return loaded multisession teamocil config as a dictionary. Also prevents re-running assertion the loads the yaml, since ordering of deep list items like panes will be inconsistent.""" teamocil_yaml_file = fixtures.layouts.teamocil_yaml_file test_config = config_reader.ConfigReader._from_file(teamocil_yaml_file) - teamocil_dict = fixtures.layouts.teamocil_dict + teamocil_dict: t.Dict[str, t.Any] = fixtures.layouts.teamocil_dict assert test_config == teamocil_dict return teamocil_dict @@ -72,7 +82,11 @@ def multisession_config(): ), ], ) -def test_multisession_config(session_name, expected, multisession_config): +def test_multisession_config( + session_name: str, + expected: t.Dict[str, t.Any], + multisession_config: t.Dict[str, t.Any], +) -> None: # teamocil can fit multiple sessions in a config assert importers.import_teamocil(multisession_config[session_name]) == expected diff --git a/tests/workspace/test_import_tmuxinator.py b/tests/workspace/test_import_tmuxinator.py index 795c48da4d5..f68f9243b3a 100644 --- a/tests/workspace/test_import_tmuxinator.py +++ b/tests/workspace/test_import_tmuxinator.py @@ -1,4 +1,6 @@ """Test for tmuxp tmuxinator configuration.""" +import typing as t + import pytest from tmuxp.config_reader import ConfigReader @@ -27,7 +29,11 @@ ), # Test importing ], ) -def test_config_to_dict(tmuxinator_yaml, tmuxinator_dict, tmuxp_dict): +def test_config_to_dict( + tmuxinator_yaml: str, + tmuxinator_dict: t.Dict[str, t.Any], + tmuxp_dict: t.Dict[str, t.Any], +) -> None: yaml_to_dict = ConfigReader._load(format="yaml", content=tmuxinator_yaml) assert yaml_to_dict == tmuxinator_dict