zkxw/backend/tests/test_oa_cc_read.py

"""OA 抄送已读：序列化、筛选语义、标记幂等"""
import unittest
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import MagicMock

from app.services import oa_engine as eng


class TestTaskToDictIsRead(unittest.TestCase):
    def test_includes_is_read_and_read_time(self):
        t = SimpleNamespace(
            id="t1",
            instance_id="i1",
            node_key="n1",
            node_name="抄送",
            task_type="cc",
            assignee_user_id="u1",
            status="done",
            comment_text=None,
            resolve_source=None,
            sign_mode=None,
            is_read=1,
            read_time=datetime(2026, 8, 6, 12, 0, 0),
            create_time=datetime(2026, 8, 6, 10, 0, 0),
            update_time=datetime(2026, 8, 6, 11, 0, 0),
        )
        d = eng._task_to_dict(t)
        self.assertEqual(d["isRead"], True)
        self.assertEqual(d["readTime"], "2026-08-06 12:00:00")

    def test_unread_defaults(self):
        t = SimpleNamespace(
            id="t2",
            instance_id="i1",
            node_key="n1",
            node_name="抄送",
            task_type="cc",
            assignee_user_id="u1",
            status="done",
            comment_text=None,
            resolve_source=None,
            sign_mode=None,
            is_read=0,
            read_time=None,
            create_time=None,
            update_time=None,
        )
        d = eng._task_to_dict(t)
        self.assertEqual(d["isRead"], False)
        self.assertEqual(d["readTime"], "")


class TestMarkCcReadHelpers(unittest.TestCase):
    def test_normalize_read_status(self):
        self.assertEqual(eng._normalize_cc_read_status(None), "unread")
        self.assertEqual(eng._normalize_cc_read_status(""), "unread")
        self.assertEqual(eng._normalize_cc_read_status("all"), "all")
        self.assertEqual(eng._normalize_cc_read_status("read"), "read")
        with self.assertRaises(eng.OaEngineError):
            eng._normalize_cc_read_status("weird")

    def test_mark_idempotent_keeps_read_time(self):
        first = datetime(2026, 8, 1, 9, 0, 0)
        task = SimpleNamespace(
            id="t1",
            task_type="cc",
            assignee_user_id="u1",
            del_flag=0,
            is_read=1,
            read_time=first,
            update_by=None,
            update_time=None,
        )
        db = MagicMock()
        out = eng._apply_cc_task_read(db, task=task, user_id="u1", now=datetime(2026, 8, 6, 12, 0, 0))
        self.assertFalse(out)  # 未发生更新
        self.assertEqual(task.read_time, first)
        self.assertEqual(task.is_read, 1)


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