PMFlow/backend/tests/test_task_persistence.py

81 lines
2.5 KiB
Python

import unittest
from app.models import SkillTag
from app.services.task_persistence import (
build_task_create_draft,
resolve_skill_tag_ids_by_names,
)
class _ScalarRows:
def __init__(self, rows: list[object]) -> None:
self._rows = rows
def scalars(self):
return self
def all(self):
return self._rows
class _FakeDB:
def __init__(self, rows: list[object]) -> None:
self._rows = rows
async def execute(self, *_args, **_kwargs):
return _ScalarRows(self._rows)
class TaskPersistenceTest(unittest.IsolatedAsyncioTestCase):
def test_build_task_create_draft_normalizes_fields(self) -> None:
draft = build_task_create_draft(
title=" 补齐统一通知入口 ",
required_skill_tag_ids=["1", 2, "2"],
priority="P0",
status="pending",
estimated_hours="4.5",
due_date="2026-03-07T10:00:00Z",
)
self.assertEqual(draft.title, "补齐统一通知入口")
self.assertEqual(draft.required_skill_tag_ids, [1, 2])
self.assertEqual(draft.priority, "high")
self.assertEqual(draft.status, "todo")
self.assertEqual(draft.estimated_hours, 4.5)
self.assertIsNotNone(draft.due_date)
self.assertIsNone(draft.due_date.tzinfo)
async def test_resolve_skill_tag_ids_by_names_preserves_unknowns_for_audit(self) -> None:
db = _FakeDB(
[
SkillTag(id=11, tenant_id=1, name="系统设计", slug="system-design", category="skill"),
SkillTag(id=12, tenant_id=1, name="开发", slug="development", category="skill"),
]
)
resolution = await resolve_skill_tag_ids_by_names(
db,
tenant_id=1,
skill_names=["系统设计", "开发", "未知技能"],
)
self.assertEqual(resolution.matched_ids, [11, 12])
self.assertEqual(resolution.unmatched_names, ["未知技能"])
self.assertEqual(
resolution.matched_items,
[
{
"input_name": "系统设计",
"skill_tag_id": 11,
"skill_tag_name": "系统设计",
"skill_tag_slug": "system-design",
},
{
"input_name": "开发",
"skill_tag_id": 12,
"skill_tag_name": "开发",
"skill_tag_slug": "development",
},
],
)