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>
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock
|
|
from src.airtable_client import PulseAirtableClient, _clean_task_name, _clean_related_item
|
|
|
|
|
|
def test_clean_task_name_strips_bom():
|
|
assert _clean_task_name("\ufeffMy Task") == "My Task"
|
|
|
|
|
|
def test_clean_task_name_strips_whitespace():
|
|
assert _clean_task_name(" Task Name ") == "Task Name"
|
|
|
|
|
|
def test_clean_related_item_strips_notion_url():
|
|
raw = "Automation (Enhancement) (https://www.notion.so/Automation-317003329b1e804c9115e1a92617fb01?pvs=21)"
|
|
assert _clean_related_item(raw) == "Automation (Enhancement)"
|
|
|
|
|
|
def test_clean_related_item_empty():
|
|
assert _clean_related_item("") == ""
|
|
assert _clean_related_item(None) == ""
|
|
|
|
|
|
def test_clean_related_item_multiple_workstreams():
|
|
raw = "Tools / Technology (https://www.notion.so/Tools?pvs=21), Syndication (Enhancement) (https://www.notion.so/Syndication?pvs=21)"
|
|
result = _clean_related_item(raw)
|
|
assert "https" not in result
|
|
assert "Tools / Technology" in result
|
|
assert "Syndication (Enhancement)" in result
|
|
assert not result.endswith(",")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_table(mocker):
|
|
mock_api = mocker.patch("src.airtable_client.Api")
|
|
mock_table = MagicMock()
|
|
mock_api.return_value.table.return_value = mock_table
|
|
return mock_table
|
|
|
|
|
|
def test_fetch_all_tasks_normalizes_fields(mock_table):
|
|
mock_table.all.return_value = [
|
|
{
|
|
"id": "rec1",
|
|
"createdTime": "2026-04-08T04:32:44.000Z",
|
|
"fields": {
|
|
"\ufeffTask": "\ufeffBuild reporting tool",
|
|
"Progress": "In Progress",
|
|
"Priority": "P1",
|
|
"RAG": "Red",
|
|
"Owner": [{"id": "usr1", "name": "Tony Coppola", "email": "tony@oliver.agency"}],
|
|
"Notes": "In good shape",
|
|
},
|
|
}
|
|
]
|
|
client = PulseAirtableClient("fake_key", "base_id", "table_id")
|
|
tasks = client.fetch_all_tasks()
|
|
|
|
assert len(tasks) == 1
|
|
t = tasks[0]
|
|
assert t["task"] == "Build reporting tool"
|
|
assert t["progress"] == "In Progress"
|
|
assert t["priority"] == "P1"
|
|
assert t["rag"] == "Red"
|
|
assert t["owner"] == "Tony Coppola"
|
|
assert t["owners"] == ["Tony Coppola"]
|
|
assert t["notes"] == "In good shape"
|
|
|
|
|
|
def test_fetch_all_tasks_handles_missing_owner(mock_table):
|
|
mock_table.all.return_value = [
|
|
{"id": "rec2", "createdTime": "2026-04-08T00:00:00.000Z",
|
|
"fields": {"\ufeffTask": "Some task", "Progress": "Not Started"}}
|
|
]
|
|
client = PulseAirtableClient("fake_key", "base_id", "table_id")
|
|
tasks = client.fetch_all_tasks()
|
|
assert tasks[0]["owner"] == "Unassigned"
|
|
assert tasks[0]["owners"] == []
|
|
|
|
|
|
def test_fetch_all_tasks_fallback_task_key(mock_table):
|
|
mock_table.all.return_value = [
|
|
{"id": "rec3", "createdTime": "2026-04-08T00:00:00.000Z",
|
|
"fields": {"Task": "Plain task name"}}
|
|
]
|
|
client = PulseAirtableClient("fake_key", "base_id", "table_id")
|
|
tasks = client.fetch_all_tasks()
|
|
assert tasks[0]["task"] == "Plain task name"
|
|
|
|
|
|
def test_fetch_all_tasks_filters_empty_owner_names(mock_table):
|
|
mock_table.all.return_value = [
|
|
{"id": "rec4", "createdTime": "2026-04-08T00:00:00.000Z",
|
|
"fields": {"\ufeffTask": "Some task",
|
|
"Owner": [{"id": "usr1"}, {"id": "usr2", "name": "Alice"}]}}
|
|
]
|
|
client = PulseAirtableClient("fake_key", "base_id", "table_id")
|
|
tasks = client.fetch_all_tasks()
|
|
assert tasks[0]["owner"] == "Alice"
|
|
assert tasks[0]["owners"] == ["Alice"]
|