Add document workflows, entry graph links, and discovery/docs updates

This commit is contained in:
thePR0M3TH3AN committed 2026-03-01 15:43:43 -05:00
1 parent 64efd1175d
commit d0ea0d96f6
41 files changed
+2904 -153

No files matched your search

+142
View File
@@ -109,6 +109,148 @@ async def test_create_and_modify_ssh_entry(client):
assert calls["modify"][1]["notes"] == "x"
@pytest.mark.anyio
async def test_create_and_modify_document_entry(client):
cl, token = client
calls = {}
def add_document(label, content, **kwargs):
calls["create"] = (label, content, kwargs)
return 11
def modify(idx, **kwargs):
calls["modify"] = (idx, kwargs)
api.app.state.pm.entry_manager.add_document = add_document
api.app.state.pm.entry_manager.modify_entry = modify
headers = {"Authorization": f"Bearer {token}"}
res = await cl.post(
"/api/v1/entry",
json={
"type": "document",
"title": "Runbook",
"content": "line1\\nline2",
"file_type": "md",
"notes": "ops",
"tags": ["doc", "ops"],
},
headers=headers,
)
assert res.status_code == 200
assert res.json() == {"id": 11}
assert calls["create"] == (
"Runbook",
"line1\\nline2",
{"file_type": "md", "notes": "ops", "archived": False, "tags": ["doc", "ops"]},
)
res = await cl.put(
"/api/v1/entry/11",
json={"content": "line1\\nline2\\nline3", "file_type": "txt"},
headers=headers,
)
assert res.status_code == 200
assert calls["modify"][0] == 11
assert calls["modify"][1]["content"].endswith("line3")
assert calls["modify"][1]["file_type"] == "txt"
@pytest.mark.anyio
async def test_document_import_export_endpoints(client):
cl, token = client
calls = {}
def import_document(path, **kwargs):
calls["import"] = (path, kwargs)
return 33
def export_document(idx, **kwargs):
calls["export"] = (idx, kwargs)
return Path("/tmp/exported.md")
api.app.state.pm.entry_manager.import_document_file = import_document
api.app.state.pm.entry_manager.export_document_file = export_document
headers = {"Authorization": f"Bearer {token}"}
imported = await cl.post(
"/api/v1/entry/document/import",
json={"path": "/tmp/source.md", "label": "Imported", "notes": "n"},
headers=headers,
)
assert imported.status_code == 200
assert imported.json() == {"id": 33}
assert calls["import"] == (
"/tmp/source.md",
{"label": "Imported", "notes": "n", "archived": False, "tags": None},
)
exported = await cl.post(
"/api/v1/entry/33/document/export",
json={"path": "/tmp/out", "overwrite": True},
headers=headers,
)
assert exported.status_code == 200
assert exported.json() == {"path": "/tmp/exported.md"}
assert calls["export"] == (33, {"output_path": "/tmp/out", "overwrite": True})
@pytest.mark.anyio
async def test_entry_links_endpoints(client):
cl, token = client
calls = {}
def add_link(entry_id, target_id, **kwargs):
calls["add"] = (entry_id, target_id, kwargs)
return [{"target_id": target_id, "relation": kwargs["relation"], "note": kwargs["note"]}]
def remove_link(entry_id, target_id, **kwargs):
calls["remove"] = (entry_id, target_id, kwargs)
return []
def get_links(entry_id):
calls["get"] = entry_id
return [
{
"target_id": 2,
"relation": "references",
"note": "used by doc",
"target_label": "API Token",
"target_kind": "key_value",
}
]
api.app.state.pm.entry_manager.add_link = add_link
api.app.state.pm.entry_manager.remove_link = remove_link
api.app.state.pm.entry_manager.get_links = get_links
headers = {"Authorization": f"Bearer {token}"}
created = await cl.post(
"/api/v1/entry/1/links",
json={"target_id": 2, "relation": "references", "note": "used by doc"},
headers=headers,
)
assert created.status_code == 200
assert created.json()["links"][0]["target_id"] == 2
assert calls["add"] == (1, 2, {"relation": "references", "note": "used by doc"})
listed = await cl.get("/api/v1/entry/1/links", headers=headers)
assert listed.status_code == 200
assert listed.json()["entry_id"] == 1
assert listed.json()["links"][0]["target_kind"] == "key_value"
assert calls["get"] == 1
removed = await cl.request(
"DELETE",
"/api/v1/entry/1/links",
json={"target_id": 2, "relation": "references"},
headers=headers,
)
assert removed.status_code == 200
assert removed.json() == {"entry_id": 1, "links": []}
assert calls["remove"] == (1, 2, {"relation": "references"})
@pytest.mark.anyio
async def test_update_entry_error(client):
cl, token = client
+202
View File
@@ -586,6 +586,207 @@ def test_agent_token_issue_and_use_limit(monkeypatch, tmp_path):
assert second_payload["reason"] == "token_exhausted"
def test_agent_document_import_with_token_scope(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
monkeypatch.setenv("SEEDPASS_PASSWORD", "pw")
monkeypatch.setattr(
agent_cli,
"PasswordManager",
lambda fingerprint, password: SimpleNamespace(
fingerprint=fingerprint, password=password
),
)
class DummyEntryService:
def __init__(self, _pm):
pass
def import_document_file(self, file_path, **kwargs):
assert str(file_path).endswith("notes.md")
assert kwargs["label"] == "My Notes"
assert kwargs["notes"] == "imported"
assert kwargs["tags"] == ["work"]
assert kwargs["archived"] is False
return 12
def retrieve_entry(self, _i):
return {"kind": EntryType.DOCUMENT.value, "label": "My Notes", "file_type": "md"}
monkeypatch.setattr(agent_cli, "EntryService", DummyEntryService)
monkeypatch.setattr(
agent_cli,
"_load_policy",
lambda **kwargs: {
"version": 1,
"default_effect": "deny",
"rules": [
{
"id": "allow_doc_read",
"effect": "allow",
"operations": ["read"],
"kinds": ["document"],
"label_regex": ".*",
"path_regex": "^entry/.*$",
"fields": ["secret"],
}
],
"approvals": {"require_for": []},
"output": {"safe_output_default": True, "redact_fields": ["secret"]},
"export": {"allow_full_vault": True},
"secret_isolation": {"enabled": False, "high_risk_kinds": [], "unlock_ttl_sec": 300},
"allow_kinds": ["document"],
"deny_private_reveal": [],
"allow_export_import": True,
},
)
source = tmp_path / "notes.md"
source.write_text("# hello\n", encoding="utf-8")
issue = runner.invoke(
app,
[
"agent",
"token-issue",
"--name",
"doc-import",
"--scope",
"import",
"--kind",
"document",
"--uses",
"1",
"--ttl",
"600",
],
)
assert issue.exit_code == 0
token = json.loads(issue.stdout)["token"]
result = runner.invoke(
app,
[
"--fingerprint",
"ABC123",
"agent",
"document-import",
"--file",
str(source),
"--label",
"My Notes",
"--notes",
"imported",
"--tag",
"work",
"--token",
token,
],
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
assert payload["index"] == 12
assert payload["kind"] == "document"
assert payload["file_type"] == "md"
assert payload["token_uses_remaining"] == 0
def test_agent_document_export_with_token_scope(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
monkeypatch.setenv("SEEDPASS_PASSWORD", "pw")
monkeypatch.setattr(
agent_cli,
"PasswordManager",
lambda fingerprint, password: SimpleNamespace(
fingerprint=fingerprint, password=password
),
)
class DummyEntryService:
def __init__(self, _pm):
pass
def retrieve_entry(self, _i):
return {"kind": EntryType.DOCUMENT.value, "label": "Spec"}
def export_document_file(self, _entry_id, out, **kwargs):
assert out == str(tmp_path)
assert kwargs["overwrite"] is True
return tmp_path / "Spec.md"
monkeypatch.setattr(agent_cli, "EntryService", DummyEntryService)
monkeypatch.setattr(
agent_cli,
"_load_policy",
lambda **kwargs: {
"version": 1,
"default_effect": "deny",
"rules": [
{
"id": "allow_doc_read",
"effect": "allow",
"operations": ["read"],
"kinds": ["document"],
"label_regex": ".*",
"path_regex": "^entry/.*$",
"fields": ["secret"],
}
],
"approvals": {"require_for": []},
"output": {"safe_output_default": True, "redact_fields": ["secret"]},
"export": {"allow_full_vault": True},
"secret_isolation": {"enabled": False, "high_risk_kinds": [], "unlock_ttl_sec": 300},
"allow_kinds": ["document"],
"deny_private_reveal": [],
"allow_export_import": True,
},
)
issue = runner.invoke(
app,
[
"agent",
"token-issue",
"--name",
"doc-export",
"--scope",
"export",
"--kind",
"document",
"--uses",
"1",
"--ttl",
"600",
],
)
assert issue.exit_code == 0
token = json.loads(issue.stdout)["token"]
result = runner.invoke(
app,
[
"--fingerprint",
"ABC123",
"agent",
"document-export",
"--entry-id",
"4",
"--out",
str(tmp_path),
"--overwrite",
"--token",
token,
],
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
assert payload["index"] == 4
assert payload["kind"] == "document"
assert payload["output_path"] == str(tmp_path / "Spec.md")
assert payload["token_uses_remaining"] == 0
def test_agent_identity_create_list_revoke(monkeypatch, tmp_path):
import seedpass.core.agent_identity as identity_core
@@ -1444,6 +1645,7 @@ def test_agent_bootstrap_context(monkeypatch, tmp_path):
assert "agent identity-create" in payload["commands"]["identities"]
assert "agent high-risk-unlock" in payload["commands"]["secret_isolation"]
assert "agent export-check" in payload["commands"]["export_controls"]
assert "agent document-export <query>" in payload["commands"]["document_io"]
assert "agent posture-check" in payload["commands"]["posture"]
assert "agent posture-remediate" in payload["commands"]["posture"]
assert "identities" in payload
+159
View File
@@ -1,5 +1,6 @@
import pytest
from types import SimpleNamespace
from pathlib import Path
from typer.testing import CliRunner
from seedpass.cli import app
@@ -133,6 +134,14 @@ runner = CliRunner()
{"index": 7, "notes": "n"},
"8",
),
(
"add-document",
"add_document",
["Doc", "--content", "hello", "--file-type", "md", "--notes", "n"],
("Doc", "hello"),
{"file_type": "md", "notes": "n", "tags": None, "archived": False},
"9",
),
],
)
def test_entry_add_commands(
@@ -161,3 +170,153 @@ def test_entry_add_commands(
assert called["args"] == expected_args
assert called["kwargs"] == expected_kwargs
assert called.get("sync") is True
def test_entry_import_document_command(monkeypatch, tmp_path):
called = {}
def import_document_file(path, **kwargs):
called["args"] = (path,)
called["kwargs"] = kwargs
return 123
def start_background_vault_sync():
called["sync"] = True
pm = SimpleNamespace(
entry_manager=SimpleNamespace(import_document_file=import_document_file),
parent_seed="seed",
select_fingerprint=lambda fp: None,
start_background_vault_sync=start_background_vault_sync,
)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
source = tmp_path / "doc.md"
source.write_text("hello", encoding="utf-8")
result = runner.invoke(
app,
[
"entry",
"import-document",
"--file",
str(source),
"--label",
"Doc",
"--notes",
"n",
],
)
assert result.exit_code == 0
assert "123" in result.stdout
assert called["args"] == (str(source),)
assert called["kwargs"] == {
"label": "Doc",
"notes": "n",
"tags": None,
"archived": False,
}
assert called.get("sync") is True
def test_entry_export_document_command(monkeypatch):
called = {}
def export_document_file(entry_id, output_path=None, **kwargs):
called["args"] = (entry_id, output_path)
called["kwargs"] = kwargs
return Path("/tmp/doc.txt")
pm = SimpleNamespace(
entry_manager=SimpleNamespace(export_document_file=export_document_file),
parent_seed="seed",
select_fingerprint=lambda fp: None,
start_background_vault_sync=lambda: None,
)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
result = runner.invoke(
app,
[
"entry",
"export-document",
"--entry-id",
"9",
"--out",
"/tmp/outdir",
"--overwrite",
],
)
assert result.exit_code == 0
assert "/tmp/doc.txt" in result.stdout
assert called["args"] == (9, "/tmp/outdir")
assert called["kwargs"] == {"overwrite": True}
def test_entry_link_commands(monkeypatch):
called = {}
def add_link(entry_id, target_id, **kwargs):
called["add"] = (entry_id, target_id, kwargs)
return [{"target_id": target_id, "relation": kwargs["relation"], "note": kwargs["note"]}]
def remove_link(entry_id, target_id, **kwargs):
called["remove"] = (entry_id, target_id, kwargs)
return []
def get_links(entry_id):
called["get"] = entry_id
return [{"target_id": 3, "relation": "references", "note": "", "target_label": "X", "target_kind": "document"}]
def start_background_vault_sync():
called["sync"] = called.get("sync", 0) + 1
pm = SimpleNamespace(
entry_manager=SimpleNamespace(
add_link=add_link,
remove_link=remove_link,
get_links=get_links,
),
parent_seed="seed",
select_fingerprint=lambda fp: None,
start_background_vault_sync=start_background_vault_sync,
)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
add_res = runner.invoke(
app,
[
"entry",
"link-add",
"--entry-id",
"1",
"--target-id",
"3",
"--relation",
"references",
"--note",
"note",
],
)
assert add_res.exit_code == 0
assert called["add"] == (1, 3, {"relation": "references", "note": "note"})
list_res = runner.invoke(app, ["entry", "links", "--entry-id", "1"])
assert list_res.exit_code == 0
assert '"target_id": 3' in list_res.stdout
assert called["get"] == 1
rm_res = runner.invoke(
app,
[
"entry",
"link-remove",
"--entry-id",
"1",
"--target-id",
"3",
"--relation",
"references",
],
)
assert rm_res.exit_code == 0
assert called["remove"] == (1, 3, {"relation": "references"})
+2 -1
View File
@@ -43,6 +43,7 @@ def _make_pm(called, locked=None):
handle_add_password=add,
handle_add_totp=lambda: None,
handle_add_managed_account=lambda: None,
handle_add_document=lambda: None,
handle_retrieve_entry=retrieve,
handle_modify_entry=modify,
update_activity=update,
@@ -81,7 +82,7 @@ def test_out_of_range_menu(monkeypatch, capsys):
def test_invalid_add_entry_submenu(monkeypatch, capsys):
called = {"add": False, "retrieve": False, "modify": False}
pm, _ = _make_pm(called)
inputs = iter(["1", "9", "", ""])
inputs = iter(["1", "10", "", ""])
monkeypatch.setattr(main, "timed_input", lambda *_: next(inputs))
monkeypatch.setattr("builtins.input", lambda *_: next(inputs))
with pytest.raises(SystemExit):
+6
View File
@@ -371,6 +371,12 @@ class TestEntryService:
digits=None,
key=None,
value=None,
content=None,
file_type=None,
custom_fields=None,
tags=None,
links=None,
archived=None,
)
mock_manager.start_background_vault_sync.assert_called_once()
+45
View File
@@ -0,0 +1,45 @@
from pathlib import Path
from tempfile import TemporaryDirectory
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
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 test_import_document_file_roundtrip():
with TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
entry_mgr = _make_entry_manager(base)
source = base / "notes.md"
source.write_text("# hello\nworld\n", encoding="utf-8")
idx = entry_mgr.import_document_file(source, notes="imported", tags=["docs"])
entry = entry_mgr.retrieve_entry(idx)
assert entry is not None
assert entry["kind"] == "document"
assert entry["label"] == "notes"
assert entry["file_type"] == "md"
assert entry["content"] == "# hello\nworld\n"
assert entry["notes"] == "imported"
assert entry["tags"] == ["docs"]
def test_export_document_file_writes_expected_content():
with TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
entry_mgr = _make_entry_manager(base)
idx = entry_mgr.add_document("Runbook", "line1\nline2", file_type="txt")
out_dir = base / "exports"
dest = entry_mgr.export_document_file(idx, output_path=out_dir)
assert dest.exists()
assert dest.suffix == ".txt"
assert dest.read_text(encoding="utf-8") == "line1\nline2"
+10 -1
View File
@@ -29,7 +29,7 @@ def test_add_and_retrieve_entry():
index = entry_mgr.add_entry("example.com", 12, "user", custom_fields=custom)
entry = entry_mgr.retrieve_entry(index)
assert entry == {
expected = {
"label": "example.com",
"length": 12,
"username": "user",
@@ -41,6 +41,9 @@ def test_add_and_retrieve_entry():
"custom_fields": custom,
"tags": [],
}
assert entry.items() >= expected.items()
assert "date_added" in entry
assert "date_modified" in entry
data = enc_mgr.load_json_data(entry_mgr.index_file)
assert str(index) in data.get("entries", {})
@@ -58,6 +61,7 @@ def test_add_and_retrieve_entry():
("add_seed", "seed"),
("add_key_value", "key_value"),
("add_managed_account", "managed_account"),
("add_document", "document"),
],
)
def test_round_trip_entry_types(method, expected_type):
@@ -74,6 +78,8 @@ def test_round_trip_entry_types(method, expected_type):
index = 0
elif method == "add_key_value":
index = entry_mgr.add_key_value("label", "k1", "val")
elif method == "add_document":
index = entry_mgr.add_document("doc", "hello", file_type="txt")
else:
if method == "add_ssh_key":
index = entry_mgr.add_ssh_key("ssh", TEST_SEED)
@@ -87,6 +93,8 @@ def test_round_trip_entry_types(method, expected_type):
entry = entry_mgr.retrieve_entry(index)
assert entry["type"] == expected_type
assert entry["kind"] == expected_type
assert "date_added" in entry
assert "date_modified" in entry
data = enc_mgr.load_json_data(entry_mgr.index_file)
assert data["entries"][str(index)]["type"] == expected_type
assert data["entries"][str(index)]["kind"] == expected_type
@@ -120,6 +128,7 @@ def test_legacy_entry_defaults_to_password():
("add_seed", ("seed", TEST_SEED)),
("add_key_value", ("label", "k1", "val")),
("add_managed_account", ("acct", TEST_SEED)),
("add_document", ("doc", "body")),
],
)
def test_add_default_archived_false(method, args):
+63
View File
@@ -0,0 +1,63 @@
import sys
from pathlib import Path
from tempfile import TemporaryDirectory
from helpers import create_vault, TEST_PASSWORD, TEST_SEED
sys.path.append(str(Path(__file__).resolve().parents[1]))
from seedpass.core.backup import BackupManager
from seedpass.core.config_manager import ConfigManager
from seedpass.core.entry_management import EntryManager
def _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 test_links_add_list_remove_and_search():
with TemporaryDirectory() as tmpdir:
em = _entry_manager(Path(tmpdir))
src = em.add_document("Project Plan", "v1", file_type="md")
dst = em.add_key_value("API Token", "token", "abc")
links = em.add_link(src, dst, relation="references", note="used by deploy")
assert links == [
{"target_id": dst, "relation": "references", "note": "used by deploy"}
]
resolved = em.get_links(src)
assert resolved[0]["target_id"] == dst
assert resolved[0]["relation"] == "references"
assert resolved[0]["target_label"] == "API Token"
assert resolved[0]["target_kind"] == "key_value"
# Query by relation, note, and target label should match source.
assert em.search_entries("references")[0][0] == src
assert em.search_entries("deploy")[0][0] == src
assert em.search_entries("api token")[0][0] == src
remaining = em.remove_link(src, dst, relation="references")
assert remaining == []
assert em.get_links(src) == []
def test_links_reject_invalid_targets_and_self_link():
with TemporaryDirectory() as tmpdir:
em = _entry_manager(Path(tmpdir))
idx = em.add_entry("example", 12)
try:
em.add_link(idx, idx)
assert False, "Expected self-link validation failure"
except ValueError as exc:
assert "Self-referential" in str(exc)
try:
em.add_link(idx, 9999)
assert False, "Expected missing target validation failure"
except ValueError as exc:
assert "Target entry not found" in str(exc)
+33
View File
@@ -68,6 +68,14 @@ def _add_entry_for_type(entry_mgr: EntryManager, entry_type: EntryType) -> int:
return entry_mgr.add_managed_account(
"Managed Entry", TEST_SEED, notes="managed-note", tags=["managed"]
)
if entry_type == EntryType.DOCUMENT:
return entry_mgr.add_document(
"Document Entry",
"alpha\nbeta\ngamma",
file_type="md",
notes="doc-note",
tags=["doc"],
)
raise AssertionError(f"Unhandled entry type in test: {entry_type}")
@@ -86,6 +94,8 @@ def test_each_entry_type_roundtrip_retrieve_and_filter(tmp_path, entry_type: Ent
assert entry["kind"] == entry_type.value
assert entry["label"]
assert "modified_ts" not in entry
assert "date_added" in entry
assert "date_modified" in entry
entries = entry_mgr.list_entries(
filter_kinds=[entry_type.value], include_archived=True, verbose=False
@@ -126,3 +136,26 @@ def test_each_entry_type_roundtrip_retrieve_and_filter(tmp_path, entry_type: Ent
if entry_type == EntryType.MANAGED_ACCOUNT:
assert entry["word_count"] == 12
assert entry["fingerprint"]
if entry_type == EntryType.DOCUMENT:
assert entry["file_type"] == "md"
assert "alpha" in entry["content"]
@pytest.mark.parametrize("entry_type", list(EntryType))
def test_each_entry_type_sets_and_updates_dates(
tmp_path, monkeypatch, entry_type: EntryType
):
entry_mgr = _make_entry_manager(tmp_path)
monkeypatch.setattr(entry_mgr, "_now_unix", lambda: 100)
index = _add_entry_for_type(entry_mgr, entry_type)
created = entry_mgr.retrieve_entry(index)
assert created is not None
assert created["date_added"] == entry_mgr._iso_from_unix(100)
assert created["date_modified"] == entry_mgr._iso_from_unix(100)
monkeypatch.setattr(entry_mgr, "_now_unix", lambda: 200)
entry_mgr.modify_entry(index, label=f"{created['label']} updated")
updated = entry_mgr.retrieve_entry(index)
assert updated is not None
assert updated["date_added"] == entry_mgr._iso_from_unix(100)
assert updated["date_modified"] == entry_mgr._iso_from_unix(200)
+32
View File
@@ -64,3 +64,35 @@ def test_full_sync_roundtrip(dummy_nostr_client):
pm_b.entry_manager.clear_cache()
labels = [e[1] for e in pm_b.entry_manager.list_entries()]
assert sorted(labels) == ["site1", "site2"]
def test_large_document_sync_roundtrip(dummy_nostr_client):
client, relay = dummy_nostr_client
with TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
dir_a = base / "A"
dir_b = base / "B"
dir_a.mkdir()
dir_b.mkdir()
pm_a = _init_pm(dir_a, client)
pm_b = _init_pm(dir_b, client)
large_content = "line\n" * 120_000
doc_id = pm_a.entry_manager.add_document(
"LargeDoc",
large_content,
file_type="txt",
notes="large-payload",
tags=["sync", "large"],
)
pm_a.sync_vault()
result = pm_b.attempt_initial_sync()
assert result is True
restored = pm_b.entry_manager.retrieve_entry(doc_id)
assert restored is not None
assert restored["kind"] == "document"
assert restored["file_type"] == "txt"
assert restored["content"] == large_content
+11 -11
View File
@@ -25,17 +25,17 @@ def test_add_and_modify_key_value():
idx = em.add_key_value("API entry", "api_key", "abc123", notes="token")
entry = em.retrieve_entry(idx)
assert entry == {
"type": "key_value",
"kind": "key_value",
"label": "API entry",
"key": "api_key",
"value": "abc123",
"notes": "token",
"archived": False,
"custom_fields": [],
"tags": [],
}
assert entry["type"] == "key_value"
assert entry["kind"] == "key_value"
assert entry["label"] == "API entry"
assert entry["key"] == "api_key"
assert entry["value"] == "abc123"
assert entry["notes"] == "token"
assert entry["archived"] is False
assert entry["custom_fields"] == []
assert entry["tags"] == []
assert "date_added" in entry
assert "date_modified" in entry
# Appears in listing
assert em.list_entries() == [(idx, "API entry", None, None, False)]
@@ -49,9 +49,11 @@ def test_migrate_iterations(tmp_path, monkeypatch, iterations):
calls: list[int] = []
orig_derive = enc_module._derive_legacy_key_from_password
def tracking_derive(password: str, iterations: int = 100_000) -> bytes:
def tracking_derive(
password: str, iterations: int = 100_000, salt: bytes = b""
) -> bytes:
calls.append(iterations)
return orig_derive(password, iterations=iterations)
return orig_derive(password, iterations=iterations, salt=salt)
monkeypatch.setattr(enc_module, "_derive_legacy_key_from_password", tracking_derive)
@@ -60,7 +62,11 @@ def test_migrate_iterations(tmp_path, monkeypatch, iterations):
vault.load_index()
assert prompts == [1]
expected = [50_000] if iterations == 50_000 else [50_000, 100_000]
expected = (
[50_000, 100_000, 50_000]
if iterations == 50_000
else [50_000, 100_000, 50_000, 100_000]
)
assert calls == expected
cfg = ConfigManager(vault, tmp_path)
+2 -2
View File
@@ -40,8 +40,8 @@ def test_sort_by_updated():
idx1 = em.add_entry("beta.com", 8, "u1")
data = em._load_index(force_reload=True)
data["entries"][str(idx0)]["updated"] = 1
data["entries"][str(idx1)]["updated"] = 2
data["entries"][str(idx0)]["modified_ts"] = 1
data["entries"][str(idx1)]["modified_ts"] = 2
em._save_index(data)
result = em.list_entries(sort_by="updated")
+24 -24
View File
@@ -74,18 +74,18 @@ def test_handle_add_password(monkeypatch, dummy_nostr_client, capsys):
assert entries == [(0, "Example", "", "", False)]
entry = entry_mgr.retrieve_entry(0)
assert entry == {
"label": "Example",
"length": DEFAULT_PASSWORD_LENGTH,
"username": "",
"url": "",
"archived": False,
"type": "password",
"kind": "password",
"notes": "",
"custom_fields": [],
"tags": [],
}
assert entry["label"] == "Example"
assert entry["length"] == DEFAULT_PASSWORD_LENGTH
assert entry["username"] == ""
assert entry["url"] == ""
assert entry["archived"] is False
assert entry["type"] == "password"
assert entry["kind"] == "password"
assert entry["notes"] == ""
assert entry["custom_fields"] == []
assert entry["tags"] == []
assert "date_added" in entry
assert "date_modified" in entry
assert f"pw-0-{DEFAULT_PASSWORD_LENGTH}" in out
@@ -194,18 +194,18 @@ def test_handle_add_password_quick_mode(monkeypatch, dummy_nostr_client, capsys)
assert entries == [(0, "Example", "", "", False)]
entry = entry_mgr.retrieve_entry(0)
assert entry == {
"label": "Example",
"length": DEFAULT_PASSWORD_LENGTH,
"username": "",
"url": "",
"archived": False,
"type": "password",
"kind": "password",
"notes": "",
"custom_fields": [],
"tags": [],
}
assert entry["label"] == "Example"
assert entry["length"] == DEFAULT_PASSWORD_LENGTH
assert entry["username"] == ""
assert entry["url"] == ""
assert entry["archived"] is False
assert entry["type"] == "password"
assert entry["kind"] == "password"
assert entry["notes"] == ""
assert entry["custom_fields"] == []
assert entry["tags"] == []
assert "date_added" in entry
assert "date_modified" in entry
assert f"pw-0-{DEFAULT_PASSWORD_LENGTH}" in out
@@ -74,6 +74,12 @@ def entry_manager():
{"url": "seed_url"},
"does not support fields: url",
),
(
EntryType.DOCUMENT,
lambda em: em.add_document("Doc", "body"),
{"username": "doc_user"},
"does not support fields: username",
),
],
)
def test_modify_entry_invalid_fields(
+9 -9
View File
@@ -24,15 +24,15 @@ def test_nostr_key_determinism():
idx = entry_mgr.add_nostr_key("main", TEST_SEED)
entry = entry_mgr.retrieve_entry(idx)
assert entry == {
"type": "nostr",
"kind": "nostr",
"index": idx,
"label": "main",
"notes": "",
"archived": False,
"tags": [],
}
assert entry["type"] == "nostr"
assert entry["kind"] == "nostr"
assert entry["index"] == idx
assert entry["label"] == "main"
assert entry["notes"] == ""
assert entry["archived"] is False
assert entry["tags"] == []
assert "date_added" in entry
assert "date_modified" in entry
npub1, nsec1 = entry_mgr.get_nostr_key_pair(idx, TEST_SEED)
npub2, nsec2 = entry_mgr.get_nostr_key_pair(idx, TEST_SEED)
@@ -12,10 +12,12 @@ import seedpass.core.encryption as enc_module
def test_legacy_password_only_fallback(monkeypatch, tmp_path, caplog):
calls: list[int] = []
def _fast_legacy_key(password: str, iterations: int = 100_000) -> bytes:
def _fast_legacy_key(
password: str, iterations: int = 100_000, salt: bytes = b""
) -> bytes:
calls.append(iterations)
normalized = unicodedata.normalize("NFKD", password).strip().encode("utf-8")
key = hashlib.pbkdf2_hmac("sha256", normalized, b"", 1, dklen=32)
key = hashlib.pbkdf2_hmac("sha256", normalized, salt, 1, dklen=32)
return base64.urlsafe_b64encode(key)
# Speed up legacy key derivation
@@ -35,4 +37,4 @@ def test_legacy_password_only_fallback(monkeypatch, tmp_path, caplog):
assert enc_mgr.decrypt_and_save_index_from_nostr(encrypted)
assert vault.load_index() == data
assert any("legacy password-only" in rec.message for rec in caplog.records)
assert calls == [50_000, 50_000]
assert calls == [50_000, 50_000, 100_000, 50_000]
+20 -20
View File
@@ -34,27 +34,27 @@ def test_seed_phrase_determinism():
entry12 = entry_mgr.retrieve_entry(idx_12)
entry24 = entry_mgr.retrieve_entry(idx_24)
assert entry12 == {
"type": "seed",
"kind": "seed",
"index": idx_12,
"label": "seed12",
"word_count": 12,
"notes": "",
"archived": False,
"tags": [],
}
assert entry12["type"] == "seed"
assert entry12["kind"] == "seed"
assert entry12["index"] == idx_12
assert entry12["label"] == "seed12"
assert entry12["word_count"] == 12
assert entry12["notes"] == ""
assert entry12["archived"] is False
assert entry12["tags"] == []
assert "date_added" in entry12
assert "date_modified" in entry12
assert entry24 == {
"type": "seed",
"kind": "seed",
"index": idx_24,
"label": "seed24",
"word_count": 24,
"notes": "",
"archived": False,
"tags": [],
}
assert entry24["type"] == "seed"
assert entry24["kind"] == "seed"
assert entry24["index"] == idx_24
assert entry24["label"] == "seed24"
assert entry24["word_count"] == 24
assert entry24["notes"] == ""
assert entry24["archived"] is False
assert entry24["tags"] == []
assert "date_added" in entry24
assert "date_modified" in entry24
assert phrase12_a not in entry12.values()
assert phrase24_a not in entry24.values()
+10 -9
View File
@@ -22,15 +22,16 @@ def test_add_and_retrieve_ssh_key_pair():
index = entry_mgr.add_ssh_key("ssh", TEST_SEED)
entry = entry_mgr.retrieve_entry(index)
assert entry == {
"type": "ssh",
"kind": "ssh",
"index": index,
"label": "ssh",
"notes": "",
"archived": False,
"tags": [],
}
assert entry["type"] == "ssh"
assert entry["kind"] == "ssh"
assert entry["index"] == index
assert entry["label"] == "ssh"
assert entry["notes"] == ""
assert entry["archived"] is False
assert entry["tags"] == []
assert entry["custom_fields"] == []
assert "date_added" in entry
assert "date_modified" in entry
priv1, pub1 = entry_mgr.get_ssh_key_pair(index, TEST_SEED)
priv2, pub2 = entry_mgr.get_ssh_key_pair(index, TEST_SEED)