mirror of
https://github.com/computerim/impactflow-discovery.git
synced 2026-08-27 06:00:35 +00:00
Joel Salmon
c4fc1cccd7
Build the boundary the ImpactFlow core time-tracker plugs into. A task maps to a foundation — one of the six stable profile elements (love, strength, mission, vocation, short_term, long_term) — so the tracker can ask "which goal does this build toward?" and post the answer back to Vision. - Models + migration 006: task_mapping (one row per logged time entry). - app/services/foundations.py: the six foundations + a pure, testable work-pattern aggregator (rollup) and a plain-language summary. - app/routers/integration.py (user-scoped; tracker calls as the user or via X-API-Key): GET /foundations, POST/GET /task-mappings, GET /work-patterns?days=N (per-foundation minutes/share/neglected). - Reminder engine now pulls from real work patterns: CheckinCoach takes an optional work-pattern summary (last 14 days) and reflects where time has gone against the person's own words — an observation, never a verdict. - Frontend: dashboard.html (time per foundation + neglected); linked from profile.html. Documented the core-tracker integration contract in the README. Phase 4 completes the Vision module's roadmap on the Discovery side; the core tracker integrates by calling these endpoints. Tests: 86 passing (added pure-aggregator, integration-endpoint, and work-pattern-into-check-in tests; run in-container). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
222 lines
7.5 KiB
Python
222 lines
7.5 KiB
Python
"""Tests for the Phase 3 coaching endpoints.
|
|
|
|
The Anthropic-backed CheckinCoach is replaced with a fake (monkeypatch on the
|
|
router) so these run offline. Auth uses the X-API-Key admin identity, which is
|
|
also what lets the admin-only batch /run be exercised.
|
|
"""
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
API_KEY = {"X-API-Key": "test-api-key"}
|
|
|
|
|
|
async def _seed_profile(locked: bool = False, triad: str = "gut") -> str:
|
|
from app.auth import API_KEY_ADMIN_ID, ensure_api_key_admin
|
|
from app.database import AsyncSessionLocal
|
|
from app.models import DiscoveryConversation, DiscoveryProfile
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
await ensure_api_key_admin(db)
|
|
conv = DiscoveryConversation(
|
|
id=str(uuid.uuid4()),
|
|
user_id=API_KEY_ADMIN_ID,
|
|
started_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(conv)
|
|
await db.commit()
|
|
profile = DiscoveryProfile(
|
|
id=str(uuid.uuid4()),
|
|
user_id=API_KEY_ADMIN_ID,
|
|
conversation_id=conv.id,
|
|
generated_at=datetime.now(timezone.utc),
|
|
triad=triad,
|
|
short_term_goals="run a pilot welding cohort",
|
|
long_term_goals="a statewide trades outfit",
|
|
overlap_narrative="you come alive protecting others",
|
|
locked=locked,
|
|
)
|
|
db.add(profile)
|
|
await db.commit()
|
|
return profile.id
|
|
|
|
|
|
class FakeCheckinCoach:
|
|
body = 'You said you want to "run a pilot welding cohort." Does that still feel true?'
|
|
|
|
last_work_patterns = "unset"
|
|
|
|
def __init__(self, api_key=None, model=None):
|
|
pass
|
|
|
|
async def generate(self, profile, prefs, work_patterns=None):
|
|
FakeCheckinCoach.last_work_patterns = work_patterns
|
|
return FakeCheckinCoach.body
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def fake_checkin(monkeypatch, app_client):
|
|
monkeypatch.setattr("app.routers.coaching.CheckinCoach", FakeCheckinCoach)
|
|
yield
|
|
|
|
|
|
async def test_get_preferences_autoderives_from_profile(app_client):
|
|
await _seed_profile(triad="gut")
|
|
r = await app_client.get(
|
|
"/discovery/coaching/preferences", headers=API_KEY
|
|
)
|
|
assert r.status_code == 200
|
|
p = r.json()
|
|
assert p["coaching_style"] == "direct" # gut default
|
|
assert p["auto_generated"] is True
|
|
|
|
|
|
async def test_get_preferences_without_profile_404(app_client):
|
|
r = await app_client.get(
|
|
"/discovery/coaching/preferences", headers=API_KEY
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
async def test_put_preferences_overrides_and_clears_auto(app_client):
|
|
await _seed_profile(triad="gut")
|
|
await app_client.get("/discovery/coaching/preferences", headers=API_KEY)
|
|
r = await app_client.put(
|
|
"/discovery/coaching/preferences",
|
|
headers=API_KEY,
|
|
json={"coaching_style": "warm", "coaching_frequency": "monthly"},
|
|
)
|
|
assert r.status_code == 200
|
|
p = r.json()
|
|
assert p["coaching_style"] == "warm"
|
|
assert p["coaching_frequency"] == "monthly"
|
|
assert p["auto_generated"] is False
|
|
|
|
|
|
async def test_put_invalid_value_rejected(app_client):
|
|
await _seed_profile()
|
|
await app_client.get("/discovery/coaching/preferences", headers=API_KEY)
|
|
r = await app_client.put(
|
|
"/discovery/coaching/preferences",
|
|
headers=API_KEY,
|
|
json={"coaching_frequency": "hourly"},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
async def test_regenerate_restores_auto_defaults(app_client):
|
|
await _seed_profile(triad="gut")
|
|
await app_client.get("/discovery/coaching/preferences", headers=API_KEY)
|
|
await app_client.put(
|
|
"/discovery/coaching/preferences",
|
|
headers=API_KEY,
|
|
json={"coaching_style": "warm"},
|
|
)
|
|
r = await app_client.post(
|
|
"/discovery/coaching/preferences/regenerate", headers=API_KEY
|
|
)
|
|
assert r.status_code == 200
|
|
p = r.json()
|
|
assert p["coaching_style"] == "direct" # back to gut default
|
|
assert p["auto_generated"] is True
|
|
|
|
|
|
async def test_create_and_respond_to_checkin(app_client):
|
|
await _seed_profile()
|
|
gen = await app_client.post(
|
|
"/discovery/coaching/checkins", headers=API_KEY
|
|
)
|
|
assert gen.status_code == 200
|
|
checkin = gen.json()
|
|
assert "run a pilot" in checkin["body"]
|
|
assert checkin["still_valid"] is None
|
|
|
|
listed = (await app_client.get(
|
|
"/discovery/coaching/checkins", headers=API_KEY
|
|
)).json()
|
|
assert len(listed) == 1
|
|
|
|
resp = await app_client.put(
|
|
f"/discovery/coaching/checkins/{checkin['id']}/respond",
|
|
headers=API_KEY,
|
|
json={"still_valid": False, "note": "my focus shifted"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["still_valid"] is False
|
|
assert body["response_note"] == "my focus shifted"
|
|
assert body["acknowledged_at"] is not None
|
|
|
|
|
|
async def test_run_batch_generates_for_due_locked_users(app_client):
|
|
await _seed_profile(locked=True, triad="gut")
|
|
# Auto-create preferences (weekly cadence for gut).
|
|
await app_client.get("/discovery/coaching/preferences", headers=API_KEY)
|
|
|
|
# No prior check-in -> due -> one generated.
|
|
r1 = (await app_client.post("/discovery/coaching/run", headers=API_KEY)).json()
|
|
assert r1["considered"] == 1
|
|
assert r1["generated"] == 1
|
|
|
|
# Immediately again -> a recent check-in exists -> not due.
|
|
r2 = (await app_client.post("/discovery/coaching/run", headers=API_KEY)).json()
|
|
assert r2["generated"] == 0
|
|
|
|
|
|
async def test_run_batch_skips_off_cadence(app_client):
|
|
await _seed_profile(locked=True)
|
|
await app_client.get("/discovery/coaching/preferences", headers=API_KEY)
|
|
await app_client.put(
|
|
"/discovery/coaching/preferences",
|
|
headers=API_KEY,
|
|
json={"coaching_frequency": "off"},
|
|
)
|
|
r = (await app_client.post("/discovery/coaching/run", headers=API_KEY)).json()
|
|
assert r["considered"] == 0
|
|
assert r["generated"] == 0
|
|
|
|
|
|
async def test_run_batch_skips_unlocked_profile(app_client):
|
|
await _seed_profile(locked=False) # not affirmed
|
|
await app_client.get("/discovery/coaching/preferences", headers=API_KEY)
|
|
r = (await app_client.post("/discovery/coaching/run", headers=API_KEY)).json()
|
|
assert r["considered"] == 0
|
|
|
|
|
|
async def test_checkin_receives_recent_work_patterns(app_client):
|
|
"""Phase 4 wiring: a generated check-in is fed the recent work-pattern
|
|
summary so the coach can reflect where time has gone."""
|
|
await _seed_profile()
|
|
await app_client.post(
|
|
"/discovery/integration/task-mappings",
|
|
headers=API_KEY,
|
|
json={"external_task_id": "t1", "foundation": "short_term", "minutes": 90},
|
|
)
|
|
FakeCheckinCoach.last_work_patterns = "unset"
|
|
await app_client.post("/discovery/coaching/checkins", headers=API_KEY)
|
|
assert FakeCheckinCoach.last_work_patterns is not None
|
|
assert "last 14 days" in FakeCheckinCoach.last_work_patterns
|
|
|
|
|
|
async def test_checkin_without_work_patterns_passes_none(app_client):
|
|
await _seed_profile()
|
|
FakeCheckinCoach.last_work_patterns = "unset"
|
|
await app_client.post("/discovery/coaching/checkins", headers=API_KEY)
|
|
assert FakeCheckinCoach.last_work_patterns is None
|
|
|
|
|
|
async def test_due_logic_unit():
|
|
from app.routers.coaching import _is_due
|
|
|
|
now = datetime(2026, 6, 16, tzinfo=timezone.utc)
|
|
assert _is_due("weekly", None, now) is True
|
|
assert _is_due("weekly", now - timedelta(days=8), now) is True
|
|
assert _is_due("weekly", now - timedelta(days=3), now) is False
|
|
assert _is_due("off", None, now) is False
|
|
|
|
|
|
async def test_coaching_requires_auth(app_client):
|
|
r = await app_client.get("/discovery/coaching/preferences")
|
|
assert r.status_code == 401
|