update test

Signed-off-by: ふぁ <yuki@yuki0311.com>

test/python/test_serialize.py

@@ -3,13 +3,18 @@ import os
 import logging
 import base64
 import openapi_client as pt
+from pydantic import BaseModel
 from pathlib import Path
+import time
 
 
 logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(levelname)s %(message)s")
 logger = logging.getLogger("test_serialize")
 
 ERROR_UNCATCHED = os.environ.get("ERROR_UNCATCHED", "false").lower() == "true"
+SLEEP_TIME = float(os.environ.get("SLEEP", "0"))
+CUESOR_TEST_COUNT = int(os.environ.get("CUESOR_TEST_COUNT", "3"))
+
 
 if Path("cookie.json").exists():
     with open("cookie.json", "r") as f:
@@ -25,11 +30,36 @@ with open("src/config/placeholder.json", "r") as f:
     placeholder = json.load(f)
 
 
-def getKey(snake_str):
+def get_key(snake_str):
     components = snake_str.split("_")
     return "".join(x.title() for x in components[1:])
 
 
+def get_cursor(obj):
+    res = []
+    if type(obj) == dict:
+        if obj.get("__typename") is pt.TypeName.TIMELINETIMELINECURSOR:
+            res.append(obj["value"])
+        else:
+            for v in obj.values():
+                res.extend(get_cursor(v))
+    elif type(obj) == list:
+        for v in obj:
+            res.extend(get_cursor(v))
+    return res
+
+
+def get_kwargs(key, additional):
+    kwargs = {"path_query_id": placeholder[key]["queryId"]}
+    if placeholder[key].get("variables") is not None:
+        kwargs["variables"] = json.dumps(placeholder[key]["variables"] | additional)
+    if placeholder[key].get("features") is not None:
+        kwargs["features"] = json.dumps(placeholder[key]["features"])
+    if placeholder[key].get("fieldToggles") is not None:
+        kwargs["field_toggles"] = json.dumps(placeholder[key]["fieldToggles"])
+    return kwargs
+
+
 api_conf = pt.Configuration(
     api_key={
         "ClientLanguage": "en",
@@ -53,22 +83,28 @@ for x in [pt.DefaultApi, pt.TweetApi, pt.UserApi, pt.UserListApi]:
         if props.startswith("__") or props.endswith("_with_http_info"):
             continue
 
-        key = getKey(props)
-        logger.info(f"Try: {key}")
-
-        kwargs = {"path_query_id": placeholder[key]["queryId"]}
-        if placeholder[key].get("variables") is not None:
-            kwargs["variables"] = json.dumps(placeholder[key]["variables"])
-        if placeholder[key].get("features") is not None:
-            kwargs["features"] = json.dumps(placeholder[key]["features"])
-        if placeholder[key].get("fieldToggles") is not None:
-            kwargs["field_toggles"] = json.dumps(placeholder[key]["fieldToggles"])
+        key = get_key(props)
+        cursor_list = set([None])
+        cursor_history = set()
 
         try:
-            res = getattr(x(api_client), props)(**kwargs)
+            for _ in range(CUESOR_TEST_COUNT):
+                cursor = cursor_list.pop()
+                cursor_history.add(cursor)
+                logger.info(f"Try: {key} {cursor}")
+
+                kwargs = get_kwargs(key, {} if cursor is None else {"cursor": cursor})
+                res: BaseModel = getattr(x(api_client), props)(**kwargs)
+
+                cursor_list.update(set(get_cursor(res.to_dict())) - cursor_history)
+
+                if len(cursor_list) == 0:
+                    break
+                time.sleep(SLEEP_TIME)
+
         except Exception as e:
             if ERROR_UNCATCHED:
-                raise e
+                raise
             import traceback
 
             logger.error("==========[STACK TRACE]==========")