02b81f0113
Security (fail-secure for production): - Refuse to start without PPM_PASSWORD/PPM_PASSWORD_HASH unless PPM_ALLOW_NO_AUTH=1 (local dev only) - Add login rate limiting (5 failures / 60s per client), enforced inside SessionManager.login so it cannot be bypassed at the endpoint layer - compose defaults PPM_COOKIE_SECURE=true for HTTPS behind reverse proxy - Add .dockerignore to keep runtime data/out of the image Features: - Task timeline gantt on home page (week/month zoom, horizontal scroll, infinite-arrow for open-ended dates, CSS Grid layout) - Duration metrics gain a second/minute/hour unit for input & display (canonical storage stays integer seconds) - Per-metric progress bars on home page (no more merged average) - Backfill historical records via date picker in the today-record card - Admin panel: stats overview, JSON import/restore, full task list mgmt Fixes: - Nullable task startDate; summarize handles missing start date - Task dialog cancel/close no longer submits the form - Import validates whole backup first, rejects path traversal, rebuilds archives, and replaces stale files Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
256 lines
12 KiB
Python
256 lines
12 KiB
Python
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, ValidationError, 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
|
||
|
||
|
||
def test_import_replaces_stale_files(tmp_path):
|
||
repo = JsonRepository(tmp_path)
|
||
first = repo.create_task(task_payload())
|
||
second = repo.create_task(task_payload().model_copy(update={"name": "多出来的任务"}))
|
||
assert repo.list_tasks()["items"] and len(repo.list_tasks()["items"]) == 2
|
||
|
||
# 用只含第一个任务的备份整体替换，第二个任务应被清除。
|
||
export = repo.export_all()
|
||
export["tasks"] = [first]
|
||
export["revisions"] = [rev for rev in export["revisions"] if rev["taskId"] == first["taskId"]]
|
||
export["daily"] = []
|
||
repo.import_all(export)
|
||
remaining = repo.list_tasks(include_deleted=True)["items"]
|
||
assert [task["taskId"] for task in remaining] == [first["taskId"]]
|
||
|
||
|
||
def test_import_rejects_path_traversal(tmp_path):
|
||
repo = JsonRepository(tmp_path)
|
||
payload = repo.export_all()
|
||
payload["tasks"] = [{"taskId": "../evil", "schemaVersion": 1, "name": "x", "version": 1, "revision": 1, "createdAt": "x", "updatedAt": "x", "deletedAt": None, "metrics": []}]
|
||
try:
|
||
repo.import_all(payload)
|
||
except ValidationError:
|
||
pass
|
||
else:
|
||
raise AssertionError("路径逃逸的 taskId 应被拒绝")
|
||
|
||
|
||
def test_import_rebuilds_archives(tmp_path):
|
||
repo = JsonRepository(tmp_path)
|
||
task = repo.create_task(task_payload())
|
||
metric_id = task["metrics"][0]["metricId"]
|
||
repo.upsert_daily(task["taskId"], date(2026, 8, 10), DailyRecordInput(values=[MetricValue(metricId=metric_id, value=1)]))
|
||
export = repo.export_all()
|
||
# 清空归档目录，模拟“从未归档”的目标状态。
|
||
for path in (tmp_path / "archive").glob("*/*.json"):
|
||
path.unlink()
|
||
assert list((tmp_path / "archive").glob("*/*.json")) == []
|
||
repo.import_all(export)
|
||
assert list((tmp_path / "archive").glob("*/*.json")), "导入后应重建归档"
|
||
|
||
|
||
def test_session_manager_fail_secure(monkeypatch):
|
||
from app.auth import SessionManager
|
||
# 清掉所有相关环境变量后，from_environment 应拒绝启动
|
||
for key in ("PPM_PASSWORD", "PPM_PASSWORD_HASH", "PPM_ALLOW_NO_AUTH"):
|
||
monkeypatch.delenv(key, raising=False)
|
||
try:
|
||
SessionManager.from_environment()
|
||
except RuntimeError:
|
||
pass
|
||
else:
|
||
raise AssertionError("未配置密码时应 fail-secure 拒绝启动")
|
||
|
||
|
||
def test_session_manager_allow_no_auth(monkeypatch):
|
||
from app.auth import SessionManager
|
||
monkeypatch.delenv("PPM_PASSWORD", raising=False)
|
||
monkeypatch.delenv("PPM_PASSWORD_HASH", raising=False)
|
||
monkeypatch.setenv("PPM_ALLOW_NO_AUTH", "1")
|
||
assert SessionManager.from_environment().enabled is False
|
||
|
||
|
||
def test_login_rate_limiting():
|
||
from app.auth import SessionManager
|
||
manager = SessionManager(make_password_hash("secret"), max_failures=3, lockout_seconds=60)
|
||
for _ in range(3):
|
||
assert manager.login("wrong", "1.2.3.4") is None
|
||
assert manager.is_login_locked("1.2.3.4") is True
|
||
# 正确密码在锁定期也应被拒
|
||
assert manager.login("secret", "1.2.3.4") is None
|