- Event-based triggers (Phase 1): - Trigger/TriggerLog models with field_change type - TriggerService for condition evaluation and action execution - Trigger CRUD API endpoints - Task integration (status, assignee, priority changes) - Frontend: TriggerList, TriggerForm components - Weekly reports (Phase 2): - ScheduledReport/ReportHistory models - ReportService for stats generation - APScheduler for Friday 16:00 job - Report preview/generate/history API - Frontend: WeeklyReportPreview, ReportHistory components - Tests: 23 new tests (14 triggers + 9 reports) - OpenSpec: add-automation change archived 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
261 lines
8.1 KiB
Python
261 lines
8.1 KiB
Python
import pytest
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
from app.models import User, Space, Project, Task, TaskStatus, ScheduledReport, ReportHistory
|
|
from app.services.report_service import ReportService
|
|
|
|
|
|
@pytest.fixture
|
|
def test_user(db):
|
|
"""Create a test user."""
|
|
user = User(
|
|
id=str(uuid.uuid4()),
|
|
email="reportuser@example.com",
|
|
name="Report User",
|
|
role_id="00000000-0000-0000-0000-000000000003",
|
|
is_active=True,
|
|
is_system_admin=False,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def test_user_token(client, mock_redis, test_user):
|
|
"""Get a token for test user."""
|
|
from app.core.security import create_access_token, create_token_payload
|
|
|
|
token_data = create_token_payload(
|
|
user_id=test_user.id,
|
|
email=test_user.email,
|
|
role="engineer",
|
|
department_id=None,
|
|
is_system_admin=False,
|
|
)
|
|
token = create_access_token(token_data)
|
|
mock_redis.setex(f"session:{test_user.id}", 900, token)
|
|
return token
|
|
|
|
|
|
@pytest.fixture
|
|
def test_space(db, test_user):
|
|
"""Create a test space."""
|
|
space = Space(
|
|
id=str(uuid.uuid4()),
|
|
name="Report Test Space",
|
|
description="Test space for reports",
|
|
owner_id=test_user.id,
|
|
)
|
|
db.add(space)
|
|
db.commit()
|
|
return space
|
|
|
|
|
|
@pytest.fixture
|
|
def test_project(db, test_space, test_user):
|
|
"""Create a test project."""
|
|
project = Project(
|
|
id=str(uuid.uuid4()),
|
|
space_id=test_space.id,
|
|
title="Report Test Project",
|
|
description="Test project for reports",
|
|
owner_id=test_user.id,
|
|
)
|
|
db.add(project)
|
|
db.commit()
|
|
return project
|
|
|
|
|
|
@pytest.fixture
|
|
def test_statuses(db, test_project):
|
|
"""Create test task statuses."""
|
|
todo = TaskStatus(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="To Do",
|
|
color="#808080",
|
|
position=0,
|
|
)
|
|
in_progress = TaskStatus(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="In Progress",
|
|
color="#0000FF",
|
|
position=1,
|
|
)
|
|
done = TaskStatus(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="Done",
|
|
color="#00FF00",
|
|
position=2,
|
|
)
|
|
db.add_all([todo, in_progress, done])
|
|
db.commit()
|
|
return {"todo": todo, "in_progress": in_progress, "done": done}
|
|
|
|
|
|
@pytest.fixture
|
|
def test_tasks(db, test_project, test_user, test_statuses):
|
|
"""Create test tasks with various statuses."""
|
|
tasks = []
|
|
|
|
# Completed task (updated this week)
|
|
completed_task = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
title="Completed Task",
|
|
status_id=test_statuses["done"].id,
|
|
created_by=test_user.id,
|
|
)
|
|
completed_task.updated_at = datetime.utcnow()
|
|
tasks.append(completed_task)
|
|
|
|
# In progress task
|
|
in_progress_task = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
title="In Progress Task",
|
|
status_id=test_statuses["in_progress"].id,
|
|
created_by=test_user.id,
|
|
)
|
|
tasks.append(in_progress_task)
|
|
|
|
# Overdue task
|
|
overdue_task = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
title="Overdue Task",
|
|
status_id=test_statuses["todo"].id,
|
|
due_date=datetime.utcnow() - timedelta(days=3),
|
|
created_by=test_user.id,
|
|
)
|
|
tasks.append(overdue_task)
|
|
|
|
db.add_all(tasks)
|
|
db.commit()
|
|
return tasks
|
|
|
|
|
|
class TestReportService:
|
|
"""Tests for ReportService."""
|
|
|
|
def test_get_week_start(self):
|
|
"""Test week start calculation."""
|
|
# Test with a known Wednesday
|
|
wednesday = datetime(2024, 12, 25, 15, 30, 0) # Wednesday
|
|
week_start = ReportService.get_week_start(wednesday)
|
|
|
|
assert week_start.weekday() == 0 # Monday
|
|
assert week_start.hour == 0
|
|
assert week_start.minute == 0
|
|
|
|
def test_get_weekly_stats_empty(self, db, test_user):
|
|
"""Test weekly stats with no projects."""
|
|
stats = ReportService.get_weekly_stats(db, test_user.id)
|
|
|
|
assert stats["summary"]["completed_count"] == 0
|
|
assert stats["summary"]["in_progress_count"] == 0
|
|
assert stats["summary"]["total_tasks"] == 0
|
|
assert len(stats["projects"]) == 0
|
|
|
|
def test_get_weekly_stats_with_tasks(self, db, test_user, test_project, test_tasks, test_statuses):
|
|
"""Test weekly stats with tasks."""
|
|
stats = ReportService.get_weekly_stats(db, test_user.id)
|
|
|
|
assert stats["summary"]["completed_count"] == 1
|
|
assert stats["summary"]["in_progress_count"] == 1
|
|
assert stats["summary"]["overdue_count"] == 1
|
|
assert stats["summary"]["total_tasks"] == 3
|
|
assert len(stats["projects"]) == 1
|
|
assert stats["projects"][0]["project_title"] == "Report Test Project"
|
|
|
|
def test_generate_weekly_report(self, db, test_user, test_project, test_tasks, test_statuses):
|
|
"""Test generating a weekly report."""
|
|
report = ReportService.generate_weekly_report(db, test_user.id)
|
|
|
|
assert report is not None
|
|
assert report.status == "sent"
|
|
assert "summary" in report.content
|
|
|
|
# Check scheduled report was created
|
|
scheduled = db.query(ScheduledReport).filter(
|
|
ScheduledReport.recipient_id == test_user.id
|
|
).first()
|
|
assert scheduled is not None
|
|
assert scheduled.last_sent_at is not None
|
|
|
|
|
|
class TestReportAPI:
|
|
"""Tests for Report API endpoints."""
|
|
|
|
def test_preview_weekly_report(self, client, test_user_token, test_project, test_tasks, test_statuses):
|
|
"""Test previewing weekly report."""
|
|
response = client.get(
|
|
"/api/reports/weekly/preview",
|
|
headers={"Authorization": f"Bearer {test_user_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "summary" in data
|
|
assert "projects" in data
|
|
assert data["summary"]["total_tasks"] == 3
|
|
|
|
def test_generate_weekly_report_api(self, client, test_user_token, test_project, test_tasks, test_statuses):
|
|
"""Test generating weekly report via API."""
|
|
response = client.post(
|
|
"/api/reports/weekly/generate",
|
|
headers={"Authorization": f"Bearer {test_user_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == "Weekly report generated successfully"
|
|
assert "report_id" in data
|
|
assert "summary" in data
|
|
|
|
def test_list_report_history_empty(self, client, test_user_token):
|
|
"""Test listing report history when empty."""
|
|
response = client.get(
|
|
"/api/reports/history",
|
|
headers={"Authorization": f"Bearer {test_user_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 0
|
|
assert len(data["reports"]) == 0
|
|
|
|
def test_list_report_history_with_reports(self, client, test_user_token, test_project, test_tasks, test_statuses, db, test_user):
|
|
"""Test listing report history with existing reports."""
|
|
# Generate a report first
|
|
ReportService.generate_weekly_report(db, test_user.id)
|
|
|
|
response = client.get(
|
|
"/api/reports/history",
|
|
headers={"Authorization": f"Bearer {test_user_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 1
|
|
assert len(data["reports"]) >= 1
|
|
assert data["reports"][0]["status"] == "sent"
|
|
|
|
def test_get_report_detail(self, client, test_user_token, test_project, test_tasks, test_statuses, db, test_user):
|
|
"""Test getting specific report detail."""
|
|
# Generate a report first
|
|
report = ReportService.generate_weekly_report(db, test_user.id)
|
|
|
|
response = client.get(
|
|
f"/api/reports/history/{report.id}",
|
|
headers={"Authorization": f"Bearer {test_user_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == report.id
|
|
assert "content" in data
|