diff --git a/CHANGELOG.md b/CHANGELOG.md index 40063dd..5336ea0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,18 @@ backfill releases from before this changelog. ## [Unreleased] +### Changed + +- Use sentence case throughout the dashboard and add persistent keyboard and sudo guidance. +- Share read-only status acquisition between the CLI and TUI, preserving unknown monitoring state and store-fault recovery guidance. +- Use one authenticated action path for the CLI and TUI; let valid collection runs finish without the former 30-second dashboard cutoff. +- Remove the unused sparkline and habit-bar rendering path while retaining the interactive history graph. +- Align all package formats on the MIT license and Python 3.10 minimum; include the license text in native packages. + +### Fixed + +- Correct observation-store directory permissions in native packages, including repair of older runit installations during upgrade. + ## [0.3.5] - 2026-09-14 ### Added diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..301ed90 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Fenris contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Makefile b/Makefile index 938df32..1bbddf3 100644 --- a/Makefile +++ b/Makefile @@ -74,7 +74,7 @@ install: check-python check-smartctl dist/fenris-*.whl @echo "=== Creating data directory (root-written, group-read) ===" @sudo groupadd -f fenris - @sudo install -d -o root -g fenris -m 2750 $(DATA_DIR) + @sudo install -d -o root -g fenris -m 2770 $(DATA_DIR) @echo "=== Installing version-neutral runtime packages ===" @sudo rm -rf $(VENV_DIR) diff --git a/README.md b/README.md index edc8aee..2b465ab 100644 --- a/README.md +++ b/README.md @@ -252,11 +252,33 @@ is retried at the next interval; it never fabricates missing observations. ## Reading the dashboard -`fenris` opens the TUI dashboard. +Run `fenris` as your normal user to open the TUI dashboard. The dashboard does +not need `sudo`. Use `sudo` for package installation and system configuration; +pause, resume, and collect-now actions normally authenticate through polkit. + +If the observation store is inaccessible, add your login user to the `fenris` +group with `sudo usermod -aG fenris "$USER"`, then log out and back in. + +If polkit authentication is unavailable, quit the dashboard and run only the +required administrative action in your terminal: + +```bash +sudo fenris monitor resume # Enable monitoring now and across reboots +sudo fenris monitor pause # Confirm a deliberate monitoring pause +sudo fenris sample # Request one collection run +``` + +Reopen the dashboard with `fenris` afterward. Press `?` for these instructions +and keyboard controls at any time; use the arrow keys to scroll and `Esc` to close. + +The CLI and TUI share the same authenticated action path. Authentication and +waiting for collection have no separate dashboard deadline; the native collector +enforces its 90-second runtime limit. An interrupted or failed action is not +automatically retried—check `fenris status` before retrying. - **Continuity** — the service strip's continuity line (and `fenris status`) reports whether monitoring survives reboots: `monitoring: active in background · persists across reboots`, or `monitoring: does not start on next boot`. - **Paused vs. quit** — a full-width `monitoring: paused — deliberate disable` block means collection is stopped (`fenris monitor pause`); resume with `fenris monitor resume`. Pressing `q` only leaves the screen — monitoring keeps running in the background. -- **Auth banner** — at launch, `privileged actions will prompt for authentication (polkit)` shows once and clears on the first refresh. Privileged actions elevate via polkit; Fenris never asks for sudo. +- **Auth banner** — the launch notice explains normal-user startup and polkit authentication, then clears on the first refresh. The `?` help screen remains available. Per-release notes live on the [releases page](https://git.bongbetic.com/xavierk/Fenris/releases): each entry is the version's `CHANGELOG.md` section — what was added, changed, and fixed — plus standing install and verification instructions. diff --git a/docs/adr/0009-mit-license.md b/docs/adr/0009-mit-license.md new file mode 100644 index 0000000..82887db --- /dev/null +++ b/docs/adr/0009-mit-license.md @@ -0,0 +1,8 @@ +# One MIT license across source and packages + +The native package metadata previously disagreed: deb/rpm declared Proprietary, +while XBPS declared MIT and the repository carried no license text. On +2026-09-16 the maintainer chose MIT for Fenris. The repository now includes the +standard MIT license, and Python, deb, rpm, and XBPS distributions must preserve +that same licensing decision; bundled third-party dependencies retain their own +license notices. diff --git a/packaging/nfpm.yaml b/packaging/nfpm.yaml index eef588c..96309a5 100644 --- a/packaging/nfpm.yaml +++ b/packaging/nfpm.yaml @@ -7,7 +7,7 @@ description: > NVMe wear monitor with persistent TUI — observes real-world drive use and translates it into an understandable endurance outlook. homepage: https://git.bongbetic.com/xavierk/Fenris -license: Proprietary +license: MIT depends: - python3 (>= 3.10) @@ -38,7 +38,7 @@ contents: - dst: /var/lib/fenris type: dir file_info: - mode: 2750 + mode: 02770 group: fenris scripts: diff --git a/packaging/postinst.sh b/packaging/postinst.sh index e92eeea..1c0d190 100755 --- a/packaging/postinst.sh +++ b/packaging/postinst.sh @@ -63,7 +63,8 @@ print(f'Fenris migration: {n} step(s) applied') if n else None else # runit: create group, set directory permissions groupadd -f fenris - install -d -o root -g fenris -m 2750 "${STORE_DIR}" 2>/dev/null || true + install -d -o root -g fenris -m 2770 "${STORE_DIR}" + chmod 02770 "${STORE_DIR}" # Mark runit service as dormant (down) for fresh install if [ -d /etc/sv/fenris-collect ] && [ ! -e /var/service/fenris-collect ]; then touch /etc/sv/fenris-collect/down diff --git a/packaging/rpm/post.sh b/packaging/rpm/post.sh index f7684e0..9502f3e 100755 --- a/packaging/rpm/post.sh +++ b/packaging/rpm/post.sh @@ -24,7 +24,8 @@ if [ "$1" -eq 1 ]; then else # runit: create group, set directory permissions groupadd -f fenris - install -d -o root -g fenris -m 2750 "${STORE_DIR}" 2>/dev/null || true + install -d -o root -g fenris -m 2770 "${STORE_DIR}" + chmod 02770 "${STORE_DIR}" # Mark runit service as dormant (down) for fresh install if [ -d /etc/sv/fenris-collect ] && [ ! -e /var/service/fenris-collect ]; then touch /etc/sv/fenris-collect/down @@ -69,6 +70,8 @@ print(f'Fenris migration: {n} step(s) applied') if n else None else # runit: repair log access when upgrading from an older package groupadd -f fenris + install -d -o root -g fenris -m 2770 "${STORE_DIR}" + chmod 02770 "${STORE_DIR}" install -d -o root -g fenris -m 2770 /var/log/fenris-collect 2>/dev/null || true fi fi diff --git a/packaging/stage.sh b/packaging/stage.sh index 6d812f7..8b5eb6e 100755 --- a/packaging/stage.sh +++ b/packaging/stage.sh @@ -58,6 +58,9 @@ python3 -m pip install --disable-pip-version-check --no-compile \ # observation store. chmod -R a+rX "${VENDOR_DIR}" +# Ship the application license in every native package. +install -D -m 0644 "${REPO_ROOT}/LICENSE" "${STAGE_DIR}/usr/share/licenses/fenris/LICENSE" + # --- Inject version into wrapper from pyproject.toml --- # The wrapper has a hardcoded version string; patch it for packaging. WRAPPER_SRC="${REPO_ROOT}/scripts/fenris" diff --git a/packaging/xbps/install.sh b/packaging/xbps/install.sh index e7ae632..f95437f 100644 --- a/packaging/xbps/install.sh +++ b/packaging/xbps/install.sh @@ -34,6 +34,10 @@ case "${ACTION}" in fi ;; post) + # Keep observation-store access consistent across fresh installs and upgrades. + groupadd -f fenris + install -d -o root -g fenris -m 2770 "${STORE_DIR}" + chmod 02770 "${STORE_DIR}" if [ "${UPDATE}" = "yes" ]; then # Upgrade — snapshot, migration, runit-aware reload if [ -f "${STORE_DB}" ]; then @@ -53,7 +57,6 @@ print(f'Fenris migration: {n} step(s) applied') if n else None else # Fresh install — runit service setup groupadd -f fenris - install -d -o root -g fenris -m 2770 "${STORE_DIR}" 2>/dev/null || true # Mark runit service as dormant (down) for fresh install if [ -d /etc/sv/fenris-collect ] && [ ! -e /var/service/fenris-collect ]; then touch /etc/sv/fenris-collect/down diff --git a/pyproject.toml b/pyproject.toml index 4b83947..e14db06 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,8 +1,9 @@ [project] name = "fenris" -version = "0.3.5" +version = "0.3.6" description = "NVMe wear monitor with persistent TUI" -requires-python = ">=3.9" +requires-python = ">=3.10" +license = {file = "LICENSE"} dependencies = [ "textual>=0.40.0", ] diff --git a/requirements.txt b/requirements.txt index 68dc4d4..fd78d97 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,3 +9,4 @@ mdurl==0.1.2 platformdirs==4.11.7 Pygments==2.21.0 linkify-it-py==2.2.0 +typing-extensions==4.16.0 diff --git a/scripts/fenris b/scripts/fenris index 3cdac9b..745972a 100755 --- a/scripts/fenris +++ b/scripts/fenris @@ -7,8 +7,6 @@ Subcommands route through fenris-monitor for privileged operations. Spec: §1.2, §8.4 """ import argparse -import os -import subprocess import sys from pathlib import Path @@ -35,35 +33,16 @@ def add_runtime_packages() -> None: add_runtime_packages() -def is_root() -> bool: - """Check if running as root.""" - return os.geteuid() == 0 - - def run_monitor(*args: str) -> None: - """Run fenris-monitor with the given arguments. + """Render the shared privileged-action outcome for the CLI.""" + from fenris.control import MonitorError, run_monitor as invoke_monitor - If not root, re-exec under pkexec. - """ - monitor_cmd = "/usr/libexec/fenris/fenris-monitor" - - if is_root(): - result = subprocess.run([monitor_cmd] + list(args)) - sys.exit(result.returncode) - else: - # Use pkexec to elevate - pkexec = subprocess.run( - ["which", "pkexec"], capture_output=True - ) - if pkexec.returncode != 0: - print( - "Error: No polkit agent available. " - "Run as root: sudo fenris-monitor ...", - file=sys.stderr, - ) - sys.exit(1) - result = subprocess.run(["pkexec", monitor_cmd] + list(args)) - sys.exit(result.returncode) + try: + invoke_monitor(*args) + except MonitorError as exc: + print(str(exc), file=sys.stderr) + sys.exit(exc.exit_code) + sys.exit(0) def cmd_tui(args: argparse.Namespace) -> None: @@ -143,7 +122,7 @@ def main() -> None: description="Fenris NVMe endurance monitor", ) parser.add_argument( - "--version", action="version", version="%(prog)s 0.3.0" + "--version", action="version", version="%(prog)s 0.3.6" ) subparsers = parser.add_subparsers(dest="command") diff --git a/src/fenris/__init__.py b/src/fenris/__init__.py index 76e7d3b..4aff423 100644 --- a/src/fenris/__init__.py +++ b/src/fenris/__init__.py @@ -1,2 +1,2 @@ """Fenris: NVMe wear monitor with persistent TUI.""" -__version__ = "0.3.1" +__version__ = "0.3.6" diff --git a/src/fenris/control.py b/src/fenris/control.py new file mode 100644 index 0000000..4a36238 --- /dev/null +++ b/src/fenris/control.py @@ -0,0 +1,56 @@ +"""Terminal-attached invocation of Fenris's fixed privileged operations. + +Both human entry points use this module. Authentication has no frontend +deadline; collection runtime is bounded by the native scheduler (ADR 0003). +""" +import os +import shlex +import subprocess + + +MONITOR_HELPER = "/usr/libexec/fenris/fenris-monitor" + + +class MonitorError(Exception): + """An action failed, with a message and exit status for either renderer.""" + + def __init__(self, message: str, exit_code: int = 1): + super().__init__(message) + self.exit_code = exit_code + + +def run_monitor(*args: str, helper_path: str = MONITOR_HELPER) -> None: + """Run one helper operation, inheriting the terminal for authentication. + + Never invoke a shell, retry an action, or fall back to sudo automatically. + The helper owns the operation allow-list and privileged state changes. + """ + helper_command = [helper_path, *args] + needs_auth = os.geteuid() != 0 + command = ["pkexec", *helper_command] if needs_auth else helper_command + root_hint = ( + " If authentication is unavailable, run in your terminal: " + + shlex.join(["sudo", *helper_command]) + ) if needs_auth else "" + + try: + # The collector owns its 90-second runtime limit. A frontend timeout + # would also count time spent authenticating or waiting for a run. + result = subprocess.run(command) + except FileNotFoundError as exc: + raise MonitorError( + "Command not found: %s.%s" % (exc.filename or command[0], root_hint), 127, + ) from exc + except OSError as exc: + raise MonitorError("Cannot run monitoring action: %s.%s" % (exc, root_hint)) from exc + except KeyboardInterrupt as exc: + raise MonitorError( + "Action interrupted. Check fenris status before retrying.", 130, + ) from exc + + if result.returncode: + exit_code = result.returncode if result.returncode > 0 else 128 - result.returncode + raise MonitorError( + "Action failed (exit %d). Check fenris status before retrying.%s" + % (exit_code, root_hint), exit_code, + ) diff --git a/src/fenris/init_system.py b/src/fenris/init_system.py index 83f57d7..aa84e6d 100644 --- a/src/fenris/init_system.py +++ b/src/fenris/init_system.py @@ -166,11 +166,11 @@ def _systemd_query_state() -> Dict[str, Any]: "ActiveState", "ExecMainStatus", "ExecMainExitTimestamp", ) - boot_enabled_str = timer_props.get("UnitFileState", "") - boot_enabled = boot_enabled_str == "enabled" + boot_enabled_str = timer_props.get("UnitFileState") + boot_enabled = boot_enabled_str == "enabled" if boot_enabled_str else None - active_state = timer_props.get("ActiveState", "inactive") - timer_active = active_state == "active" + active_state = timer_props.get("ActiveState") + timer_active = active_state == "active" if active_state else None last_collect_ok = None last_collect_age_s = None diff --git a/src/fenris/status.py b/src/fenris/status.py index c7e9374..3e868bf 100644 --- a/src/fenris/status.py +++ b/src/fenris/status.py @@ -14,10 +14,14 @@ Freshness constants are defined once here and shared with the TUI (§8.9): Criteria: LC-9, CI-2, CI-4, FL-4, FL-5, FL-7. """ import sqlite3 +from contextlib import contextmanager import sys from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING + +if TYPE_CHECKING: + from .status_composition import StatusComposition from .projection import compute_projection, ConfidenceState, DISCLOSURES from .store import SCHEMA_VERSION @@ -98,7 +102,7 @@ def open_store_readonly(store_path: Path) -> sqlite3.Connection: # PermissionError before any StoreFault can be raised (issue #54). raise StoreFault("observation store not readable: %s" % e) if not exists: - raise StoreFault("observation store not found at %s" % store_path) + raise MissingStore("observation store not found at %s" % store_path) try: conn = sqlite3.connect("file:%s?mode=ro" % store_path, uri=True) @@ -125,6 +129,10 @@ class StoreFault(Exception): pass +class MissingStore(StoreFault): + """No observation history has been created yet.""" + + class NewerSchema(Exception): """Store has a newer user_version (§9.5).""" def __init__(self, version: int): @@ -246,27 +254,12 @@ def check_retired_flag(flag: str) -> Optional[str]: # Formatting # --------------------------------------------------------------------------- -def _format_projection(proj, freshness: str, service: Dict[str, Any], - drive_facts: List[str], config_error: Optional[str], - store_fault: Optional[str], newer_schema: Optional[str], - journal_hint: Optional[str], +def _format_projection(proj, freshness: str, drive_facts: List[str], + config_error: Optional[str], sample_count: int = 0, day_count: int = 0) -> str: - """Format the complete status output.""" + """Format projection details; monitoring status has its own renderer.""" lines = [] - # --- Store/system fault overrides (§9.4, §9.5) --- - if store_fault: - lines.append("observation store unreadable") - if journal_hint: - lines.append("") - lines.append("Recent journal entries:") - lines.append(journal_hint) - return "\n".join(lines) - - if newer_schema: - lines.append("observation store written by a newer Fenris — upgrade Fenris") - return "\n".join(lines) - # --- Configuration error (§8.3) --- if config_error: lines.append("configuration error: %s" % config_error) @@ -277,7 +270,6 @@ def _format_projection(proj, freshness: str, service: Dict[str, Any], lines.append("no observations yet") lines.append("") lines.append("Enable monitoring: fenris monitor resume") - _append_service_facts(lines, service) return "\n".join(lines) # --- Single sample: awaiting another sample (issue #73 AC3) --- @@ -287,7 +279,10 @@ def _format_projection(proj, freshness: str, service: Dict[str, Any], lines.append("awaiting another sample") lines.append("") lines.append("Collecting usage data — the first projection requires at least two samples.") - _append_service_facts(lines, service) + return "\n".join(lines) + + if proj is None: + lines.append("no projection available") return "\n".join(lines) # --- Projection headline --- @@ -332,16 +327,6 @@ def _format_projection(proj, freshness: str, service: Dict[str, Any], lines.append(fact) lines.append("") - # --- Four separate service facts (§7.3, LC-9) --- - _append_service_facts(lines, service) - - # --- Journal hint on failure or staleness (§8.8) --- - if journal_hint: - if freshness in ("missed", "stale"): - lines.append("") - lines.append("Recent journal entries:") - lines.append(journal_hint) - return "\n".join(lines) @@ -386,36 +371,6 @@ def _format_headline(proj) -> str: return headline -def _append_service_facts(lines: List[str], service: Dict[str, Any]) -> None: - """Append service facts and dashboard-clarity monitoring state.""" - boot = "enabled" if service.get("boot_enabled") else "disabled" - activity = "active" if service.get("timer_active") else "inactive" - - if service.get("last_collect_ok") is True: - collect = "ok" - elif service.get("last_collect_ok") is False: - collect = "FAILED" - if service.get("last_collect_reason"): - collect += " (%s)" % service["last_collect_reason"] - else: - collect = "unknown" - - collect_age = "" - if service.get("last_collect_age_s") is not None: - collect_age = " %s" % freshness_age_human(service["last_collect_age_s"]) - - freshness_str = service.get("freshness", "unknown") - freshness_age = "" - if service.get("freshness_age_s") is not None: - freshness_age = " (%s)" % freshness_age_human(service["freshness_age_s"]) - - lines.append("boot: %s · timer: %s · last collect: %s%s · freshness: %s%s" - % (boot, activity, collect, collect_age, freshness_str, freshness_age)) - lines.append("CONTINUITY: %s" % monitoring_continuity(service)) - if service.get("deliberately_paused"): - lines.extend(deliberate_pause_lines()) - - # --------------------------------------------------------------------------- # Dashboard clarity parity wording (DC-2, DC-3) # --------------------------------------------------------------------------- @@ -430,7 +385,9 @@ _PAUSED_CONSEQUENCE = ( def monitoring_continuity(service: Dict[str, Any]) -> str: """Return the boot-persistence wording, independent of timer runtime.""" - return _CONTINUITY_ACTIVE if service.get("boot_enabled") else _CONTINUITY_DISABLED + if service.get("boot_enabled") is None: + return "monitoring: boot persistence unknown" + return _CONTINUITY_ACTIVE if service["boot_enabled"] else _CONTINUITY_DISABLED def deliberate_pause_lines() -> List[str]: @@ -476,121 +433,92 @@ def format_disclosures() -> str: # Main status entry point # --------------------------------------------------------------------------- -def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None, - query_services: bool = True, query_journal: bool = True) -> str: - """Render the complete read-only status (§8.8, LC-9). +@contextmanager +def read_status( + store_path: Optional[Path] = None, + clock_now: Optional[datetime] = None, + query_services: bool = True, + collecting: bool = False, + reduced_motion: bool = False, +) -> Iterator[Tuple[Optional[sqlite3.Connection], "StatusComposition"]]: + """Yield a read-only store snapshot and its composed monitoring status. - This is the single entry point for 'fenris status'. It never auto-samples, - never prompts, and never writes to the store. + Own acquisition, fault classification, and connection lifetime for both + renderers. An absent store is empty; an unreadable or newer store exposes + no connection. Unknown monitoring facts are never coerced to disabled. """ - if clock_now is None: - clock_now = datetime.now(timezone.utc) + from .status_composition import compose_status - # --- Configuration (§8.3) --- - config_error = None - device = None - try: - config = read_config() - device = config["device"] - except ConfigError as e: - config_error = str(e) - - # --- Service state --- - service = {} + clock_now = clock_now or datetime.now(timezone.utc) + service = None if query_services: - service = query_service_state() - - # --- Store open --- - store_fault = None - newer_schema = None - conn = None - - if store_path is None: - store_path = Path("/var/lib/fenris/observations.db") - - try: - conn = open_store_readonly(store_path) - except StoreFault as e: - store_fault = str(e) - except NewerSchema as e: - newer_schema = str(e) - - # --- Store fault / newer schema short-circuit --- - if store_fault or newer_schema: - journal_hint = None - if query_journal: - journal_hint = _journalctl_hint() - service["freshness"] = "unknown" - service["freshness_age_s"] = None - return _format_projection( - None, "unknown", service, [], config_error, store_fault, newer_schema, journal_hint - ) - - # --- Freshness grading (§8.9) --- - try: - cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1") - row = cursor.fetchone() - newest_ts = row[0] if row else None - except sqlite3.Error: - newest_ts = None - - freshness = grade_freshness(newest_ts, clock_now) - - # --- Sample count for single-sample state (issue #73 AC3) --- - sample_count = 0 - day_count = 0 - try: - cursor = conn.execute("SELECT COUNT(*) FROM samples") - sample_count = cursor.fetchone()[0] - cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates") - day_count = cursor.fetchone()[0] - except sqlite3.Error: - pass - - # Freshness age for the service fact - freshness_age_s = None - if newest_ts: try: - ts = datetime.fromisoformat(newest_ts) - if ts.tzinfo is None: - ts = ts.replace(tzinfo=timezone.utc) - freshness_age_s = int((clock_now - ts).total_seconds()) - except (ValueError, TypeError): + service = query_service_state() + except (OSError, RuntimeError): pass - service["freshness"] = freshness - service["freshness_age_s"] = freshness_age_s + conn = None + store_fault = newer_schema = None try: - service["deliberately_paused"] = is_deliberately_paused(conn, service) - except sqlite3.Error: - service["deliberately_paused"] = False + try: + conn = open_store_readonly(store_path or Path("/var/lib/fenris/observations.db")) + conn.execute("BEGIN") + except MissingStore: + pass + except (StoreFault, sqlite3.Error) as exc: + store_fault = str(exc) + except NewerSchema as exc: + newer_schema = str(exc) - # --- Drive anomalies (§9.7, FL-7) --- - drive_facts = [] + comp = compose_status( + conn, service, clock_now, store_fault=store_fault, + newer_schema=newer_schema, collecting=collecting, + reduced_motion=reduced_motion, + ) + if conn is not None and (comp.store_fault or comp.newer_schema): + conn.close() + conn = None + yield conn, comp + finally: + if conn is not None: + conn.close() + + +def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None, + query_services: bool = True, query_journal: bool = True) -> str: + """Render CLI status through the shared read-only acquisition path.""" + from .status_composition import render_status_cli + + clock_now = clock_now or datetime.now(timezone.utc) + config_error = None try: - drive_facts = _query_drive_facts(conn) - except sqlite3.Error: - pass + read_config() + except ConfigError as exc: + config_error = str(exc) - # --- Projection (§6 — recomputed on read, never stored) --- - try: - proj = compute_projection(conn, clock_now) - except Exception: - proj = None - - # --- Journal hint on failure or staleness (§8.8) --- - journal_hint = None - if query_journal and freshness in ("missed", "stale"): - journal_hint = _journalctl_hint() - - # --- Compose output --- - result = _format_projection( - proj, freshness, service, drive_facts, config_error, - None, None, journal_hint, sample_count, day_count, - ) - - conn.close() - return result + with read_status(store_path, clock_now, query_services) as (conn, comp): + parts = [render_status_cli(comp)] + if not comp.store_fault and not comp.newer_schema: + drive_facts = [] + proj = None + if conn is not None: + try: + drive_facts = _query_drive_facts(conn) + proj = compute_projection(conn, clock_now) + except (sqlite3.Error, ValueError, TypeError): + pass + parts.append(_format_projection( + proj, comp.freshness, drive_facts, config_error, + comp.sample_count, comp.day_count, + )) + if query_journal and ( + comp.store_fault or comp.last_collect_ok is False + or comp.freshness in ("missed", "stale") + ): + hint = _journalctl_hint() + if hint: + parts.append("Recent collector logs:\n" + hint) + return "\n\n".join(part for part in parts if part) def render_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None, @@ -619,65 +547,8 @@ def get_status_composition( collecting: bool = False, reduced_motion: bool = False, ) -> 'StatusComposition': - """Get the shared status composition consumed by both TUI and CLI. - - This is the new entry point that centralizes the status lattice. - """ - # Lazy import to avoid circular dependency - from .status_composition import ( - StatusComposition, - compose_status, - ) - - if clock_now is None: - clock_now = datetime.now(timezone.utc) - - # --- Configuration (§8.3) --- - # Config errors are surfaced through the store fault mechanism - - # --- Service state --- - service = {} - if query_services: - try: - service = query_service_state() - except Exception: - service = None - - # --- Store open --- - store_fault = None - newer_schema = None - conn = None - - if store_path is None: - store_path = Path("/var/lib/fenris/observations.db") - - try: - conn = open_store_readonly(store_path) - except StoreFault as e: - store_fault = str(e) - except NewerSchema as e: - newer_schema = str(e) - - # --- Use shared composition --- - if conn is not None: - try: - comp = compose_status( - conn, service, clock_now, - store_fault=store_fault, - newer_schema=newer_schema, - collecting=collecting, - reduced_motion=reduced_motion, - ) - finally: - conn.close() - else: - # Store fault or newer schema - compose without store data - comp = compose_status( - None, service, clock_now, - store_fault=store_fault, - newer_schema=newer_schema, - collecting=collecting, - reduced_motion=reduced_motion, - ) - - return comp + """Return monitoring status without retaining the read-only snapshot.""" + with read_status( + store_path, clock_now, query_services, collecting, reduced_motion, + ) as (_, comp): + return comp diff --git a/src/fenris/status_composition.py b/src/fenris/status_composition.py index 4d0cacc..bb39452 100644 --- a/src/fenris/status_composition.py +++ b/src/fenris/status_composition.py @@ -188,7 +188,7 @@ def _determine_explanation( ) -> str: """Determine the explanation line for the status state.""" if store_fault: - return "observation store unreadable — see journal" + return "observation store unreadable — see collector logs" if newer_schema: return "observation store written by a newer Fenris — upgrade Fenris" @@ -200,7 +200,7 @@ def _determine_explanation( if last_collect_reason: parts.append("(%s)" % last_collect_reason) if freshness_age_s is not None and freshness != "empty": - parts.append("· last good sample %s ago" % freshness_age_human(freshness_age_s)) + parts.append("· last good sample %s" % freshness_age_human(freshness_age_s)) return " ".join(parts) if parts else "last run failed" if state == StatusState.INTERRUPTED: @@ -211,7 +211,7 @@ def _determine_explanation( if state == StatusState.STALE: if freshness_age_s is not None: - return "last sample %s ago" % freshness_age_human(freshness_age_s) + return "last sample %s" % freshness_age_human(freshness_age_s) return "data is stale" if state == StatusState.WAITING: @@ -223,7 +223,7 @@ def _determine_explanation( if state == StatusState.MONITORING: if freshness_age_s is not None: - return "last sample %s ago" % freshness_age_human(freshness_age_s) + return "last sample %s" % freshness_age_human(freshness_age_s) return "monitoring active" if state == StatusState.UNKNOWN: @@ -257,7 +257,7 @@ def _determine_collecting_overlay( def compose_status( - conn: sqlite3.Connection, + conn: Optional[sqlite3.Connection], service: Optional[Dict[str, Any]], clock_now: datetime, store_fault: Optional[str] = None, @@ -269,14 +269,17 @@ def compose_status( This is the single entry point consumed by both TUI and CLI. """ - service_available = service is not None and len(service) > 0 + service_available = bool(service) and any( + service.get(key) is not None + for key in ("boot_enabled", "timer_active") + ) # --- Separate facts from service --- - boot_enabled = service.get("boot_enabled") if service_available else None - timer_active = service.get("timer_active") if service_available else None - last_collect_ok = service.get("last_collect_ok") if service_available else None - last_collect_age_s = service.get("last_collect_age_s") if service_available else None - last_collect_reason = service.get("last_collect_reason") if service_available else None + boot_enabled = service.get("boot_enabled") if service else None + timer_active = service.get("timer_active") if service else None + last_collect_ok = service.get("last_collect_ok") if service else None + last_collect_age_s = service.get("last_collect_age_s") if service else None + last_collect_reason = service.get("last_collect_reason") if service else None # --- Freshness from store --- freshness = "unknown" @@ -285,7 +288,10 @@ def compose_status( day_count = 0 deliberately_paused = False - if store_fault is None and newer_schema is None: + if conn is None and store_fault is None and newer_schema is None: + freshness = "empty" + + if conn is not None and store_fault is None and newer_schema is None: try: cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1") row = cursor.fetchone() @@ -301,28 +307,35 @@ def compose_status( freshness_age_s = int((clock_now - ts).total_seconds()) except (ValueError, TypeError): pass - except sqlite3.Error: - freshness = "unknown" + except sqlite3.Error as exc: + store_fault = str(exc) try: cursor = conn.execute("SELECT COUNT(*) FROM samples") sample_count = cursor.fetchone()[0] cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates") day_count = cursor.fetchone()[0] - except sqlite3.Error: - pass + except sqlite3.Error as exc: + store_fault = str(exc) try: svc_for_pause = service if service_available else {} deliberately_paused = is_deliberately_paused(conn, svc_for_pause) - except sqlite3.Error: - deliberately_paused = False + except sqlite3.Error as exc: + store_fault = str(exc) + + if store_fault or newer_schema: + freshness = "unknown" + freshness_age_s = None + sample_count = day_count = 0 + deliberately_paused = False # --- External stop detection --- # External stop = timer inactive + boot disabled + NOT deliberately paused # + period still open (the timer was stopped but Fenris didn't close the period) external_stop_reason = None - if not deliberately_paused and timer_active is False and boot_enabled is False: + if (conn is not None and not store_fault and not newer_schema + and not deliberately_paused and timer_active is False and boot_enabled is False): # Check if there's an open monitoring period (external stop left it open) try: open_period = conn.execute( @@ -428,10 +441,9 @@ def render_status_cli(comp: StatusComposition) -> str: # Separate facts facts = [] - if comp.freshness != "unknown": - facts.append("freshness: %s" % comp.freshness) - if comp.freshness_age_s is not None: - facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s) if facts else "freshness: %s" % freshness_age_human(comp.freshness_age_s) + facts.append("freshness: %s" % comp.freshness) + if comp.freshness_age_s is not None and facts: + facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s) if comp.last_collect_ok is True: facts.append("last collect: ok") elif comp.last_collect_ok is False: @@ -441,10 +453,12 @@ def render_status_cli(comp: StatusComposition) -> str: facts.append(collect_str) else: facts.append("last collect: unknown") - if comp.boot_enabled is not None: - facts.append("boot: %s" % ("enabled" if comp.boot_enabled else "disabled")) - if comp.timer_active is not None: - facts.append("timer: %s" % ("active" if comp.timer_active else "inactive")) + facts.append("boot: %s" % ( + "unknown" if comp.boot_enabled is None else "enabled" if comp.boot_enabled else "disabled" + )) + facts.append("timer: %s" % ( + "unknown" if comp.timer_active is None else "active" if comp.timer_active else "inactive" + )) if facts: lines.append(" · ".join(facts)) @@ -486,29 +500,30 @@ def render_status_tui(comp: StatusComposition) -> str: # Explanation if comp.explanation: - lines.append(comp.explanation) + lines.append(comp.explanation[:1].upper() + comp.explanation[1:]) lines.append("") # Separate facts facts = [] - if comp.freshness != "unknown": - facts.append("freshness: %s" % comp.freshness) + facts.append("Freshness: %s" % comp.freshness) if comp.freshness_age_s is not None and facts: facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s) if comp.last_collect_ok is True: - facts.append("last collect: ok") + facts.append("Last collect: ok") elif comp.last_collect_ok is False: - collect_str = "last collect: FAILED" + collect_str = "Last collect: failed" if comp.last_collect_reason: collect_str += " (%s)" % comp.last_collect_reason facts.append(collect_str) else: - facts.append("last collect: unknown") - if comp.boot_enabled is not None: - facts.append("boot: %s" % ("enabled" if comp.boot_enabled else "disabled")) - if comp.timer_active is not None: - facts.append("timer: %s" % ("active" if comp.timer_active else "inactive")) + facts.append("Last collect: unknown") + facts.append("Boot: %s" % ( + "unknown" if comp.boot_enabled is None else "enabled" if comp.boot_enabled else "disabled" + )) + facts.append("Timer: %s" % ( + "unknown" if comp.timer_active is None else "active" if comp.timer_active else "inactive" + )) if facts: lines.append(" · ".join(facts)) @@ -516,11 +531,11 @@ def render_status_tui(comp: StatusComposition) -> str: # Continuity if comp.continuity: lines.append("") - lines.append("[bold]CONTINUITY[/bold] %s" % comp.continuity) + lines.append("[bold]Continuity[/bold] %s" % comp.continuity) # Deliberate pause if comp.paused_lines: for pl in comp.paused_lines: - lines.append(pl) + lines.append(pl[:1].upper() + pl[1:]) return "\n".join(lines) diff --git a/src/fenris/tui.py b/src/fenris/tui.py index d2a2724..2e5249b 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -14,14 +14,13 @@ Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8. """ from __future__ import annotations -import os import sqlite3 -import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional +from rich.text import Text from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container, Horizontal, VerticalScroll @@ -39,27 +38,19 @@ from .status import ( CADENCE_DEFAULT_S, FRESH_THRESHOLD_S, STALENESS_THRESHOLD_S, - ConfigError, - NewerSchema, - StoreFault, format_disclosures, freshness_age_human, grade_freshness, deliberate_pause_lines, - is_deliberately_paused, - monitoring_continuity, - open_store_readonly, - query_service_state, - read_config, + read_status, + _journalctl_hint, ) -from .monitoring_periods import get_open_period from .status_composition import ( - StatusState, StatusComposition, - compose_status, render_status_tui, STATUS_POLL_INTERVAL_S, ) +from .control import MONITOR_HELPER, MonitorError, run_monitor from .preferences import load_preferences, save_preferences from .themes import THEMES, THEME_NAMES, get_theme, get_graph_colors @@ -68,9 +59,10 @@ from .themes import THEMES, THEME_NAMES, get_theme, get_graph_colors # Helpers # --------------------------------------------------------------------------- -_RESUME_HINT = "r resume — enable monitoring and future boots" +_RESUME_HINT = "r Resume — enable monitoring and future boots" _ACTION_LEGEND = ( - "p pause · " + _RESUME_HINT + " · c collect · t theme · m motion · d disclosures" + "p Pause · " + _RESUME_HINT + "\n" + "c Collect now · t Theme · m Motion · d Disclosures · ? Help" ) def _format_remaining(seconds: float) -> str: @@ -91,43 +83,6 @@ def _format_remaining(seconds: float) -> str: return " ".join(parts) -def _sparkline(values: List[float], width: int = 40) -> str: - """Render a sparkline from daily bytes-written values.""" - if not values: - return "" - mx = max(values) or 1.0 - blocks = " ▁▂▃▄▅▆▇█" - step = max(1, len(values) // width or 1) - picked = values[-width * step :][::step][-width:] - return "".join( - blocks[min(len(blocks) - 1, int(v / mx * (len(blocks) - 1)) + (1 if v > 0 else 0))] - for v in picked - ) - - -def _habit_bar(a: float, i: float, o: float, u: float, width: int = 40) -> str: - """Render the habit-split bar with legend.""" - total = a + i + o + u or 1.0 - segs = [ - ("a", a, "#33ff33"), - ("i", i, "#ffff33"), - ("o", o, "#33ffff"), - ("?", u, "#ff33ff"), - ] - parts = [] - for label, v, _color in segs: - n = max(1 if v > 0 else 0, round(v / total * width)) - parts.append(label * n) - bar = "".join(parts) - legend = " active %d%% · idle %d%% · powered-off %d%% · unknown %d%%" % ( - round(a / total * 100), - round(i / total * 100), - round(o / total * 100), - round(u / total * 100), - ) - return bar + "\n" + legend - - # --------------------------------------------------------------------------- # Identity and wolf glyph detection (issue #79) # --------------------------------------------------------------------------- @@ -274,8 +229,8 @@ class DailyBarGraph(Widget): self._render_readout() def _show_empty(self) -> None: - self.query_one("#bar-range").update("[dim]usage history[/dim]") - self.query_one("#bar-render").update("[dim]awaiting first sample[/dim]") + self.query_one("#bar-range").update("[dim]Usage history[/dim]") + self.query_one("#bar-render").update("[dim]Awaiting first sample[/dim]") self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update("") @@ -296,9 +251,9 @@ class DailyBarGraph(Widget): def _show_constrained_summary(self) -> None: """Textual fallback for terminals below 80×24.""" - self.query_one("#bar-range").update("[dim]usage history[/dim]") + self.query_one("#bar-range").update("[dim]Usage history[/dim]") if not self._day_data: - self.query_one("#bar-render").update("[dim]graph needs ≥80×24[/dim]") + self.query_one("#bar-render").update("[dim]Graph needs ≥80×24[/dim]") self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update("") return @@ -310,7 +265,7 @@ class DailyBarGraph(Widget): first = self._day_data[0].get("local_label", "") last = self._day_data[-1].get("local_label", "") self.query_one("#bar-render").update( - "[dim]graph needs ≥80×24[/dim]\n" + "[dim]Graph needs ≥80×24[/dim]\n" " %d days · %d with writes · %.3f GB total\n" " %s → %s" % (n, days_with_data, total_bytes / 1e9, first, last) ) @@ -324,7 +279,7 @@ class DailyBarGraph(Widget): % (day.get("local_label", ""), day.get("total_bytes", 0) / 1e9) ) else: - self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select[/dim]") + self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select[/dim]") def _show_constrained_hourly_summary(self) -> None: """Textual fallback for hourly view when terminal is too small.""" @@ -334,7 +289,7 @@ class DailyBarGraph(Widget): ) n = len(self._hour_data) self.query_one("#bar-render").update( - "[dim]graph needs ≥80×24[/dim]\n" + "[dim]Graph needs ≥80×24[/dim]\n" " %d hours · %d with writes · %.3f GB total" % (n, hours_with_data, total_bytes / 1e9) ) self.query_one("#bar-legend").update("") @@ -346,7 +301,7 @@ class DailyBarGraph(Widget): % (h.get("local_label", ""), h.get("bytes_written", 0) / 1e9) ) else: - self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]") + self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]") def on_resize(self) -> None: """Re-render when terminal size changes.""" @@ -362,9 +317,9 @@ class DailyBarGraph(Widget): parts.append("[bold]%d[/bold]" % r) else: parts.append(str(r)) - label = "range: " + " / ".join(parts) + label = "Range: " + " / ".join(parts) if self.view_mode == "hourly": - label += " \u00b7 [bold]%s[/bold] \u00b7 esc back" % (self.drill_day or "") + label += " \u00b7 [bold]%s[/bold] \u00b7 Esc Back" % (self.drill_day or "") self.query_one("#bar-range").update(label) def _render_bars(self) -> None: @@ -436,7 +391,7 @@ class DailyBarGraph(Widget): def _render_legend(self) -> None: legend = ( - "%s alloc %s unalloc %s gap %s zero %s partial" + "%s Alloc %s Unalloc %s Gap %s Zero %s Partial" % ( _GLYPH_ALLOCATED, _GLYPH_UNALLOCATED, @@ -450,7 +405,7 @@ class DailyBarGraph(Widget): def _render_readout(self) -> None: if self.selected_index < 0 or self.selected_index >= len(self._day_data): self.query_one("#bar-readout").update( - "[dim]\u2190 \u2192 select \u00b7 1-4 range \u00b7 Enter drill[/dim]" + "[dim]\u2190 \u2192 Select \u00b7 1-4 Range \u00b7 Enter Hourly view[/dim]" ) return @@ -473,7 +428,7 @@ class DailyBarGraph(Widget): ] if unallocated > 0: parts.append( - " allocated %.3f GB \u00b7 unallocated %.3f GB" + " Allocated %.3f GB \u00b7 unallocated %.3f GB" % (allocated / 1e9, unallocated / 1e9) ) self.query_one("#bar-readout").update("\n".join(parts)) @@ -482,7 +437,7 @@ class DailyBarGraph(Widget): def _refresh_hourly(self) -> None: if not self._hour_data: - self.query_one("#bar-render").update("[dim]no hourly data[/dim]") + self.query_one("#bar-render").update("[dim]No hourly data[/dim]") return self._render_range() @@ -533,7 +488,7 @@ class DailyBarGraph(Widget): self.query_one("#bar-render").update("\n".join(lines)) self.query_one("#bar-legend").update( - "%s writes \u00b7 %s zero" % (_GLYPH_ALLOCATED, _GLYPH_ZERO) + "%s Writes \u00b7 %s Zero" % (_GLYPH_ALLOCATED, _GLYPH_ZERO) ) # Hourly readout @@ -548,7 +503,7 @@ class DailyBarGraph(Widget): ) ) else: - self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]") + self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]") # -- Event handling -- @@ -754,51 +709,6 @@ def _query_hourly_graph_data( # Data queries for TUI regions # --------------------------------------------------------------------------- -def _query_usage_history(conn: sqlite3.Connection) -> Dict[str, Any]: - """Query usage-history data for the left pane.""" - cursor = conn.execute( - "SELECT day, active_seconds, idle_seconds, powered_off_seconds, " - "unknown_seconds, bytes_written_delta, coverage " - "FROM day_aggregates ORDER BY day" - ) - days = cursor.fetchall() - if not days: - return { - "sparkline": "", - "habit_bar": "", - "num_days": 0, - "min_gb": 0, - "max_gb": 0, - "habit_change": False, - "gap_days": [], - "day_labels": [], - } - - bw_values = [d[5] for d in days] - total_a = sum(d[1] for d in days) - total_i = sum(d[2] for d in days) - total_o = sum(d[3] for d in days) - total_u = sum(d[4] for d in days) - total = total_a + total_i + total_o + total_u or 1 - - spark = _sparkline([b / 1e9 for b in bw_values]) # Convert to GB for display - bar = _habit_bar(total_a / total, total_i / total, total_o / total, total_u / total) - - min_gb = min(bw_values) / 1e9 if bw_values else 0 - max_gb = max(bw_values) / 1e9 if bw_values else 0 - - return { - "sparkline": spark, - "habit_bar": bar, - "num_days": len(days), - "min_gb": min_gb, - "max_gb": max_gb, - "habit_change": False, - "gap_days": [], - "day_labels": [d[0] for d in days], - } - - def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]: """Query drive health data for the right pane.""" cursor = conn.execute( @@ -840,41 +750,6 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]: } -def _query_service_facts(conn: sqlite3.Connection, clock_now: datetime) -> Dict[str, Any]: - """Query service facts for the bottom strip.""" - # Get freshness - cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1") - row = cursor.fetchone() - newest_ts = row[0] if row else None - freshness = grade_freshness(newest_ts, clock_now) - - # Get monitoring period - period = get_open_period(conn) - period_info = "no monitoring period" - if period: - period_info = "open since %s" % period["started_at"][:10] - - # Get service state - try: - svc = query_service_state() - except Exception: - svc = { - "boot_enabled": False, - "timer_active": False, - "last_collect_ok": None, - "last_collect_age_s": None, - "last_collect_reason": None, - } - - return { - "freshness": freshness, - "freshness_age_s": None, - "period": period_info, - "deliberately_paused": is_deliberately_paused(conn, svc), - **svc, - } - - # --------------------------------------------------------------------------- # Modal screens # --------------------------------------------------------------------------- @@ -894,7 +769,7 @@ class ConfirmPause(ModalScreen[bool]): "This closes the current monitoring period.\n" "Paused time is [bold]excluded[/bold] from your usage habit\n" "(powered-off time would still count).\n\n" - "[dim]y pause · n cancel[/dim]", + "[dim]y Pause · n Cancel[/dim]", id="confirm-text", ) @@ -915,7 +790,44 @@ class DisclosuresScreen(ModalScreen[None]): def compose(self) -> ComposeResult: text = format_disclosures() - yield Static(text + "\n\n[dim]esc to close[/dim]", id="disc-text") + with VerticalScroll(): + yield Static(text + "\n\n[dim]Esc Close[/dim]", id="disc-text") + + def action_close(self) -> None: + self.dismiss() + + +class HelpScreen(ModalScreen[None]): + """Persistent keyboard and privilege guidance.""" + + BINDINGS = [Binding("escape", "close", "Close"), Binding("?", "close", "Close")] + + def compose(self) -> ComposeResult: + with VerticalScroll(): + yield Static( + "[bold]Using Fenris[/bold]\n\n" + "Open the dashboard with [bold]fenris[/bold] as your normal user.\n" + "The dashboard does not need sudo. Pause, resume, and collect now\n" + "authenticate through polkit when needed.\n\n" + "[bold]When to use sudo[/bold]\n" + "Use sudo for installation and system configuration.\n" + "For observation store access, an administrator can run:\n" + ' sudo usermod -aG fenris "$USER"\n' + "Then log out and back in before opening Fenris.\n\n" + "If polkit authentication is unavailable, quit with q and run\n" + "only the required action in your terminal:\n" + " sudo fenris monitor resume\n" + " sudo fenris monitor pause\n" + " sudo fenris sample\n" + "Then reopen the dashboard with fenris.\n\n" + "[bold]Keyboard controls[/bold]\n" + + _ACTION_LEGEND + "\n" + "q Quit the dashboard — monitoring continues\n" + "Tab Focus the graph · ← → Select · 1-4 Change range\n" + "Enter Show hours · Esc Return to days\n\n" + "[dim]↑ ↓ Scroll · Esc Close[/dim]", + id="help-text", + ) def action_close(self) -> None: self.dismiss() @@ -936,12 +848,12 @@ class FenrisTuiApp(App): layout: grid; grid-size: 2 4; grid-columns: 3fr 2fr; - grid-rows: 8 10 7 3; + grid-rows: auto 10 auto 3; height: auto; } #main-grid.paused { grid-size: 2 5; - grid-rows: 8 5 10 7 3; + grid-rows: auto auto 10 auto 3; } #main-grid.constrained { grid-size: 1 4; @@ -966,15 +878,16 @@ class FenrisTuiApp(App): min-height: 3; } #dashboard-scroll { height: 1fr; } - #headline-band { column-span: 2; } + #headline-band { column-span: 2; height: auto; min-height: 8; } #paused-banner { column-span: 2; display: none; background: $error 20%; color: $text; - height: 100%; + height: auto; + min-height: 5; } - #service-strip { column-span: 2; height: 100%; } + #service-strip { column-span: 2; height: auto; } #quit-rail { column-span: 2; border: heavy $accent; @@ -984,15 +897,17 @@ class FenrisTuiApp(App): .pane { border: round #555555; padding: 0 1; height: 100%; } #confirm-text { padding: 1 2; } #disc-text { padding: 1 2; } + #help-text { padding: 1 2; } """ BINDINGS = [ Binding("p", "pause", "Pause", show=False), Binding("r", "resume", "Resume", show=False), - Binding("c", "collect", "Collect Now", show=False), + Binding("c", "collect", "Collect now", show=False), Binding("d", "disclose", "Disclosures", show=False), Binding("t", "toggle_theme", "Theme", show=False), Binding("m", "toggle_motion", "Motion", show=False), + Binding("?", "help", "Help", show=False), Binding("q", "quit", "Quit", show=False), ] @@ -1012,10 +927,9 @@ class FenrisTuiApp(App): super().__init__(**kwargs) self.store_path = store_path or Path("/var/lib/fenris/observations.db") self.config_path = config_path - self.helper_path = helper_path or "/usr/libexec/fenris/fenris-monitor" + self.helper_path = helper_path or MONITOR_HELPER self.refresh_interval_s = refresh_interval_s self._show_auth_notice = True - self._conn: Optional[sqlite3.Connection] = None self._clock_now = datetime.now(timezone.utc) # Register all Fenris themes @@ -1034,7 +948,7 @@ class FenrisTuiApp(App): yield Static("", id="constrained-summary", classes="pane") yield Static("", id="drive-health", classes="pane") yield Static("", id="service-strip", classes="pane") - yield Static("q QUIT TUI", id="quit-rail") + yield Static("q Quit TUI", id="quit-rail") def on_mount(self) -> None: """Set border titles, apply theme, and render initial state.""" @@ -1042,34 +956,19 @@ class FenrisTuiApp(App): # Preference uses underscores (high_contrast); Textual theme uses dashes (fenris-high-contrast) self.theme = "fenris-%s" % self._current_theme_name.replace("_", "-") - self.query_one("#headline-band").border_title = "headline" - self.query_one("#usage-history").border_title = "usage history" - self.query_one("#constrained-summary").border_title = "usage history" - self.query_one("#drive-health").border_title = "drive" - self.query_one("#service-strip").border_title = "service + actions" + self.query_one("#headline-band").border_title = "Headline" + self.query_one("#usage-history").border_title = "Usage history" + self.query_one("#constrained-summary").border_title = "Usage history" + self.query_one("#drive-health").border_title = "Drive" + self.query_one("#service-strip").border_title = "Monitoring and actions" self._refresh_timer = self.set_interval( self.refresh_interval_s, self.on_refresh_tick ) self._refresh() def on_resize(self) -> None: - """Handle terminal resize to switch between constrained and normal layout (issue #81).""" - width, height = self.size - was_constrained = self._is_constrained_mode - self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT - - main_grid = self.query_one("#main-grid") - if self._is_constrained_mode: - main_grid.add_class("constrained") - else: - main_grid.remove_class("constrained") - - # Re-render constrained summary when entering constrained mode - if self._is_constrained_mode and not was_constrained: - self._update_constrained_summary() - # Re-render full layout when leaving constrained mode - elif not self._is_constrained_mode and was_constrained: - self._refresh() + """Re-render with the new terminal size after Textual applies it.""" + self.call_after_refresh(self._refresh) def _update_constrained_summary(self) -> None: """Update the textual summary shown when terminal is too small for graph.""" @@ -1077,7 +976,7 @@ class FenrisTuiApp(App): summary = self.query_one("#constrained-summary") if not graph._day_data: - summary.update("[dim]graph needs ≥80×24[/dim]\nawaiting first sample") + summary.update("[dim]Graph needs ≥80×24[/dim]\nAwaiting first sample") return # Summarise visible days as text (issue #81 AC2) @@ -1087,7 +986,7 @@ class FenrisTuiApp(App): first = graph._day_data[0].get("local_label", "") last = graph._day_data[-1].get("local_label", "") text = ( - "[dim]graph needs ≥80×24[/dim]\n" + "[dim]Graph needs ≥80×24[/dim]\n" " %d days · %d with writes · %.3f GB total\n" " %s → %s" % (n, days_with_data, total_bytes / 1e9, first, last) ) @@ -1116,7 +1015,7 @@ class FenrisTuiApp(App): identity = _IDENTITY_FALLBACK lines = ["[bold]%s[/bold]" % identity] if self._show_auth_notice: - lines.append("[dim]privileged actions will prompt for authentication (polkit)[/dim]") + lines.append("[dim]Open with fenris (no sudo). Actions authenticate via polkit. ? Help[/dim]") return "\n".join(lines) def _render_headline(self, body: str = "") -> None: @@ -1126,94 +1025,67 @@ class FenrisTuiApp(App): text += "\n\n" + body self.query_one("#headline-band").update(text) - def _open_store(self) -> Optional[sqlite3.Connection]: - """Open store read-only, handling faults.""" - try: - return open_store_readonly(self.store_path) - except (StoreFault, NewerSchema): - return None - def _refresh(self) -> None: """Refresh all four regions from store data.""" self._clock_now = datetime.now(timezone.utc) - # Also update constrained state on each refresh (issue #81) - try: - width, height = self.size - was_constrained = self._is_constrained_mode - self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT - main_grid = self.query_one("#main-grid") - if self._is_constrained_mode: - main_grid.add_class("constrained") - else: - main_grid.remove_class("constrained") - if self._is_constrained_mode and not was_constrained: - self._update_constrained_summary() - except Exception: - pass - conn = self._open_store() - - if conn is None: - self._render_empty_or_fault() - return - - try: - self._conn = conn - self._render_all_regions(conn) - finally: - conn.close() - self._conn = None - - def _render_empty_or_fault(self) -> None: - """Render empty store greeting or store fault.""" - self._hide_paused_banner() - if not self.store_path.exists(): - # Empty store — greeting with enable hint (IN-3) - self._render_headline( - "[bold]No observations yet[/bold]\n\n" - "[bold]%s[/bold]" % _RESUME_HINT - ) - self.query_one("#usage-history").set_data([]) - if self._is_constrained_mode: - self.query_one("#constrained-summary").update( - "[dim]graph needs ≥80×24[/dim]\nawaiting first sample" - ) - self.query_one("#drive-health").update("") - self.query_one("#service-strip").update( - ( - "boot: disabled · timer: inactive · last collect: unknown · freshness: empty\n" - "[bold]CONTINUITY[/bold] %s\n" + _ACTION_LEGEND - ) % monitoring_continuity({"boot_enabled": False}) - ) + width, height = self.size + self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT + main_grid = self.query_one("#main-grid") + if self._is_constrained_mode: + main_grid.add_class("constrained") else: - # Store fault (FL-4) - self._render_headline( - "[bold red]Observation store unreadable[/bold red]\n" - "Check journalctl -u fenris-collect.service" - ) - self.query_one("#usage-history").set_data([]) - if self._is_constrained_mode: - self.query_one("#constrained-summary").update( - "[dim]graph needs ≥80×24[/dim]\nobservation store unreadable" - ) - self.query_one("#drive-health").update("") + main_grid.remove_class("constrained") + with read_status( + self.store_path, self._clock_now, reduced_motion=self._reduced_motion, + ) as (conn, comp): + if conn is None: + self._render_empty_or_fault(comp) + else: + self._render_all_regions(conn, comp) self.query_one("#service-strip").update( - _ACTION_LEGEND + "%s\n%s" % (render_status_tui(comp), _ACTION_LEGEND) + ) + self._render_paused_banner(comp) + + def _render_empty_or_fault(self, comp: StatusComposition) -> None: + """Render empty history or the shared store-fault classification.""" + if comp.newer_schema or comp.store_fault: + message = comp.explanation[:1].upper() + comp.explanation[1:] + self._render_headline("[bold red]%s[/bold red]" % message) + if comp.store_fault: + hint = _journalctl_hint() + if hint: + self.query_one("#drive-health").update(Text(hint)) + else: + self.query_one("#drive-health").update( + "Check observation store permissions and collector logs. ? Help" + ) + else: + self.query_one("#drive-health").update("") + else: + message = "Awaiting first sample" + self._render_headline( + "[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT + ) + self.query_one("#drive-health").update("") + + graph = self.query_one("#usage-history") + graph.set_data([]) + if comp.store_fault or comp.newer_schema: + self.query_one("#bar-render").update(message) + if self._is_constrained_mode: + self.query_one("#constrained-summary").update( + "[dim]Graph needs ≥80×24[/dim]\n" + message ) - def _render_all_regions(self, conn: sqlite3.Connection) -> None: + def _render_all_regions(self, conn: sqlite3.Connection, comp: StatusComposition) -> None: """Render all four regions from live store data.""" - # --- Sample count for single-sample state (issue #73 AC3) --- - try: - cursor = conn.execute("SELECT COUNT(*) FROM samples") - sample_count = cursor.fetchone()[0] - cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates") - day_count = cursor.fetchone()[0] - except Exception: - sample_count = 0 - day_count = 0 - # --- Headline band (§7.2) --- - if sample_count <= 1 and day_count == 0: + if comp.sample_count == 0 and comp.day_count == 0: + self._render_headline( + "[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT + ) + elif comp.sample_count <= 1 and comp.day_count == 0: # Single sample: awaiting another sample self._render_headline( "[bold]Awaiting another sample[/bold]\n\n" @@ -1253,10 +1125,10 @@ class FenrisTuiApp(App): # Vendor wear grouped under Drive health with context health_text = ( "[bold]Drive health[/bold] · %s\n" - " temperature %d°C · spare %d%%\n" - " media errors %d · unsafe shutdowns %d\n" - " power-on %d h · %d cycles · %s\n" - " vendor wear: %d%% used · %.1f TB written" + " Temperature %d°C · spare %d%%\n" + " Media errors %d · unsafe shutdowns %d\n" + " Power-on %d h · %d cycles · %s\n" + " Vendor wear: %d%% used · %.1f TB written" ) % ( health["model"], health["temp"], @@ -1271,30 +1143,6 @@ class FenrisTuiApp(App): ) self.query_one("#drive-health").update(health_text) - # --- Service strip (§7.2 bottom) --- - try: - svc = _query_service_facts(conn, self._clock_now) - # Use shared status composition - comp = compose_status( - conn, svc, self._clock_now, - store_fault=None, - newer_schema=None, - reduced_motion=self._reduced_motion, - ) - # Render using the shared composition with TUI styling - status_text = render_status_tui(comp) - # Add TUI-only actions - self.query_one("#service-strip").update( - "%s\n%s" % (status_text, _ACTION_LEGEND) - ) - self._render_paused_banner(comp) - except Exception: - self._hide_paused_banner() - self.query_one("#service-strip").update( - "boot: unknown · timer: unknown · last collect: unknown · freshness: unknown\n" - + _ACTION_LEGEND - ) - def _render_paused_banner(self, comp) -> None: """Show the high-contrast Deliberate disable block only when sanctioned. @@ -1312,7 +1160,7 @@ class FenrisTuiApp(App): if is_paused and paused_lines: banner.update( "[bold black on red]%s[/bold black on red]\n%s" - % tuple(paused_lines) + % tuple(line[:1].upper() + line[1:] for line in paused_lines) ) banner.styles.display = "block" main_grid = self.query_one("#main-grid") @@ -1331,16 +1179,12 @@ class FenrisTuiApp(App): def _on_graph_drill(self, day: str) -> None: """Load hourly data when the graph enters drill-down mode.""" graph = self.query_one("#usage-history") - try: - conn = self._open_store() - if conn is not None: - try: - hour_data = _query_hourly_graph_data(conn, day) - graph.set_hour_data(hour_data) - finally: - conn.close() - except Exception: - graph.set_hour_data([]) + with read_status( + self.store_path, self._clock_now, query_services=False, + ) as (conn, _): + graph.set_hour_data( + _query_hourly_graph_data(conn, day) if conn is not None else [] + ) def _format_headline(self, proj: ProjectionResult) -> str: """Format the lifespan headline (spec §6.11).""" @@ -1357,7 +1201,7 @@ class FenrisTuiApp(App): regime = " · sustained regime: %d days" % proj.regime_days return ( "[bold]Usage-adjusted theoretical lifespan: [white]%s remaining[/white][/bold]" - "\n if current habits continue%s" % (remaining, regime) + "\n If current habits continue%s" % (remaining, regime) ) def _format_confidence(self, proj: ProjectionResult) -> str: @@ -1377,7 +1221,7 @@ class FenrisTuiApp(App): color, proj.confidence_state.value, color, - facts, + facts[:1].upper() + facts[1:], ) def _format_scenario(self, proj: ProjectionResult) -> str: @@ -1436,48 +1280,29 @@ class FenrisTuiApp(App): def action_collect(self) -> None: """Collect now — synchronous outcome (spec §8.7, LC-8).""" - self._run_helper("collect", blocking=True) + self._run_helper("collect") def action_disclose(self) -> None: """Show disclosures (spec §6.11, CI-4).""" self.push_screen(DisclosuresScreen()) + def action_help(self) -> None: + """Show keyboard controls and sudo guidance.""" + self.push_screen(HelpScreen()) + def _run_helper( self, operation: str, extra_args: Optional[List[str]] = None, - blocking: bool = False, ) -> None: - """Run fenris-monitor as terminal-attached subprocess (LC-6, LC-8). - - The TUI suspends, polkit agent prompts on real terminal, control returns. - """ - cmd = [self.helper_path, operation] - if extra_args: - cmd.extend(extra_args) - if os.geteuid() != 0: - cmd.insert(0, "pkexec") - + """Suspend the TUI while the shared control module owns the action.""" try: with self.suspend(): - proc = subprocess.run(cmd, timeout=30) - if proc.returncode != 0: - self.notify( - "Operation failed (exit %d)" % proc.returncode, - severity="error", - ) - except FileNotFoundError: - self.notify( - "Helper not found: %s" % self.helper_path, - severity="error", - ) - except subprocess.TimeoutExpired: - self.notify("Operation timed out", severity="error") - except Exception as e: - self.notify("Error: %s" % e, severity="error") - - # Refresh after action - self._refresh() + run_monitor(operation, *(extra_args or []), helper_path=self.helper_path) + except MonitorError as exc: + self.notify(str(exc), severity="error") + finally: + self._refresh() def run_tui( diff --git a/tests/test_acceptance_sweep.py b/tests/test_acceptance_sweep.py index 009bfd0..8bbbc68 100644 --- a/tests/test_acceptance_sweep.py +++ b/tests/test_acceptance_sweep.py @@ -461,7 +461,7 @@ class TestCI2Parity: } with patch("fenris.status.query_service_state", return_value=service_state), patch( - "fenris.tui.query_service_state", return_value=service_state + "fenris.status.query_service_state", return_value=service_state ): status = get_status( store_path=db, clock_now=_clock(), query_services=True, query_journal=False @@ -502,18 +502,6 @@ class TestCI2Parity: query_services=True, query_journal=False) assert "no observations yet" in status.lower() - def test_store_fault_phrase_both_views(self, tmp_path): - status_src = (FENRIS_PKG / "status.py").read_text() - tui_src = (FENRIS_PKG / "tui.py").read_text() - phrase = "observation store unreadable" - assert phrase in status_src - assert phrase.lower() in tui_src.lower() - - def test_newer_schema_phrase_both_views(self): - status_src = (FENRIS_PKG / "status.py").read_text() - phrase = "observation store written by a newer Fenris" - assert phrase in status_src - def test_status_never_prompts(self): status_src = (FENRIS_PKG / "status.py").read_text() assert "input(" not in status_src @@ -603,12 +591,6 @@ class TestCI3ProhibitionSet: table_names.append(m.group(1)) assert "projection" not in [t.lower() for t in table_names] - def test_no_partial_newer_schema_interpretation(self): - """Readers refuse newer-schema stores. [3.6, 9.5]""" - status_src = (FENRIS_PKG / "status.py").read_text() - assert "NewerSchema" in status_src - assert "upgrade Fenris" in status_src - def test_polkit_authorizes_one_binary(self): """Polkit authorizes exactly one binary: fenris-monitor. [8.5]""" monitor_src = (FENRIS_PKG / "monitor.py").read_text() @@ -688,18 +670,6 @@ class TestCI4WordingAndDisclosures: proj_src = (FENRIS_PKG / "projection.py").read_text() assert phrase in proj_src - def test_store_fault_phrase(self): - phrase = "observation store unreadable" - status_src = (FENRIS_PKG / "status.py").read_text() - assert phrase in status_src - tui_src = (FENRIS_PKG / "tui.py").read_text() - assert phrase.lower() in tui_src.lower() - - def test_newer_schema_phrase(self): - phrase = "observation store written by a newer Fenris" - status_src = (FENRIS_PKG / "status.py").read_text() - assert phrase in status_src - def test_no_observations_phrase(self): phrase = "no observations yet" status_src = (FENRIS_PKG / "status.py").read_text() diff --git a/tests/test_collector_history_tracer.py b/tests/test_collector_history_tracer.py index 14c4ba0..4385eb4 100644 --- a/tests/test_collector_history_tracer.py +++ b/tests/test_collector_history_tracer.py @@ -412,7 +412,7 @@ class TestDisplayStates: def test_one_sample_awaiting_another_in_tui(self, tmp_path): """One sample → TUI shows awaiting state.""" - from fenris.tui import FenrisTuiApp, _query_service_facts + from fenris.tui import FenrisTuiApp db = tmp_path / "test.db" conn = init_store(db) conn.execute( diff --git a/tests/test_control.py b/tests/test_control.py new file mode 100644 index 0000000..3d503e1 --- /dev/null +++ b/tests/test_control.py @@ -0,0 +1,77 @@ +"""The CLI and TUI cross the same terminal-attached action interface.""" +import argparse +import runpy +import shlex +import subprocess +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +from fenris.control import MONITOR_HELPER, MonitorError, run_monitor +from fenris.tui import FenrisTuiApp + + +@pytest.mark.parametrize("uid", [0, 1000]) +@pytest.mark.parametrize("args", [("enable", "--now"), ("disable", "--now"), ("collect",)]) +def test_fixed_helper_and_terminal_attachment(uid, args): + with patch("fenris.control.os.geteuid", return_value=uid), \ + patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run: + run_monitor(*args) + expected = [MONITOR_HELPER, *args] + if uid: + expected.insert(0, "pkexec") + # Inherited stdin/out/err keep authentication on the user's terminal. + # No frontend deadline can cut short a valid 90-second Collection run. + run.assert_called_once_with(expected) + + +@pytest.mark.parametrize("code", [1, 126, 127, -15]) +def test_failure_is_not_retried_and_root_equivalent_preserves_arguments(code): + args = ("baseline", "set", '{"model": "Drive $(whoami)", "tbw": 100}') + with patch("fenris.control.os.geteuid", return_value=1000), \ + patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], code)) as run: + with pytest.raises(MonitorError) as failure: + run_monitor(*args) + run.assert_called_once() + assert failure.value.exit_code == (code if code > 0 else 128 - code) + root_command = str(failure.value).split("run in your terminal: ")[1] + assert shlex.split(root_command) == ["sudo", MONITOR_HELPER, *args] + + +@pytest.mark.parametrize("uid,missing", [(0, MONITOR_HELPER), (1000, "pkexec")]) +def test_missing_command_is_identified(uid, missing): + with patch("fenris.control.os.geteuid", return_value=uid), \ + patch("fenris.control.subprocess.run", side_effect=FileNotFoundError(2, "Missing", missing)): + with pytest.raises(MonitorError, match="Command not found") as failure: + run_monitor("collect") + assert missing in str(failure.value) + assert failure.value.exit_code == 127 + assert ("sudo" in str(failure.value)) == bool(uid) + + +def test_interrupt_reports_uncertain_outcome(): + with patch("fenris.control.subprocess.run", side_effect=KeyboardInterrupt): + with pytest.raises(MonitorError, match="Check fenris status") as failure: + run_monitor("collect") + assert failure.value.exit_code == 130 + + +def test_cli_and_tui_show_the_same_failure(tmp_path, capsys): + # The launcher may prepend an installed runtime while loading; isolate it. + with patch.object(sys, "path", sys.path.copy()): + cli = runpy.run_path(str(Path(__file__).parent.parent / "scripts" / "fenris")) + with patch("fenris.control.os.geteuid", return_value=1000), \ + patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], 126)): + with pytest.raises(SystemExit) as exit_info: + cli["cmd_monitor_resume"](argparse.Namespace()) + assert exit_info.value.code == 126 + cli_message = capsys.readouterr().err.strip() + app = FenrisTuiApp(store_path=tmp_path / "missing.db") + with patch.object(app, "suspend"), patch.object(app, "_refresh") as refresh, \ + patch.object(app, "notify") as notify: + app.action_resume() + notify.assert_called_once_with(cli_message, severity="error") + refresh.assert_called_once() diff --git a/tests/test_init_system.py b/tests/test_init_system.py index a3f576a..d8d60c9 100644 --- a/tests/test_init_system.py +++ b/tests/test_init_system.py @@ -283,7 +283,7 @@ class TestSystemdQueryState: def test_query_state_disabled_inactive(self): """Query state for disabled and inactive timer.""" with patch("fenris.init_system._systemctl_show") as mock_show: - mock_show.return_value = {} + mock_show.return_value = {"UnitFileState": "disabled", "ActiveState": "inactive"} result = _systemd_query_state() assert result["boot_enabled"] is False assert result["timer_active"] is False @@ -688,5 +688,5 @@ class TestEdgeCases: with patch("fenris.init_system._systemctl_show") as mock_show: mock_show.return_value = {} result = _systemd_query_state() - assert result["boot_enabled"] is False - assert result["timer_active"] is False + assert result["boot_enabled"] is None + assert result["timer_active"] is None diff --git a/tests/test_issue_79.py b/tests/test_issue_79.py index d387cbe..80b2e72 100644 --- a/tests/test_issue_79.py +++ b/tests/test_issue_79.py @@ -260,7 +260,7 @@ class TestPreservedBehavior: conn.close() app = FenrisTuiApp(store_path=tmp_path / "test.db") - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": True, "timer_active": True, "last_collect_ok": True, "last_collect_age_s": 60, "last_collect_reason": None, @@ -272,11 +272,11 @@ class TestPreservedBehavior: @pytest.mark.asyncio async def test_quit_rail_preserved(self, tmp_path): - """Separate q QUIT TUI rail preserved.""" + """Separate q Quit TUI rail preserved.""" app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") async with app.run_test(size=(120, 24)) as pilot: rail = str(app.query_one("#quit-rail").render()) - assert "q QUIT TUI" in rail + assert "q Quit TUI" in rail @pytest.mark.asyncio async def test_auth_banner_preserved(self, tmp_path): @@ -307,7 +307,7 @@ class TestPreservedBehavior: conn.close() app = FenrisTuiApp(store_path=db) - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": False, "timer_active": False, "last_collect_ok": None, "last_collect_age_s": None, "last_collect_reason": None, diff --git a/tests/test_packaging.py b/tests/test_packaging.py index b0f565e..a136c85 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -135,7 +135,7 @@ RPM_TARGETS = [ ] XBPS_TARGETS = [ - ("voidlinux/void-linux:latest", "xbps"), + ("ghcr.io/void-linux/void-glibc-full:latest", "xbps"), ] ALL_TARGETS = DEB_TARGETS + RPM_TARGETS + XBPS_TARGETS @@ -176,9 +176,8 @@ def _build_xbps_dockerfile(image: str, pkg_name: str) -> str: """Dockerfile for testing xbps installation.""" return textwrap.dedent(f"""\ FROM {image} - RUN xbps-install -Suyn void-repo-nonfree && \\ - xbps-install -Syun python3 smartmontools runit && \\ - xbps-remove -Oy void-repo-nonfree || true + RUN xbps-install -Sy python3 smartmontools runit shadow && \\ + xbps-remove -O COPY dist/{pkg_name} /pkg/{pkg_name} """) @@ -1713,7 +1712,7 @@ def test_xbps_dormant_install(skip_no_docker, version): check=True, ) (build_dir / "Dockerfile").write_text( - _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name) + _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name) ) tag = "fenris-test-xbps" @@ -1731,8 +1730,17 @@ def test_xbps_dormant_install(skip_no_docker, version): ) try: - rc, out = _xbps_extract_and_install(container, pkg_name, action="install") + # Exercise XBPS itself, including its file ownership and lifecycle hooks. + # The temporary local repository does not need release signatures. + rc, out = _container_exec( + container, + f"mkdir -p /tmp/fenris-repo && cp /pkg/{pkg_name} /tmp/fenris-repo/ && " + f"xbps-rindex -a /tmp/fenris-repo/{pkg_name} && " + "xbps-install -y -R /tmp/fenris-repo fenris", + ) assert rc == 0, f"XBPS install failed: {out}" + rc, out = _container_exec(container, "xbps-query -p pkgver fenris") + assert rc == 0 and out.strip() == f"fenris-{version}_1" _assert_runit_dormant_layout(container, version) finally: subprocess.run( @@ -1755,7 +1763,7 @@ def test_xbps_migration_guard(skip_no_docker, version): check=True, ) (build_dir / "Dockerfile").write_text( - _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name) + _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name) ) tag = "fenris-guard-xbps" @@ -1803,7 +1811,7 @@ def test_xbps_upgrade_semantics(skip_no_docker, version): check=True, ) (build_dir / "Dockerfile").write_text( - _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name) + _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name) ) tag = "fenris-upgrade-xbps" @@ -1889,7 +1897,7 @@ def test_xbps_remove_preserves_config_and_store(skip_no_docker, version): check=True, ) (build_dir / "Dockerfile").write_text( - _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name) + _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name) ) tag = "fenris-remove-xbps" @@ -1960,7 +1968,7 @@ def test_xbps_purge_removes_everything(skip_no_docker, version): check=True, ) (build_dir / "Dockerfile").write_text( - _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name) + _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name) ) tag = "fenris-purge-xbps" diff --git a/tests/test_status.py b/tests/test_status.py index 9c801f0..bddc614 100644 --- a/tests/test_status.py +++ b/tests/test_status.py @@ -622,6 +622,7 @@ class TestProjectionInStatus: from fenris.status import get_status nonexistent = tmp_path / "nonexistent.db" + nonexistent.write_bytes(b"not a SQLite database") now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) with patch("fenris.status.query_service_state", return_value={ diff --git a/tests/test_status_composition.py b/tests/test_status_composition.py index 0a57559..2d6a8b5 100644 --- a/tests/test_status_composition.py +++ b/tests/test_status_composition.py @@ -587,7 +587,7 @@ class TestCLIStatusRendering: newer_schema=None) cli_text = render_status_cli(comp) assert "observation store unreadable" in cli_text - assert "see journal" in cli_text.lower() + assert "see collector logs" in cli_text.lower() conn.close() @@ -653,7 +653,7 @@ class TestTUIStatusRendering: comp = compose_status(conn, svc, now, store_fault="observation store unreadable", newer_schema=None) tui_text = render_status_tui(comp) - assert "observation store unreadable" in tui_text + assert "Observation store unreadable" in tui_text conn.close() @@ -966,7 +966,7 @@ class TestTUIIntegration: conn.close() app = FenrisTuiApp(store_path=tmp_path / "test.db") - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": True, "timer_active": True, "last_collect_ok": True, "last_collect_age_s": 120, "last_collect_reason": None, @@ -992,7 +992,7 @@ class TestTUIIntegration: conn.close() app = FenrisTuiApp(store_path=db) - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": False, "timer_active": False, "last_collect_ok": None, "last_collect_age_s": None, "last_collect_reason": None, @@ -1019,7 +1019,7 @@ class TestTUIIntegration: conn.close() app = FenrisTuiApp(store_path=tmp_path / "test.db") - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": True, "timer_active": True, "last_collect_ok": False, "last_collect_age_s": 60, "last_collect_reason": "exit code 3", diff --git a/tests/test_status_read.py b/tests/test_status_read.py new file mode 100644 index 0000000..59e0e78 --- /dev/null +++ b/tests/test_status_read.py @@ -0,0 +1,150 @@ +"""Exercise shared status acquisition through the CLI and running TUI.""" +import sqlite3 +import sys +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from fenris.status import get_status, get_status_composition, read_status +from fenris.status_composition import StatusState +from fenris.store import init_store, SCHEMA_VERSION +from fenris.tui import FenrisTuiApp + + +NOW = datetime(2026, 9, 16, 12, tzinfo=timezone.utc) +ACTIVE = {"boot_enabled": True, "timer_active": True, "last_collect_ok": True} +DISABLED = {"boot_enabled": False, "timer_active": False, "last_collect_ok": None} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("kind", ["missing", "corrupt", "newer", "incomplete", "denied"]) +@pytest.mark.parametrize("service", [ACTIVE, DISABLED, {}]) +async def test_cli_tui_acquisition_parity(tmp_path, kind, service): + path = tmp_path / "observations.db" + if kind == "corrupt": + path.write_bytes(b"Not a SQLite database") + elif kind == "incomplete": + sqlite3.connect(path).close() + elif kind != "missing": + conn = init_store(path) + if kind == "newer": + conn.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION + 1)) + conn.close() + original_exists = Path.exists + + def exists(target): + if kind == "denied" and target == path: + raise PermissionError("Observation store access denied") + return original_exists(target) + + before = path.read_bytes() if original_exists(path) else None + with patch("fenris.status.query_service_state", return_value=service), \ + patch("fenris.status._journalctl_hint", return_value="[bold]Native collector log[/bold]"), \ + patch("fenris.tui._journalctl_hint", return_value="[bold]Native collector log[/bold]"), \ + patch.object(Path, "exists", exists), \ + patch("fenris.tui.compute_projection") as projection: + comp = get_status_composition(path, NOW) + cli = get_status(path, NOW).lower() + app = FenrisTuiApp(store_path=path) + async with app.run_test(size=(80, 24)) as pilot: + headline = str(app.query_one("#headline-band").render()).lower() + strip = str(app.query_one("#service-strip").render()).lower() + if kind == "missing": + assert comp.store_fault is None + assert comp.freshness == "empty" + assert "no observations yet" in cli and "no observations yet" in headline + else: + assert comp.state == StatusState.ERROR + assert comp.freshness == "unknown" + phrase = "newer fenris — upgrade fenris" if kind == "newer" else "observation store unreadable" + assert phrase in cli and phrase in headline + if kind != "newer": + assert str(app.query_one("#drive-health").render()) == "[bold]Native collector log[/bold]" + assert not app.query_one("#main-grid").has_class("paused") + await pilot.resize_terminal(60, 18) + await pilot.pause() + assert phrase in str(app.query_one("#constrained-summary").render()).lower() + for text in (cli, strip): + assert ("boot: enabled" if service == ACTIVE else "boot: disabled" if service == DISABLED else "boot: unknown") in text + assert ("timer: active" if service == ACTIVE else "timer: inactive" if service == DISABLED else "timer: unknown") in text + if not service: + assert "does not start on next boot" not in text + projection.assert_not_called() + assert (path.read_bytes() if original_exists(path) else None) == before + + +def test_reader_owns_readonly_snapshot_and_closes_on_error(tmp_path): + path = tmp_path / "observations.db" + writer = init_store(path) + with patch("fenris.status.query_service_state", return_value=ACTIVE) as query: + with pytest.raises(RuntimeError, match="renderer failed"): + with read_status(path, NOW) as (conn, comp): + assert comp.sample_count == 0 + assert conn.in_transaction + with pytest.raises(sqlite3.OperationalError, match="readonly"): + conn.execute("DELETE FROM samples") + writer.execute("INSERT INTO monitoring_periods (started_at) VALUES ('2026-09-16')") + writer.commit() + assert conn.execute("SELECT COUNT(*) FROM monitoring_periods").fetchone()[0] == 0 + raise RuntimeError("renderer failed") + query.assert_called_once() + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + conn.execute("SELECT 1") + writer.close() + + +def test_monitoring_query_failure_is_unknown(tmp_path): + path = tmp_path / "observations.db" + init_store(path).close() + with patch("fenris.status.query_service_state", side_effect=OSError("Unavailable")): + comp = get_status_composition(path, NOW) + cli = get_status(path, NOW, query_journal=False) + assert comp.state == StatusState.UNKNOWN + assert comp.boot_enabled is None and comp.timer_active is None + assert "boot: unknown" in cli and "timer: unknown" in cli + assert "does not start on next boot" not in cli + + +def test_native_systemd_query_failure_is_unknown(tmp_path): + from fenris.init_system import InitSystem + + with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \ + patch("fenris.init_system._systemctl_show", return_value={}): + comp = get_status_composition(tmp_path / "missing.db", NOW) + assert comp.state == StatusState.UNKNOWN + assert comp.boot_enabled is None and comp.timer_active is None + + +@pytest.mark.asyncio +async def test_new_store_fault_clears_paused_views_and_can_recover(tmp_path): + path = tmp_path / "observations.db" + conn = init_store(path) + conn.execute( + "INSERT INTO monitoring_periods (started_at, ended_at, end_cause) " + "VALUES ('2026-09-15', '2026-09-16', 'user_disabled')" + ) + conn.commit() + conn.close() + with patch("fenris.status.query_service_state", return_value=DISABLED): + app = FenrisTuiApp(store_path=path) + async with app.run_test() as pilot: + assert app.query_one("#main-grid").has_class("paused") + writer = sqlite3.connect(path) + writer.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION + 1)) + writer.close() + app.on_refresh_tick() + await pilot.pause() + assert not app.query_one("#main-grid").has_class("paused") + assert str(app.query_one("#drive-health").render()) == "" + assert "upgrade Fenris" in str(app.query_one("#headline-band").render()) + assert app.query_one("#usage-history")._day_data == [] + writer = sqlite3.connect(path) + writer.execute("PRAGMA user_version = %d" % SCHEMA_VERSION) + writer.close() + app.on_refresh_tick() + await pilot.pause() + assert app.query_one("#main-grid").has_class("paused") diff --git a/tests/test_tui.py b/tests/test_tui.py index 8b94f4e..f5302c2 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -42,11 +42,7 @@ from fenris.tui import ( FenrisTuiApp, DailyBarGraph, _format_remaining, - _sparkline, - _habit_bar, - _query_usage_history, _query_drive_health, - _query_service_facts, _query_daily_graph_data, _query_hourly_graph_data, _RANGE_OPTIONS, @@ -142,64 +138,6 @@ class TestFormatRemaining: assert _format_remaining(-100) == "endurance exhausted" -class TestSparkline: - def test_empty(self): - assert _sparkline([]) == "" - - def test_single_value(self): - result = _sparkline([100.0]) - assert len(result) == 1 - - def test_multiple_values(self): - result = _sparkline([1.0, 2.0, 3.0, 4.0, 5.0]) - assert len(result) > 0 - assert all(c in " ▁▂▃▄▅▆▇█" for c in result) - - def test_width_limit(self): - result = _sparkline([1.0] * 100, width=20) - assert len(result) <= 20 - - -class TestHabitBar: - def test_all_active(self): - result = _habit_bar(1.0, 0.0, 0.0, 0.0) - assert "active 100%" in result - - def test_mixed(self): - result = _habit_bar(0.5, 0.3, 0.1, 0.1) - assert "active 50%" in result - assert "idle 30%" in result - - def test_all_unknown(self): - result = _habit_bar(0.0, 0.0, 0.0, 1.0) - assert "unknown 100%" in result - - -# --------------------------------------------------------------------------- -# Data query tests -# --------------------------------------------------------------------------- - -class TestQueryUsageHistory: - def test_empty_store(self, tmp_path): - conn = init_store(tmp_path / "test.db") - result = _query_usage_history(conn) - assert result["num_days"] == 0 - assert result["sparkline"] == "" - conn.close() - - def test_with_days(self, tmp_path): - conn = init_store(tmp_path / "test.db") - _insert_segment(conn) - _open_period(conn) - for i in range(14): - d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") - _insert_day(conn, d, bw=1024*1024*100) - result = _query_usage_history(conn) - assert result["num_days"] == 14 - assert result["sparkline"] != "" - conn.close() - - class TestQueryDriveHealth: def test_empty_store(self, tmp_path): conn = init_store(tmp_path / "test.db") @@ -215,27 +153,6 @@ class TestQueryDriveHealth: conn.close() -class TestQueryServiceFacts: - def test_empty_store(self, tmp_path): - conn = init_store(tmp_path / "test.db") - now = _clock() - result = _query_service_facts(conn, now) - assert result["freshness"] == "empty" - conn.close() - - def test_fresh_sample(self, tmp_path): - conn = init_store(tmp_path / "test.db") - _insert_segment(conn) - _open_period(conn) - _insert_day(conn, "2026-09-29", bw=1024*1024*100) - now = _clock() - ts = (now - timedelta(minutes=2)).isoformat() - _insert_sample(conn, ts) - result = _query_service_facts(conn, now) - assert result["freshness"] == "fresh" - conn.close() - - # --------------------------------------------------------------------------- # CI-1: Exhaustive state matrix from synthetic stores # --------------------------------------------------------------------------- @@ -397,10 +314,10 @@ class TestDenseScreen: app = FenrisTuiApp(store_path=tmp_path / "test.db") async with app.run_test() as pilot: strip = str(app.query_one("#service-strip").render()) - assert "boot:" in strip - assert "timer:" in strip - assert "last collect:" in strip - assert "freshness:" in strip + assert "Boot:" in strip + assert "Timer:" in strip + assert "Last collect:" in strip + assert "Freshness:" in strip # Maker credit now in titlebox only (issue #79) assert "by Bongbetic" not in strip @@ -412,7 +329,7 @@ class TestDenseScreen: conn.close() app = FenrisTuiApp(store_path=tmp_path / "test.db") - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": True, "timer_active": True, "last_collect_ok": True, "last_collect_age_s": 60, "last_collect_reason": None, @@ -427,7 +344,7 @@ class TestDenseScreen: assert "monitoring: active in background · persists across reboots" in strip assert "r resume — enable monitoring and future boots" in strip assert "q quit" not in strip - assert rail == "q QUIT TUI" + assert rail == "q Quit TUI" assert usage.region.y < service.region.y < quit_rail.region.y assert usage.region.bottom <= service.region.y assert service.region.bottom <= quit_rail.region.y @@ -446,11 +363,11 @@ class TestDenseScreen: conn.close() app = FenrisTuiApp(store_path=db) - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": False, "timer_active": False, "last_collect_ok": None, "last_collect_age_s": None, "last_collect_reason": None, - }), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object( + }), patch("fenris.control.subprocess.run") as subprocess_run, patch.object( app, "_run_helper" ) as run_helper: async with app.run_test(size=(80, 24)) as pilot: @@ -472,7 +389,7 @@ class TestDenseScreen: visible_text = " ".join( "".join(ElementTree.fromstring(screenshot).itertext()).split() ) - assert "paused time is excluded from your usage habit" in visible_text + assert "Paused time is excluded from your usage habit" in visible_text assert "resume: fenris monitor resume" in visible_text assert "monitoring: does not start on next boot" in str( app.query_one("#service-strip").render() @@ -487,7 +404,7 @@ class TestDenseScreen: ElementTree.fromstring(app.export_screenshot()).itertext() ).split() ) - assert "q QUIT TUI" in footer_text + assert "q Quit TUI" in footer_text await pilot.press("q") assert not app.is_running subprocess_run.assert_not_called() @@ -509,11 +426,11 @@ class TestDenseScreen: conn.close() app = FenrisTuiApp(store_path=db) - with patch("fenris.tui.query_service_state", return_value={ + with patch("fenris.status.query_service_state", return_value={ "boot_enabled": True, "timer_active": True, "last_collect_ok": None, "last_collect_age_s": None, "last_collect_reason": None, - }), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object( + }), patch("fenris.control.subprocess.run") as subprocess_run, patch.object( app, "_run_helper" ) as run_helper: async with app.run_test() as pilot: @@ -539,7 +456,7 @@ class TestDenseScreen: store_path=tmp_path / "nonexistent.db", refresh_interval_s=0.2, ) - auth_notice = "privileged actions will prompt for authentication (polkit)" + auth_notice = "Open with fenris (no sudo). Actions authenticate via polkit. ? Help" async with app.run_test() as pilot: headline = str(app.query_one("#headline-band").render()) @@ -595,12 +512,47 @@ class TestDisclosuresAndGreeting: # --------------------------------------------------------------------------- class TestFirstRun: + @pytest.mark.asyncio + @pytest.mark.parametrize("size", [(80, 24), (60, 18)]) + async def test_sudo_help_remains_available_after_launch(self, tmp_path, size): + app = FenrisTuiApp(store_path=tmp_path / "missing.db") + with patch.object(app, "_run_helper") as run_helper: + async with app.run_test(size=size) as pilot: + app.on_refresh_tick() + await pilot.press("?") + help_text = str(app.screen.query_one("#help-text").render()) + assert "The dashboard does not need sudo" in help_text + assert "sudo fenris monitor resume" in help_text + assert "sudo fenris monitor pause" in help_text + assert "sudo fenris sample" in help_text + assert 'sudo usermod -aG fenris "$USER"' in help_text + assert "log out and back in" in help_text + await pilot.press("end") + visible = " ".join("".join( + ElementTree.fromstring(app.export_screenshot()).itertext() + ).split()) + assert "Esc Close" in visible + await pilot.press("escape") + assert len(app.screen_stack) == 1 + run_helper.assert_not_called() + + def test_missing_polkit_identifies_command_and_points_to_help(self, tmp_path): + app = FenrisTuiApp(store_path=tmp_path / "missing.db") + with patch("fenris.control.os.geteuid", return_value=1000), \ + patch("fenris.control.subprocess.run", side_effect=FileNotFoundError(2, "Missing", "pkexec")), \ + patch.object(app, "suspend"), patch.object(app, "_refresh"), \ + patch.object(app, "notify") as notify: + app.action_resume() + message = notify.call_args.args[0] + assert "Command not found: pkexec" in message + assert "sudo /usr/libexec/fenris/fenris-monitor enable --now" in message + def test_unprivileged_resume_uses_polkit(self, tmp_path): """TUI controls use the same authenticated path as the CLI.""" app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") - with patch("fenris.tui.os.geteuid", return_value=1000), \ - patch("fenris.tui.subprocess.run") as run, \ + with patch("fenris.control.os.geteuid", return_value=1000), \ + patch("fenris.control.subprocess.run") as run, \ patch.object(app, "suspend"), \ patch.object(app, "_refresh"): run.return_value.returncode = 0 @@ -608,7 +560,6 @@ class TestFirstRun: run.assert_called_once_with( ["pkexec", "/usr/libexec/fenris/fenris-monitor", "enable", "--now"], - timeout=30, ) @pytest.mark.asyncio @@ -924,8 +875,8 @@ class TestBarGraphTUI: assert len(graph._day_data) > 0 # Legend should be visible legend = str(app.query_one("#bar-legend").render()) - assert "alloc" in legend - assert "zero" in legend + assert "Alloc" in legend + assert "Zero" in legend @pytest.mark.asyncio async def test_arrow_selection(self, tmp_path): @@ -1091,7 +1042,7 @@ class TestBarGraphTUI: app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") async with app.run_test(size=(80, 24)) as pilot: graph = app.query_one("#usage-history") - assert graph.border_title == "usage history" + assert graph.border_title == "Usage history" @pytest.mark.asyncio async def test_glyphs_in_legend(self, tmp_path): @@ -1172,7 +1123,7 @@ class TestConstrainedLayout: # Constrained summary should be visible summary = app.query_one("#constrained-summary") summary_text = str(summary.render()) - assert "graph needs ≥80×24" in summary_text + assert "Graph needs ≥80×24" in summary_text assert "days" in summary_text assert "GB total" in summary_text @@ -1196,7 +1147,7 @@ class TestConstrainedLayout: # Constrained summary should be visible summary = app.query_one("#constrained-summary") summary_text = str(summary.render()) - assert "graph needs ≥80×24" in summary_text + assert "Graph needs ≥80×24" in summary_text @pytest.mark.asyncio async def test_resize_from_constrained_to_normal(self, tmp_path): @@ -1273,7 +1224,7 @@ class TestConstrainedLayout: # Constrained summary should show selected day context summary = app.query_one("#constrained-summary") summary_text = str(summary.render()) - assert "graph needs ≥80×24" in summary_text + assert "Graph needs ≥80×24" in summary_text # Selected day context should survive assert "2026-09" in summary_text @@ -1285,7 +1236,7 @@ class TestConstrainedLayout: assert app._is_constrained_mode summary = app.query_one("#constrained-summary") summary_text = str(summary.render()) - assert "graph needs ≥80×24" in summary_text + assert "Graph needs ≥80×24" in summary_text assert "awaiting" in summary_text.lower() @pytest.mark.asyncio @@ -1315,13 +1266,13 @@ class TestConstrainedLayout: # Service strip should have actions strip = str(app.query_one("#service-strip").render()) - assert "p pause" in strip - assert "r resume" in strip + assert "p Pause" in strip + assert "r Resume" in strip assert "q quit" not in strip # Quit is in quit-rail # Quit rail should be visible rail = str(app.query_one("#quit-rail").render()) - assert "QUIT" in rail + assert "Quit" in rail @pytest.mark.asyncio async def test_constrained_long_reasons_visible(self, tmp_path):