zkxw/backend/tests/test_oa_instance_comment.py

"""OA 实例评论：附件枚举与引用校验"""
import json
import unittest
from datetime import datetime

from app.services.oa_instance_comment import (
    enumerate_form_attachments,
    normalize_attachment_refs,
)


class TestEnumerateFormAttachments(unittest.TestCase):
    def test_groups_upload_fields(self):
        schema = {
            "rule": [
                {
                    "type": "upload",
                    "field": "files",
                    "title": "附件",
                },
                {"type": "input", "field": "name", "title": "名称"},
            ]
        }
        form = {
            "files": [
                {"url": "https://x/a.pdf", "name": "a.pdf"},
                {"url": "https://x/b.pdf", "name": "b.pdf"},
            ],
            "name": "demo",
        }
        items = enumerate_form_attachments(schema, form)
        self.assertEqual(len(items), 1)
        self.assertEqual(items[0]["fieldKey"], "files")
        self.assertEqual(items[0]["fieldLabel"], "附件")
        self.assertEqual(len(items[0]["files"]), 2)
        self.assertEqual(items[0]["files"][0]["fileName"], "a.pdf")

    def test_skips_empty_upload(self):
        schema = {"rule": [{"type": "fcUpload", "field": "f", "title": "F"}]}
        self.assertEqual(enumerate_form_attachments(schema, {"f": []}), [])


class TestNormalizeAttachmentRefs(unittest.TestCase):
    def test_accepts_valid_ref(self):
        schema = {"rule": [{"type": "upload", "field": "files", "title": "附件"}]}
        form = {"files": [{"url": "https://x/a.pdf", "name": "a.pdf"}]}
        out = normalize_attachment_refs(
            schema,
            form,
            [{"fieldKey": "files", "url": "https://x/a.pdf", "fileName": "wrong"}],
        )
        self.assertEqual(out[0]["fileName"], "a.pdf")
        self.assertEqual(out[0]["fieldLabel"], "附件")

    def test_rejects_foreign_url(self):
        from app.services.oa_engine import OaEngineError

        schema = {"rule": [{"type": "upload", "field": "files", "title": "附件"}]}
        form = {"files": [{"url": "https://x/a.pdf", "name": "a.pdf"}]}
        with self.assertRaises(OaEngineError):
            normalize_attachment_refs(
                schema,
                form,
                [{"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):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker
        from app.models.models import OaAppVersion, OaInstance, OaInstanceComment, SysUser

        self.engine = create_engine("sqlite:///:memory:")
        OaAppVersion.__table__.create(self.engine)
        OaInstance.__table__.create(self.engine)
        OaInstanceComment.__table__.create(self.engine)
        SysUser.__table__.create(self.engine)
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        now = datetime.now()
        schema = {
            "rule": [
                {"type": "upload", "field": "files", "title": "附件"},
            ]
        }
        self.db.add(
            OaAppVersion(
                id="v1",
                app_id="app1",
                version_no=1,
                status="published",
                is_latest_published="1",
                form_schema_json=json.dumps(schema, ensure_ascii=False),
                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=json.dumps(
                    {"files": [{"url": "https://x/a.pdf", "name": "a.pdf"}]},
                    ensure_ascii=False,
                ),
                tenant_no="zkxw",
                create_dept_id="d1",
                create_time=now,
                update_time=now,
                del_flag=0,
                current_round=1,
            )
        )
        self.db.add(
            SysUser(
                id="u1",
                tenant_no="zkxw",
                user_no="u1",
                nick_name="张三",
                email="",
                phone_no="",
                password="",
                status="1",
                create_dept_id="",
                del_flag=0,
                create_time=now,
                update_time=now,
            )
        )
        self.db.commit()

    def test_create_list_revoke(self):
        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=[{"fieldKey": "files", "url": "https://x/a.pdf"}],
        )
        self.assertEqual(row["status"], "normal")
        self.assertTrue(row["canRevoke"])
        self.assertEqual(len(row["attachmentRefs"]), 1)

        listed = svc.list_comments(
            self.db, instance_id="i1", tenant_no="zkxw", viewer_user_id="u1"
        )
        self.assertEqual(len(listed), 1)

        revoked = svc.revoke_comment(
            self.db,
            instance_id="i1",
            comment_id=row["id"],
            tenant_no="zkxw",
            user_id="u1",
        )
        self.assertEqual(revoked["status"], "revoked")
        self.assertFalse(revoked["canRevoke"])

        again = svc.revoke_comment(
            self.db,
            instance_id="i1",
            comment_id=row["id"],
            tenant_no="zkxw",
            user_id="u1",
        )
        self.assertEqual(again["status"], "revoked")

    def test_cannot_comment_when_not_running(self):
        from app.services import oa_instance_comment as svc
        from app.services.oa_engine import OaEngineError
        from app.models.models import OaInstance

        inst = self.db.query(OaInstance).filter(OaInstance.id == "i1").one()
        inst.status = "completed"
        self.db.commit()
        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,
            )


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"],
        )