提交

backend/tests/test_oa_instance_comment.py

@@ -64,6 +64,83 @@ class TestNormalizeAttachmentRefs(unittest.TestCase):
                 [{"fieldKey": "files", "url": "https://evil/x.pdf"}],
             )
 
+    def test_marks_form_ref_source(self):
+        schema = {"rule": [{"type": "upload", "field": "files", "title": "附件"}]}
+        form = {"files": [{"url": "https://x/a.pdf", "name": "a.pdf"}]}
+        out = normalize_attachment_refs(
+            schema,
+            form,
+            [{"url": "https://x/a.pdf"}],
+        )
+        self.assertEqual(out[0]["source"], "form")
+
+    def test_accepts_tenant_upload_and_uses_stored_name(self):
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+        from app.models.models import BizAttachment
+
+        engine = create_engine("sqlite:///:memory:")
+        BizAttachment.__table__.create(engine)
+        db = sessionmaker(bind=engine)()
+        db.add(
+            BizAttachment(
+                id="att1",
+                original_name="note.pdf",
+                file_url="oa-approval/note.pdf",
+                tenant_no="t1",
+                create_dept_id="",
+            )
+        )
+        db.commit()
+        out = normalize_attachment_refs(
+            None,
+            {},
+            [{"source": "upload", "url": "oa-approval/note.pdf", "fileName": "hack.pdf"}],
+            db=db,
+            tenant_no="t1",
+        )
+        self.assertEqual(out[0]["source"], "upload")
+        self.assertEqual(out[0]["fileName"], "note.pdf")
+        self.assertEqual(out[0]["fieldKey"], "")
+        db.close()
+
+    def test_rejects_upload_from_other_tenant_or_unknown_url(self):
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+        from app.models.models import BizAttachment
+        from app.services.oa_engine import OaEngineError
+
+        engine = create_engine("sqlite:///:memory:")
+        BizAttachment.__table__.create(engine)
+        db = sessionmaker(bind=engine)()
+        db.add(
+            BizAttachment(
+                id="att2",
+                original_name="note.pdf",
+                file_url="oa-approval/note.pdf",
+                tenant_no="other",
+                create_dept_id="",
+            )
+        )
+        db.commit()
+        with self.assertRaises(OaEngineError):
+            normalize_attachment_refs(
+                None,
+                {},
+                [{"source": "upload", "url": "oa-approval/note.pdf"}],
+                db=db,
+                tenant_no="t1",
+            )
+        with self.assertRaises(OaEngineError):
+            normalize_attachment_refs(
+                None,
+                {},
+                [{"source": "upload", "url": "https://evil.example/a.pdf"}],
+                db=db,
+                tenant_no="t1",
+            )
+        db.close()
+
 
 class TestCommentCrud(unittest.TestCase):
     def setUp(self):
@@ -198,3 +275,253 @@ class TestCommentCrud(unittest.TestCase):
                 content="x",
                 attachment_refs=None,
             )
+
+
+class TestCommentMentionGrant(unittest.TestCase):
+    def setUp(self):
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+        from app.models.models import (
+            OaAppVersion,
+            OaInstance,
+            OaInstanceComment,
+            OaInstanceViewGrant,
+            OaTask,
+            SysUser,
+        )
+
+        self.engine = create_engine("sqlite:///:memory:")
+        for table in (
+            OaAppVersion.__table__,
+            OaInstance.__table__,
+            OaInstanceComment.__table__,
+            OaInstanceViewGrant.__table__,
+            OaTask.__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, status in (("u1", "张三", "1"), ("u2", "李四", "1"), ("u3", "王五", "0")):
+            self.db.add(
+                SysUser(
+                    id=uid,
+                    tenant_no="zkxw",
+                    user_no=uid,
+                    nick_name=name,
+                    email="",
+                    phone_no="",
+                    password="",
+                    status=status,
+                    create_dept_id="",
+                    del_flag=0,
+                    create_time=now,
+                    update_time=now,
+                )
+            )
+        self.db.commit()
+
+    def test_mention_grants_view_and_is_idempotent(self):
+        from app.models.models import OaInstanceViewGrant
+        from app.services import oa_instance_comment as svc
+
+        row = 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", "u2", "u1"],
+        )
+        self.assertEqual(row["content"], "")
+        self.assertEqual([m["userId"] for m in row["mentions"]], ["u2"])
+        self.assertEqual(row["mentions"][0]["userName"], "李四")
+        grants = (
+            self.db.query(OaInstanceViewGrant)
+            .filter(OaInstanceViewGrant.del_flag == 0)
+            .all()
+        )
+        self.assertEqual(len(grants), 1)
+        self.assertEqual(grants[0].user_id, "u2")
+        self.assertEqual(grants[0].source, "comment_mention")
+
+        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"],
+        )
+        grants = (
+            self.db.query(OaInstanceViewGrant)
+            .filter(OaInstanceViewGrant.del_flag == 0)
+            .all()
+        )
+        self.assertEqual(len(grants), 1)
+
+    def test_rejects_empty_without_mention(self):
+        from app.services import oa_instance_comment as svc
+        from app.services.oa_engine import OaEngineError
+
+        with self.assertRaises(OaEngineError):
+            svc.create_comment(
+                self.db,
+                instance_id="i1",
+                tenant_no="zkxw",
+                user_id="u1",
+                dept_id="d1",
+                content="  ",
+                attachment_refs=None,
+                mention_user_ids=None,
+            )
+
+    def test_allows_upload_only_comment(self):
+        from app.models.models import BizAttachment
+        from app.services import oa_instance_comment as svc
+
+        BizAttachment.__table__.create(self.engine)
+        self.db.add(
+            BizAttachment(
+                id="att-c1",
+                original_name="note.pdf",
+                file_url="oa-approval/note.pdf",
+                tenant_no="zkxw",
+                create_dept_id="",
+            )
+        )
+        self.db.commit()
+        row = svc.create_comment(
+            self.db,
+            instance_id="i1",
+            tenant_no="zkxw",
+            user_id="u1",
+            dept_id="d1",
+            content="  ",
+            attachment_refs=[{"source": "upload", "url": "oa-approval/note.pdf"}],
+            mention_user_ids=None,
+        )
+        self.assertEqual(row["content"], "")
+        self.assertEqual(row["attachmentRefs"][0]["source"], "upload")
+        self.assertEqual(row["attachmentRefs"][0]["fileName"], "note.pdf")
+
+    def test_rejects_disabled_user(self):
+        from app.services import oa_instance_comment as svc
+        from app.services.oa_engine import OaEngineError
+
+        with self.assertRaises(OaEngineError):
+            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=["u3"],
+            )
+
+    def test_revoke_keeps_grant_and_list_skips_participants(self):
+        from app.models.models import OaInstance, OaInstanceViewGrant, OaTask
+        from app.services import oa_instance_comment as svc
+
+        row = 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.revoke_comment(
+            self.db,
+            instance_id="i1",
+            comment_id=row["id"],
+            tenant_no="zkxw",
+            user_id="u1",
+        )
+        self.assertEqual(
+            self.db.query(OaInstanceViewGrant)
+            .filter(OaInstanceViewGrant.del_flag == 0)
+            .count(),
+            1,
+        )
+        self.assertEqual(
+            svc.list_mentioned_instance_ids(self.db, tenant_no="zkxw", user_id="u2"),
+            ["i1"],
+        )
+
+        now = datetime.now()
+        self.db.add(
+            OaTask(
+                id="t-cc",
+                instance_id="i1",
+                task_type="cc",
+                assignee_user_id="u2",
+                status="done",
+                tenant_no="zkxw",
+                create_dept_id="",
+                create_time=now,
+                update_time=now,
+                del_flag=0,
+                round_no=1,
+                is_read=0,
+            )
+        )
+        self.db.commit()
+        # 即使已是抄送人，被@仍应出现在「提到我的」
+        self.assertEqual(
+            svc.list_mentioned_instance_ids(self.db, tenant_no="zkxw", user_id="u2"),
+            ["i1"],
+        )
+
+        inst = self.db.query(OaInstance).filter(OaInstance.id == "i1").one()
+        inst.initiator_user_id = "u2"
+        self.db.commit()
+        # 即使已是发起人，被@仍应出现在「提到我的」
+        self.assertEqual(
+            svc.list_mentioned_instance_ids(self.db, tenant_no="zkxw", user_id="u2"),
+            ["i1"],
+        )