improved tests

This commit is contained in:
thePR0M3TH3AN committed 2026-02-25 18:07:22 -05:00
1 parent f7f81a29d7
commit c3eef44ab7
13 files changed
+1271 -50

No files matched your search

+59 -48
View File
@@ -352,6 +352,51 @@ class PasswordManager:
self.KEY_PW_DERIVE = kd(master, b"seedpass:v1:pw")
self.KEY_TOTP_DET = kd(master, b"seedpass:v1:totp")
def _get_kdf_mode(self) -> str:
if getattr(self, "config_manager", None):
return self.config_manager.get_kdf_mode()
return "pbkdf2"
def _get_kdf_iterations(self) -> int:
if getattr(self, "config_manager", None):
return self.config_manager.get_kdf_iterations()
return 50_000
def _get_argon2_time_cost(self) -> int:
if getattr(self, "config_manager", None):
return self.config_manager.get_argon2_time_cost()
return 2
def _derive_seed_key(
self,
password: str,
fingerprint: str,
*,
mode: str | None = None,
iterations: int | None = None,
) -> bytes:
"""Derive the profile seed-encryption key from current KDF settings."""
chosen_mode = mode or self._get_kdf_mode()
if chosen_mode == "argon2":
salt = hashlib.sha256(fingerprint.encode()).digest()[:16]
cfg = KdfConfig(
params={
"time_cost": self._get_argon2_time_cost(),
"memory_cost": 64 * 1024,
"parallelism": 8,
},
salt_b64=base64.b64encode(salt).decode(),
)
return derive_key_from_password_argon2(password, cfg)
if chosen_mode != "pbkdf2":
logger.warning("Unknown kdf_mode '%s'; falling back to pbkdf2", chosen_mode)
iter_count = (
iterations if iterations is not None else self._get_kdf_iterations()
)
return derive_key_from_password(password, fingerprint, iterations=iter_count)
def ensure_key_hierarchy(self) -> None:
"""Ensure sub-keys are derived from the current parent seed."""
if (
@@ -733,14 +778,12 @@ class PasswordManager:
seed_mgr: EncryptionManager | None = None
for iter_try in dict.fromkeys(iter_candidates):
try:
if mode == "argon2":
salt = hashlib.sha256(salt_fp.encode()).digest()[:16]
cfg = KdfConfig(salt_b64=base64.b64encode(salt).decode())
seed_key = derive_key_from_password_argon2(password, cfg)
else:
seed_key = derive_key_from_password(
password, salt_fp, iterations=iter_try
)
seed_key = self._derive_seed_key(
password,
salt_fp,
mode=mode,
iterations=iter_try if mode != "argon2" else None,
)
seed_mgr = EncryptionManager(seed_key, fingerprint_dir)
print("Decrypting seed...")
self.parent_seed = seed_mgr.decrypt_parent_seed()
@@ -814,14 +857,9 @@ class PasswordManager:
else 50_000
)
salt_fp = fingerprint_dir.name
if mode == "argon2":
salt = hashlib.sha256(salt_fp.encode()).digest()[:16]
cfg = KdfConfig(salt_b64=base64.b64encode(salt).decode())
seed_key = derive_key_from_password_argon2(password, cfg)
else:
seed_key = derive_key_from_password(
password, salt_fp, iterations=iterations
)
seed_key = self._derive_seed_key(
password, salt_fp, mode=mode, iterations=iterations
)
seed_mgr = EncryptionManager(seed_key, fingerprint_dir)
self.parent_seed = seed_mgr.decrypt_parent_seed()
seed_bytes = Bip39SeedGenerator(self.parent_seed).Generate()
@@ -948,14 +986,7 @@ class PasswordManager:
raise SeedPassError("Seed profile directory not found.")
# Derive encryption key from password using selected fingerprint
iterations = (
self.config_manager.get_kdf_iterations()
if getattr(self, "config_manager", None)
else 50_000
)
key = derive_key_from_password(
password, selected_fingerprint, iterations=iterations
)
key = self._derive_seed_key(password, selected_fingerprint)
seed_mgr = EncryptionManager(key, fingerprint_dir)
self.vault = Vault(seed_mgr, fingerprint_dir)
@@ -1125,14 +1156,7 @@ class PasswordManager:
seed_bytes = Bip39SeedGenerator(parent_seed).Generate()
self.derive_key_hierarchy(seed_bytes)
index_key = base64.urlsafe_b64encode(self.KEY_STORAGE)
iterations = (
self.config_manager.get_kdf_iterations()
if getattr(self, "config_manager", None)
else 50_000
)
seed_key = derive_key_from_password(
password, fingerprint, iterations=iterations
)
seed_key = self._derive_seed_key(password, fingerprint)
self.encryption_manager = EncryptionManager(index_key, fingerprint_dir)
seed_mgr = EncryptionManager(seed_key, fingerprint_dir)
@@ -1323,14 +1347,7 @@ class PasswordManager:
seed_bytes = Bip39SeedGenerator(seed).Generate()
self.derive_key_hierarchy(seed_bytes)
index_key = base64.urlsafe_b64encode(self.KEY_STORAGE)
iterations = (
self.config_manager.get_kdf_iterations()
if getattr(self, "config_manager", None)
else 50_000
)
seed_key = derive_key_from_password(
password, fingerprint_dir.name, iterations=iterations
)
seed_key = self._derive_seed_key(password, fingerprint_dir.name)
self.encryption_manager = EncryptionManager(index_key, fingerprint_dir)
seed_mgr = EncryptionManager(seed_key, fingerprint_dir)
@@ -4282,10 +4299,7 @@ class PasswordManager:
if mode == PortableMode.NONE.value:
try:
password = prompt_new_password()
iterations = self.config_manager.get_kdf_iterations()
seed_key = derive_key_from_password(
password, self.current_fingerprint, iterations=iterations
)
seed_key = self._derive_seed_key(password, self.current_fingerprint)
seed_mgr = EncryptionManager(seed_key, self.fingerprint_dir)
seed_mgr.encrypt_parent_seed(self.parent_seed)
self.store_hashed_password(password)
@@ -4584,10 +4598,7 @@ class PasswordManager:
# Create a new encryption manager with the new password
new_key = base64.urlsafe_b64encode(self.KEY_STORAGE)
iterations = self.config_manager.get_kdf_iterations()
seed_key = derive_key_from_password(
new_password, self.current_fingerprint, iterations=iterations
)
seed_key = self._derive_seed_key(new_password, self.current_fingerprint)
seed_mgr = EncryptionManager(seed_key, self.fingerprint_dir)
new_enc_mgr = EncryptionManager(new_key, self.fingerprint_dir)
+154
View File
@@ -0,0 +1,154 @@
from __future__ import annotations
import shutil
from pathlib import Path
from helpers import TEST_PASSWORD, TEST_SEED, create_vault
from seedpass.core.backup import BackupManager
from seedpass.core.config_manager import ConfigManager
from seedpass.core.entry_management import EntryManager
from seedpass.core.manager import EncryptionMode, PasswordManager
from seedpass.core.portable_backup import export_backup, import_backup
def _init_pm(dir_path: Path, nostr_client=None) -> PasswordManager:
vault, enc_mgr = create_vault(dir_path, TEST_SEED, TEST_PASSWORD)
cfg_mgr = ConfigManager(vault, dir_path)
backup_mgr = BackupManager(dir_path, cfg_mgr)
entry_mgr = EntryManager(vault, backup_mgr)
pm = PasswordManager.__new__(PasswordManager)
pm.encryption_mode = EncryptionMode.SEED_ONLY
pm.encryption_manager = enc_mgr
pm.vault = vault
pm.entry_manager = entry_mgr
pm.backup_manager = backup_mgr
pm.config_manager = cfg_mgr
pm.nostr_client = nostr_client
pm.fingerprint_dir = dir_path
pm.current_fingerprint = "fp"
pm.parent_seed = TEST_SEED
pm.is_dirty = False
pm.secret_mode_enabled = False
pm.state_manager = None
pm.offline_mode = False
pm.notify = lambda *a, **k: None
return pm
def _add_sample_entries(entry_mgr: EntryManager) -> None:
entry_mgr.add_entry(
"example.com",
14,
username="alice",
url="https://example.com",
notes="password note",
tags=["web"],
)
entry_mgr.add_totp(
"example-totp",
TEST_SEED,
deterministic=True,
period=45,
digits=8,
notes="totp note",
tags=["otp"],
)
entry_mgr.add_key_value(
"api-token",
"token",
"abc123",
notes="kv note",
tags=["api"],
)
def _assert_sample_entries(entry_mgr: EntryManager) -> None:
pw = entry_mgr.retrieve_entry(0)
totp = entry_mgr.retrieve_entry(1)
kv = entry_mgr.retrieve_entry(2)
assert pw is not None
assert pw["kind"] == "password"
assert pw["label"] == "example.com"
assert pw["username"] == "alice"
assert pw["url"] == "https://example.com"
assert pw["length"] == 14
assert pw["notes"] == "password note"
assert totp is not None
assert totp["kind"] == "totp"
assert totp["label"] == "example-totp"
assert totp["deterministic"] is True
assert totp["index"] == 0
assert totp["period"] == 45
assert totp["digits"] == 8
assert totp["notes"] == "totp note"
assert kv is not None
assert kv["kind"] == "key_value"
assert kv["label"] == "api-token"
assert kv["key"] == "token"
assert kv["value"] == "abc123"
assert kv["notes"] == "kv note"
labels = [item[1] for item in entry_mgr.list_entries(verbose=False)]
assert labels == ["example.com", "example-totp", "api-token"]
def test_account_roundtrip_via_nostr_seed_restore(tmp_path, make_dummy_nostr_client):
client, relay = make_dummy_nostr_client(tmp_path / "relay")
source_dir = tmp_path / "account-source"
source_dir.mkdir()
pm_source = _init_pm(source_dir, client)
_add_sample_entries(pm_source.entry_manager)
sync_result = pm_source.sync_vault()
assert sync_result is not None
assert relay.manifests
shutil.rmtree(source_dir)
assert not source_dir.exists()
restored_dir = tmp_path / "account-restored"
restored_dir.mkdir()
pm_restored = _init_pm(restored_dir, client)
restored = pm_restored.attempt_initial_sync()
assert restored is True
pm_restored.entry_manager.clear_cache()
_assert_sample_entries(pm_restored.entry_manager)
def test_account_roundtrip_via_export_import_seed_restore(tmp_path):
source_dir = tmp_path / "account-source"
source_dir.mkdir()
pm_source = _init_pm(source_dir, None)
_add_sample_entries(pm_source.entry_manager)
export_path = tmp_path / "seedpass-portable-export.json"
path = export_backup(
pm_source.vault,
pm_source.backup_manager,
dest_path=export_path,
parent_seed=TEST_SEED,
encrypt=True,
)
assert path.exists()
shutil.rmtree(source_dir)
assert not source_dir.exists()
restored_dir = tmp_path / "account-restored"
restored_dir.mkdir()
pm_restored = _init_pm(restored_dir, None)
import_backup(
pm_restored.vault,
pm_restored.backup_manager,
path,
parent_seed=TEST_SEED,
)
pm_restored.entry_manager.clear_cache()
_assert_sample_entries(pm_restored.entry_manager)
+126
View File
@@ -0,0 +1,126 @@
from __future__ import annotations
from pathlib import Path
import pytest
from helpers import TEST_PASSWORD, TEST_SEED, create_vault
from seedpass.core.backup import BackupManager
from seedpass.core.config_manager import ConfigManager
from seedpass.core.entry_management import EntryManager
from seedpass.core.entry_types import ALL_ENTRY_TYPES, EntryType
def _make_entry_manager(tmp_path: Path) -> EntryManager:
vault, _ = create_vault(tmp_path, TEST_SEED, TEST_PASSWORD)
cfg_mgr = ConfigManager(vault, tmp_path)
backup_mgr = BackupManager(tmp_path, cfg_mgr)
return EntryManager(vault, backup_mgr)
def _add_entry_for_type(entry_mgr: EntryManager, entry_type: EntryType) -> int:
if entry_type == EntryType.PASSWORD:
return entry_mgr.add_entry(
"Password Entry",
16,
username="alice",
url="https://example.com",
notes="pw-note",
tags=["auth"],
)
if entry_type == EntryType.TOTP:
entry_mgr.add_totp(
"TOTP Entry",
TEST_SEED,
deterministic=True,
period=45,
digits=8,
notes="totp-note",
tags=["otp"],
)
return 0
if entry_type == EntryType.SSH:
return entry_mgr.add_ssh_key("SSH Entry", TEST_SEED, notes="ssh-note")
if entry_type == EntryType.SEED:
return entry_mgr.add_seed(
"Seed Entry", TEST_SEED, words_num=12, notes="seed-note", tags=["seed"]
)
if entry_type == EntryType.PGP:
return entry_mgr.add_pgp_key(
"PGP Entry",
TEST_SEED,
key_type="ed25519",
user_id="[email protected]",
notes="pgp-note",
tags=["pgp"],
)
if entry_type == EntryType.NOSTR:
return entry_mgr.add_nostr_key("Nostr Entry", TEST_SEED, notes="nostr-note")
if entry_type == EntryType.KEY_VALUE:
return entry_mgr.add_key_value(
"KV Entry",
"api_key",
"secret-value",
notes="kv-note",
tags=["kv"],
)
if entry_type == EntryType.MANAGED_ACCOUNT:
return entry_mgr.add_managed_account(
"Managed Entry", TEST_SEED, notes="managed-note", tags=["managed"]
)
raise AssertionError(f"Unhandled entry type in test: {entry_type}")
def test_all_entry_type_values_are_enum_backed():
assert set(ALL_ENTRY_TYPES) == {entry_type.value for entry_type in EntryType}
@pytest.mark.parametrize("entry_type", list(EntryType))
def test_each_entry_type_roundtrip_retrieve_and_filter(tmp_path, entry_type: EntryType):
entry_mgr = _make_entry_manager(tmp_path)
index = _add_entry_for_type(entry_mgr, entry_type)
entry = entry_mgr.retrieve_entry(index)
assert entry is not None
assert entry["type"] == entry_type.value
assert entry["kind"] == entry_type.value
assert entry["label"]
assert "modified_ts" not in entry
entries = entry_mgr.list_entries(
filter_kinds=[entry_type.value], include_archived=True, verbose=False
)
assert len(entries) == 1
listed_index, listed_label, listed_username, listed_url, listed_archived = entries[0]
assert listed_index == index
assert listed_label == entry["label"]
assert listed_archived is False
if entry_type == EntryType.PASSWORD:
assert listed_username == "alice"
assert listed_url == "https://example.com"
assert entry["length"] == 16
assert entry["notes"] == "pw-note"
else:
assert listed_username is None
assert listed_url is None
if entry_type == EntryType.TOTP:
assert entry["deterministic"] is True
assert entry["period"] == 45
assert entry["digits"] == 8
if entry_type == EntryType.SSH:
assert "index" in entry
if entry_type == EntryType.SEED:
assert entry["word_count"] == 12
if entry_type == EntryType.PGP:
assert entry["key_type"] == "ed25519"
assert entry["user_id"] == "[email protected]"
if entry_type == EntryType.NOSTR:
assert "index" in entry
if entry_type == EntryType.KEY_VALUE:
assert entry["key"] == "api_key"
assert entry["value"] == "secret-value"
if entry_type == EntryType.MANAGED_ACCOUNT:
assert entry["word_count"] == 12
assert entry["fingerprint"]
+102
View File
@@ -0,0 +1,102 @@
import io
import pytest
from utils import input_utils
def test_timed_input_no_timeout_reads_stdin(monkeypatch):
monkeypatch.setattr(input_utils.sys, "stdin", io.StringIO(" hello \n"))
assert input_utils.timed_input("Prompt: ", None) == "hello"
def test_timed_input_dispatches_windows(monkeypatch):
monkeypatch.setattr(input_utils.sys, "platform", "win32")
monkeypatch.setattr(
input_utils, "_timed_input_windows", lambda timeout: f"windows:{timeout}"
)
assert input_utils.timed_input("Prompt: ", 1.5) == "windows:1.5"
def test_timed_input_dispatches_posix(monkeypatch):
monkeypatch.setattr(input_utils.sys, "platform", "linux")
monkeypatch.setattr(
input_utils, "_timed_input_posix", lambda timeout: f"posix:{timeout}"
)
assert input_utils.timed_input("Prompt: ", 2.0) == "posix:2.0"
def test_timed_input_posix_fileno_fallback(monkeypatch):
monkeypatch.setattr(input_utils.sys, "stdin", io.StringIO(" via-input \n"))
assert input_utils._timed_input_posix(timeout=5.0) == "via-input"
def test_timed_input_posix_returns_line_when_ready(monkeypatch):
class DummyStdin:
def fileno(self):
return 0
def readline(self):
return "value\n"
dummy_stdin = DummyStdin()
monkeypatch.setattr(input_utils.sys, "stdin", dummy_stdin)
monkeypatch.setattr(
input_utils.select, "select", lambda r, _w, _x, _t: ([dummy_stdin], [], [])
)
assert input_utils._timed_input_posix(timeout=0.1) == "value"
def test_timed_input_posix_raises_timeout(monkeypatch):
class DummyStdin:
def fileno(self):
return 0
dummy_stdin = DummyStdin()
monkeypatch.setattr(input_utils.sys, "stdin", dummy_stdin)
monkeypatch.setattr(
input_utils.select, "select", lambda _r, _w, _x, _t: ([], [], [])
)
with pytest.raises(TimeoutError, match="input timed out"):
input_utils._timed_input_posix(timeout=0.1)
def test_timed_input_windows_fallback_when_msvcrt_missing(monkeypatch):
monkeypatch.setattr(input_utils, "msvcrt", None)
monkeypatch.setattr("builtins.input", lambda: " typed ")
assert input_utils._timed_input_windows(timeout=0.5) == "typed"
def test_timed_input_windows_handles_backspace_and_enter(monkeypatch):
class DummyMSVCRT:
def __init__(self):
self._chars = iter(["a", "\b", "b", "\r"])
def kbhit(self):
return True
def getwche(self):
return next(self._chars)
monkeypatch.setattr(input_utils, "msvcrt", DummyMSVCRT())
monkeypatch.setattr(input_utils.time, "sleep", lambda _s: None)
monkeypatch.setattr(input_utils.time, "time", lambda: 0.0)
assert input_utils._timed_input_windows(timeout=10.0) == "b"
def test_timed_input_windows_raises_timeout(monkeypatch):
class DummyMSVCRT:
def kbhit(self):
return False
def getwche(self):
return ""
timestamps = iter([0.0, 2.0])
monkeypatch.setattr(input_utils, "msvcrt", DummyMSVCRT())
monkeypatch.setattr(input_utils.time, "sleep", lambda _s: None)
monkeypatch.setattr(input_utils.time, "time", lambda: next(timestamps))
with pytest.raises(TimeoutError, match="input timed out"):
input_utils._timed_input_windows(timeout=1.0)
+52
View File
@@ -116,3 +116,55 @@ def test_vault_kdf_migration(tmp_path):
vault.load_index()
new_kdf = mgr.get_file_kdf(vault.index_file)
assert new_kdf.version == KdfConfig().version
def test_derive_seed_key_argon2_uses_configured_time_cost(monkeypatch):
pm = PasswordManager.__new__(PasswordManager)
pm.config_manager = SimpleNamespace(
get_kdf_mode=lambda: "argon2",
get_argon2_time_cost=lambda: 5,
get_kdf_iterations=lambda: 50_000,
)
captured = {}
def fake_argon2(password, cfg):
captured["password"] = password
captured["cfg"] = cfg
return b"derived-argon2-key"
monkeypatch.setattr(
"seedpass.core.manager.derive_key_from_password_argon2", fake_argon2
)
key = pm._derive_seed_key("pw", "fingerprint123")
assert key == b"derived-argon2-key"
assert captured["password"] == "pw"
assert captured["cfg"].params["time_cost"] == 5
assert captured["cfg"].params["memory_cost"] == 64 * 1024
assert captured["cfg"].params["parallelism"] == 8
expected_salt = hashlib.sha256("fingerprint123".encode()).digest()[:16]
assert base64.b64decode(captured["cfg"].salt_b64) == expected_salt
def test_derive_seed_key_pbkdf2_respects_iteration_override(monkeypatch):
pm = PasswordManager.__new__(PasswordManager)
pm.config_manager = SimpleNamespace(
get_kdf_mode=lambda: "pbkdf2",
get_argon2_time_cost=lambda: 2,
get_kdf_iterations=lambda: 777,
)
captured = {}
def fake_pbkdf2(password, fingerprint, iterations):
captured["password"] = password
captured["fingerprint"] = fingerprint
captured["iterations"] = iterations
return b"derived-pbkdf2-key"
monkeypatch.setattr("seedpass.core.manager.derive_key_from_password", fake_pbkdf2)
key = pm._derive_seed_key("pw", "fp", iterations=123)
assert key == b"derived-pbkdf2-key"
assert captured == {"password": "pw", "fingerprint": "fp", "iterations": 123}
+17 -2
View File
@@ -11,7 +11,6 @@ import uuid
import pytest
import base64
import os
sys.path.append(str(Path(__file__).resolve().parents[1]))
@@ -23,8 +22,18 @@ from seedpass.core.config_manager import ConfigManager
from nostr.client import NostrClient, Kind, KindStandard
def _relays_from_env() -> list[str]:
raw = os.getenv("NOSTR_RELAYS", "")
relays = [value.strip() for value in raw.split(",") if value.strip()]
return relays or ["wss://relay.snort.social"]
@pytest.mark.desktop
@pytest.mark.network
@pytest.mark.skipif(
os.getenv("NOSTR_INDEX_SIZE_E2E") != "1",
reason="Set NOSTR_INDEX_SIZE_E2E=1 to run index size E2E test",
)
def test_nostr_index_size_limits(pytestconfig: pytest.Config):
"""Manually explore maximum index size for Nostr backups."""
seed = (
@@ -36,16 +45,22 @@ def test_nostr_index_size_limits(pytestconfig: pytest.Config):
key = base64.urlsafe_b64encode(os.urandom(32))
enc_mgr = EncryptionManager(key, Path(tmpdir))
with patch.object(enc_mgr, "decrypt_parent_seed", return_value=seed):
account_index = int(uuid.uuid4().int % (2**31 - 1))
client = NostrClient(
enc_mgr,
f"size_test_{uuid.uuid4().hex}",
relays=["wss://relay.snort.social"],
relays=_relays_from_env(),
account_index=account_index,
)
npub = client.key_manager.get_npub()
vault = Vault(enc_mgr, tmpdir)
cfg_mgr = ConfigManager(vault, Path(tmpdir))
backup_mgr = BackupManager(Path(tmpdir), cfg_mgr)
entry_mgr = EntryManager(vault, backup_mgr)
# This test targets payload growth and relay round-trip integrity.
# Disable per-entry local backup/checksum work to keep runtime bounded.
backup_mgr.create_backup = lambda: None
entry_mgr.update_checksum = lambda: None
delay = float(os.getenv("NOSTR_TEST_DELAY", "5"))
max_entries = pytestconfig.getoption("--max-entries")
+222
View File
@@ -14,7 +14,13 @@ import base64
sys.path.append(str(Path(__file__).resolve().parents[1]))
from seedpass.core.encryption import EncryptionManager
from seedpass.core.backup import BackupManager
from seedpass.core.config_manager import ConfigManager
from seedpass.core.entry_management import EntryManager
from seedpass.core.manager import EncryptionMode, PasswordManager
from seedpass.core.portable_backup import export_backup, import_backup
from nostr.client import NostrClient
from helpers import create_vault, TEST_SEED, TEST_PASSWORD
@pytest.mark.network
@@ -41,3 +47,219 @@ def test_nostr_publish_and_retrieve():
retrieved = gzip.decompress(b"".join(result[1])) if result else None
client.close_client_pool()
assert retrieved == payload
def _relays_from_env() -> list[str]:
raw = os.getenv("NOSTR_RELAYS", "")
relays = [value.strip() for value in raw.split(",") if value.strip()]
return relays or ["wss://relay.snort.social"]
def _init_pm_for_real_nostr(dir_path: Path, client: NostrClient) -> PasswordManager:
vault, enc_mgr = create_vault(dir_path, TEST_SEED, TEST_PASSWORD)
cfg_mgr = ConfigManager(vault, dir_path)
backup_mgr = BackupManager(dir_path, cfg_mgr)
entry_mgr = EntryManager(vault, backup_mgr)
pm = PasswordManager.__new__(PasswordManager)
pm.encryption_mode = EncryptionMode.SEED_ONLY
pm.encryption_manager = enc_mgr
pm.vault = vault
pm.entry_manager = entry_mgr
pm.backup_manager = backup_mgr
pm.config_manager = cfg_mgr
pm.nostr_client = client
pm.fingerprint_dir = dir_path
pm.current_fingerprint = "fp"
pm.parent_seed = TEST_SEED
pm.is_dirty = False
pm.secret_mode_enabled = False
pm.state_manager = None
pm.offline_mode = False
pm.notify = lambda *a, **k: None
return pm
@pytest.mark.network
@pytest.mark.skipif(not os.getenv("NOSTR_E2E"), reason="NOSTR_E2E not set")
def test_account_roundtrip_sync_delete_restore_real_relays():
"""Real relay roundtrip:
create account -> add entries -> sync -> delete local -> restore by seed.
"""
relays = _relays_from_env()
account_index = int(uuid.uuid4().int % (2**31 - 1))
fingerprint = f"test_fp_{uuid.uuid4().hex}"
with TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
source_dir = root / "source"
restore_dir = root / "restore"
source_dir.mkdir()
restore_dir.mkdir()
_, source_enc = create_vault(source_dir, TEST_SEED, TEST_PASSWORD)
source_client = NostrClient(
source_enc,
fingerprint,
relays=relays,
parent_seed=TEST_SEED,
account_index=account_index,
)
pm_source = _init_pm_for_real_nostr(source_dir, source_client)
pm_source.entry_manager.add_entry(
"roundtrip-site", 15, username="alice", url="https://example.com"
)
pm_source.entry_manager.add_totp(
"roundtrip-totp",
TEST_SEED,
deterministic=True,
period=45,
digits=8,
)
pm_source.entry_manager.add_key_value("roundtrip-kv", "token", "abc123")
sync = pm_source.sync_vault()
assert sync is not None
index_file = source_dir / "seedpass_entries_db.json.enc"
if index_file.exists():
index_file.unlink()
assert not index_file.exists()
restore_vault, restore_enc = create_vault(restore_dir, TEST_SEED, TEST_PASSWORD)
restore_client = NostrClient(
restore_enc,
fingerprint,
relays=relays,
parent_seed=TEST_SEED,
account_index=account_index,
)
pm_restore = _init_pm_for_real_nostr(restore_dir, restore_client)
restored = False
for attempt in range(6):
restored = pm_restore.attempt_initial_sync()
if restored:
break
time.sleep(1.5 * (attempt + 1))
assert restored is True
pm_restore.entry_manager.clear_cache()
labels = [item[1] for item in pm_restore.entry_manager.list_entries(verbose=False)]
assert sorted(labels) == ["roundtrip-kv", "roundtrip-site", "roundtrip-totp"]
pw = pm_restore.entry_manager.retrieve_entry(0)
totp = pm_restore.entry_manager.retrieve_entry(1)
kv = pm_restore.entry_manager.retrieve_entry(2)
assert pw and pw["kind"] == "password" and pw["username"] == "alice"
assert totp and totp["kind"] == "totp" and totp["deterministic"] is True
assert kv and kv["kind"] == "key_value" and kv["value"] == "abc123"
source_client.close_client_pool()
restore_client.close_client_pool()
@pytest.mark.network
@pytest.mark.skipif(not os.getenv("NOSTR_E2E"), reason="NOSTR_E2E not set")
def test_real_relays_with_portable_export_import_roundtrip():
"""Live relay + portable backup chain in one E2E flow."""
relays = _relays_from_env()
account_index = int(uuid.uuid4().int % (2**31 - 1))
fingerprint = f"test_fp_{uuid.uuid4().hex}"
with TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
source_dir = root / "source"
import_dir = root / "imported"
relay_restore_dir = root / "relay-restore"
source_dir.mkdir()
import_dir.mkdir()
relay_restore_dir.mkdir()
source_vault, source_enc = create_vault(source_dir, TEST_SEED, TEST_PASSWORD)
source_client = NostrClient(
source_enc,
fingerprint,
relays=relays,
parent_seed=TEST_SEED,
account_index=account_index,
)
pm_source = _init_pm_for_real_nostr(source_dir, source_client)
pm_source.entry_manager.add_entry(
"portable-site", 18, username="carol", url="https://portable.example"
)
pm_source.entry_manager.add_totp(
"portable-totp",
TEST_SEED,
deterministic=True,
period=30,
digits=6,
)
pm_source.entry_manager.add_key_value("portable-kv", "api", "xyz789")
first_sync = pm_source.sync_vault()
assert first_sync is not None
export_path = root / "portable-export.json"
path = export_backup(
pm_source.vault,
pm_source.backup_manager,
dest_path=export_path,
parent_seed=TEST_SEED,
encrypt=True,
)
assert path.exists()
_, import_enc = create_vault(import_dir, TEST_SEED, TEST_PASSWORD)
import_client = NostrClient(
import_enc,
fingerprint,
relays=relays,
parent_seed=TEST_SEED,
account_index=account_index,
)
pm_import = _init_pm_for_real_nostr(import_dir, import_client)
import_backup(
pm_import.vault,
pm_import.backup_manager,
path,
parent_seed=TEST_SEED,
)
pm_import.entry_manager.clear_cache()
imported_labels = [
item[1] for item in pm_import.entry_manager.list_entries(verbose=False)
]
assert sorted(imported_labels) == ["portable-kv", "portable-site", "portable-totp"]
resync = pm_import.sync_vault()
assert resync is not None
_, relay_enc = create_vault(relay_restore_dir, TEST_SEED, TEST_PASSWORD)
relay_client = NostrClient(
relay_enc,
fingerprint,
relays=relays,
parent_seed=TEST_SEED,
account_index=account_index,
)
pm_relay_restore = _init_pm_for_real_nostr(relay_restore_dir, relay_client)
restored = False
for attempt in range(6):
restored = pm_relay_restore.attempt_initial_sync()
if restored:
break
time.sleep(1.5 * (attempt + 1))
assert restored is True
pm_relay_restore.entry_manager.clear_cache()
relay_labels = [
item[1] for item in pm_relay_restore.entry_manager.list_entries(verbose=False)
]
assert sorted(relay_labels) == ["portable-kv", "portable-site", "portable-totp"]
source_client.close_client_pool()
import_client.close_client_pool()
relay_client.close_client_pool()
+115
View File
@@ -0,0 +1,115 @@
import queue
from types import SimpleNamespace
from utils import terminal_utils
class _NamedFingerprintManager:
def get_name(self, fingerprint):
return {"parent": "Parent Name", "child": "Child Name"}.get(fingerprint)
def test_format_profile_returns_custom_name():
pm = SimpleNamespace(fingerprint_manager=_NamedFingerprintManager())
assert terminal_utils.format_profile("parent", pm) == "Parent Name (parent)"
def test_format_profile_returns_raw_when_name_missing():
pm = SimpleNamespace(fingerprint_manager=_NamedFingerprintManager())
assert terminal_utils.format_profile("unknown", pm) == "unknown"
def test_clear_and_print_fingerprint_parent_child_breadcrumb(monkeypatch, capsys):
pm = SimpleNamespace(fingerprint_manager=_NamedFingerprintManager())
monkeypatch.setattr(terminal_utils, "clear_screen", lambda: None)
monkeypatch.setattr(terminal_utils, "colored", lambda text, _color: text)
terminal_utils.clear_and_print_fingerprint(
parent_fingerprint="parent",
child_fingerprint="child",
breadcrumb="Settings",
pm=pm,
)
out = capsys.readouterr().out
assert "Seed Profile: Parent Name (parent) > Managed Account > Child Name (child)" in out
assert "> Settings" in out
def test_clear_and_print_profile_chain_with_breadcrumb(monkeypatch, capsys):
pm = SimpleNamespace(fingerprint_manager=_NamedFingerprintManager())
monkeypatch.setattr(terminal_utils, "clear_screen", lambda: None)
monkeypatch.setattr(terminal_utils, "colored", lambda text, _color: text)
terminal_utils.clear_and_print_profile_chain(
["parent", "child"], breadcrumb="Retrieve Entry", pm=pm
)
out = capsys.readouterr().out
assert "Seed Profile: Parent Name (parent) > Managed Account > Child Name (child)" in out
assert "> Retrieve Entry" in out
def test_clear_header_with_notification_invalid_level_falls_back_to_info(
monkeypatch, capsys
):
pm = SimpleNamespace(
fingerprint_manager=_NamedFingerprintManager(),
get_current_notification=lambda: SimpleNamespace(
level="DEBUG", message="sync complete"
),
)
monkeypatch.setattr(terminal_utils, "clear_screen", lambda: None)
monkeypatch.setattr(terminal_utils, "colored", lambda text, _color: text)
monkeypatch.setattr(
terminal_utils, "color_text", lambda message, category: f"[{category}] {message}"
)
terminal_utils.clear_header_with_notification(pm, fingerprint="parent")
out = capsys.readouterr().out
assert "Seed Profile: Parent Name (parent)" in out
assert "[info] sync complete" in out
def test_clear_header_with_notification_handles_queue_empty(monkeypatch, capsys):
pm = SimpleNamespace(
fingerprint_manager=_NamedFingerprintManager(),
get_current_notification=lambda: (_ for _ in ()).throw(queue.Empty()),
)
monkeypatch.setattr(terminal_utils, "clear_screen", lambda: None)
monkeypatch.setattr(terminal_utils, "colored", lambda text, _color: text)
terminal_utils.clear_header_with_notification(pm, fingerprint="parent")
out = capsys.readouterr().out
assert "Seed Profile: Parent Name (parent)" in out
def test_pause_returns_early_when_not_tty(monkeypatch):
monkeypatch.setattr(
terminal_utils.sys,
"stdin",
SimpleNamespace(isatty=lambda: False),
)
called = {"input": False}
monkeypatch.setattr(
"builtins.input",
lambda _msg: called.__setitem__("input", True),
)
terminal_utils.pause("ignored")
assert called["input"] is False
def test_pause_swallows_eoferror(monkeypatch):
monkeypatch.setattr(
terminal_utils.sys,
"stdin",
SimpleNamespace(isatty=lambda: True),
)
monkeypatch.setattr(
"builtins.input",
lambda _msg: (_ for _ in ()).throw(EOFError()),
)
terminal_utils.pause("Press Enter")