19526335899

Committed 20 Nov 2025 05:09AM UTC coverage: 56.979% (-0.04%) from 57.017%

Build # 19526335899

Build Type

push

github

Committed by

haliphax

Commit Message

🎨 remove unnecessary type hint

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

645 of 1132 relevant lines covered (56.98%)

0.57 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

40.48

/xthulu/ssh/console/app.py

"""Textual application wrapper"""

# stdlib
from asyncio import sleep

# 3rd party
from rich.segment import Segments
from textual import events
from textual.app import App, ReturnType
from textual.geometry import Size

# local
from ...logger import log
from ..context import SSHContext


class _ErrorConsoleProxy:
    def print(self, what, **kwargs):
        if isinstance(what, Segments):
            log.error("".join([s.text for s in what.segments]))
            return

        log.error(what)


class XthuluApp(App[ReturnType]):
    """SSH wrapper for Textual apps"""

    ENABLE_COMMAND_PALETTE = False
    """Command palette is disabled by default"""

    context: SSHContext
    """The current SSH context"""

    def __init__(self, context: SSHContext, ansi_color=True, **kwargs):
        ""  # empty docstring
        # avoid cyclic import
        from .internal.driver import SSHDriver

        super(XthuluApp, self).__init__(
            driver_class=SSHDriver, ansi_color=ansi_color, **kwargs
        )
        self.context = context
        self.console = context.console
        self.error_console = _ErrorConsoleProxy()  # type: ignore
        self.run_worker(self._watch_for_resize, exclusive=True)

    async def _watch_for_resize(self):
        # avoid cyclic import
        from .internal.driver import SSHDriver

        while True:
            ev = self.context.events.get("resize")

            if not ev:
                await sleep(0.5)
                continue

            new_size = Size(*ev[-1].data)
            d: SSHDriver = self._driver  # type: ignore
            d.process_message(events.Resize(new_size, new_size))

    def exit(self, **kwargs) -> None:  # type: ignore
        ""  # empty docstring
        # avoid cyclic import
        from .internal.driver import SSHDriver

        super(XthuluApp, self).exit(**kwargs)
        d: SSHDriver = self._driver  # type: ignore
        d._disable_bracketed_paste()
        d._disable_mouse_support()
        d.exit_event.set()

1	"""Textual application wrapper"""
2
3	# stdlib
4	from asyncio import sleep	1✔
5
6	# 3rd party
7	from rich.segment import Segments	1✔
8	from textual import events	1✔
9	from textual.app import App, ReturnType	1✔
10	from textual.geometry import Size	1✔
11
12	# local
13	from ...logger import log	1✔
14	from ..context import SSHContext	1✔
15
16
17	class _ErrorConsoleProxy:	1✔
18	def print(self, what, **kwargs):	1✔
19	if isinstance(what, Segments):	×
20	log.error("".join([s.text for s in what.segments]))	×
21	return	×
22
23	log.error(what)	×
24
25
26	class XthuluApp(App[ReturnType]):	1✔
27	"""SSH wrapper for Textual apps"""
28
29	ENABLE_COMMAND_PALETTE = False	1✔
30	"""Command palette is disabled by default"""	1✔
31
32	context: SSHContext	1✔
33	"""The current SSH context"""	1✔
34
35	def __init__(self, context: SSHContext, ansi_color=True, **kwargs):	1✔
36	"" # empty docstring
37	# avoid cyclic import
38	from .internal.driver import SSHDriver	×
39
40	super(XthuluApp, self).__init__(	×
41	driver_class=SSHDriver, ansi_color=ansi_color, **kwargs
42	)
43	self.context = context	×
44	self.console = context.console	×
45	self.error_console = _ErrorConsoleProxy() # type: ignore	×
46	self.run_worker(self._watch_for_resize, exclusive=True)	×
47
48	async def _watch_for_resize(self):	1✔
49	# avoid cyclic import
50	from .internal.driver import SSHDriver	×
51
52	while True:	×
NEW 53	ev = self.context.events.get("resize")	×
54
55	if not ev:	×
56	await sleep(0.5)	×
57	continue	×
58
59	new_size = Size(*ev[-1].data)	×
60	d: SSHDriver = self._driver # type: ignore	×
61	d.process_message(events.Resize(new_size, new_size))	×
62
63	def exit(self, **kwargs) -> None: # type: ignore	1✔
64	"" # empty docstring
65	# avoid cyclic import
66	from .internal.driver import SSHDriver	×
67
68	super(XthuluApp, self).exit(**kwargs)	×
69	d: SSHDriver = self._driver # type: ignore	×
70	d._disable_bracketed_paste()	×
71	d._disable_mouse_support()	×
72	d.exit_event.set()	×

haliphax / xthulu / 19526335899

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous