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


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