201 lines
7.7 KiB
Python
201 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import urllib.error
|
|
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")
|
|
|
|
|
|
def admin_cookie(user_id: str = "u_1", name: str = "Lin") -> str:
|
|
return "daily_report_session=" + create_session_cookie(
|
|
{"feishu_user_id": user_id, "name": name}, "session-secret"
|
|
)
|
|
|
|
|
|
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("今日状态", submit)
|
|
self.assertIn('data-list="today_done"', submit)
|
|
self.assertIn('data-add-list="today_done"', submit)
|
|
self.assertIn('resetItems("today_done")', submit)
|
|
self.assertIn("我的历史日报", submit)
|
|
|
|
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_and_previous_reference(self) -> None:
|
|
server, db, base_url = make_server()
|
|
try:
|
|
first_request = urllib.request.Request(
|
|
f"{base_url}/api/reports",
|
|
data=json.dumps(
|
|
{
|
|
"feishu_user_id": "u_1",
|
|
"report_date": "2026-05-06",
|
|
"today_done": "1. 完成 API",
|
|
"tomorrow_plan": "1. 完善界面",
|
|
},
|
|
ensure_ascii=False,
|
|
).encode("utf-8"),
|
|
headers={"content-type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(first_request, timeout=5) as response:
|
|
self.assertEqual(response.status, 200)
|
|
|
|
second_request = urllib.request.Request(
|
|
f"{base_url}/api/reports",
|
|
data=json.dumps(
|
|
{
|
|
"feishu_user_id": "u_1",
|
|
"report_date": "2026-05-07",
|
|
"report_status": "need_help",
|
|
"today_done": "1. 完善界面",
|
|
"tomorrow_plan": "1. 上线测试",
|
|
},
|
|
ensure_ascii=False,
|
|
).encode("utf-8"),
|
|
headers={"content-type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(second_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",
|
|
}
|
|
)
|
|
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"], "1. 完善界面")
|
|
self.assertEqual(summary["reports"][0]["report_status"], "need_help")
|
|
|
|
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"], "1. 上线测试")
|
|
|
|
status, reference_body = get(f"{base_url}/api/reports/previous?feishu_user_id=u_1&date=2026-05-07")
|
|
reference = json.loads(reference_body)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(reference["report"]["report_date"], "2026-05-06")
|
|
self.assertEqual(reference["report"]["tomorrow_plan"], "1. 完善界面")
|
|
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:
|
|
status, body = get_with_cookie(f"{base_url}/submit", admin_cookie())
|
|
self.assertEqual(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 = admin_cookie("u_staff", "Staff")
|
|
with self.assertRaises(urllib.error.HTTPError) as error:
|
|
get_with_cookie(f"{base_url}/manager", staff_cookie)
|
|
self.assertEqual(error.exception.code, 403)
|
|
|
|
status, body = get_with_cookie(f"{base_url}/manager", admin_cookie("u_admin", "Admin"))
|
|
self.assertEqual(status, 200)
|
|
self.assertIn("日报浏览", body)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|