from datetime import date, timedelta from app.auth import SessionManager, make_password_hash from app.models import DailyRecordInput, MetricValue, TaskInput from app.repository import ConflictError, JsonRepository, period_window def task_payload(): return TaskInput.model_validate({ "name": "学习计划", "startDate": "2026-08-01", "priority": 2, "metrics": [ {"name": "打卡", "kind": "checkin", "period": "day"}, {"name": "刷题", "kind": "quantity", "period": "lifetime", "target": 100, "unit": "题"}, {"name": "专注", "kind": "duration", "period": "day", "target": 3600}, ], }) def test_crud_revision_soft_delete_and_purge(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(task_payload()) assert len(task["metrics"]) == 3 updated = repo.update_task(task["taskId"], task_payload().model_copy(update={"name": "改名后的计划"}), task["version"]) assert updated["taskId"] == task["taskId"] and updated["revision"] == 2 assert repo.history(task["taskId"], date(2026, 8, 1), date(2026, 8, 1))["revisions"] deleted = repo.soft_delete_task(task["taskId"], updated["version"]) assert repo.list_tasks()["items"] == [] repo.restore_task(task["taskId"], deleted["version"]) repo.soft_delete_task(task["taskId"], deleted["version"] + 1) repo.purge_task(task["taskId"], task["taskId"]) assert repo.list_tasks(include_deleted=True)["items"] == [] def test_daily_mixed_metrics_history_and_idempotent_archive(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(task_payload()) values = [MetricValue(metricId=item["metricId"], value=1 if item["kind"] == "checkin" else 120) for item in task["metrics"]] record = repo.upsert_daily(task["taskId"], date(2026, 8, 10), DailyRecordInput(note="不错", values=values)) assert record["version"] == 1 repo.close_day(date(2026, 8, 10)); repo.close_day(date(2026, 8, 10)) history = repo.history(task["taskId"], date(2026, 8, 10), date(2026, 8, 10)) assert history["records"][0]["note"] == "不错" try: repo.upsert_daily(task["taskId"], date(2026, 8, 10), DailyRecordInput(expectedVersion=0, values=[])) except ConflictError: pass else: raise AssertionError("expected version conflict") def test_first_daily_record_accepts_expected_version_zero(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(task_payload()) assert repo.get_daily(task["taskId"], date(2026, 8, 11))["version"] == 0 record = repo.upsert_daily(task["taskId"], date(2026, 8, 11), DailyRecordInput(expectedVersion=0, values=[])) assert record["version"] == 1 def test_period_window_boundaries(): ref = date(2026, 8, 12) assert period_window("day", ref) == (ref, ref) week_start, week_end = period_window("week", ref) assert week_start.weekday() == 0 and (week_end - week_start).days == 6 assert week_start <= ref <= week_end assert period_window("month", ref) == (date(2026, 8, 1), date(2026, 8, 31)) assert period_window("month", date(2026, 2, 5)) == (date(2026, 2, 1), date(2026, 2, 28)) assert period_window("month", date(2028, 2, 5)) == (date(2028, 2, 1), date(2028, 2,29)) assert period_window("month", date(2026, 12, 9)) == (date(2026, 12, 1), date(2026, 12, 31)) assert period_window("lifetime", ref, date(2026, 8, 1)) == (date(2026, 8, 1), ref) assert period_window("lifetime", date(2026, 7, 1), date(2026, 8, 1)) is None, "任务开始前的 lifetime 窗口应为 None" def test_period_window_date_extremes_do_not_overflow(): assert period_window("day", date.max) == (date.max, date.max) assert period_window("day", date.min) == (date.min, date.min) assert period_window("month", date(9999, 12, 1)) == (date(9999, 12, 1), date(9999, 12, 31)) week = period_window("week", date(9999, 12, 31)) assert week[0] <= date(9999, 12, 31) <= week[1] # 1999-12-31 是周五；week 起点应是周一 9999-12-27，终点 clip 到 date.max assert week[0] == date(9999, 12, 27) assert week[1] == date.max assert period_window("month", date(1, 1, 1)) == (date(1, 1, 1), date(1, 1, 31)) def test_summary_counts_each_metric_inside_its_own_period(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(TaskInput.model_validate({ "name": "周期检查", "startDate": "2026-08-01", "metrics": [ {"name": "每日打卡", "kind": "checkin", "period": "day"}, {"name": "本月刷题", "kind": "quantity", "period": "month", "target": 100, "unit": "题"}, {"name": "累计专注", "kind": "duration", "period": "lifetime", "target": 7200}, ], })) day_id, month_id, life_id = [metric["metricId"] for metric in task["metrics"]] for day, quantity, seconds in [(date(2026, 7, 25), 10, 600), (date(2026, 8, 5), 30, 1200), (date(2026, 8, 20), 20, 1800)]: repo.upsert_daily(task["taskId"], day, DailyRecordInput(values=[ MetricValue(metricId=day_id, value=1), MetricValue(metricId=month_id, value=quantity), MetricValue(metricId=life_id, value=seconds), ])) summary = repo.summarize(date(2026, 8, 20))["tasks"][task["taskId"]] by_id = {metric["metricId"]: metric for metric in summary["metrics"]} assert by_id[day_id]["value"] == 1, "day 周期只能统计当天" assert by_id[month_id]["value"] == 50 and by_id[month_id]["percent"] == 50, "month 周期不能带上 7 月" assert by_id[life_id]["value"] == 3000 and by_id[life_id]["percent"] == 42, "lifetime 不能早于任务开始日期" assert summary["progress"] == 64 and summary["today"]["recorded"] is True def test_summary_without_targets_has_no_progress(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(TaskInput.model_validate({ "name": "无上限", "startDate": "2026-08-01", "metrics": [{"name": "写作", "kind": "quantity", "period": "lifetime", "unit": "字"}], })) summary = repo.summarize(date(2026, 8, 20))["tasks"][task["taskId"]] assert summary["progress"] is None assert summary["metrics"][0]["percent"] is None assert summary["today"]["recorded"] is False def test_summary_skips_soft_deleted_tasks(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(task_payload()) repo.soft_delete_task(task["taskId"], task["version"]) assert repo.summarize(date(2026, 8, 20))["tasks"] == {} def test_summary_excludes_facts_after_reference_date(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(TaskInput.model_validate({ "name": "月中查询", "startDate": "2026-08-01", "metrics": [{"name": "本月", "kind": "quantity", "period": "month", "target": 100, "unit": "次"}], })) metric_id = task["metrics"][0]["metricId"] # 8 月 5 日一条，8 月 25 日（在 ref=8-20 之后）一条 for day, value in [(date(2026, 8, 5), 30), (date(2026, 8, 25), 20)]: repo.upsert_daily(task["taskId"], day, DailyRecordInput(values=[MetricValue(metricId=metric_id, value=value)])) metric = repo.summarize(date(2026, 8, 20))["tasks"][task["taskId"]]["metrics"][0] assert metric["value"] == 30, "ref 之后的事实不得计入历史汇总" def test_summary_lifetime_ignores_facts_before_task_start(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(TaskInput.model_validate({ "name": "未来开始", "startDate": "2026-08-15", "metrics": [{"name": "累计", "kind": "quantity", "period": "lifetime", "target": 100, "unit": "次"}], })) metric_id = task["metrics"][0]["metricId"] repo.upsert_daily(task["taskId"], date(2026, 8, 10), DailyRecordInput(values=[MetricValue(metricId=metric_id, value=50)])) metric = repo.summarize(date(2026, 8, 20))["tasks"][task["taskId"]]["metrics"][0] assert metric["value"] == 0, "任务开始日期之前的事实不得计入 lifetime" assert metric["windowStart"] == "2026-08-15" def test_percent_half_up_rounding(tmp_path): repo = JsonRepository(tmp_path) task = repo.create_task(TaskInput.model_validate({ "name": "舍入", "startDate": "2026-08-01", "metrics": [{"name": "进度", "kind": "quantity", "period": "lifetime", "target": 8, "unit": "次"}], })) metric_id = task["metrics"][0]["metricId"] repo.upsert_daily(task["taskId"], date(2026, 8, 1), DailyRecordInput(values=[MetricValue(metricId=metric_id, value=1)])) metric = repo.summarize(date(2026, 8, 1))["tasks"][task["taskId"]]["metrics"][0] assert metric["percent"] == 13, "1/8 应四舍五入进位到 13%，而非四舍六入的 12%" def test_session_expiry(): manager = SessionManager(make_password_hash("secret"), ttl_seconds=10) session_id, session = manager.login("secret") assert manager.get(session_id) is session session.created_at -= 11 # 模拟 11 秒前的会话 assert manager.get(session_id) is None, "过期会话应当失效" def test_session_ttl_keeps_fresh_sessions(): manager = SessionManager(make_password_hash("secret"), ttl_seconds=10) session_id, session = manager.login("secret") session.created_at -= 5 assert manager.get(session_id) is session