feat(selection): 支持选股结果服务端分页

zhixing-server/tests/unit/selection/test_postgres_runs.py

@@ -11,6 +11,7 @@ from zhixing_server.modules.selection.domain.models import SelectionSignal, Zhix
 from zhixing_server.modules.selection.domain.runs import (
     SelectionExecutionSource,
     SelectionRerunRequired,
+    SelectionResultQuery,
     SelectionRunInProgress,
     SelectionRunItem,
     SelectionStock,
@@ -163,6 +164,9 @@ def test_record_item_persists_independent_signals_as_jsonb(monkeypatch: pytest.M
 
 
 class LoadConnection:
+    def __init__(self) -> None:
+        self.statements: list[tuple[str, tuple[object, ...]]] = []
+
     def __enter__(self) -> "LoadConnection":
         return self
 
@@ -170,6 +174,7 @@ class LoadConnection:
         return None
 
     def execute(self, query: str, parameters: tuple[object, ...]) -> "LoadResult":
+        self.statements.append((query, parameters))
         if "FROM selection_run\n" in query:
             return LoadResult(
                 row=(
@@ -193,6 +198,8 @@ class LoadConnection:
             )
         if "FROM selection_run_item" in query:
             return LoadResult(rows=[("000001.SZ", "平安银行", "selected", 2, None)])
+        if "COUNT(*) FROM selection_signal" in query:
+            return LoadResult(row=(2,))
         return LoadResult(
             rows=[
                 (
@@ -242,10 +249,32 @@ def test_get_run_orders_signals_by_formula_priority(monkeypatch: pytest.MonkeyPa
         return connection
 
     monkeypatch.setattr(psycopg, "connect", connect)
-    run = PostgresSelectionRunRepository("postgresql://test").get_run("run-1")
+    run = PostgresSelectionRunRepository("postgresql://test").get_run(
+        "run-1",
+        query=SelectionResultQuery(
+            page=2,
+            page_size=1,
+            search="100%",
+            category="pullback",
+        ),
+    )
 
     assert run is not None
     assert [signal.category for signal in run.signals] == [
         ZHIXING_B1_SIGNAL_ORDER[0],
         ZHIXING_B1_SIGNAL_ORDER[-1],
     ]
+    count_query, count_parameters = next(
+        (query, parameters)
+        for query, parameters in connection.statements
+        if "COUNT(*) FROM selection_signal" in query
+    )
+    assert "name ILIKE %s ESCAPE" in count_query
+    assert count_parameters == ("run-1", "%100\\%%", "%100\\%%", "zhixing_b1_pullback_%")
+    page_query, page_parameters = next(
+        (query, parameters)
+        for query, parameters in connection.statements
+        if "LIMIT %s OFFSET %s" in query
+    )
+    assert "ORDER BY ts_code, CASE category" in page_query
+    assert page_parameters[-2:] == (1, 1)