fix: support linked planning dual filters

tests/test_planning_dual_filter.py

@@ -0,0 +1,175 @@
+"""计划预测“产品类型 + 跟单小组”组合筛选回归测试。"""
+
+import json
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+APP_JS = (ROOT / "static" / "js" / "app.js").read_text(encoding="utf-8")
+
+
+class PlanningDualFilterFrontendTest(unittest.TestCase):
+    def test_product_type_options_come_from_linked_backend_query(self):
+        self.assertIn("if (field === 'product_type')", APP_JS)
+        self.assertIn(
+            "fetch('/api/planning/column-values?field=product_type'",
+            APP_JS,
+        )
+        self.assertIn("var linkedProductTypeFilters = _getPlanLinkedFilters(field);", APP_JS)
+
+    def test_follower_dropdown_comes_from_all_linked_rows(self):
+        self.assertIn(
+            "if (field === '采购跟单' || field === '采购计划')",
+            APP_JS,
+        )
+        self.assertIn(
+            "fetch('/api/planning/column-values?field=' + encodeURIComponent(field)",
+            APP_JS,
+        )
+
+    def test_top_filters_are_included_in_linked_filter_state(self):
+        self.assertIn("function _getPlanLinkedFilters(field)", APP_JS)
+        self.assertIn("linked['采购跟单'] = {type: 'list', values: [follower]};", APP_JS)
+
+
+class PlanningDualFilterApiTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        # test_wsgi supplies the isolated local-test database credentials.
+        import test_wsgi  # noqa: F401
+        from app import app
+        from routes.planning import _personnel_group_for_value, _personnel_group_members
+        from services.db import query
+
+        cls.query = staticmethod(query)
+        cls.group_for_value = staticmethod(_personnel_group_for_value)
+        cls.groups = _personnel_group_members()
+        cls.app = app
+        cls.app.config.update(TESTING=True)
+        cls.user_id = query(
+            "SELECT id FROM dashboard_users WHERE status=1 ORDER BY id LIMIT 1"
+        )[0]["id"]
+
+        intersections = {}
+        rows = query(
+            """
+            SELECT product_type, procurement_follower
+            FROM products
+            WHERE deleted_at IS NULL
+              AND product_type NOT IN ('包材', '其他')
+              AND product_code NOT IN (SELECT product_code FROM plan_hidden)
+              AND procurement_follower IS NOT NULL
+              AND procurement_follower != ''
+            """
+        )
+        for row in rows:
+            group = cls.group_for_value(row["procurement_follower"], cls.groups)
+            if group:
+                key = (row["product_type"], group)
+                intersections[key] = intersections.get(key, 0) + 1
+        if not intersections:
+            raise unittest.SkipTest("测试库没有产品类型与跟单小组的交集数据")
+        cls.product_type, cls.group = max(intersections, key=intersections.get)
+
+    def setUp(self):
+        self.client = self.app.test_client()
+        with self.client.session_transaction() as session:
+            session.update(
+                user_id=self.user_id,
+                username="planning-filter-test",
+                real_name="planning-filter-test",
+                role="super_admin",
+                must_change_password=False,
+            )
+
+    def _planning_list(self, filters=None, follower=""):
+        query_string = {"page": 1, "size": 100}
+        if filters:
+            query_string.update(
+                custom_field="__multi__",
+                custom_value="__multi__:" + json.dumps(filters, ensure_ascii=False),
+            )
+        if follower:
+            query_string["procurement_follower"] = follower
+        response = self.client.get("/api/planning/list", query_string=query_string)
+        self.assertEqual(response.status_code, 200)
+        body = response.get_json()
+        self.assertEqual(body["code"], 0)
+        return body["data"]
+
+    def test_header_filters_return_exact_intersection(self):
+        type_filter = {
+            "field": "product_type",
+            "type": "list",
+            "values": [self.product_type],
+        }
+        group_filter = {
+            "field": "采购跟单",
+            "type": "list",
+            "values": [self.group],
+        }
+        by_type = self._planning_list([type_filter])
+        by_group = self._planning_list([group_filter])
+        combined = self._planning_list([type_filter, group_filter])
+
+        expected = set(by_type["all_codes"]) & set(by_group["all_codes"])
+        self.assertEqual(set(combined["all_codes"]), expected)
+        self.assertEqual(combined["total"], len(expected))
+
+    def test_top_follower_and_header_product_type_can_be_combined(self):
+        type_filter = {
+            "field": "product_type",
+            "type": "list",
+            "values": [self.product_type],
+        }
+        header_combined = self._planning_list(
+            [
+                type_filter,
+                {"field": "采购跟单", "type": "list", "values": [self.group]},
+            ]
+        )
+        top_combined = self._planning_list([type_filter], follower=self.group)
+        self.assertEqual(
+            set(top_combined["all_codes"]),
+            set(header_combined["all_codes"]),
+        )
+
+    def test_linked_options_keep_the_real_intersection(self):
+        group_response = self.client.get(
+            "/api/planning/column-values",
+            query_string={
+                "field": "采购跟单",
+                "plan_filter": json.dumps(
+                    {
+                        "product_type": {
+                            "type": "list",
+                            "values": [self.product_type],
+                        }
+                    },
+                    ensure_ascii=False,
+                ),
+            },
+        ).get_json()
+        type_response = self.client.get(
+            "/api/planning/column-values",
+            query_string={
+                "field": "product_type",
+                "plan_filter": json.dumps(
+                    {"采购跟单": {"type": "list", "values": [self.group]}},
+                    ensure_ascii=False,
+                ),
+            },
+        ).get_json()
+        self.assertEqual(group_response["code"], 0)
+        self.assertEqual(type_response["code"], 0)
+        self.assertIn(self.product_type, type_response["data"])
+        normalized_groups = {
+            self.group_for_value(value, self.groups)
+            for value in group_response["data"]
+        }
+        self.assertIn(self.group, normalized_groups)
+
+
+if __name__ == "__main__":
+    unittest.main()