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

1"""The line a command shows on stderr while a first open upgrades the library.""" 

2 

3from __future__ import annotations 

4 

5from contextlib import ExitStack 

6from time import monotonic 

7 

8from rich.text import Text 

9 

10from lilbee.core.config import cfg 

11from lilbee.runtime.console import PlainConsole 

12from lilbee.runtime.progress import UPGRADE_HEADLINE, EventType, LibraryUpgradeEvent, ProgressEvent 

13 

14 

15class UpgradeLine: 

16 """Shows library upgrade events on *console*: a live line on a terminal, else plain lines. 

17 

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 """ 

21 

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 

29 

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) 

38 

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)) 

43 

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)) 

54 

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) 

61 

62 def _write(self, line: str) -> None: 

63 self._console.print(Text(line), soft_wrap=True, highlight=False)