diff --git a/src/seedpass/tui_v3/app.py b/src/seedpass/tui_v3/app.py index e200071c..a71e76f2 100644 --- a/src/seedpass/tui_v3/app.py +++ b/src/seedpass/tui_v3/app.py @@ -63,7 +63,18 @@ class CommandProcessor: if cmd == "help": self.app.notify( - "v3 commands: help, stats, atlas, wayfinder, session-status, lock, unlock , refresh, search , search-mode , sort , filter , archived, open , settings, profiles, relay-list, npub, nostr-pubkey, nostr-reset-sync-state, nostr-fresh-namespace, change-password, backup-parent-seed (optional: password), maximize, copy, edit, export, add, seed-plus, archive, restore, delete, ml, mx, db-export , db-import " + "v3 commands: help, stats, atlas, wayfinder, session-status, lock, unlock , refresh, " + "search , search-mode , sort , " + "filter , archived, archive-filter , " + "open , settings, profiles, relay-list, relay-add , relay-rm , relay-reset, " + "npub, nostr-pubkey, nostr-reset-sync-state, nostr-fresh-namespace, " + "sync-now, sync-bg, " + "checksum-verify, checksum-update, totp-export [path], " + "setting-secret , setting-quick-unlock , setting-offline , " + "setting-timeout , setting-kdf-mode , setting-kdf-iterations , " + "density , onboarding, " + "change-password, backup-parent-seed (optional: password), maximize, copy, edit, export, " + "add, seed-plus, archive, restore, delete, ml, mx, db-export , db-import " ) elif cmd == "stats": self.app.notify("Calculating stats...") @@ -182,6 +193,84 @@ class CommandProcessor: self.app.notify("Usage: db-import ", severity="warning") return self.app.action_db_import(args[0]) + elif cmd == "checksum-verify": + self.app.action_checksum_verify() + elif cmd == "checksum-update": + self.app.action_checksum_update() + elif cmd == "totp-export": + path = args[0] if args else None + self.app.action_totp_export(path) + elif cmd == "sync-now": + self.app.action_sync_now() + elif cmd == "sync-bg": + self.app.action_sync_bg() + elif cmd == "relay-add": + if not args: + self.app.notify("Usage: relay-add ", severity="warning") + return + self.app.action_relay_add(args[0]) + elif cmd == "relay-rm": + if not args: + self.app.notify("Usage: relay-rm ", severity="warning") + return + try: + idx = int(args[0]) + except ValueError: + self.app.notify("relay-rm: index must be an integer", severity="error") + return + self.app.action_relay_rm(idx) + elif cmd == "relay-reset": + self.app.action_relay_reset() + elif cmd == "setting-secret": + if not args: + self.app.notify("Usage: setting-secret ", severity="warning") + return + self.app.action_setting_secret(args[0]) + elif cmd == "setting-quick-unlock": + if not args: + self.app.notify( + "Usage: setting-quick-unlock ", severity="warning" + ) + return + self.app.action_setting_quick_unlock(args[0]) + elif cmd == "setting-offline": + if not args: + self.app.notify("Usage: setting-offline ", severity="warning") + return + self.app.action_setting_offline(args[0]) + elif cmd == "setting-timeout": + if not args: + self.app.notify("Usage: setting-timeout ", severity="warning") + return + self.app.action_setting_timeout(args[0]) + elif cmd == "setting-kdf-mode": + if not args: + self.app.notify("Usage: setting-kdf-mode ", severity="warning") + return + self.app.action_setting_kdf_mode(args[0]) + elif cmd == "setting-kdf-iterations": + if not args: + self.app.notify( + "Usage: setting-kdf-iterations ", severity="warning" + ) + return + self.app.action_setting_kdf_iterations(args[0]) + elif cmd == "archive-filter": + if not args: + self.app.notify( + "Usage: archive-filter ", severity="warning" + ) + return + self.app.action_archive_filter(args[0]) + elif cmd == "density": + if not args: + self.app.notify( + "Usage: density ", severity="warning" + ) + return + self.app.action_set_density(args[0]) + elif cmd in {"onboarding", "welcome", "quickstart"}: + self.app.action_show_onboarding() else: self.app.notify(f"Unknown v3 command: {cmd}", severity="warning") @@ -837,6 +926,8 @@ class SeedPassTuiV3(App[None]): search_sort = reactive("relevance") filter_kind = reactive("all") show_archived = reactive(False) + filter_archived_only = reactive(False) + density_mode = reactive("comfortable") # Internal state for sensitive actions _pending_sensitive_confirm: tuple[str, int, float] | None = None @@ -1884,3 +1975,231 @@ class SeedPassTuiV3(App[None]): board.reveal_data = data except Exception: pass + + # --------------------------------------------------------------------------- + # Utility / maintenance actions (legacy parity) + # --------------------------------------------------------------------------- + + def action_checksum_verify(self) -> None: + """Verify the application checksum.""" + utility = self.services.get("utility") + if not utility: + self.notify("Utility service offline", severity="error") + return + try: + utility.verify_checksum() + self.notify("Checksum verified") + except Exception as e: + self.notify(f"Checksum verify failed: {e}", severity="error") + + def action_checksum_update(self) -> None: + """Update / regenerate the application checksum.""" + utility = self.services.get("utility") + if not utility: + self.notify("Utility service offline", severity="error") + return + try: + utility.update_checksum() + self.notify("Checksum updated") + except Exception as e: + self.notify(f"Checksum update failed: {e}", severity="error") + + def action_totp_export(self, path: str | None = None) -> None: + """Export all TOTP entries, optionally writing to a file.""" + entry = self.services.get("entry") + if not entry: + self.notify("Entry service offline", severity="error") + return + try: + data = entry.export_totp_entries() + if path: + import json + from pathlib import Path + + Path(path).write_text(json.dumps(data, indent=2), encoding="utf-8") + self.notify(f"TOTP entries exported to {path}") + else: + count = len(data) if isinstance(data, dict) else 0 + self.notify( + f"TOTP export: {count} entries (provide a path to save to file)" + ) + except Exception as e: + self.notify(f"TOTP export failed: {e}", severity="error") + + def action_sync_now(self) -> None: + """Synchronise vault to Nostr immediately.""" + sync = self.services.get("sync") + if not sync: + self.notify("Sync service offline", severity="error") + return + try: + self.notify("Syncing vault to Nostr…") + result = sync.sync() + if result: + self.notify("Sync complete") + else: + self.notify("Sync returned no events", severity="warning") + except Exception as e: + self.notify(f"Sync failed: {e}", severity="error") + + def action_sync_bg(self) -> None: + """Start a background Nostr vault sync.""" + sync = self.services.get("sync") + if not sync: + self.notify("Sync service offline", severity="error") + return + try: + sync.start_background_vault_sync() + self.notify("Background sync started") + except Exception as e: + self.notify(f"Background sync failed: {e}", severity="error") + + def action_relay_add(self, url: str) -> None: + """Add a Nostr relay.""" + nostr = self.services.get("nostr") + if not nostr: + self.notify("Nostr service offline", severity="error") + return + try: + nostr.add_relay(url) + self.notify(f"Relay added: {url}") + except Exception as e: + self.notify(f"Relay add failed: {e}", severity="error") + + def action_relay_rm(self, idx: int) -> None: + """Remove a Nostr relay by index.""" + nostr = self.services.get("nostr") + if not nostr: + self.notify("Nostr service offline", severity="error") + return + try: + nostr.remove_relay(idx) + self.notify(f"Relay #{idx} removed") + except Exception as e: + self.notify(f"Relay remove failed: {e}", severity="error") + + def action_relay_reset(self) -> None: + """Reset Nostr relays to built-in defaults.""" + nostr = self.services.get("nostr") + if not nostr: + self.notify("Nostr service offline", severity="error") + return + try: + relays = nostr.reset_relays() + self.notify(f"Relays reset to defaults ({len(relays)} relays)") + except Exception as e: + self.notify(f"Relay reset failed: {e}", severity="error") + + def action_setting_secret(self, value: str) -> None: + """Toggle secret mode on or off.""" + config = self.services.get("config") + if not config: + self.notify("Config service offline", severity="error") + return + try: + enabled = value.lower() in {"on", "1", "true", "yes", "y"} + delay = config.get_clipboard_clear_delay() + config.set_secret_mode(enabled, delay) + self.notify(f"Secret mode: {'on' if enabled else 'off'}") + except Exception as e: + self.notify(f"Setting failed: {e}", severity="error") + + def action_setting_quick_unlock(self, value: str) -> None: + """Enable or disable quick-unlock.""" + config = self.services.get("config") + if not config: + self.notify("Config service offline", severity="error") + return + try: + config.set("quick_unlock", value) + self.notify(f"Quick unlock: {value}") + except Exception as e: + self.notify(f"Setting failed: {e}", severity="error") + + def action_setting_offline(self, value: str) -> None: + """Enable or disable offline mode.""" + config = self.services.get("config") + if not config: + self.notify("Config service offline", severity="error") + return + try: + enabled = value.lower() in {"on", "1", "true", "yes", "y"} + config.set_offline_mode(enabled) + self.notify(f"Offline mode: {'on' if enabled else 'off'}") + except Exception as e: + self.notify(f"Setting failed: {e}", severity="error") + + def action_setting_timeout(self, seconds: str) -> None: + """Set the inactivity auto-lock timeout in seconds.""" + config = self.services.get("config") + if not config: + self.notify("Config service offline", severity="error") + return + try: + config.set("inactivity_timeout", seconds) + self.notify(f"Inactivity timeout: {seconds}s") + except Exception as e: + self.notify(f"Setting failed: {e}", severity="error") + + def action_setting_kdf_mode(self, mode: str) -> None: + """Set the KDF algorithm (e.g. argon2id, pbkdf2).""" + config = self.services.get("config") + if not config: + self.notify("Config service offline", severity="error") + return + try: + config.set("kdf_mode", mode) + self.notify(f"KDF mode: {mode}") + except Exception as e: + self.notify(f"Setting failed: {e}", severity="error") + + def action_setting_kdf_iterations(self, n: str) -> None: + """Set the number of KDF iterations.""" + config = self.services.get("config") + if not config: + self.notify("Config service offline", severity="error") + return + try: + config.set("kdf_iterations", n) + self.notify(f"KDF iterations: {n}") + except Exception as e: + self.notify(f"Setting failed: {e}", severity="error") + + def action_archive_filter(self, mode: str) -> None: + """Set the archive visibility scope: active, all, or archived.""" + normalized = mode.lower() + if normalized not in {"active", "all", "archived"}: + self.notify( + "Usage: archive-filter ", severity="warning" + ) + return + if normalized == "archived": + self.show_archived = True + self.filter_archived_only = True + elif normalized == "all": + self.show_archived = True + self.filter_archived_only = False + else: + self.show_archived = False + self.filter_archived_only = False + self.notify(f"Archive filter: {normalized}") + self.action_refresh() + + def action_set_density(self, mode: str) -> None: + """Switch display density between compact and comfortable.""" + normalized = mode.lower() + if normalized not in {"compact", "comfortable"}: + self.notify( + "Usage: density ", severity="warning" + ) + return + self.density_mode = normalized + self.notify(f"Density: {normalized}") + + def action_show_onboarding(self) -> None: + """Show onboarding / quickstart guidance.""" + self.notify( + "Onboarding: add your first entry with 'add', reveal it with 'v', " + "inspect fields in the inspector pane, and use Ctrl+P to open the " + "command palette for all available operations." + ) diff --git a/src/tests/test_tui_v3_parity.py b/src/tests/test_tui_v3_parity.py index 894050d5..65ea93f2 100644 --- a/src/tests/test_tui_v3_parity.py +++ b/src/tests/test_tui_v3_parity.py @@ -164,6 +164,73 @@ class MockVaultService: } +class MockUtilityService: + def __init__(self): + self.checksum_verified = False + self.checksum_updated = False + + def verify_checksum(self): + self.checksum_verified = True + + def update_checksum(self): + self.checksum_updated = True + + +class MockSyncService: + def __init__(self): + self.synced = False + self.bg_synced = False + + def sync(self): + self.synced = True + return {"events": 1} + + def start_background_vault_sync(self, summary=None): + self.bg_synced = True + + +class MockNostrService: + def __init__(self): + self.relays: list[str] = ["wss://relay.example.com"] + self.added: list[str] = [] + self.removed: list[int] = [] + self.reset_called = False + + def add_relay(self, url: str): + self.added.append(url) + self.relays.append(url) + + def remove_relay(self, idx: int): + self.removed.append(idx) + + def reset_relays(self) -> list[str]: + self.reset_called = True + self.relays = ["wss://default.relay"] + return list(self.relays) + + +class MockConfigService: + def __init__(self): + self._cfg: dict = {} + self.secret_mode_calls: list[tuple] = [] + self.offline_mode_calls: list[bool] = [] + + def get(self, key: str): + return self._cfg.get(key) + + def set(self, key: str, value: str): + self._cfg[key] = value + + def get_clipboard_clear_delay(self) -> int: + return 30 + + def set_secret_mode(self, enabled: bool, delay: int): + self.secret_mode_calls.append((enabled, delay)) + + def set_offline_mode(self, enabled: bool): + self.offline_mode_calls.append(enabled) + + @pytest.mark.anyio async def test_v3_reveal_copy_parity(): app = SeedPassTuiV3( @@ -312,3 +379,183 @@ async def test_v3_action_bar_only_shows_context_actions_for_selected_kind() -> N assert "Load" in rendered assert "Delete" in rendered assert "Export" not in rendered + + +# --------------------------------------------------------------------------- +# Parity tests: legacy utility/maintenance commands migrated to v3 +# --------------------------------------------------------------------------- + + +def _make_full_app(**extra_services): + """Build a SeedPassTuiV3 with all mock services wired up.""" + entry_svc = MockEntryService() + vault_svc = MockVaultService() + utility_svc = MockUtilityService() + sync_svc = MockSyncService() + nostr_svc = MockNostrService() + config_svc = MockConfigService() + + app = SeedPassTuiV3( + entry_service_factory=lambda: entry_svc, + vault_service_factory=lambda: vault_svc, + utility_service_factory=lambda: utility_svc, + sync_service_factory=lambda: sync_svc, + nostr_service_factory=lambda: nostr_svc, + config_service_factory=lambda: config_svc, + ) + return app, entry_svc, vault_svc, utility_svc, sync_svc, nostr_svc, config_svc + + +@pytest.mark.anyio +async def test_v3_checksum_commands() -> None: + app, *_, utility_svc, _, _, _ = _make_full_app() + async with app.run_test() as pilot: + await pilot.pause() + app.action_checksum_verify() + assert utility_svc.checksum_verified is True + + app.action_checksum_update() + assert utility_svc.checksum_updated is True + + +@pytest.mark.anyio +async def test_v3_totp_export_no_path() -> None: + app, entry_svc, *_ = _make_full_app() + # Provide a basic export_totp_entries method + entry_svc.export_totp_entries = lambda: {"1": "otpauth://totp/label?secret=ABC"} + async with app.run_test() as pilot: + await pilot.pause() + # Should not raise; notifies count + app.action_totp_export(None) + + +@pytest.mark.anyio +async def test_v3_sync_commands() -> None: + app, _, _, _, sync_svc, _, _ = _make_full_app() + async with app.run_test() as pilot: + await pilot.pause() + app.action_sync_now() + assert sync_svc.synced is True + + app.action_sync_bg() + assert sync_svc.bg_synced is True + + +@pytest.mark.anyio +async def test_v3_relay_commands() -> None: + app, _, _, _, _, nostr_svc, _ = _make_full_app() + async with app.run_test() as pilot: + await pilot.pause() + + app.action_relay_add("wss://new.relay") + assert "wss://new.relay" in nostr_svc.added + + app.action_relay_rm(0) + assert 0 in nostr_svc.removed + + app.action_relay_reset() + assert nostr_svc.reset_called is True + + +@pytest.mark.anyio +async def test_v3_setting_secret() -> None: + app, _, _, _, _, _, config_svc = _make_full_app() + async with app.run_test() as pilot: + await pilot.pause() + app.action_setting_secret("on") + assert config_svc.secret_mode_calls[-1][0] is True + + app.action_setting_secret("off") + assert config_svc.secret_mode_calls[-1][0] is False + + +@pytest.mark.anyio +async def test_v3_setting_offline() -> None: + app, _, _, _, _, _, config_svc = _make_full_app() + async with app.run_test() as pilot: + await pilot.pause() + app.action_setting_offline("on") + assert config_svc.offline_mode_calls[-1] is True + + app.action_setting_offline("off") + assert config_svc.offline_mode_calls[-1] is False + + +@pytest.mark.anyio +async def test_v3_setting_config_keys() -> None: + app, _, _, _, _, _, config_svc = _make_full_app() + async with app.run_test() as pilot: + await pilot.pause() + app.action_setting_quick_unlock("on") + assert config_svc._cfg.get("quick_unlock") == "on" + + app.action_setting_timeout("600") + assert config_svc._cfg.get("inactivity_timeout") == "600" + + app.action_setting_kdf_mode("argon2id") + assert config_svc._cfg.get("kdf_mode") == "argon2id" + + app.action_setting_kdf_iterations("200000") + assert config_svc._cfg.get("kdf_iterations") == "200000" + + +@pytest.mark.anyio +async def test_v3_archive_filter() -> None: + app = SeedPassTuiV3( + entry_service_factory=lambda: MockEntryService(), + vault_service_factory=lambda: MockVaultService(), + ) + async with app.run_test() as pilot: + await pilot.pause() + + app.action_archive_filter("active") + assert app.show_archived is False + assert app.filter_archived_only is False + + app.action_archive_filter("all") + assert app.show_archived is True + assert app.filter_archived_only is False + + app.action_archive_filter("archived") + assert app.show_archived is True + assert app.filter_archived_only is True + + +@pytest.mark.anyio +async def test_v3_density() -> None: + app = SeedPassTuiV3( + entry_service_factory=lambda: MockEntryService(), + vault_service_factory=lambda: MockVaultService(), + ) + async with app.run_test() as pilot: + await pilot.pause() + app.action_set_density("compact") + assert app.density_mode == "compact" + + app.action_set_density("comfortable") + assert app.density_mode == "comfortable" + + +@pytest.mark.anyio +async def test_v3_command_processor_routes_utility_commands() -> None: + """Verify the CommandProcessor correctly routes new utility commands.""" + app, *_, utility_svc, sync_svc, nostr_svc, _ = _make_full_app() + async with app.run_test() as pilot: + await pilot.pause() + proc = app.processor + + proc.execute("checksum-verify") + assert utility_svc.checksum_verified is True + + proc.execute("sync-now") + assert sync_svc.synced is True + + proc.execute("relay-add wss://proc.relay") + assert "wss://proc.relay" in nostr_svc.added + + proc.execute("archive-filter archived") + assert app.show_archived is True + assert app.filter_archived_only is True + + proc.execute("density compact") + assert app.density_mode == "compact"