kaikai_test/tests/test_web.py

from __future__ import annotations

import json
import tempfile
import threading
import unittest
import urllib.request
from pathlib import Path

from daily_report.config import Config
from daily_report.db import Database
from daily_report.feishu_auth import create_session_cookie
from daily_report.web import create_server


def make_server(employees: list[dict] | None = None) -> tuple:
    temp_dir = Path(tempfile.mkdtemp(prefix="daily-report-web-"))
    seed_path = temp_dir / "employees.json"
    seed_path.write_text(
        json.dumps(employees or [{"feishu_user_id": "u_1", "name": "Lin", "department": "Design", "active": True}]),
        encoding="utf-8",
    )
    config = Config(
        root_dir=Path(__file__).resolve().parent.parent,
        port=0,
        base_url="http://localhost:8787",
        database_path=temp_dir / "test.sqlite",
        employee_seed_path=seed_path,
        workday_calendar_path=temp_dir / "workday-calendar.json",
        feishu_webhook_url="",
        feishu_webhook_secret="",
        feishu_app_id="",
        feishu_app_secret="",
        session_secret="session-secret",
    )
    db = Database(config.database_path, config.employee_seed_path)
    server = create_server(config, db)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, db, f"http://127.0.0.1:{server.server_address[1]}"


def get(url: str) -> tuple[int, str]:
    with urllib.request.urlopen(url, timeout=5) as response:
        return response.status, response.read().decode("utf-8")


def get_with_cookie(url: str, cookie: str) -> tuple[int, str]:
    opener = urllib.request.build_opener()
    opener.addheaders = [("Cookie", cookie)]
    with opener.open(url, timeout=5) as response:
        return response.status, response.read().decode("utf-8")


class WebTest(unittest.TestCase):
    def test_serves_submit_and_manager_pages(self) -> None:
        server, db, base_url = make_server()
        try:
            status, submit = get(f"{base_url}/submit")
            self.assertEqual(status, 200)
            self.assertIn("每日报告", submit)
            self.assertIn("1. \n2. \n3. \n4.", submit)
            self.assertIn("我的历史日报", submit)

            admin_cookie = "daily_report_session=" + create_session_cookie(
                {"feishu_user_id": "u_1", "name": "Lin"}, "session-secret"
            )
            db.upsert_employee(
                {
                    "feishu_user_id": "u_1",
                    "name": "Lin",
                    "department": "Design",
                    "manager": "",
                    "active": True,
                    "role": "admin",
                }
            )
            status, manager = get_with_cookie(f"{base_url}/manager", admin_cookie)
            self.assertEqual(status, 200)
            self.assertIn("日报浏览", manager)
        finally:
            server.shutdown()
            server.server_close()
            db.close()

    def test_accepts_report_submission_and_returns_summary(self) -> None:
        server, db, base_url = make_server()
        try:
            request = urllib.request.Request(
                f"{base_url}/api/reports",
                data=json.dumps(
                    {
                        "feishu_user_id": "u_1",
                        "report_date": "2026-05-07",
                        "today_done": "完成 API",
                        "tomorrow_plan": "完善界面",
                    },
                    ensure_ascii=False,
                ).encode("utf-8"),
                headers={"content-type": "application/json"},
                method="POST",
            )
            with urllib.request.urlopen(request, timeout=5) as response:
                self.assertEqual(response.status, 200)

            db.upsert_employee(
                {
                    "feishu_user_id": "u_1",
                    "name": "Lin",
                    "department": "Design",
                    "manager": "",
                    "active": True,
                    "role": "admin",
                }
            )
            admin_cookie = "daily_report_session=" + create_session_cookie(
                {"feishu_user_id": "u_1", "name": "Lin"}, "session-secret"
            )
            status, body = get_with_cookie(f"{base_url}/api/reports?date=2026-05-07", admin_cookie)
            summary = json.loads(body)
            self.assertEqual(status, 200)
            self.assertEqual(summary["submittedCount"], 1)
            self.assertEqual(summary["reports"][0]["today_done"], "完成 API")

            status, history_body = get(f"{base_url}/api/reports/history?feishu_user_id=u_1")
            history = json.loads(history_body)
            self.assertEqual(status, 200)
            self.assertEqual(history["employee"]["name"], "Lin")
            self.assertEqual(history["reports"][0]["tomorrow_plan"], "完善界面")
        finally:
            server.shutdown()
            server.server_close()
            db.close()

    def test_submit_page_uses_logged_in_feishu_identity(self) -> None:
        server, db, base_url = make_server()
        try:
            opener = urllib.request.build_opener()
            opener.addheaders = [
                (
                    "Cookie",
                    "daily_report_session="
                    + create_session_cookie({"feishu_user_id": "u_1", "name": "Lin"}, "session-secret"),
                )
            ]
            with opener.open(f"{base_url}/submit", timeout=5) as response:
                body = response.read().decode("utf-8")

            self.assertEqual(response.status, 200)
            self.assertIn("Lin", body)
            self.assertIn('type="hidden" name="feishu_user_id" value="u_1"', body)
            self.assertNotIn("员工 ID<input", body)
        finally:
            server.shutdown()
            server.server_close()
            db.close()

    def test_manager_page_requires_admin_role(self) -> None:
        server, db, base_url = make_server(
            [
                {"feishu_user_id": "u_admin", "name": "Admin", "active": True, "role": "admin"},
                {"feishu_user_id": "u_staff", "name": "Staff", "active": True, "role": "staff"},
            ]
        )
        try:
            staff_cookie = "daily_report_session=" + create_session_cookie(
                {"feishu_user_id": "u_staff", "name": "Staff"}, "session-secret"
            )
            with self.assertRaises(urllib.error.HTTPError) as error:
                get_with_cookie(f"{base_url}/manager", staff_cookie)
            self.assertEqual(error.exception.code, 403)

            admin_cookie = "daily_report_session=" + create_session_cookie(
                {"feishu_user_id": "u_admin", "name": "Admin"}, "session-secret"
            )
            status, body = get_with_cookie(f"{base_url}/manager", admin_cookie)
            self.assertEqual(status, 200)
            self.assertIn("日报浏览", body)
        finally:
            server.shutdown()
            server.server_close()
            db.close()


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