giteadmin/lifeos-dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a2183af6e2a630dbf289c28ffbd926ae10697129
lifeos-dev/tests/conftest.py
Michael a2183af6e2 fix: test suite — seed reset, FK mappings, standalone focus tests 
- Change all seed inserts from ON CONFLICT DO NOTHING to DO UPDATE SET
  so seeds always reset to canonical values between test runs
- Add meeting_id, decision_id, link_id to FK_FIELD_MAP in form_factory
- Add item_ids, direction, label, content to NAME_PATTERNS
- Add TestStandaloneFocusItems (7 tests): quick-add, edit, project tab
- Add TestFocusConversion (6 tests): convert to task/note/link/list item
- Add focus_standalone seed with fixture
- Update focus_page_loads test for permanent (non-date-scoped) items
- All 398 tests passing

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-04 03:08:53 +00:00

387 lines
19 KiB
Python
Raw Blame History

"""
Test fixtures - uses psycopg2 (sync) for seed data to avoid event loop conflicts.
Seeds are session-scoped and committed so the app can see them via its own engine.
"""
import asyncio
import uuid
import pytest
import psycopg2
from httpx import AsyncClient, ASGITransport
from tests.registry import app
TEST_DB_DSN = "host=lifeos-db port=5432 dbname=lifeos_test user=postgres password=UCTOQDZiUhN8U"
# ── Fixed seed UUIDs (stable across test runs) ──────────────
SEED_IDS = {
"domain": "a0000000-0000-0000-0000-000000000001",
"area": "a0000000-0000-0000-0000-000000000002",
"project": "a0000000-0000-0000-0000-000000000003",
"task": "a0000000-0000-0000-0000-000000000004",
"contact": "a0000000-0000-0000-0000-000000000005",
"note": "a0000000-0000-0000-0000-000000000006",
"meeting": "a0000000-0000-0000-0000-000000000007",
"decision": "a0000000-0000-0000-0000-000000000008",
"appointment": "a0000000-0000-0000-0000-000000000009",
"link_folder": "a0000000-0000-0000-0000-00000000000a",
"list": "a0000000-0000-0000-0000-00000000000b",
"link": "a0000000-0000-0000-0000-00000000000c",
"capture": "a0000000-0000-0000-0000-00000000000e",
"focus": "a0000000-0000-0000-0000-00000000000f",
"process": "a0000000-0000-0000-0000-000000000010",
"process_step": "a0000000-0000-0000-0000-000000000011",
"process_run": "a0000000-0000-0000-0000-000000000012",
"time_budget": "a0000000-0000-0000-0000-000000000013",
"file": "a0000000-0000-0000-0000-000000000014",
"focus_standalone": "a0000000-0000-0000-0000-000000000015",
}
# ── Reinitialize the async engine within the test event loop ──
@pytest.fixture(scope="session", autouse=True)
async def _reinit_engine():
"""
Replace the engine created at import time with a fresh one created
within the test event loop. This ensures all connections use the right loop.
"""
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from core import database
# Dispose the import-time engine (might have stale loop references)
await database.engine.dispose()
# Create a brand new engine on the current (test) event loop
new_engine = create_async_engine(
database.DATABASE_URL,
echo=False,
pool_size=5,
max_overflow=10,
pool_pre_ping=True,
)
new_session_factory = async_sessionmaker(
new_engine,
class_=AsyncSession,
expire_on_commit=False,
)
# Patch the module so all app code uses the new engine
database.engine = new_engine
database.async_session_factory = new_session_factory
yield
await new_engine.dispose()
# ── Sync DB connection for seed management ──────────────────
@pytest.fixture(scope="session")
def sync_conn():
conn = psycopg2.connect(TEST_DB_DSN)
conn.autocommit = False
yield conn
conn.close()
# ── Seed data (session-scoped, committed) ───────────────────
@pytest.fixture(scope="session")
def all_seeds(sync_conn):
"""Insert all seed data once. Committed so the app's engine can see it."""
cur = sync_conn.cursor()
d = SEED_IDS
try:
# Domain
cur.execute("""
INSERT INTO domains (id, name, color, description, sort_order, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Domain', '#FF5733', 'Auto test domain', 0, false, now(), now())
ON CONFLICT (id) DO UPDATE SET name='Test Domain', color='#FF5733', description='Auto test domain',
sort_order=0, is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["domain"],))
# Area
cur.execute("""
INSERT INTO areas (id, name, domain_id, description, status, sort_order, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Area', %s, 'Auto test area', 'active', 0, false, now(), now())
ON CONFLICT (id) DO UPDATE SET name='Test Area', domain_id=%s, description='Auto test area',
status='active', sort_order=0, is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["area"], d["domain"], d["domain"]))
# Project
cur.execute("""
INSERT INTO projects (id, name, domain_id, area_id, description, status, priority, sort_order, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Project', %s, %s, 'Auto test project', 'active', 2, 0, false, now(), now())
ON CONFLICT (id) DO UPDATE SET name='Test Project', domain_id=%s, area_id=%s,
description='Auto test project', status='active', priority=2, sort_order=0,
is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["project"], d["domain"], d["area"], d["domain"], d["area"]))
# Task (status='open' matches DB default, not 'todo')
cur.execute("""
INSERT INTO tasks (id, title, domain_id, project_id, description, priority, status, sort_order, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Task', %s, %s, 'Auto test task', 2, 'open', 0, false, now(), now())
ON CONFLICT (id) DO UPDATE SET title='Test Task', domain_id=%s, project_id=%s,
description='Auto test task', priority=2, status='open', sort_order=0,
is_deleted=false, deleted_at=NULL, completed_at=NULL, updated_at=now()
""", (d["task"], d["domain"], d["project"], d["domain"], d["project"]))
# Contact
cur.execute("""
INSERT INTO contacts (id, first_name, last_name, company, email, is_deleted, created_at, updated_at)
VALUES (%s, 'Test', 'Contact', 'TestCorp', 'test@example.com', false, now(), now())
ON CONFLICT (id) DO UPDATE SET first_name='Test', last_name='Contact', company='TestCorp',
email='test@example.com', is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["contact"],))
# Note
cur.execute("""
INSERT INTO notes (id, title, domain_id, body, content_format, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Note', %s, 'Test body content', 'markdown', false, now(), now())
ON CONFLICT (id) DO UPDATE SET title='Test Note', domain_id=%s, body='Test body content',
content_format='markdown', is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["note"], d["domain"], d["domain"]))
# Meeting
cur.execute("""
INSERT INTO meetings (id, title, meeting_date, status, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Meeting', '2025-06-15', 'scheduled', false, now(), now())
ON CONFLICT (id) DO UPDATE SET title='Test Meeting', meeting_date='2025-06-15',
status='scheduled', is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["meeting"],))
# Decision
cur.execute("""
INSERT INTO decisions (id, title, status, impact, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Decision', 'decided', 'high', false, now(), now())
ON CONFLICT (id) DO UPDATE SET title='Test Decision', status='decided', impact='high',
is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["decision"],))
# Appointment
cur.execute("""
INSERT INTO appointments (id, title, start_at, end_at, all_day, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Appointment', '2025-06-15 10:00:00', '2025-06-15 11:00:00', false, false, now(), now())
ON CONFLICT (id) DO UPDATE SET title='Test Appointment', start_at='2025-06-15 10:00:00',
end_at='2025-06-15 11:00:00', all_day=false, is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["appointment"],))
# Link folder
cur.execute("""
INSERT INTO link_folders (id, name, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Folder', false, now(), now())
ON CONFLICT (id) DO UPDATE SET name='Test Folder', is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["link_folder"],))
# List
cur.execute("""
INSERT INTO lists (id, name, domain_id, project_id, list_type, is_deleted, created_at, updated_at)
VALUES (%s, 'Test List', %s, %s, 'checklist', false, now(), now())
ON CONFLICT (id) DO UPDATE SET name='Test List', domain_id=%s, project_id=%s,
list_type='checklist', is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["list"], d["domain"], d["project"], d["domain"], d["project"]))
# Link
cur.execute("""
INSERT INTO links (id, label, url, domain_id, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Link', 'https://example.com', %s, false, now(), now())
ON CONFLICT (id) DO UPDATE SET label='Test Link', url='https://example.com', domain_id=%s,
is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["link"], d["domain"], d["domain"]))
# Link folder junction
cur.execute("""
INSERT INTO folder_links (folder_id, link_id)
VALUES (%s, %s) ON CONFLICT DO NOTHING
""", (d["link_folder"], d["link"]))
# Capture
cur.execute("""
INSERT INTO capture (id, raw_text, processed, is_deleted, created_at, updated_at)
VALUES (%s, 'Test capture item', false, false, now(), now())
ON CONFLICT (id) DO UPDATE SET raw_text='Test capture item', processed=false,
is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["capture"],))
# Daily focus (task-linked)
cur.execute("""
INSERT INTO daily_focus (id, task_id, focus_date, completed, created_at, updated_at)
VALUES (%s, %s, CURRENT_DATE, false, now(), now())
ON CONFLICT (id) DO UPDATE SET task_id=%s, focus_date=CURRENT_DATE, completed=false,
is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["focus"], d["task"], d["task"]))
# Daily focus (standalone text item)
cur.execute("""
INSERT INTO daily_focus (id, focus_date, completed, title, domain_id, project_id, created_at, updated_at)
VALUES (%s, CURRENT_DATE, false, 'Test Standalone Focus', %s, %s, now(), now())
ON CONFLICT (id) DO UPDATE SET focus_date=CURRENT_DATE, completed=false,
title='Test Standalone Focus', domain_id=%s, project_id=%s,
is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["focus_standalone"], d["domain"], d["project"], d["domain"], d["project"]))
# Junction table seeds for contact tabs
cur.execute("""
INSERT INTO contact_tasks (contact_id, task_id, role)
VALUES (%s, %s, 'assignee') ON CONFLICT DO NOTHING
""", (d["contact"], d["task"]))
cur.execute("""
INSERT INTO contact_projects (contact_id, project_id, role)
VALUES (%s, %s, 'stakeholder') ON CONFLICT DO NOTHING
""", (d["contact"], d["project"]))
cur.execute("""
INSERT INTO contact_meetings (contact_id, meeting_id, role)
VALUES (%s, %s, 'attendee') ON CONFLICT DO NOTHING
""", (d["contact"], d["meeting"]))
cur.execute("""
INSERT INTO contact_lists (contact_id, list_id, role)
VALUES (%s, %s, 'contributor') ON CONFLICT DO NOTHING
""", (d["contact"], d["list"]))
# Process
cur.execute("""
INSERT INTO processes (id, name, process_type, status, category, is_deleted, created_at, updated_at)
VALUES (%s, 'Test Process', 'checklist', 'active', 'Testing', false, now(), now())
ON CONFLICT (id) DO UPDATE SET name='Test Process', process_type='checklist', status='active',
category='Testing', is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["process"],))
# Process step
cur.execute("""
INSERT INTO process_steps (id, process_id, title, instructions, sort_order, is_deleted, created_at, updated_at)
VALUES (%s, %s, 'Test Step', 'Do the thing', 0, false, now(), now())
ON CONFLICT (id) DO UPDATE SET process_id=%s, title='Test Step', instructions='Do the thing',
sort_order=0, is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["process_step"], d["process"], d["process"]))
# Process run
cur.execute("""
INSERT INTO process_runs (id, process_id, title, status, process_type, task_generation, is_deleted, created_at, updated_at)
VALUES (%s, %s, 'Test Run', 'not_started', 'checklist', 'all_at_once', false, now(), now())
ON CONFLICT (id) DO UPDATE SET process_id=%s, title='Test Run', status='not_started',
process_type='checklist', task_generation='all_at_once', is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["process_run"], d["process"], d["process"]))
# Time budget
cur.execute("""
INSERT INTO time_budgets (id, domain_id, weekly_hours, effective_from, is_deleted, created_at, updated_at)
VALUES (%s, %s, 10, CURRENT_DATE, false, now(), now())
ON CONFLICT (id) DO UPDATE SET domain_id=%s, weekly_hours=10, effective_from=CURRENT_DATE,
is_deleted=false, deleted_at=NULL, updated_at=now()
""", (d["time_budget"], d["domain"], d["domain"]))
# File (create a dummy file on disk for download/serve tests)
import os
from pathlib import Path
file_storage = os.getenv("FILE_STORAGE_PATH", "/opt/lifeos/files/dev")
Path(file_storage).mkdir(parents=True, exist_ok=True)
dummy_file_path = os.path.join(file_storage, "test_seed_file.txt")
with open(dummy_file_path, "w") as f:
f.write("test seed file content")
cur.execute("""
INSERT INTO files (id, filename, original_filename, storage_path, mime_type, size_bytes, is_deleted, created_at, updated_at)
VALUES (%s, 'test_seed_file.txt', 'test_seed_file.txt', %s, 'text/plain', 22, false, now(), now())
ON CONFLICT (id) DO NOTHING
""", (d["file"], dummy_file_path))
sync_conn.commit()
except Exception as e:
sync_conn.rollback()
raise RuntimeError(f"Seed data insertion failed: {e}") from e
yield d
# Cleanup: delete all seed data (reverse dependency order)
try:
# Junction tables first
cur.execute("DELETE FROM contact_tasks WHERE task_id = %s", (d["task"],))
cur.execute("DELETE FROM contact_projects WHERE project_id = %s", (d["project"],))
cur.execute("DELETE FROM contact_meetings WHERE meeting_id = %s", (d["meeting"],))
cur.execute("DELETE FROM contact_lists WHERE list_id = %s", (d["list"],))
cur.execute("DELETE FROM files WHERE id = %s", (d["file"],))
if os.path.exists(dummy_file_path):
os.remove(dummy_file_path)
cur.execute("DELETE FROM time_budgets WHERE id = %s", (d["time_budget"],))
cur.execute("DELETE FROM process_runs WHERE id = %s", (d["process_run"],))
cur.execute("DELETE FROM process_steps WHERE id = %s", (d["process_step"],))
cur.execute("DELETE FROM processes WHERE id = %s", (d["process"],))
cur.execute("DELETE FROM daily_focus WHERE id = %s", (d["focus_standalone"],))
cur.execute("DELETE FROM daily_focus WHERE id = %s", (d["focus"],))
cur.execute("DELETE FROM capture WHERE id = %s", (d["capture"],))
cur.execute("DELETE FROM folder_links WHERE link_id = %s", (d["link"],))
cur.execute("DELETE FROM links WHERE id = %s", (d["link"],))
cur.execute("DELETE FROM lists WHERE id = %s", (d["list"],))
cur.execute("DELETE FROM link_folders WHERE id = %s", (d["link_folder"],))
cur.execute("DELETE FROM appointments WHERE id = %s", (d["appointment"],))
cur.execute("DELETE FROM decisions WHERE id = %s", (d["decision"],))
cur.execute("DELETE FROM meetings WHERE id = %s", (d["meeting"],))
cur.execute("DELETE FROM notes WHERE id = %s", (d["note"],))
cur.execute("DELETE FROM contacts WHERE id = %s", (d["contact"],))
cur.execute("DELETE FROM tasks WHERE id = %s", (d["task"],))
cur.execute("DELETE FROM projects WHERE id = %s", (d["project"],))
cur.execute("DELETE FROM areas WHERE id = %s", (d["area"],))
cur.execute("DELETE FROM domains WHERE id = %s", (d["domain"],))
sync_conn.commit()
except Exception:
sync_conn.rollback()
finally:
cur.close()
# ── HTTP client (function-scoped) ───────────────────────────
@pytest.fixture
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
# ── Async DB session for business logic tests ───────────────
@pytest.fixture
async def db_session():
"""Yields an async DB session for direct SQL in tests."""
from core.database import async_session_factory
async with async_session_factory() as session:
yield session
# ── Individual seed entity fixtures (for test_business_logic.py) ──
@pytest.fixture(scope="session")
def seed_domain(all_seeds):
return {"id": all_seeds["domain"], "name": "Test Domain", "color": "#FF5733"}
@pytest.fixture(scope="session")
def seed_area(all_seeds):
return {"id": all_seeds["area"], "name": "Test Area"}
@pytest.fixture(scope="session")
def seed_project(all_seeds):
return {"id": all_seeds["project"], "name": "Test Project"}
@pytest.fixture(scope="session")
def seed_task(all_seeds):
return {"id": all_seeds["task"], "title": "Test Task"}
@pytest.fixture(scope="session")
def seed_contact(all_seeds):
return {"id": all_seeds["contact"], "first_name": "Test", "last_name": "Contact"}
@pytest.fixture(scope="session")
def seed_note(all_seeds):
return {"id": all_seeds["note"], "title": "Test Note"}
@pytest.fixture(scope="session")
def seed_meeting(all_seeds):
return {"id": all_seeds["meeting"], "title": "Test Meeting"}
@pytest.fixture(scope="session")
def seed_list(all_seeds):
return {"id": all_seeds["list"], "name": "Test List"}
@pytest.fixture(scope="session")
def seed_appointment(all_seeds):
return {"id": all_seeds["appointment"], "title": "Test Appointment"}
@pytest.fixture(scope="session")
def seed_focus_standalone(all_seeds):
return {"id": all_seeds["focus_standalone"], "title": "Test Standalone Focus"}
Reference in New Issue View Git Blame Copy Permalink