kaikai_test/tests/test_report_service.py

from __future__ import annotations

import json
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path

from daily_report.db import Database
from daily_report.report_service import ReportService


def make_database(employees: list[dict]) -> Database:
    temp_dir = Path(tempfile.mkdtemp(prefix="daily-report-"))
    seed_path = temp_dir / "employees.json"
    seed_path.write_text(json.dumps(employees), encoding="utf-8")
    return Database(temp_dir / "test.sqlite", seed_path)


class ReportServiceTest(unittest.TestCase):
    def test_creates_schema_and_loads_active_employees(self) -> None:
        db = make_database(
            [
                {"feishu_user_id": "u_1", "name": "Lin", "department": "Design", "manager": "Mo", "active": True},
                {"feishu_user_id": "u_2", "name": "Chen", "department": "Sales", "manager": "Mo", "active": False},
            ]
        )
        try:
            employees = db.list_active_employees()
            self.assertEqual(len(employees), 1)
            self.assertEqual(employees[0]["name"], "Lin")
        finally:
            db.close()

    def test_validates_required_report_fields(self) -> None:
        db = make_database([{"feishu_user_id": "u_1", "name": "Lin", "active": True}])
        service = ReportService(db)
        try:
            with self.assertRaisesRegex(ValueError, "today_done is required"):
                service.upsert_report(
                    {
                        "feishu_user_id": "u_1",
                        "report_date": "2026-05-07",
                        "today_done": "",
                        "tomorrow_plan": "完善管理页",
                    }
                )
        finally:
            db.close()

    def test_upserts_one_report_per_employee_and_date(self) -> None:
        db = make_database([{"feishu_user_id": "u_1", "name": "Lin", "department": "Design", "active": True}])
        service = ReportService(db, clock=lambda: datetime(2026, 5, 7, 10, tzinfo=timezone.utc))
        try:
            service.upsert_report(
                {
                    "feishu_user_id": "u_1",
                    "report_date": "2026-05-07",
                    "report_status": "risk",
                    "today_done": "1. 完成原型初稿",
                    "tomorrow_plan": "1. 和主管评审",
                }
            )
            service.upsert_report(
                {
                    "feishu_user_id": "u_1",
                    "report_date": "2026-05-07",
                    "report_status": "need_help",
                    "today_done": "1. 完成原型初稿\n2. 调整文案",
                    "tomorrow_plan": "1. 和主管评审",
                    "blockers": "需要确认 logo",
                    "help_needed": "确认视觉方向",
                }
            )
            result = service.list_reports_for_date("2026-05-07")
            self.assertEqual(len(result["reports"]), 1)
            self.assertEqual(result["reports"][0]["report_status"], "need_help")
            self.assertEqual(result["reports"][0]["today_done"], "1. 完成原型初稿\n2. 调整文案")
            self.assertEqual(result["reports"][0]["blockers"], "需要确认 logo")
            self.assertEqual(len(result["missing"]), 0)
        finally:
            db.close()

    def test_returns_missing_active_employees_csv_and_previous_reference(self) -> None:
        db = make_database(
            [
                {"feishu_user_id": "u_1", "name": "Lin", "department": "Design", "active": True},
                {"feishu_user_id": "u_2", "name": "Chen", "department": "Sales", "active": True},
            ]
        )
        service = ReportService(db, clock=lambda: datetime(2026, 5, 7, 10, tzinfo=timezone.utc))
        try:
            service.upsert_report(
                {
                    "feishu_user_id": "u_1",
                    "report_date": "2026-05-06",
                    "today_done": "1. 完成上线检查清单",
                    "tomorrow_plan": "1. 准备发布",
                }
            )
            service.upsert_report(
                {
                    "feishu_user_id": "u_1",
                    "report_date": "2026-05-07",
                    "today_done": "1. 准备发布",
                    "tomorrow_plan": "1. 复盘问题",
                }
            )
            result = service.list_reports_for_date("2026-05-07")
            self.assertEqual(result["submittedCount"], 1)
            self.assertEqual(result["missing"][0]["name"], "Chen")

            csv_text = service.export_reports_csv("2026-05-07")
            self.assertIn("employee_name,report_date,today_done,tomorrow_plan,blockers,help_needed,report_status,submitted_at", csv_text)
            self.assertIn("Lin,2026-05-07,1. 准备发布,1. 复盘问题,,,normal,", csv_text)

            reference = service.previous_report_reference("u_1", "2026-05-07")
            self.assertEqual(reference["report"]["report_date"], "2026-05-06")
            self.assertEqual(reference["report"]["tomorrow_plan"], "1. 准备发布")
        finally:
            db.close()

    def test_admins_are_not_required_to_submit_reports(self) -> None:
        db = make_database(
            [
                {"feishu_user_id": "u_admin", "name": "Admin", "active": True, "role": "admin"},
                {"feishu_user_id": "u_staff", "name": "Staff", "active": True, "role": "staff"},
            ]
        )
        service = ReportService(db, clock=lambda: datetime(2026, 5, 7, 10, tzinfo=timezone.utc))
        try:
            service.upsert_report(
                {
                    "feishu_user_id": "u_admin",
                    "report_date": "2026-05-07",
                    "today_done": "1. 查看汇总",
                    "tomorrow_plan": "1. 继续跟进",
                }
            )
            result = service.list_reports_for_date("2026-05-07")
            self.assertEqual(result["expectedCount"], 1)
            self.assertEqual(result["submittedCount"], 0)
            self.assertEqual(result["missingCount"], 1)
            self.assertEqual(result["missing"][0]["name"], "Staff")
            self.assertEqual(result["reports"], [])
        finally:
            db.close()


if __name__ == "__main__":
    unittest.main()