mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-06-15 17:25:26 -04:00
test: move area_cli tests into cli directory (#3842)
* test: move area_cli tests into cli directory * test: include research CLI status in cli test move
This commit is contained in:
committed by
GitHub
parent
3e65326c3f
commit
a79c0bd369
@@ -0,0 +1,13 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_calendar_name_handles_missing_relation(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["CalendarCal", "CalendarEvent"])
|
||||
cli = load_script("odysseus-calendar")
|
||||
|
||||
assert cli._calendar_name(SimpleNamespace(calendar=None)) == ""
|
||||
assert cli._calendar_name(SimpleNamespace(calendar=SimpleNamespace(name=123))) == ""
|
||||
assert cli._calendar_name(SimpleNamespace(calendar=SimpleNamespace(name="Work"))) == "Work"
|
||||
@@ -0,0 +1,24 @@
|
||||
import sys
|
||||
import types
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_cli(monkeypatch):
|
||||
routes = types.ModuleType("routes.contacts_routes")
|
||||
routes._get_carddav_config = MagicMock()
|
||||
routes._fetch_contacts = MagicMock()
|
||||
routes._create_contact = MagicMock()
|
||||
monkeypatch.setitem(sys.modules, "routes.contacts_routes", routes)
|
||||
return load_script("odysseus-contacts")
|
||||
|
||||
|
||||
def test_contact_rows_skips_invalid_rows(monkeypatch):
|
||||
cli = _load_cli(monkeypatch)
|
||||
|
||||
assert cli._contact_rows([
|
||||
{"name": "Ada", "email": "ada@example.test"},
|
||||
"bad-row",
|
||||
None,
|
||||
]) == [{"name": "Ada", "email": "ada@example.test"}]
|
||||
@@ -0,0 +1,17 @@
|
||||
import io
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def test_state_set_rejects_non_object_json(tmp_path, monkeypatch, capsys):
|
||||
cli = load_script("odysseus-cookbook")
|
||||
cli._STATE_PATH = tmp_path / "cookbook_state.json"
|
||||
monkeypatch.setattr(cli.sys, "stdin", io.StringIO("[]"))
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli.cmd_state_set(type("Args", (), {})())
|
||||
|
||||
assert "expected a JSON object" in capsys.readouterr().err
|
||||
assert not cli._STATE_PATH.exists()
|
||||
@@ -0,0 +1,11 @@
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_text_len_ignores_non_string_values(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["Document", "DocumentVersion"])
|
||||
cli = load_script("odysseus-docs")
|
||||
|
||||
assert cli._text_len("hello") == 5
|
||||
assert cli._text_len(None) == 0
|
||||
assert cli._text_len({"bad": "row"}) == 0
|
||||
@@ -0,0 +1,13 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_album_image_count_handles_missing_relationship(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["GalleryImage", "GalleryAlbum"])
|
||||
cli = load_script("odysseus-gallery")
|
||||
|
||||
assert cli._album_image_count(SimpleNamespace(images=[1, 2])) == 2
|
||||
assert cli._album_image_count(SimpleNamespace(images=None)) == 0
|
||||
assert cli._album_image_count(SimpleNamespace(images=object())) == 0
|
||||
@@ -0,0 +1,30 @@
|
||||
"""Regression: gallery CLI image serialization must tolerate a non-string prompt.
|
||||
|
||||
`_serialize_image` did `(i.prompt or "")[:200]`. A non-string prompt is truthy,
|
||||
so `123[:200]` raised TypeError. `_preview_text` coerces non-strings to "".
|
||||
"""
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_preview_text_ignores_non_string(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["GalleryImage", "GalleryAlbum"])
|
||||
cli = load_script("odysseus-gallery")
|
||||
assert cli._preview_text(None) == ""
|
||||
assert cli._preview_text(123) == ""
|
||||
assert cli._preview_text("p" * 250) == "p" * 200
|
||||
|
||||
|
||||
def test_serialize_image_does_not_crash_on_non_string_prompt(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["GalleryImage", "GalleryAlbum"])
|
||||
cli = load_script("odysseus-gallery")
|
||||
img = SimpleNamespace(
|
||||
id="i1", filename="a.png", prompt=123, model=None, size=None, tags=None,
|
||||
favorite=0, album_id=None, session_id=None, width=1, height=1, file_size=1,
|
||||
taken_at=None, camera_make=None, camera_model=None, created_at=None,
|
||||
)
|
||||
out = cli._serialize_image(img)
|
||||
assert out["prompt"] == ""
|
||||
assert out["id"] == "i1"
|
||||
@@ -0,0 +1,13 @@
|
||||
"""Regression: logs CLI _resolve must tolerate a non-string name.
|
||||
|
||||
`_resolve` did `name in p.name` and `p.name == name`; a non-string `name`
|
||||
(e.g. None) raised TypeError once any *.log file existed. Non-strings now
|
||||
return None (no match).
|
||||
"""
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def test_non_string_name_returns_none():
|
||||
cli = load_script("odysseus-logs")
|
||||
assert cli._resolve(None) is None
|
||||
assert cli._resolve(123) is None
|
||||
@@ -0,0 +1,57 @@
|
||||
import sys
|
||||
from types import ModuleType, SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
class _Conn:
|
||||
def select(self, folder, readonly=True):
|
||||
return "OK", [b"1"]
|
||||
|
||||
def fetch(self, uid, spec):
|
||||
# IMAP can return OK with an empty payload (UID expunged mid-session).
|
||||
return "OK", []
|
||||
|
||||
|
||||
class _ImapCtx:
|
||||
def __init__(self, account):
|
||||
pass
|
||||
|
||||
def __enter__(self):
|
||||
return _Conn()
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
|
||||
def _load_mail_cli(monkeypatch):
|
||||
helpers = ModuleType("routes.email_helpers")
|
||||
helpers._imap = _ImapCtx
|
||||
helpers._get_email_config = lambda account=None: {}
|
||||
helpers._decode_header = lambda value: value
|
||||
helpers._extract_text = lambda msg: ""
|
||||
helpers._extract_html = lambda msg: ""
|
||||
helpers._list_attachments_from_msg = lambda msg: []
|
||||
pollers = ModuleType("routes.email_pollers")
|
||||
pollers._scheduled_poll_once = lambda: {}
|
||||
pollers._run_auto_summarize_once = lambda **kwargs: ""
|
||||
monkeypatch.setitem(sys.modules, "routes.email_helpers", helpers)
|
||||
monkeypatch.setitem(sys.modules, "routes.email_pollers", pollers)
|
||||
make_core_db_stub(
|
||||
monkeypatch,
|
||||
attributes={"SessionLocal": object, "EmailAccount": object},
|
||||
install_core_package=True,
|
||||
)
|
||||
return load_script("odysseus-mail")
|
||||
|
||||
|
||||
def test_cmd_read_handles_empty_fetch_payload(monkeypatch):
|
||||
cli = _load_mail_cli(monkeypatch)
|
||||
args = SimpleNamespace(account="acc", folder="INBOX", uid="5", html=False)
|
||||
# old code did raw = msg_data[0][1] on the empty list and raised IndexError;
|
||||
# the guard turns it into a clean fail() (SystemExit).
|
||||
with pytest.raises(SystemExit):
|
||||
cli.cmd_read(args)
|
||||
@@ -0,0 +1,50 @@
|
||||
import sys
|
||||
from types import ModuleType
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def _load_mail_cli(monkeypatch):
|
||||
helpers = ModuleType("routes.email_helpers")
|
||||
helpers._imap = object
|
||||
helpers._get_email_config = lambda account=None: {}
|
||||
helpers._decode_header = lambda value: value
|
||||
helpers._extract_text = lambda msg: ""
|
||||
helpers._extract_html = lambda msg: ""
|
||||
helpers._list_attachments_from_msg = lambda msg: []
|
||||
|
||||
pollers = ModuleType("routes.email_pollers")
|
||||
pollers._scheduled_poll_once = lambda: {}
|
||||
pollers._run_auto_summarize_once = lambda **kwargs: ""
|
||||
|
||||
monkeypatch.setitem(sys.modules, "routes.email_helpers", helpers)
|
||||
monkeypatch.setitem(sys.modules, "routes.email_pollers", pollers)
|
||||
make_core_db_stub(
|
||||
monkeypatch,
|
||||
attributes={"SessionLocal": object, "EmailAccount": object},
|
||||
install_core_package=True,
|
||||
)
|
||||
|
||||
return load_script("odysseus-mail")
|
||||
|
||||
|
||||
def test_recipient_list_trims_to_cc_and_bcc(monkeypatch):
|
||||
cli = _load_mail_cli(monkeypatch)
|
||||
|
||||
assert cli._recipient_list(" a@example.com, ", "b@example.com", " c@example.com ") == [
|
||||
"a@example.com",
|
||||
"b@example.com",
|
||||
"c@example.com",
|
||||
]
|
||||
|
||||
|
||||
def test_recipient_list_rejects_empty_envelope(monkeypatch):
|
||||
cli = _load_mail_cli(monkeypatch)
|
||||
|
||||
try:
|
||||
cli._recipient_list(" , ", "", "")
|
||||
except SystemExit as exc:
|
||||
assert exc.code == 1
|
||||
else:
|
||||
raise AssertionError("expected empty recipient list to exit")
|
||||
@@ -0,0 +1,29 @@
|
||||
"""Regression: mcp CLI _serialize must not crash when env JSON is not an object.
|
||||
|
||||
`env_obj = json.loads(s.env)` can yield a list (e.g. env stored as "[1,2]").
|
||||
`if redact_env and env_obj:` then called `env_obj.items()` -> AttributeError.
|
||||
Guard with isinstance(dict).
|
||||
"""
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def _srv(env):
|
||||
return SimpleNamespace(id="s1", name="n", transport="stdio", command="c", args="[]",
|
||||
env=env, url=None, is_enabled=1, oauth_config=None, created_at=None)
|
||||
|
||||
|
||||
def test_serialize_handles_list_env(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["McpServer"])
|
||||
cli = load_script("odysseus-mcp")
|
||||
out = cli._serialize(_srv("[1, 2]")) # JSON array, not object
|
||||
assert out["id"] == "s1"
|
||||
|
||||
|
||||
def test_serialize_redacts_dict_env(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["McpServer"])
|
||||
cli = load_script("odysseus-mcp")
|
||||
out = cli._serialize(_srv('{"API_KEY": "secret"}'))
|
||||
assert out["env"] == {"API_KEY": "***"}
|
||||
@@ -0,0 +1,14 @@
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_mcp_json_helpers_reject_wrong_shapes(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["McpServer"])
|
||||
cli = load_script("odysseus-mcp")
|
||||
|
||||
assert cli._json_list('["a"]') == ["a"]
|
||||
assert cli._json_list('{"not":"list"}') == []
|
||||
assert cli._json_list("{bad") == []
|
||||
assert cli._json_dict('{"A":"B"}') == {"A": "B"}
|
||||
assert cli._json_dict('["bad"]') == {}
|
||||
assert cli._json_dict("{bad") == {}
|
||||
@@ -0,0 +1,22 @@
|
||||
import sys
|
||||
import types
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_cli(monkeypatch):
|
||||
svc = types.ModuleType("services.memory.memory")
|
||||
svc.MemoryManager = MagicMock()
|
||||
monkeypatch.setitem(sys.modules, "services.memory.memory", svc)
|
||||
return load_script("odysseus-memory")
|
||||
|
||||
|
||||
def test_memory_entries_skips_invalid_rows(monkeypatch):
|
||||
cli = _load_cli(monkeypatch)
|
||||
|
||||
assert cli._memory_entries([
|
||||
{"id": "m1", "text": "ok"},
|
||||
"bad-row",
|
||||
None,
|
||||
]) == [{"id": "m1", "text": "ok"}]
|
||||
@@ -0,0 +1,48 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_serialize_ignores_invalid_note_items(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["Note"])
|
||||
cli = load_script("odysseus-notes")
|
||||
note = SimpleNamespace(
|
||||
id="n1",
|
||||
title="Checklist",
|
||||
content="",
|
||||
items="{bad json",
|
||||
note_type="checklist",
|
||||
color=None,
|
||||
label=None,
|
||||
pinned=False,
|
||||
archived=False,
|
||||
due_date=None,
|
||||
source=None,
|
||||
created_at=None,
|
||||
updated_at=None,
|
||||
)
|
||||
|
||||
assert cli._serialize(note)["items"] == []
|
||||
|
||||
|
||||
def test_serialize_keeps_list_note_items(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["Note"])
|
||||
cli = load_script("odysseus-notes")
|
||||
note = SimpleNamespace(
|
||||
id="n1",
|
||||
title="Checklist",
|
||||
content="",
|
||||
items='[{"text": "done"}]',
|
||||
note_type="checklist",
|
||||
color=None,
|
||||
label=None,
|
||||
pinned=False,
|
||||
archived=False,
|
||||
due_date=None,
|
||||
source=None,
|
||||
created_at=None,
|
||||
updated_at=None,
|
||||
)
|
||||
|
||||
assert cli._serialize(note)["items"] == [{"text": "done"}]
|
||||
@@ -0,0 +1,22 @@
|
||||
import sys
|
||||
import types
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_cli(monkeypatch):
|
||||
personal_docs = types.ModuleType("src.personal_docs")
|
||||
personal_docs.PersonalDocsManager = MagicMock()
|
||||
monkeypatch.setitem(sys.modules, "src.personal_docs", personal_docs)
|
||||
return load_script("odysseus-personal")
|
||||
|
||||
|
||||
def test_file_rows_skips_invalid_rows(monkeypatch):
|
||||
cli = _load_cli(monkeypatch)
|
||||
|
||||
assert cli._file_rows([
|
||||
{"name": "notes.txt", "path": "/tmp/notes.txt"},
|
||||
"bad-row",
|
||||
None,
|
||||
]) == [{"name": "notes.txt", "path": "/tmp/notes.txt"}]
|
||||
@@ -0,0 +1,18 @@
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def test_entry_or_fail_rejects_non_object_entries():
|
||||
cli = load_script("odysseus-preset")
|
||||
|
||||
try:
|
||||
cli._entry_or_fail({"broken": "raw prompt"}, "broken")
|
||||
except SystemExit as exc:
|
||||
assert exc.code == 1
|
||||
else:
|
||||
raise AssertionError("expected invalid preset entry to exit")
|
||||
|
||||
|
||||
def test_entry_or_fail_returns_valid_entry():
|
||||
cli = load_script("odysseus-preset")
|
||||
|
||||
assert cli._entry_or_fail({"ok": {"name": "ok"}}, "ok") == {"name": "ok"}
|
||||
@@ -0,0 +1,34 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_preset_cli():
|
||||
return load_script("odysseus-preset")
|
||||
|
||||
|
||||
def test_set_replaces_corrupt_existing_entry(monkeypatch):
|
||||
cli = _load_preset_cli()
|
||||
saved = {}
|
||||
emitted = {}
|
||||
|
||||
monkeypatch.setattr(cli, "_load", lambda: {"broken": "raw prompt"})
|
||||
monkeypatch.setattr(cli, "_save", lambda data: saved.update(data))
|
||||
monkeypatch.setattr(cli, "emit", lambda payload, _args: emitted.update(payload))
|
||||
|
||||
args = SimpleNamespace(
|
||||
name="broken",
|
||||
prompt="new prompt",
|
||||
prompt_file=None,
|
||||
temperature=0.7,
|
||||
display_name=None,
|
||||
)
|
||||
|
||||
cli.cmd_set(args)
|
||||
|
||||
assert saved["broken"] == {
|
||||
"name": "broken",
|
||||
"system_prompt": "new prompt",
|
||||
"temperature": 0.7,
|
||||
}
|
||||
assert emitted["ok"] is True
|
||||
@@ -0,0 +1,14 @@
|
||||
import pytest
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def test_load_rejects_non_object_preset_store(tmp_path, capsys):
|
||||
cli = load_script("odysseus-preset")
|
||||
cli._PATH = tmp_path / "presets.json"
|
||||
cli._PATH.write_text("[]")
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli._load()
|
||||
|
||||
assert "expected an object" in capsys.readouterr().err
|
||||
@@ -0,0 +1,25 @@
|
||||
"""Regression: research CLI summary must tolerate a non-string query.
|
||||
|
||||
`_summarize` did `(data.get("query") or "")[:200]`. A non-string query from a
|
||||
legacy/corrupt research JSON is truthy, so `123[:200]` raised TypeError.
|
||||
"""
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_cli():
|
||||
return load_script("odysseus-research")
|
||||
|
||||
|
||||
def test_preview_text_ignores_non_string():
|
||||
cli = _load_cli()
|
||||
assert cli._preview_text(None) == ""
|
||||
assert cli._preview_text(123) == ""
|
||||
assert cli._preview_text(["x"]) == ""
|
||||
assert cli._preview_text("q" * 250) == "q" * 200
|
||||
|
||||
|
||||
def test_summarize_does_not_crash_on_non_string_query():
|
||||
cli = _load_cli()
|
||||
out = cli._summarize("rp1", {"query": 123, "status": "done"})
|
||||
assert out["query"] == ""
|
||||
assert out["id"] == "rp1"
|
||||
@@ -0,0 +1,57 @@
|
||||
"""`odysseus-research list --status complete` must match completed runs.
|
||||
|
||||
Completed research runs are persisted with status "done" (research_handler),
|
||||
but the user-facing CLI value is the friendlier "complete". The CLI offered
|
||||
"complete" yet filtered `status != args.status`, so `--status complete` never
|
||||
matched any record. The fix keeps "complete" as the CLI value and maps it to
|
||||
the stored "done" at filter time, so the on-disk corpus stays the source of
|
||||
truth and the documented CLI surface keeps working.
|
||||
"""
|
||||
import importlib.machinery
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _load_cli():
|
||||
path = ROOT / "scripts" / "odysseus-research"
|
||||
loader = importlib.machinery.SourceFileLoader("odysseus_research_cli_status", str(path))
|
||||
spec = importlib.util.spec_from_loader(loader.name, loader)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def test_complete_is_a_valid_status_choice():
|
||||
cli = _load_cli()
|
||||
parser = cli._build_parser()
|
||||
ns = parser.parse_args(["list", "--status", "complete"])
|
||||
assert ns.status == "complete"
|
||||
|
||||
|
||||
def test_filter_returns_completed_runs(tmp_path, monkeypatch):
|
||||
cli = _load_cli(); cli._DATA_DIR = tmp_path
|
||||
(tmp_path / "r1.json").write_text(json.dumps({"query": "q1", "status": "done"}))
|
||||
(tmp_path / "r2.json").write_text(json.dumps({"query": "q2", "status": "running"}))
|
||||
emitted = []
|
||||
monkeypatch.setattr(cli, "emit", lambda value, args: emitted.append(value))
|
||||
# CLI "complete" must map to the stored "done" and match r1.
|
||||
cli.cmd_list(SimpleNamespace(status="complete", limit=50))
|
||||
ids = [r["id"] for r in emitted[0]]
|
||||
assert ids == ["r1"] # only the completed run
|
||||
|
||||
|
||||
def test_verbatim_status_still_filters(tmp_path, monkeypatch):
|
||||
cli = _load_cli(); cli._DATA_DIR = tmp_path
|
||||
(tmp_path / "r1.json").write_text(json.dumps({"query": "q1", "status": "done"}))
|
||||
(tmp_path / "r2.json").write_text(json.dumps({"query": "q2", "status": "running"}))
|
||||
emitted = []
|
||||
monkeypatch.setattr(cli, "emit", lambda value, args: emitted.append(value))
|
||||
cli.cmd_list(SimpleNamespace(status="running", limit=50))
|
||||
ids = [r["id"] for r in emitted[0]]
|
||||
assert ids == ["r2"] # verbatim choices pass through unchanged
|
||||
@@ -0,0 +1,106 @@
|
||||
"""`odysseus-research list --status complete` was returning nothing.
|
||||
|
||||
The CLI's `--status` argparse choice is "complete" — that is the user-facing
|
||||
label — but the writer in `services/research/research_handler.py` stores
|
||||
`status="done"` for a finished run (and the older `src/research_handler.py`
|
||||
copy does the same). The list filter was a literal string compare, so
|
||||
`--status complete` matched zero records on any real on-disk corpus.
|
||||
|
||||
These tests pin the alias so the friendlier CLI word keeps matching the
|
||||
stored value. The other choices (`running`, `cancelled`, `error`) are
|
||||
stored verbatim, so they must NOT be rewritten by the alias map.
|
||||
|
||||
Part of #2122 (odysseus-* CLI list/search bugs).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.machinery
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _load_cli():
|
||||
path = ROOT / "scripts" / "odysseus-research"
|
||||
loader = importlib.machinery.SourceFileLoader("odysseus_research_cli", str(path))
|
||||
spec = importlib.util.spec_from_loader(loader.name, loader)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _run_list(cli, tmp_path, monkeypatch, status, records):
|
||||
cli._DATA_DIR = tmp_path
|
||||
for name, blob in records.items():
|
||||
(tmp_path / f"{name}.json").write_text(json.dumps(blob))
|
||||
emitted = []
|
||||
monkeypatch.setattr(cli, "emit", lambda value, args: emitted.append(value))
|
||||
cli.cmd_list(SimpleNamespace(status=status, limit=50))
|
||||
assert emitted, "cmd_list emitted nothing"
|
||||
return [r["id"] for r in emitted[0]]
|
||||
|
||||
|
||||
def test_status_complete_matches_writer_done_records(tmp_path, monkeypatch):
|
||||
"""`--status complete` must return the records the writer marked `done`.
|
||||
Without the alias this filter is silently empty on any real corpus."""
|
||||
cli = _load_cli()
|
||||
ids = _run_list(cli, tmp_path, monkeypatch, status="complete", records={
|
||||
"rp-done": {"query": "finished one", "status": "done", "started_at": "2026-01-02"},
|
||||
"rp-running": {"query": "still running", "status": "running", "started_at": "2026-01-01"},
|
||||
"rp-cancelled": {"query": "user stopped", "status": "cancelled","started_at": "2025-12-31"},
|
||||
})
|
||||
assert ids == ["rp-done"], (
|
||||
"--status complete should alias to the writer's stored 'done' value; "
|
||||
f"got {ids}. The alias map in `_STATUS_CLI_TO_STORED` was bypassed."
|
||||
)
|
||||
|
||||
|
||||
def test_status_running_still_matches_verbatim(tmp_path, monkeypatch):
|
||||
"""`running` is stored verbatim, so the alias must NOT rewrite it.
|
||||
A blanket map that turned every CLI choice into a stored variant would
|
||||
re-introduce the empty-result bug on the running/cancelled/error paths."""
|
||||
cli = _load_cli()
|
||||
ids = _run_list(cli, tmp_path, monkeypatch, status="running", records={
|
||||
"rp-done": {"query": "finished", "status": "done"},
|
||||
"rp-running": {"query": "still running", "status": "running"},
|
||||
})
|
||||
assert ids == ["rp-running"], f"--status running must match verbatim; got {ids}"
|
||||
|
||||
|
||||
def test_status_cancelled_still_matches_verbatim(tmp_path, monkeypatch):
|
||||
cli = _load_cli()
|
||||
ids = _run_list(cli, tmp_path, monkeypatch, status="cancelled", records={
|
||||
"rp-done": {"query": "finished", "status": "done"},
|
||||
"rp-cancelled": {"query": "user stop", "status": "cancelled"},
|
||||
})
|
||||
assert ids == ["rp-cancelled"]
|
||||
|
||||
|
||||
def test_status_error_still_matches_verbatim(tmp_path, monkeypatch):
|
||||
cli = _load_cli()
|
||||
ids = _run_list(cli, tmp_path, monkeypatch, status="error", records={
|
||||
"rp-done": {"query": "finished", "status": "done"},
|
||||
"rp-error": {"query": "crashed", "status": "error"},
|
||||
})
|
||||
assert ids == ["rp-error"]
|
||||
|
||||
|
||||
def test_status_filter_tolerates_missing_or_non_string_status(tmp_path, monkeypatch):
|
||||
"""A corrupt record with no `status` (or a non-string status) must not
|
||||
crash the filter and must not falsely match `--status complete`. The
|
||||
existing `_load_path` already drops non-dict blobs; this guards the
|
||||
next layer."""
|
||||
cli = _load_cli()
|
||||
ids = _run_list(cli, tmp_path, monkeypatch, status="complete", records={
|
||||
"rp-good": {"query": "ok", "status": "done"},
|
||||
"rp-blank": {"query": "no status field"},
|
||||
"rp-typed": {"query": "non-string", "status": 42},
|
||||
})
|
||||
assert ids == ["rp-good"], (
|
||||
"--status complete should only match the writer's 'done' string; "
|
||||
f"got {ids}."
|
||||
)
|
||||
@@ -0,0 +1,32 @@
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_cli():
|
||||
return load_script("odysseus-research")
|
||||
|
||||
|
||||
def test_list_skips_non_object_research_records(tmp_path, monkeypatch):
|
||||
cli = _load_cli()
|
||||
cli._DATA_DIR = tmp_path
|
||||
(tmp_path / "good.json").write_text(json.dumps({"query": "hello", "status": "complete"}))
|
||||
(tmp_path / "list.json").write_text("[]")
|
||||
(tmp_path / "broken.json").write_text("{")
|
||||
|
||||
emitted = []
|
||||
monkeypatch.setattr(cli, "emit", lambda value, args: emitted.append(value))
|
||||
|
||||
cli.cmd_list(SimpleNamespace(status=None, limit=50))
|
||||
|
||||
assert emitted == [[{
|
||||
"id": "good",
|
||||
"query": "hello",
|
||||
"category": "",
|
||||
"status": "complete",
|
||||
"started_at": "",
|
||||
"completed_at": "",
|
||||
"sources": 0,
|
||||
"stats": {},
|
||||
}]]
|
||||
@@ -0,0 +1,39 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def _load_sessions_cli(monkeypatch):
|
||||
make_core_db_stub(
|
||||
monkeypatch,
|
||||
attributes={"SessionLocal": object, "Session": object},
|
||||
install_core_package=True,
|
||||
)
|
||||
return load_script("odysseus-sessions")
|
||||
|
||||
|
||||
def test_serialize_normalizes_numeric_counters(monkeypatch):
|
||||
cli = _load_sessions_cli(monkeypatch)
|
||||
session = SimpleNamespace(
|
||||
id="s1",
|
||||
name="chat",
|
||||
model="m",
|
||||
endpoint_url="",
|
||||
owner=None,
|
||||
folder=None,
|
||||
archived=False,
|
||||
rag=False,
|
||||
is_important=False,
|
||||
message_count="12",
|
||||
total_input_tokens="bad",
|
||||
total_output_tokens=None,
|
||||
last_accessed=None,
|
||||
created_at=None,
|
||||
)
|
||||
|
||||
out = cli._serialize(session)
|
||||
|
||||
assert out["message_count"] == 12
|
||||
assert out["total_input_tokens"] == 0
|
||||
assert out["total_output_tokens"] == 0
|
||||
@@ -0,0 +1,45 @@
|
||||
import sys
|
||||
from types import ModuleType
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_signature_cli(monkeypatch):
|
||||
sqlalchemy_mod = ModuleType("sqlalchemy")
|
||||
sqlalchemy_mod.text = lambda value: value
|
||||
core_mod = ModuleType("core")
|
||||
database_mod = ModuleType("core.database")
|
||||
database_mod.engine = object()
|
||||
monkeypatch.setitem(sys.modules, "sqlalchemy", sqlalchemy_mod)
|
||||
monkeypatch.setitem(sys.modules, "core", core_mod)
|
||||
monkeypatch.setitem(sys.modules, "core.database", database_mod)
|
||||
return load_script("odysseus-signature")
|
||||
|
||||
|
||||
def test_decode_png_data_accepts_data_url(monkeypatch):
|
||||
cli = _load_signature_cli(monkeypatch)
|
||||
|
||||
png = b"\x89PNG\r\n\x1a\nrest"
|
||||
assert cli._decode_png_data("data:image/png;base64,iVBORw0KGgpyZXN0") == png
|
||||
|
||||
|
||||
def test_decode_png_data_rejects_invalid_base64(monkeypatch):
|
||||
cli = _load_signature_cli(monkeypatch)
|
||||
|
||||
try:
|
||||
cli._decode_png_data("not valid!!!")
|
||||
except SystemExit as exc:
|
||||
assert exc.code == 1
|
||||
else:
|
||||
raise AssertionError("expected invalid base64 to exit")
|
||||
|
||||
|
||||
def test_decode_png_data_rejects_non_png_bytes(monkeypatch):
|
||||
cli = _load_signature_cli(monkeypatch)
|
||||
|
||||
try:
|
||||
cli._decode_png_data("aGVsbG8=")
|
||||
except SystemExit as exc:
|
||||
assert exc.code == 1
|
||||
else:
|
||||
raise AssertionError("expected non-PNG bytes to exit")
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Regression: the skills CLI summary must tolerate a non-string description.
|
||||
|
||||
`_summary` did `(skill.get("description") or "")[:200]`. A non-string
|
||||
description (e.g. a number from a hand-edited/legacy skill store) is truthy, so
|
||||
`123[:200]` raised TypeError. `_preview_text` coerces non-strings to "".
|
||||
"""
|
||||
import sys
|
||||
import types
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_cli(monkeypatch):
|
||||
mod = types.ModuleType("services.memory.skills")
|
||||
mod.SkillsManager = MagicMock()
|
||||
monkeypatch.setitem(sys.modules, "services.memory.skills", mod)
|
||||
return load_script("odysseus-skills")
|
||||
|
||||
|
||||
def test_preview_text_ignores_non_string(monkeypatch):
|
||||
cli = _load_cli(monkeypatch)
|
||||
assert cli._preview_text(None) == ""
|
||||
assert cli._preview_text(123) == ""
|
||||
assert cli._preview_text({"x": 1}) == ""
|
||||
assert cli._preview_text("y" * 250) == "y" * 200
|
||||
|
||||
|
||||
def test_summary_does_not_crash_on_non_string_description(monkeypatch):
|
||||
cli = _load_cli(monkeypatch)
|
||||
out = cli._summary({"name": "n", "description": 123})
|
||||
assert out["description"] == ""
|
||||
@@ -0,0 +1,22 @@
|
||||
import sys
|
||||
import types
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
def _load_cli(monkeypatch):
|
||||
svc = types.ModuleType("services.memory.skills")
|
||||
svc.SkillsManager = MagicMock()
|
||||
monkeypatch.setitem(sys.modules, "services.memory.skills", svc)
|
||||
return load_script("odysseus-skills")
|
||||
|
||||
|
||||
def test_skill_entries_skips_invalid_rows(monkeypatch):
|
||||
cli = _load_cli(monkeypatch)
|
||||
|
||||
assert cli._skill_entries([
|
||||
{"name": "deploy", "category": "ops"},
|
||||
"bad-row",
|
||||
None,
|
||||
]) == [{"name": "deploy", "category": "ops"}]
|
||||
@@ -0,0 +1,11 @@
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_preview_text_ignores_non_string_values(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["ScheduledTask", "TaskRun"])
|
||||
cli = load_script("odysseus-tasks")
|
||||
|
||||
assert cli._preview_text(None) == ""
|
||||
assert cli._preview_text({"bad": "row"}) == ""
|
||||
assert cli._preview_text("x" * 201) == ("x" * 200) + "…"
|
||||
@@ -0,0 +1,15 @@
|
||||
import pytest
|
||||
|
||||
from tests.helpers.cli_loader import load_script
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", ["[]", '{"_users": []}'])
|
||||
def test_load_prefs_rejects_non_object_user_store(tmp_path, capsys, payload):
|
||||
cli = load_script("odysseus-theme")
|
||||
cli._USER_PREFS_PATH = tmp_path / "user_prefs.json"
|
||||
cli._USER_PREFS_PATH.write_text(payload)
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli._load_prefs()
|
||||
|
||||
assert "is corrupt" in capsys.readouterr().err
|
||||
@@ -0,0 +1,12 @@
|
||||
from tests.helpers.cli_loader import load_script
|
||||
from tests.helpers.db_stubs import make_core_db_stub
|
||||
|
||||
|
||||
def test_mask_token_handles_short_values(monkeypatch):
|
||||
make_core_db_stub(monkeypatch, models=["ScheduledTask"])
|
||||
cli = load_script("odysseus-webhook")
|
||||
|
||||
assert cli._mask_token("") == ""
|
||||
assert cli._mask_token("short") == "***"
|
||||
assert cli._mask_token("abcdef1234567890") == "abcdef…7890"
|
||||
assert cli._mask_token("short", reveal=True) == "short"
|
||||
Reference in New Issue
Block a user