Coverage for src/lilbee/cli/upgrade_line.py: 100%
41 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""The line a command shows on stderr while a first open upgrades the library."""
3from __future__ import annotations
5from contextlib import ExitStack
6from time import monotonic
8from rich.text import Text
10from lilbee.core.config import cfg
11from lilbee.runtime.console import PlainConsole
12from lilbee.runtime.progress import UPGRADE_HEADLINE, EventType, LibraryUpgradeEvent, ProgressEvent
15class UpgradeLine:
16 """Shows library upgrade events on *console*: a live line on a terminal, else plain lines.
18 The live line is Rich's status spinner, whose own refresh reads the elapsed time
19 again at each redraw. With *live* off, under ``--json`` or piped, each event is a line.
20 """
22 def __init__(self, console: PlainConsole, *, live: bool = True) -> None:
23 self._console = console
24 self.live = live
25 self._event: LibraryUpgradeEvent | None = None
26 self._started = 0.0
27 # Holds Rich's status spinner while a live line shows.
28 self._status: ExitStack | None = None
30 def __call__(self, _event_type: EventType, event: ProgressEvent) -> None:
31 # The reporter takes the whole progress union; the store passes it upgrade events only.
32 if not isinstance(event, LibraryUpgradeEvent):
33 return
34 if event.running:
35 self._show_running(event)
36 else:
37 self._show_end(event)
39 def __rich__(self) -> Text:
40 """The progress line of the latest event, timed to now."""
41 event = self._event
42 return Text("" if event is None else event.line_at(monotonic() - self._started))
44 def _show_running(self, event: LibraryUpgradeEvent) -> None:
45 if self._event is None:
46 self._write(UPGRADE_HEADLINE)
47 self._event = event
48 self._started = monotonic() - event.elapsed_s
49 if not (self.live and self._console.is_terminal and not cfg.json_mode):
50 self._write(event.text)
51 elif self._status is None:
52 self._status = ExitStack()
53 self._status.enter_context(self._console.status(self))
55 def _show_end(self, event: LibraryUpgradeEvent) -> None:
56 if self._status is not None:
57 self._status.close()
58 self._status = None
59 self._event = None
60 self._write(event.text)
62 def _write(self, line: str) -> None:
63 self._console.print(Text(line), soft_wrap=True, highlight=False)