Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions core/wren/src/wren/genbi/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,10 @@ def list_apps(path: ProjectPathOpt = None) -> None:
return

for name, entry in apps.items():
# Hand-edited non-dict entries: skip rather than AttributeError on .get
# (same lenient stance as get_app treating them as unregistered).
if not isinstance(entry, dict):
continue
deploy = entry.get("deploy") or {}
suffix = f" → {deploy['last_url']}" if deploy.get("last_url") else ""
typer.echo(
Expand Down
27 changes: 20 additions & 7 deletions core/wren/src/wren/genbi/index.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,11 +80,17 @@ def save_index(project_path: Path, index: dict) -> None:
def register_app(project_path: Path, name: str, *, data_mode: str) -> dict:
"""Create or update the entry for ``name``. Returns the entry."""
index = load_index(project_path)
entry = index["apps"].get(name) or {
"source": f"apps/{name}",
"status": "scaffolded",
"created_at": date.today().isoformat(),
}
existing = index["apps"].get(name)
# Hand-edited truthy non-dicts (e.g. a string) must not fall through the
# ``or`` default — assignment would raise TypeError. Treat like missing.
if not isinstance(existing, dict):
entry = {
"source": f"apps/{name}",
"status": "scaffolded",
"created_at": date.today().isoformat(),
}
else:
entry = existing
entry["data_mode"] = data_mode
index["apps"][name] = entry
save_index(project_path, index)
Expand All @@ -103,13 +109,20 @@ def remove_app(project_path: Path, name: str) -> bool:

def get_app(project_path: Path, name: str) -> dict | None:
"""Return the entry for ``name`` or None if not registered."""
return load_index(project_path)["apps"].get(name)
entry = load_index(project_path)["apps"].get(name)
# Hand-edited apps.yml may put a scalar/list under an app key; treat as
# unregistered rather than letting callers AttributeError on .get keys.
if entry is not None and not isinstance(entry, dict):
return None
return entry


def update_app(project_path: Path, name: str, **fields) -> dict:
"""Merge ``fields`` into the entry for ``name`` and persist."""
index = load_index(project_path)
entry = index["apps"][name]
entry = index["apps"].get(name)
if not isinstance(entry, dict):
raise KeyError(name)
entry.update(fields)
save_index(project_path, index)
return entry
82 changes: 82 additions & 0 deletions core/wren/tests/unit/test_genbi_index.py
Original file line number Diff line number Diff line change
Expand Up @@ -192,3 +192,85 @@ def test_load_index_preserves_falsy_schema_version(tmp_path: Path) -> None:

data = load_index(tmp_path)
assert data["schema_version"] == 0


def test_get_app_skips_non_dict_entry(tmp_path: Path) -> None:
from wren.genbi.index import get_app, save_index

idx = {
"schema_version": 1,
"apps": {"bad": "not-a-map", "good": {"source": "apps/good"}},
}
save_index(tmp_path, idx)
assert get_app(tmp_path, "bad") is None
assert get_app(tmp_path, "good") == {"source": "apps/good"}
assert get_app(tmp_path, "missing") is None


def test_register_app_replaces_non_dict_entry(tmp_path: Path) -> None:
from wren.genbi.index import load_index, register_app, save_index

project = _make_project(tmp_path, with_app="bad")
save_index(
project,
{"schema_version": 1, "apps": {"bad": "not-a-map"}},
)
entry = register_app(project, "bad", data_mode="snapshot")
assert isinstance(entry, dict)
assert entry["data_mode"] == "snapshot"
assert entry["source"] == "apps/bad"
assert entry["status"] == "scaffolded"
assert load_index(project)["apps"]["bad"] == entry

Comment thread
coderabbitai[bot] marked this conversation as resolved.

def test_update_app_rejects_missing_and_non_dict(tmp_path: Path) -> None:
from wren.genbi.index import save_index, update_app

save_index(
tmp_path,
{
"schema_version": 1,
"apps": {
"bad": "not-a-map",
"good": {"source": "apps/good", "status": "scaffolded"},
},
},
)
with pytest.raises(KeyError):
update_app(tmp_path, "missing", status="built")
with pytest.raises(KeyError):
update_app(tmp_path, "bad", status="built")
updated = update_app(tmp_path, "good", status="built")
assert updated["status"] == "built"
assert updated["source"] == "apps/good"
# Persistence: re-load and confirm
data = load_index(tmp_path)
assert data["apps"]["good"]["status"] == "built"
assert data["apps"]["bad"] == "not-a-map"


def test_list_skips_non_dict_entries(tmp_path: Path) -> None:
from wren.genbi.index import save_index

project = _make_project(tmp_path, with_app="good")
save_index(
project,
{
"schema_version": 1,
"apps": {
"bad": "not-a-map",
"good": {
"source": "apps/good",
"data_mode": "snapshot",
"status": "scaffolded",
},
},
},
)
result = runner.invoke(app, ["genbi", "list", "-p", str(project)])
assert result.exit_code == 0, result.output
assert "good" in result.output
assert "snapshot" in result.output
assert "bad" not in result.output
# Must not traceback; bad entry is skipped (lenient listing).
assert "Traceback" not in result.output
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Loading