test: isolate database settings from production

.codex/memory/MEMORY.md

@@ -8,6 +8,7 @@
 - [合批原料字段历史问题](project_material_ofrom_merge_bug.md)：`material_ofrom` 不一致的既有排查结论。
 - [前端独立发版流程](reference_ehs_web_release.md)：发布 `ehs_web` 时使用。
 - [项目 Python 虚拟环境](reference_python_venv.md)：运行 Django、pytest 和脚本时使用。
+- [项目测试数据库](reference_test_database.md)：测试与可切换的生产查询连接解耦，始终使用固定的 `test_ehs_develop`。
 - [两个 WebView 套壳 App](reference_wrapper_apps.md)：修改 h5x 与扫码、返回键交互时使用。
 
 这些文件记录的是长期约定或历史上下文。执行任务前应结合当前代码和数据重新验证，尤其不要把历史缺陷结论直接当成当前故障原因。

.codex/memory/reference_test_database.md

@@ -0,0 +1,15 @@
+# 项目测试数据库
+
+- 生产问题查询时，默认数据库连接可以根据工厂或环境切换到不同 IP 和业务库；这类连接只用于授权范围内的生产数据只读验证。
+- 所有 Django、pytest 及其他自动化测试必须与当前生产查询连接解耦，始终使用固定的 `test_ehs_develop` 测试数据库连接。
+- 不能仅依赖当前默认连接的 Django 自动 `test_<NAME>` 命名；即使当前业务库是 `bxerp` 或其他库，测试也不应转而使用 `test_bxerp` 或其他派生库。
+- 不得在生产数据库上运行会建表、迁移、写入或清理数据的测试。
+- 生产查询与固定测试库的连接参数均从本机已忽略配置中读取，不在项目记忆、源码或提交信息中记录凭据。
+
+当前实现：
+
+- 固定测试连接保存在已忽略的 `config/conf_test.py`。
+- `server/test_settings.py` 加载该连接，`manage.py test` 会自动选择测试 settings；其他 `manage.py` 命令仍使用当前业务库连接。
+- 测试 settings 中的基础 `NAME` 和 `TEST.NAME` 都必须是 `test_ehs_develop`，并在启动时校验，防止通过测试 settings 误操作 `ehs_develop` 或任何生产库。
+- pytest-django 通过项目根目录 `pytest.ini` 固定使用 `server.test_settings`。
+- 本地重复运行测试时优先使用 `.venv\\Scripts\\python.exe manage.py test --keepdb --noinput`。

apps/wpm/serializers.py

@@ -1265,7 +1265,26 @@ class Handoverbwserializer(CustomModelSerializer):
         read_only_fields = EXCLUDE_FIELDS_BASE + ["handoverb", "number"]
         extra_kwargs = {'wpr': {'required': True}}
 
+
+class CachedWMaterialPrimaryKeyRelatedField(serializers.PrimaryKeyRelatedField):
+    def to_internal_value(self, data):
+        cache = getattr(self.root, "_handover_wmaterial_cache", None)
+        if cache is None:
+            return super().to_internal_value(data)
+        if not isinstance(data, (str, int)):
+            self.fail("incorrect_type", data_type=type(data).__name__)
+        try:
+            return cache[str(data)]
+        except KeyError:
+            self.fail("does_not_exist", pk_value=data)
+
+
 class HandoverbSerializer(CustomModelSerializer):
+    wm = CachedWMaterialPrimaryKeyRelatedField(
+        queryset=WMaterial.objects.select_related(
+            "material", "defect", "mgroup", "belong_dept"
+        )
+    )
     notok_sign = serializers.CharField(source='wm.notok_sign', read_only=True)
     defect_name = serializers.CharField(source="wm.defect.name", read_only=True)
     handoverbw = Handoverbwserializer(many=True, required=False)
@@ -1302,6 +1321,27 @@ class HandoverSerializer(CustomModelSerializer):
     handoverb = HandoverbSerializer(many=True, required=False)
     ticket_ = TicketSimpleSerializer(source='ticket', read_only=True)
 
+    def to_internal_value(self, data):
+        handoverb = data.get("handoverb", []) if hasattr(data, "get") else []
+        wm_ids = {
+            str(item["wm"])
+            for item in handoverb
+            if isinstance(item, dict) and item.get("wm") is not None
+        }
+        if not wm_ids:
+            return super().to_internal_value(data)
+
+        queryset = WMaterial.objects.select_related(
+            "material", "defect", "mgroup", "belong_dept"
+        )
+        self._handover_wmaterial_cache = {
+            str(pk): instance for pk, instance in queryset.in_bulk(wm_ids).items()
+        }
+        try:
+            return super().to_internal_value(data)
+        finally:
+            del self._handover_wmaterial_cache
+    
     def validate(self, attrs):
         if "mtype" not in attrs:
             attrs['mtype'] = Handover.H_NORMAL

apps/wpm/tests/test_handover_serializer.py

@@ -0,0 +1,58 @@
+from django.db import connection
+from django.test import TestCase
+from django.test.utils import CaptureQueriesContext
+
+from apps.mtm.models import Material
+from apps.qm.models import Defect
+from apps.system.models import Dept, User
+from apps.wpm.models import Handover, WMaterial
+from apps.wpm.serializers import HandoverSerializer
+
+
+class HandoverSerializerQueryTests(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.dept = Dept.objects.create(name="合批查询测试车间")
+        cls.user = User.objects.create_user(username="handover-query-user")
+        cls.material = Material.objects.create(name="合批查询测试物料")
+        cls.defect_b = Defect.objects.create(
+            name="合批查询测试B类缺陷",
+            cate=Defect.cate_list[0],
+            okcate=Defect.DEFECT_OK_B,
+        )
+        cls.inventories = [
+            WMaterial.objects.create(
+                material=cls.material,
+                batch=f"QUERY-{index}",
+                count=1,
+                state=WMaterial.WM_OK,
+                defect=cls.defect_b if index == 2 else None,
+                belong_dept=cls.dept,
+            )
+            for index in range(3)
+        ]
+
+    def test_handover_inventory_is_loaded_in_one_query(self):
+        serializer = HandoverSerializer(data={
+            "send_date": "2026-08-07",
+            "send_user": self.user.id,
+            "send_dept": self.dept.id,
+            "recive_dept": self.dept.id,
+            "handoverb": [
+                {"wm": inventory.id, "count": 1}
+                for inventory in self.inventories
+            ],
+            "new_batch": "QUERY-MERGED",
+            "clear_defect": True,
+            "type": Handover.H_NORMAL,
+            "mtype": Handover.H_MERGE,
+        })
+
+        with CaptureQueriesContext(connection) as queries:
+            self.assertTrue(serializer.is_valid(), serializer.errors)
+
+        inventory_queries = [
+            query["sql"] for query in queries
+            if 'FROM "wpm_wmaterial"' in query["sql"]
+        ]
+        self.assertEqual(len(inventory_queries), 1, inventory_queries)

manage.py

@@ -5,7 +5,12 @@ import sys
 
 
 def main():
-    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'server.settings')
+    settings_module = (
+        'server.test_settings'
+        if sys.argv[1:2] == ['test']
+        else 'server.settings'
+    )
+    os.environ.setdefault('DJANGO_SETTINGS_MODULE', settings_module)
     try:
         from django.core.management import execute_from_command_line
     except ImportError as exc:

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+DJANGO_SETTINGS_MODULE = server.test_settings
+python_files = tests.py test_*.py *_tests.py

server/test_settings.py

@@ -0,0 +1,16 @@
+from server.settings import *  # noqa: F403
+
+from config.conf_test import TEST_DATABASES
+from django.core.exceptions import ImproperlyConfigured
+
+
+DATABASES = TEST_DATABASES
+
+if (
+    DATABASES["default"].get("NAME") != "test_ehs_develop"
+    or DATABASES["default"].get("TEST", {}).get("NAME")
+    != "test_ehs_develop"
+):
+    raise ImproperlyConfigured(
+        "Test settings must only use the test_ehs_develop database."
+    )