zkxw/backend/tests/test_oa_mention_unread.py

"""OA 提到我的未读计数 / 标记已读 / 再@重置。"""
import unittest
from datetime import datetime
from unittest.mock import patch


class TestMentionUnread(unittest.TestCase):
    def setUp(self):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker
        from app.models.models import (
            OaAppVersion,
            OaInstance,
            OaInstanceComment,
            OaInstanceViewGrant,
            SysUser,
        )

        self.engine = create_engine("sqlite:///:memory:")
        for table in (
            OaAppVersion.__table__,
            OaInstance.__table__,
            OaInstanceComment.__table__,
            OaInstanceViewGrant.__table__,
            SysUser.__table__,
        ):
            table.create(self.engine)
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        now = datetime.now()
        self.db.add(
            OaAppVersion(
                id="v1",
                app_id="app1",
                version_no=1,
                status="published",
                is_latest_published="1",
                form_schema_json="{}",
                flow_tree_json="{}",
                advanced_settings_json="{}",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaInstance(
                id="i1",
                app_id="app1",
                app_version_id="v1",
                title="t",
                status="running",
                initiator_user_id="u0",
                initiator_dept_id="d1",
                form_data_json="{}",
                tenant_no="zkxw",
                create_dept_id="d1",
                create_time=now,
                update_time=now,
                del_flag=0,
                current_round=1,
            )
        )
        for uid, name in (("u1", "张三"), ("u2", "李四")):
            self.db.add(
                SysUser(
                    id=uid,
                    tenant_no="zkxw",
                    user_no=uid,
                    nick_name=name,
                    email="",
                    phone_no="",
                    password="",
                    status="1",
                    create_dept_id="",
                    del_flag=0,
                    create_time=now,
                    update_time=now,
                )
            )
        self.db.commit()

    def tearDown(self):
        self.db.close()
        self.engine.dispose()

    def test_mention_starts_unread_and_count(self):
        from app.services import oa_instance_comment as svc

        with patch("app.services.oa_wechat_notify.schedule_oa_mention_notifies"):
            svc.create_comment(
                self.db,
                instance_id="i1",
                tenant_no="zkxw",
                user_id="u1",
                dept_id="d1",
                content="@你",
                attachment_refs=None,
                mention_user_ids=["u2"],
            )
        self.assertEqual(
            svc.count_mention_unread(self.db, tenant_no="zkxw", user_id="u2"), 1
        )
        self.assertEqual(
            svc.count_mention_unread(self.db, tenant_no="zkxw", user_id="u1"), 0
        )

    def test_mark_read_clears_unread(self):
        from app.services import oa_instance_comment as svc

        with patch("app.services.oa_wechat_notify.schedule_oa_mention_notifies"):
            svc.create_comment(
                self.db,
                instance_id="i1",
                tenant_no="zkxw",
                user_id="u1",
                dept_id="d1",
                content="x",
                attachment_refs=None,
                mention_user_ids=["u2"],
            )
        out = svc.mark_mention_instance_read(
            self.db, tenant_no="zkxw", user_id="u2", instance_id="i1"
        )
        self.assertTrue(out.get("isRead"))
        self.assertEqual(
            svc.count_mention_unread(self.db, tenant_no="zkxw", user_id="u2"), 0
        )

    def test_remention_resets_unread(self):
        from app.services import oa_instance_comment as svc

        with patch("app.services.oa_wechat_notify.schedule_oa_mention_notifies"):
            svc.create_comment(
                self.db,
                instance_id="i1",
                tenant_no="zkxw",
                user_id="u1",
                dept_id="d1",
                content="第一次",
                attachment_refs=None,
                mention_user_ids=["u2"],
            )
            svc.mark_mention_instance_read(
                self.db, tenant_no="zkxw", user_id="u2", instance_id="i1"
            )
            self.assertEqual(
                svc.count_mention_unread(self.db, tenant_no="zkxw", user_id="u2"), 0
            )
            svc.create_comment(
                self.db,
                instance_id="i1",
                tenant_no="zkxw",
                user_id="u1",
                dept_id="d1",
                content="再提",
                attachment_refs=None,
                mention_user_ids=["u2"],
            )
        self.assertEqual(
            svc.count_mention_unread(self.db, tenant_no="zkxw", user_id="u2"), 1
        )


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