fix(prax): slim dashboard to lean glance — drop session/todo/ai_mail sections, quick_status is the count home
This commit is contained in:
@@ -30,6 +30,18 @@ PyPI version — not the changelog header.
|
||||
constant), so a refresh strips them. Verified: `bulletin_board` removed from
|
||||
the devpulse dashboard; 116/116 prax tests, seedgo 100%. (Follow-up: `@trigger`
|
||||
still has a `bulletin_created` writer to retire separately.)
|
||||
- **Dashboard slimmed to a lean glance — removed duplicated/dead sections.**
|
||||
Dropped three sections from the devpulse dashboard: `session` (broken since
|
||||
May — read keys `id`/`d`/`sum` vs the actual `session`/`date`/`summary`, so it
|
||||
always wrote empty strings — and it duplicated `local.json`, which loads at
|
||||
startup), `todo` (carried only `todo_count`, already in `quick_status`; now
|
||||
sourced directly from `local.json`), and `ai_mail` (its counts live in
|
||||
`quick_status`; the section is removed from output *after* quick_status is
|
||||
computed from it). End state: 4 sections (`flow`, `memory`, `git`, `dispatch`)
|
||||
+ the `quick_status` glance. `session_section.py`/`todo_section.py` archived
|
||||
(not deleted). `DASHBOARD.local.json` overall 6.8 KB → 2.4 KB. Verified: seedgo
|
||||
100%, 108 prax tests. (Follow-up: `@ai_mail`'s `dashboard_sync.py` section
|
||||
writer to retire separately.)
|
||||
- **`.backupignore` is now a true `.gitignore` for the backup system — a single
|
||||
source of truth (FPLAN-0269).** Replaced the hand-rolled `fnmatch`+part-loop
|
||||
matcher (which broke leading-slash anchoring, `*`-crossing-`/`, dir-only `foo/`,
|
||||
|
||||
@@ -22,7 +22,7 @@ from aipass.prax.apps.modules.logger import get_direct_logger
|
||||
|
||||
logger = get_direct_logger()
|
||||
|
||||
from aipass.prax.apps.handlers.json import json_handler
|
||||
from aipass.prax.apps.handlers.json import json_handler # noqa: E402
|
||||
|
||||
# Resolve prax root from this file's location
|
||||
_PRAX_ROOT = Path(__file__).resolve().parents[3] # .../prax/
|
||||
@@ -139,7 +139,11 @@ def create_fresh_dashboard(branch_path: Path) -> Dict:
|
||||
# Fallback: hardcoded (backward compat)
|
||||
now = datetime.now().isoformat()
|
||||
return {
|
||||
"_warning": "AUTO-GENERATED FILE - DO NOT MANUALLY EDIT. This file is 100% automated and will be overwritten. Services update their own sections.",
|
||||
"_warning": (
|
||||
"AUTO-GENERATED FILE - DO NOT MANUALLY EDIT."
|
||||
" This file is 100% automated and will be overwritten."
|
||||
" Services update their own sections."
|
||||
),
|
||||
"branch": branch_path.name.upper(),
|
||||
"last_updated": now,
|
||||
"quick_status": {"action_required": False},
|
||||
@@ -183,12 +187,25 @@ def update_section(
|
||||
dashboard["sections"][section_name] = section_data
|
||||
|
||||
# Recalculate quick status
|
||||
dashboard["quick_status"] = calculate_status_func(dashboard["sections"])
|
||||
dashboard["quick_status"] = calculate_status_func(dashboard["sections"], branch_path)
|
||||
|
||||
return save_dashboard(branch_path, dashboard)
|
||||
|
||||
|
||||
def _calculate_quick_status_standalone(sections: Dict) -> Dict:
|
||||
def _read_todo_count(branch_path: Path) -> int:
|
||||
"""Read todos[] length from .trinity/local.json."""
|
||||
local_path = branch_path / ".trinity" / "local.json"
|
||||
if not local_path.exists():
|
||||
return 0
|
||||
try:
|
||||
data = json.loads(local_path.read_text())
|
||||
return len(data.get("todos", []))
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to read todos from %s: %s", local_path, exc)
|
||||
return 0
|
||||
|
||||
|
||||
def _calculate_quick_status_standalone(sections: Dict, branch_path: Path) -> Dict:
|
||||
"""
|
||||
Calculate quick_status from live section data.
|
||||
|
||||
@@ -197,18 +214,18 @@ def _calculate_quick_status_standalone(sections: Dict) -> Dict:
|
||||
|
||||
Args:
|
||||
sections: All dashboard sections dict
|
||||
branch_path: Path to branch root (for sourcing todo_count from local.json)
|
||||
|
||||
Returns:
|
||||
Quick status dict with summary, action flags, and counts
|
||||
"""
|
||||
ai_mail = sections.get("ai_mail", {})
|
||||
flow = sections.get("flow", {})
|
||||
todo = sections.get("todo", {})
|
||||
|
||||
new_mail_raw = ai_mail.get("new", ai_mail.get("unread", 0))
|
||||
opened_raw = ai_mail.get("opened", 0)
|
||||
active_plans_raw = flow.get("active_plans", 0)
|
||||
todo_count = int(todo.get("todo_count", 0) or 0)
|
||||
todo_count = _read_todo_count(branch_path)
|
||||
|
||||
new_mail = len(new_mail_raw) if isinstance(new_mail_raw, list) else int(new_mail_raw or 0)
|
||||
opened_mail = len(opened_raw) if isinstance(opened_raw, list) else int(opened_raw or 0)
|
||||
@@ -296,7 +313,7 @@ def write_section(branch_path: Path, section_name: str, section_data: Dict) -> b
|
||||
dashboard["sections"][section_name] = section_data
|
||||
|
||||
# Recalculate quick_status from live data
|
||||
dashboard["quick_status"] = _calculate_quick_status_standalone(dashboard["sections"])
|
||||
dashboard["quick_status"] = _calculate_quick_status_standalone(dashboard["sections"], branch_path)
|
||||
|
||||
# Save
|
||||
saved = save_dashboard(branch_path, dashboard)
|
||||
|
||||
@@ -153,12 +153,26 @@ def _extract_memory_section(centrals: Dict, branch_path: Path) -> Dict:
|
||||
return {"managed_by": "memory", "vectors_stored": local_vectors, "notes": {}, "last_updated": mb_last_updated}
|
||||
|
||||
|
||||
def _calculate_quick_status(sections: Dict) -> Dict:
|
||||
def _read_todo_count(branch_path: Path) -> int:
|
||||
"""Read todos[] length from .trinity/local.json."""
|
||||
local_path = branch_path / ".trinity" / "local.json"
|
||||
if not local_path.exists():
|
||||
return 0
|
||||
try:
|
||||
data = json.loads(local_path.read_text())
|
||||
return len(data.get("todos", []))
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to read todos from %s: %s", local_path, exc)
|
||||
return 0
|
||||
|
||||
|
||||
def _calculate_quick_status(sections: Dict, branch_path: Path) -> Dict:
|
||||
"""
|
||||
Calculate quick_status from live section data (v3 schema).
|
||||
|
||||
Args:
|
||||
sections: All dashboard sections dict
|
||||
branch_path: Path to branch root (for sourcing todo_count from local.json)
|
||||
|
||||
Returns:
|
||||
Quick status dict with counts, action flag, and summary
|
||||
@@ -169,6 +183,7 @@ def _calculate_quick_status(sections: Dict) -> Dict:
|
||||
new_mail = ai_mail.get("new", ai_mail.get("unread", 0))
|
||||
opened_mail = ai_mail.get("opened", 0)
|
||||
active_plans = flow.get("active_plans", 0)
|
||||
todo_count = _read_todo_count(branch_path)
|
||||
|
||||
action_required = new_mail > 0 or active_plans > 0
|
||||
|
||||
@@ -179,11 +194,14 @@ def _calculate_quick_status(sections: Dict) -> Dict:
|
||||
parts.append(f"{opened_mail} opened")
|
||||
if active_plans > 0:
|
||||
parts.append(f"{active_plans} active plans")
|
||||
if todo_count > 0:
|
||||
parts.append(f"{todo_count} todos")
|
||||
|
||||
return {
|
||||
"new_mail": new_mail,
|
||||
"opened_mail": opened_mail,
|
||||
"active_plans": active_plans,
|
||||
"todo_count": todo_count,
|
||||
"action_required": action_required,
|
||||
"summary": ", ".join(parts) if parts else "All clear",
|
||||
}
|
||||
@@ -260,8 +278,9 @@ def refresh_all_dashboards() -> Dict:
|
||||
_preserve_write_through_sections(dashboard, branch_path, branch_name)
|
||||
_prune_deprecated_sections(dashboard)
|
||||
|
||||
# Calculate quick status
|
||||
dashboard["quick_status"] = _calculate_quick_status(dashboard["sections"])
|
||||
# Calculate quick status (ai_mail section still present for counts)
|
||||
dashboard["quick_status"] = _calculate_quick_status(dashboard["sections"], branch_path)
|
||||
dashboard["sections"].pop("ai_mail", None)
|
||||
|
||||
# Save
|
||||
save_dashboard(branch_path, dashboard)
|
||||
@@ -325,7 +344,8 @@ def refresh_single_dashboard(branch_path: Path) -> Dict:
|
||||
_preserve_write_through_sections(dashboard, branch_path, branch_name)
|
||||
_prune_deprecated_sections(dashboard)
|
||||
|
||||
dashboard["quick_status"] = _calculate_quick_status(dashboard["sections"])
|
||||
dashboard["quick_status"] = _calculate_quick_status(dashboard["sections"], branch_path)
|
||||
dashboard["sections"].pop("ai_mail", None)
|
||||
|
||||
save_dashboard(branch_path, dashboard)
|
||||
|
||||
|
||||
@@ -17,8 +17,11 @@ import json
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
|
||||
from aipass.prax.apps.modules.logger import get_direct_logger
|
||||
from aipass.prax.apps.handlers.json import json_handler
|
||||
|
||||
logger = get_direct_logger()
|
||||
|
||||
|
||||
def _find_repo_root() -> Path:
|
||||
"""Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json)."""
|
||||
@@ -32,7 +35,20 @@ def _find_repo_root() -> Path:
|
||||
AIPASS_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json"
|
||||
|
||||
|
||||
def calculate_quick_status(sections: Dict) -> Dict:
|
||||
def _read_todo_count(branch_path: Path) -> int:
|
||||
"""Read todos[] length from .trinity/local.json."""
|
||||
local_path = branch_path / ".trinity" / "local.json"
|
||||
if not local_path.exists():
|
||||
return 0
|
||||
try:
|
||||
data = json.loads(local_path.read_text())
|
||||
return len(data.get("todos", []))
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to read todos from %s: %s", local_path, exc)
|
||||
return 0
|
||||
|
||||
|
||||
def calculate_quick_status(sections: Dict, branch_path: "Path | None" = None) -> Dict:
|
||||
"""
|
||||
Calculate quick status from live section data.
|
||||
|
||||
@@ -40,18 +56,18 @@ def calculate_quick_status(sections: Dict) -> Dict:
|
||||
|
||||
Args:
|
||||
sections: All dashboard sections
|
||||
branch_path: Optional path to branch root (for sourcing todo_count)
|
||||
|
||||
Returns:
|
||||
Quick status dict with summary data
|
||||
"""
|
||||
ai_mail = sections.get("ai_mail", {})
|
||||
flow = sections.get("flow", {})
|
||||
todo = sections.get("todo", {})
|
||||
|
||||
new_mail = ai_mail.get("new", ai_mail.get("unread", 0))
|
||||
opened_mail = ai_mail.get("opened", 0)
|
||||
active_plans = flow.get("active_plans", 0)
|
||||
todo_count = todo.get("todo_count", 0)
|
||||
todo_count = _read_todo_count(branch_path) if branch_path else 0
|
||||
|
||||
action_required = new_mail > 0 or active_plans > 0
|
||||
|
||||
|
||||
@@ -57,7 +57,15 @@ TEMPLATE_FILE = TEMPLATE_DIR / "DASHBOARD.template.json"
|
||||
AIPASS_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json"
|
||||
|
||||
# Deprecated sections that should be flagged for removal
|
||||
DEPRECATED_SECTIONS = ["bulletin_board", "devpulse", "commons_activity", "agent_status", "memory_bank"]
|
||||
DEPRECATED_SECTIONS = [
|
||||
"bulletin_board",
|
||||
"devpulse",
|
||||
"commons_activity",
|
||||
"agent_status",
|
||||
"memory_bank",
|
||||
"session",
|
||||
"todo",
|
||||
]
|
||||
|
||||
# Deprecated quick_status keys that should be flagged
|
||||
DEPRECATED_QUICK_STATUS_KEYS = ["pending_bulletins", "commons_mentions"]
|
||||
|
||||
@@ -33,7 +33,7 @@ from aipass.prax.apps.modules.logger import get_direct_logger
|
||||
|
||||
logger = get_direct_logger()
|
||||
|
||||
from aipass.prax.apps.handlers.json import json_handler
|
||||
from aipass.prax.apps.handlers.json import json_handler # noqa: E402
|
||||
|
||||
# =============================================================================
|
||||
# PATH RESOLUTION
|
||||
@@ -61,7 +61,15 @@ VERSION_FILE = TEMPLATE_DIR / ".dashboard_version.json"
|
||||
AIPASS_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json"
|
||||
|
||||
# Deprecated sections to REMOVE during push
|
||||
DEPRECATED_SECTIONS = ["bulletin_board", "devpulse", "commons_activity", "agent_status", "memory_bank"]
|
||||
DEPRECATED_SECTIONS = [
|
||||
"bulletin_board",
|
||||
"devpulse",
|
||||
"commons_activity",
|
||||
"agent_status",
|
||||
"memory_bank",
|
||||
"session",
|
||||
"todo",
|
||||
]
|
||||
|
||||
# Deprecated quick_status keys to REMOVE during push
|
||||
DEPRECATED_QUICK_STATUS_KEYS = ["pending_bulletins", "commons_mentions"]
|
||||
|
||||
@@ -39,9 +39,7 @@ def refresh(branch_path: Optional[Path] = None) -> Dict:
|
||||
|
||||
builders: List[tuple] = [
|
||||
("git", _refresh_git),
|
||||
("session", _refresh_session),
|
||||
("dispatch", _refresh_dispatch),
|
||||
("todo", _refresh_todo),
|
||||
]
|
||||
|
||||
for name, builder_fn in builders:
|
||||
@@ -61,19 +59,7 @@ def _refresh_git(branch_path: Path) -> None:
|
||||
build_git_section(branch_path)
|
||||
|
||||
|
||||
def _refresh_session(branch_path: Path) -> None:
|
||||
from .session_section import build_session_section
|
||||
|
||||
build_session_section(branch_path)
|
||||
|
||||
|
||||
def _refresh_dispatch(branch_path: Path) -> None:
|
||||
from .dispatch_section import build_dispatch_section
|
||||
|
||||
build_dispatch_section(branch_path)
|
||||
|
||||
|
||||
def _refresh_todo(branch_path: Path) -> None:
|
||||
from .todo_section import build_todo_section
|
||||
|
||||
build_todo_section(branch_path)
|
||||
|
||||
@@ -1,86 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: session_section.py
|
||||
# Description: Session info section builder for devpulse dashboard
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-05-16
|
||||
# Modified: 2026-05-16
|
||||
# =============================================
|
||||
|
||||
"""Session section builder for devpulse dashboard plugin.
|
||||
|
||||
Reads .trinity/local.json to extract current session ID, date,
|
||||
active tasks, and last session summary. Writes to dashboard
|
||||
via write_section().
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.prax.apps.modules.dashboard import write_section
|
||||
from aipass.prax.apps.modules.logger import system_logger as logger
|
||||
|
||||
|
||||
def build_session_section(branch_path: Path) -> bool:
|
||||
"""Build session section data and write to dashboard.
|
||||
|
||||
Args:
|
||||
branch_path: Path to devpulse branch root.
|
||||
|
||||
Returns:
|
||||
True if write_section succeeded, False otherwise.
|
||||
"""
|
||||
local_json_path = branch_path / ".trinity" / "local.json"
|
||||
|
||||
if not local_json_path.exists():
|
||||
section_data: Dict = {
|
||||
"managed_by": "devpulse",
|
||||
"current_session": "unknown",
|
||||
"session_date": "",
|
||||
"today_focus": "",
|
||||
"active_tasks": [],
|
||||
"last_session_summary": "",
|
||||
}
|
||||
return write_section(branch_path, "session", section_data)
|
||||
|
||||
try:
|
||||
data = json.loads(local_json_path.read_text())
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to read local.json at %s: %s", local_json_path, exc)
|
||||
section_data = {
|
||||
"managed_by": "devpulse",
|
||||
"current_session": "error",
|
||||
"session_date": "",
|
||||
"today_focus": "",
|
||||
"active_tasks": [],
|
||||
"last_session_summary": "Failed to read local.json",
|
||||
}
|
||||
return write_section(branch_path, "session", section_data)
|
||||
|
||||
# Extract latest session (first in list = newest)
|
||||
sessions = data.get("sessions", [])
|
||||
current_session = ""
|
||||
session_date = ""
|
||||
last_session_summary = ""
|
||||
|
||||
if sessions:
|
||||
latest = sessions[0]
|
||||
current_session = latest.get("id", "")
|
||||
session_date = latest.get("d", "")
|
||||
last_session_summary = latest.get("sum", "")
|
||||
|
||||
# Extract active tasks
|
||||
active_tasks_data = data.get("active_tasks", {})
|
||||
today_focus = active_tasks_data.get("today_focus", "")
|
||||
pending = active_tasks_data.get("pending", [])
|
||||
|
||||
section_data = {
|
||||
"managed_by": "devpulse",
|
||||
"current_session": current_session,
|
||||
"session_date": session_date,
|
||||
"today_focus": today_focus,
|
||||
"active_tasks": pending,
|
||||
"last_session_summary": last_session_summary,
|
||||
}
|
||||
|
||||
return write_section(branch_path, "session", section_data)
|
||||
@@ -1,59 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: todo_section.py
|
||||
# Description: Todo section builder for devpulse dashboard
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-06-07
|
||||
# Modified: 2026-06-07
|
||||
# =============================================
|
||||
|
||||
"""Todo section builder for devpulse dashboard plugin.
|
||||
|
||||
Reads todos[] from .trinity/local.json and writes a 'todo' section
|
||||
(managed_by: devpulse, todo_count, todos) via write_section().
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.prax.apps.modules.dashboard import write_section
|
||||
from aipass.prax.apps.modules.logger import system_logger as logger
|
||||
|
||||
|
||||
def build_todo_section(branch_path: Path) -> bool:
|
||||
"""Build todo section data and write to dashboard.
|
||||
|
||||
Args:
|
||||
branch_path: Path to devpulse branch root.
|
||||
|
||||
Returns:
|
||||
True if write_section succeeded, False otherwise.
|
||||
"""
|
||||
local_json_path = branch_path / ".trinity" / "local.json"
|
||||
|
||||
if not local_json_path.exists():
|
||||
section_data: Dict = {
|
||||
"managed_by": "devpulse",
|
||||
"todo_count": 0,
|
||||
}
|
||||
return write_section(branch_path, "todo", section_data)
|
||||
|
||||
try:
|
||||
data = json.loads(local_json_path.read_text())
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to read local.json at %s: %s", local_json_path, exc)
|
||||
section_data = {
|
||||
"managed_by": "devpulse",
|
||||
"todo_count": 0,
|
||||
}
|
||||
return write_section(branch_path, "todo", section_data)
|
||||
|
||||
todos = data.get("todos", [])
|
||||
|
||||
section_data = {
|
||||
"managed_by": "devpulse",
|
||||
"todo_count": len(todos),
|
||||
# "todos": todos, # bodies live in .trinity/local.json — count-only here
|
||||
}
|
||||
|
||||
return write_section(branch_path, "todo", section_data)
|
||||
@@ -114,52 +114,6 @@ class TestGitSection:
|
||||
build_git_section(branch_path)
|
||||
|
||||
|
||||
class TestSessionSection:
|
||||
"""Tests for session_section.py."""
|
||||
|
||||
def test_build_session_section_success(self, branch_with_trinity):
|
||||
"""Test session section reads local.json correctly."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.session_section import build_session_section
|
||||
|
||||
result = build_session_section(branch_with_trinity)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_with_trinity / "DASHBOARD.local.json").read_text())
|
||||
session = dash["sections"]["session"]
|
||||
assert session["managed_by"] == "devpulse"
|
||||
assert session["current_session"] == "S162"
|
||||
assert session["session_date"] == "2026-05-16"
|
||||
assert session["today_focus"] == "Testing the plugin"
|
||||
assert session["active_tasks"] == ["Task 1", "Task 2"]
|
||||
assert session["last_session_summary"] == "Test session summary"
|
||||
|
||||
def test_build_session_section_no_local_json(self, branch_path):
|
||||
"""Test session section when local.json doesn't exist."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.session_section import build_session_section
|
||||
|
||||
result = build_session_section(branch_path)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
|
||||
session = dash["sections"]["session"]
|
||||
assert session["current_session"] == "unknown"
|
||||
|
||||
def test_build_session_section_corrupt_json(self, branch_path):
|
||||
"""Test session section handles corrupt local.json."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.session_section import build_session_section
|
||||
|
||||
trinity = branch_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
(trinity / "local.json").write_text("not valid json{{{")
|
||||
|
||||
result = build_session_section(branch_path)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
|
||||
session = dash["sections"]["session"]
|
||||
assert session["current_session"] == "error"
|
||||
|
||||
|
||||
class TestDispatchSection:
|
||||
"""Tests for dispatch_section.py."""
|
||||
|
||||
@@ -219,118 +173,24 @@ class TestDispatchSection:
|
||||
assert dispatch["details"]["seedgo"]["subject"] == "unknown"
|
||||
|
||||
|
||||
class TestTodoSection:
|
||||
"""Tests for todo_section.py."""
|
||||
|
||||
def test_build_todo_section_with_todos(self, branch_path):
|
||||
"""Test todo section reads todos[] from local.json correctly."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.todo_section import build_todo_section
|
||||
|
||||
trinity = branch_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
local_data = {
|
||||
"todos": [
|
||||
{"id": "t1", "text": "Fix the bug", "created": "2026-06-07"},
|
||||
{"id": "t2", "text": "Write tests", "created": "2026-06-07", "priority": "high"},
|
||||
],
|
||||
}
|
||||
(trinity / "local.json").write_text(json.dumps(local_data))
|
||||
|
||||
result = build_todo_section(branch_path)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
|
||||
todo = dash["sections"]["todo"]
|
||||
assert todo["managed_by"] == "devpulse"
|
||||
assert todo["todo_count"] == 2
|
||||
assert "todos" not in todo
|
||||
|
||||
def test_build_todo_section_empty_todos(self, branch_path):
|
||||
"""Test todo section with empty todos list."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.todo_section import build_todo_section
|
||||
|
||||
trinity = branch_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
(trinity / "local.json").write_text(json.dumps({"todos": []}))
|
||||
|
||||
result = build_todo_section(branch_path)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
|
||||
todo = dash["sections"]["todo"]
|
||||
assert todo["todo_count"] == 0
|
||||
assert "todos" not in todo
|
||||
|
||||
def test_build_todo_section_no_local_json(self, branch_path):
|
||||
"""Test todo section when local.json doesn't exist."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.todo_section import build_todo_section
|
||||
|
||||
result = build_todo_section(branch_path)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
|
||||
todo = dash["sections"]["todo"]
|
||||
assert todo["todo_count"] == 0
|
||||
assert "todos" not in todo
|
||||
|
||||
def test_build_todo_section_corrupt_json(self, branch_path):
|
||||
"""Test todo section handles corrupt local.json."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.todo_section import build_todo_section
|
||||
|
||||
trinity = branch_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
(trinity / "local.json").write_text("not valid json{{{")
|
||||
|
||||
result = build_todo_section(branch_path)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
|
||||
todo = dash["sections"]["todo"]
|
||||
assert todo["todo_count"] == 0
|
||||
|
||||
def test_build_todo_section_no_todos_key(self, branch_path):
|
||||
"""Test todo section when local.json has no todos key."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.todo_section import build_todo_section
|
||||
|
||||
trinity = branch_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
(trinity / "local.json").write_text(json.dumps({"sessions": []}))
|
||||
|
||||
result = build_todo_section(branch_path)
|
||||
assert result is True
|
||||
|
||||
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
|
||||
todo = dash["sections"]["todo"]
|
||||
assert todo["todo_count"] == 0
|
||||
assert "todos" not in todo
|
||||
|
||||
|
||||
class TestRefresh:
|
||||
"""Tests for refresh.py orchestrator."""
|
||||
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.todo_section.build_todo_section")
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.git_section.build_git_section")
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.session_section.build_session_section")
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.dispatch_section.build_dispatch_section")
|
||||
def test_refresh_all_success(self, mock_dispatch, mock_session, mock_git, mock_todo, branch_path):
|
||||
def test_refresh_all_success(self, mock_dispatch, mock_git, branch_path):
|
||||
"""Test refresh orchestrator calls all builders."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.refresh import refresh
|
||||
|
||||
results = refresh(branch_path)
|
||||
assert results["git"]["success"] is True
|
||||
assert results["session"]["success"] is True
|
||||
assert results["dispatch"]["success"] is True
|
||||
assert results["todo"]["success"] is True
|
||||
mock_git.assert_called_once_with(branch_path)
|
||||
mock_session.assert_called_once_with(branch_path)
|
||||
mock_dispatch.assert_called_once_with(branch_path)
|
||||
mock_todo.assert_called_once_with(branch_path)
|
||||
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.todo_section.build_todo_section")
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.git_section.build_git_section")
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.session_section.build_session_section")
|
||||
@patch("aipass.prax.apps.plugins.devpulse_dashboard.dispatch_section.build_dispatch_section")
|
||||
def test_refresh_partial_failure(self, mock_dispatch, mock_session, mock_git, mock_todo, branch_path):
|
||||
def test_refresh_partial_failure(self, mock_dispatch, mock_git, branch_path):
|
||||
"""Test refresh continues when one section fails."""
|
||||
from aipass.prax.apps.plugins.devpulse_dashboard.refresh import refresh
|
||||
|
||||
@@ -338,9 +198,7 @@ class TestRefresh:
|
||||
results = refresh(branch_path)
|
||||
assert results["git"]["success"] is False
|
||||
assert "No .git" in results["git"]["error"]
|
||||
assert results["session"]["success"] is True
|
||||
assert results["dispatch"]["success"] is True
|
||||
assert results["todo"]["success"] is True
|
||||
|
||||
def test_refresh_default_path(self):
|
||||
"""Test refresh uses DEVPULSE_PATH by default."""
|
||||
|
||||
@@ -310,52 +310,52 @@ class TestWriteSection:
|
||||
class TestCalculateQuickStatusStandalone:
|
||||
"""Tests for _calculate_quick_status_standalone -- pure calculation."""
|
||||
|
||||
def test_empty_sections_returns_defaults(self):
|
||||
def test_empty_sections_returns_defaults(self, tmp_path):
|
||||
"""Empty sections produce zeroed counters and 'All clear' summary."""
|
||||
ops = _load_ops()
|
||||
result = ops._calculate_quick_status_standalone({})
|
||||
result = ops._calculate_quick_status_standalone({}, tmp_path)
|
||||
assert result["new_mail"] == 0
|
||||
assert result["opened_mail"] == 0
|
||||
assert result["active_plans"] == 0
|
||||
assert result["action_required"] is False
|
||||
assert result["summary"] == "All clear"
|
||||
|
||||
def test_new_mail_triggers_action_required(self):
|
||||
def test_new_mail_triggers_action_required(self, tmp_path):
|
||||
"""New mail count > 0 sets action_required to True."""
|
||||
ops = _load_ops()
|
||||
sections = {"ai_mail": {"new": 3, "opened": 0}}
|
||||
result = ops._calculate_quick_status_standalone(sections)
|
||||
result = ops._calculate_quick_status_standalone(sections, tmp_path)
|
||||
assert result["new_mail"] == 3
|
||||
assert result["action_required"] is True
|
||||
assert "3 new emails" in result["summary"]
|
||||
|
||||
def test_active_plans_triggers_action_required(self):
|
||||
def test_active_plans_triggers_action_required(self, tmp_path):
|
||||
"""Active plans > 0 sets action_required to True."""
|
||||
ops = _load_ops()
|
||||
sections = {"flow": {"active_plans": 2}}
|
||||
result = ops._calculate_quick_status_standalone(sections)
|
||||
result = ops._calculate_quick_status_standalone(sections, tmp_path)
|
||||
assert result["active_plans"] == 2
|
||||
assert result["action_required"] is True
|
||||
assert "2 active plans" in result["summary"]
|
||||
|
||||
def test_combined_summary_includes_all_parts(self):
|
||||
def test_combined_summary_includes_all_parts(self, tmp_path):
|
||||
"""Summary string includes all active counts."""
|
||||
ops = _load_ops()
|
||||
sections = {
|
||||
"ai_mail": {"new": 2, "opened": 1},
|
||||
"flow": {"active_plans": 3},
|
||||
}
|
||||
result = ops._calculate_quick_status_standalone(sections)
|
||||
result = ops._calculate_quick_status_standalone(sections, tmp_path)
|
||||
assert result["action_required"] is True
|
||||
assert "2 new emails" in result["summary"]
|
||||
assert "1 opened" in result["summary"]
|
||||
assert "3 active plans" in result["summary"]
|
||||
|
||||
def test_unread_field_falls_back_from_new(self):
|
||||
def test_unread_field_falls_back_from_new(self, tmp_path):
|
||||
"""ai_mail may use 'unread' instead of 'new' -- code checks both."""
|
||||
ops = _load_ops()
|
||||
sections = {"ai_mail": {"unread": 7}}
|
||||
result = ops._calculate_quick_status_standalone(sections)
|
||||
result = ops._calculate_quick_status_standalone(sections, tmp_path)
|
||||
assert result["new_mail"] == 7
|
||||
assert result["action_required"] is True
|
||||
|
||||
@@ -468,7 +468,7 @@ class TestUpdateSectionLegacy:
|
||||
}
|
||||
status_called_with: dict[str, object] = {}
|
||||
|
||||
def mock_status(sections):
|
||||
def mock_status(sections, branch_path=None):
|
||||
"""Capture sections passed to status calculator."""
|
||||
status_called_with.update(sections)
|
||||
return {"action_required": True, "summary": "test"}
|
||||
@@ -505,7 +505,7 @@ class TestUpdateSectionLegacy:
|
||||
"flow",
|
||||
{"active_plans": 2},
|
||||
template,
|
||||
lambda s: {"action_required": False},
|
||||
lambda s, bp=None: {"action_required": False},
|
||||
)
|
||||
assert result is True
|
||||
data = json.loads((branch_dir / "DASHBOARD.local.json").read_text(encoding="utf-8"))
|
||||
|
||||
Reference in New Issue
Block a user