zlwuandCursor
4654678231
ACL 表与 sys_user JOIN/条件比较统一 utf8mb4_unicode_ci，并补建表 COLLATE 与 CONVERT 修复脚本（#97）。 Co-authored-by: Cursor <cursoragent@cursor.com>
102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
"""储存证质检字段 ACL：合法字段与 scope。"""
|
|
import unittest
|
|
from unittest.mock import MagicMock
|
|
|
|
from sqlalchemy.dialects import mysql
|
|
|
|
from app.crud import sample_storage_cert_qc_acl as acl_svc
|
|
from app.models.models import BizStorageCertQcAcl, BizStorageCertQcAclUser, SysUser
|
|
from app.services.sample_storage_cert_qc import (
|
|
ACL_SCOPE_PRODUCTION,
|
|
ACL_SCOPE_QC,
|
|
ALL_STORAGE_CERT_QC_FIELD_KEYS,
|
|
SAMPLE_TYPE_KEY_UC_MSC,
|
|
dumps_field_keys,
|
|
legal_field_keys_for_type,
|
|
loads_field_keys,
|
|
)
|
|
|
|
|
|
class TestAclFieldKeys(unittest.TestCase):
|
|
def test_uc_msc_excludes_mother_nucleic_acids(self):
|
|
keys = legal_field_keys_for_type(SAMPLE_TYPE_KEY_UC_MSC)
|
|
self.assertNotIn("mother_hbv_dna", keys)
|
|
self.assertIn("mother_hbsag", keys)
|
|
|
|
def test_roundtrip_field_keys(self):
|
|
raw = dumps_field_keys(["receive_time", "bacteria_test", "receive_time", ""])
|
|
self.assertEqual(loads_field_keys(raw), ["bacteria_test", "receive_time"])
|
|
|
|
def test_all_keys_cover_common_fields(self):
|
|
self.assertIn("receive_time", ALL_STORAGE_CERT_QC_FIELD_KEYS)
|
|
self.assertIn("mother_anti_hiv", ALL_STORAGE_CERT_QC_FIELD_KEYS)
|
|
|
|
def test_scopes(self):
|
|
self.assertEqual(ACL_SCOPE_QC, "qc")
|
|
self.assertEqual(ACL_SCOPE_PRODUCTION, "production")
|
|
|
|
|
|
class TestAclUserJoinCollation(unittest.TestCase):
|
|
def test_detail_user_join_uses_unicode_collate(self):
|
|
"""JOIN sys_user 时统一 utf8mb4_unicode_ci，避免 1267 mix of collations。"""
|
|
acl = MagicMock()
|
|
acl.id = "acl-1"
|
|
acl.field_keys = "[]"
|
|
acl.configured_flag = "1"
|
|
|
|
captured = {}
|
|
|
|
class _Query:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def filter(self, *args, **kwargs):
|
|
captured.setdefault("filters", []).extend(args)
|
|
return self
|
|
|
|
def first(self):
|
|
return acl
|
|
|
|
def outerjoin(self, *args, **kwargs):
|
|
captured["join_on"] = args[1] if len(args) > 1 else None
|
|
return self
|
|
|
|
def order_by(self, *args, **kwargs):
|
|
return self
|
|
|
|
def all(self):
|
|
return []
|
|
|
|
db = MagicMock()
|
|
db.query.side_effect = lambda *args, **kwargs: _Query()
|
|
|
|
detail = acl_svc.get_acl_detail(db, "t1", "qc")
|
|
self.assertEqual(detail["scope"], "qc")
|
|
join_on = captured.get("join_on")
|
|
self.assertIsNotNone(join_on)
|
|
sql = str(
|
|
join_on.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})
|
|
).lower()
|
|
self.assertIn("utf8mb4_unicode_ci", sql)
|
|
filter_sql = " ".join(
|
|
str(f.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})).lower()
|
|
for f in captured.get("filters", [])
|
|
)
|
|
self.assertIn("utf8mb4_unicode_ci", filter_sql)
|
|
self.assertIn("sys_user", SysUser.__tablename__)
|
|
self.assertIn("biz_storage_cert_qc_acl_user", BizStorageCertQcAclUser.__tablename__)
|
|
self.assertEqual(BizStorageCertQcAcl.__tablename__, "biz_storage_cert_qc_acl")
|
|
|
|
def test_eq_unicode_literal_only_collates_column(self):
|
|
from app.core.db_collation import eq_unicode
|
|
|
|
expr = eq_unicode(BizStorageCertQcAcl.tenant_no, "t1")
|
|
sql = str(
|
|
expr.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})
|
|
).lower()
|
|
self.assertIn("utf8mb4_unicode_ci", sql)
|
|
self.assertEqual(sql.count("utf8mb4_unicode_ci"), 1)
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|