b70d148b94
Backend - Routes moved under /api/, JWT bearer auth via @before_request - DEV_AUTH_BYPASS escape hatch for local dev - In-memory chat history and report state replaced with Postgres tables (preferences, chat_messages, reports, feedback_events) keyed on user - SQLAlchemy 2.x + Alembic migrations run on container start - Graceful Airtable failure handling — bad creds no longer 500 the API - Per-user data isolation via g.user_email from validated token Frontend - React + Vite + TypeScript SPA at /programme-pulse/ - MSAL.js (PKCE, sessionStorage, ID token to backend) - VITE_DEV_AUTH_BYPASS mirrors backend bypass for local dev - Streaming chat via fetch ReadableStream + SSE parsing - Charts via chart.js, markdown via react-markdown + remark-gfm - Full UI parity with the original templates/index.html Deploy (optical-dev split-build pattern) - Dockerfile + docker-compose.yml (name: programme-pulse pinned; app + Postgres; 127.0.0.1 binding only) - deploy/apache-programme-pulse.conf.tmpl with flushpackets=on for SSE - deploy/deploy.sh mirrors OSOP — port auto-pick (5051..5099), apache conf render, frontend build in throwaway node container, rsync to /var/www/html/programme-pulse, /api/health poll Tests - 49 passing; new tests for DB-backed preferences and JWT auth helpers - SQLite-backed test fixture in tests/conftest.py Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
import pytest
|
|
from datetime import date, timedelta
|
|
from src.analyzer import analyze
|
|
|
|
TODAY = date.today().isoformat()
|
|
YESTERDAY = (date.today() - timedelta(days=1)).isoformat()
|
|
NEXT_WEEK = (date.today() + timedelta(days=7)).isoformat()
|
|
|
|
|
|
def _task(**kwargs):
|
|
defaults = {
|
|
"id": "rec1", "task": "Test task", "progress": "In Progress",
|
|
"priority": "P1", "rag": "Red", "owner": "Tony Coppola",
|
|
"owners": ["Tony Coppola"], "related_item": "", "category": [],
|
|
"start_date": "", "end_date": "", "deadline": "",
|
|
"doing": "", "blocked_by": "", "blocking": "", "notes": "", "hours": None,
|
|
}
|
|
defaults.update(kwargs)
|
|
return defaults
|
|
|
|
|
|
def test_analyze_counts_total():
|
|
tasks = [_task(id="r1"), _task(id="r2"), _task(id="r3", progress="Complete")]
|
|
result = analyze(tasks)
|
|
assert result["total"] == 3
|
|
assert result["active_total"] == 2
|
|
|
|
|
|
def test_analyze_excludes_complete_and_cancelled_from_active():
|
|
tasks = [
|
|
_task(id="r1", progress="Complete"),
|
|
_task(id="r2", progress="Cancelled"),
|
|
_task(id="r3", progress="In Progress"),
|
|
]
|
|
result = analyze(tasks)
|
|
assert result["active_total"] == 1
|
|
|
|
|
|
def test_analyze_red_flags_blocked_and_pending():
|
|
tasks = [
|
|
_task(id="r1", progress="Blocked"),
|
|
_task(id="r2", progress="Pending Feedback"),
|
|
_task(id="r3", progress="In Progress"),
|
|
]
|
|
result = analyze(tasks)
|
|
assert len(result["red_flags"]) == 2
|
|
flags = {t["id"] for t in result["red_flags"]}
|
|
assert flags == {"r1", "r2"}
|
|
|
|
|
|
def test_analyze_p1_watchlist_excludes_complete():
|
|
tasks = [
|
|
_task(id="r1", priority="P1", progress="In Progress", rag="Red"),
|
|
_task(id="r2", priority="P1", progress="Complete", rag="Red"),
|
|
_task(id="r3", priority="P2", progress="In Progress", rag="Red"),
|
|
]
|
|
result = analyze(tasks)
|
|
assert len(result["p1_watchlist"]) == 1
|
|
assert result["p1_watchlist"][0]["id"] == "r1"
|
|
|
|
|
|
def test_analyze_p1_watchlist_ordered_by_rag():
|
|
tasks = [
|
|
_task(id="r1", priority="P1", progress="In Progress", rag="Green"),
|
|
_task(id="r2", priority="P1", progress="In Progress", rag="Red"),
|
|
_task(id="r3", priority="P1", progress="In Progress", rag="Amber"),
|
|
]
|
|
result = analyze(tasks)
|
|
rags = [t["rag"] for t in result["p1_watchlist"]]
|
|
assert rags == ["Red", "Amber", "Green"]
|
|
|
|
|
|
def test_analyze_groups_by_owner():
|
|
tasks = [
|
|
_task(id="r1", owner="Alice", progress="In Progress"),
|
|
_task(id="r2", owner="Bob", progress="In Progress"),
|
|
_task(id="r3", owner="Alice", progress="Not Started"),
|
|
]
|
|
result = analyze(tasks)
|
|
assert set(result["by_owner"].keys()) == {"Alice", "Bob"}
|
|
assert len(result["by_owner"]["Alice"]) == 2
|
|
assert len(result["by_owner"]["Bob"]) == 1
|
|
|
|
|
|
def test_analyze_detects_overdue_by_deadline():
|
|
tasks = [
|
|
_task(id="r1", progress="In Progress", deadline=YESTERDAY),
|
|
_task(id="r2", progress="In Progress", deadline=NEXT_WEEK),
|
|
_task(id="r3", progress="Complete", deadline=YESTERDAY),
|
|
]
|
|
result = analyze(tasks)
|
|
assert len(result["overdue"]) == 1
|
|
assert result["overdue"][0]["id"] == "r1"
|
|
assert result["overdue"][0]["_days_overdue"] >= 1
|
|
|
|
|
|
def test_analyze_overdue_falls_back_to_end_date():
|
|
tasks = [_task(id="r1", progress="In Progress", deadline="", end_date=YESTERDAY)]
|
|
result = analyze(tasks)
|
|
assert len(result["overdue"]) == 1
|
|
|
|
|
|
def test_analyze_progress_counts():
|
|
tasks = [
|
|
_task(id="r1", progress="In Progress"),
|
|
_task(id="r2", progress="In Progress"),
|
|
_task(id="r3", progress="Complete"),
|
|
]
|
|
result = analyze(tasks)
|
|
assert result["progress_counts"]["In Progress"] == 2
|
|
assert result["progress_counts"]["Complete"] == 1
|