feat(ci): add unit tests

Author: kevinlin09

tests/legacy/test_add_resources.py tests/unit/test_add_resources.pytests/legacy/test_add_resources.py renamed to tests/unit/test_add_resources.py

@@ -45,25 +45,29 @@ def test_rag_call(self, mock_server: MockServer):
                         "action_type": "api",
                         "action_name": "文档检索",
                         "action": "searchDocument",
-                        "action_input_stream": '{"query":"API接口说明中, TopP参数改如何传递?"}',
+                        "action_input_stream": (
+                            '{"query":"API接口说明中, ' 'TopP参数改如何传递?"}'
+                        ),
                         "action_input": {
-                            "query": "API接口说明中, TopP参数改如何传递?",
+                            "query": ("API接口说明中, TopP参数改如何传递?"),
                         },
-                        "observation": """{"data": [
-                                                {
-                                                  "docId": "1234",
-                                                  "docName": "API接口说明",
-                                                  "docUrl": "https://127.0.0.1/dl/API接口说明.pdf",
-                                                  "indexId": "1",
-                                                  "score": 0.11992252,
-                                                  "text": "填(0,1.0),取值越大,生成的随机性越高;启用文档检索后,文档引用类型,取值包括:simple|indexed。",
-                                                  "title": "API接口说明",
-                                                  "titlePath": "API接口说明>>>接口说明>>>是否必   说明>>>填"
-                                                }
-                                              ],
-                                              "status": "SUCCESS"
-                                            }""",
-                        "response": "API接口说明中, TopP参数是一个float类型的参数,取值范围为0到1.0,默认为1.0。取值越大,生成的随机性越高。[5]",
+                        "observation": (
+                            '{"data": [{"docId": "1234", '
+                            '"docName": "API接口说明", '
+                            '"docUrl": "https://127.0.0.1/dl/'
+                            'API接口说明.pdf", "indexId": "1", '
+                            '"score": 0.11992252, "text": "填(0,1.0),'
+                            "取值越大,生成的随机性越高;启用文档检索后,"
+                            '文档引用类型,取值包括:simple|indexed。", '
+                            '"title": "API接口说明", "titlePath": '
+                            '"API接口说明>>>接口说明>>>是否必   说明>>>填"}], '
+                            '"status": "SUCCESS"}'
+                        ),
+                        "response": (
+                            "API接口说明中, TopP参数是一个float类型的"
+                            "参数,取值范围为0到1.0,默认为1.0。取值越大,"
+                            "生成的随机性越高。[5]"
+                        ),
                     },
                 ],
             },
@@ -86,11 +90,12 @@ def test_rag_call(self, mock_server: MockServer):
             top_p=0.2,
             temperature=1.0,
             doc_tag_codes=["t1234", "t2345"],
-            doc_reference_type=Application.DocReferenceType.simple,
+            doc_reference_type=(Application.DocReferenceType.simple),
             has_thoughts=True,
         )
 
-        self.check_result(resp, test_response)
+        # Test mock response type
+        self.check_result(resp, test_response)  # type: ignore[arg-type]
 
     def test_flow_call(self, mock_server: MockServer):
         test_response = {
@@ -105,13 +110,19 @@ def test_flow_call(self, mock_server: MockServer):
                         "action_type": "api",
                         "action_name": "plugin",
                         "action": "api",
-                        "action_input_stream": '{"userId": "123", "date": "202402", "city": "hangzhou"}',
+                        "action_input_stream": (
+                            '{"userId": "123", "date": "202402", '
+                            '"city": "hangzhou"}'
+                        ),
                         "action_input": {
                             "userId": "123",
                             "date": "202402",
                             "city": "hangzhou",
                         },
-                        "observation": """{"quantity": 102, "type": "resident", "date": "202402", "unit": "千瓦"}""",
+                        "observation": (
+                            '{"quantity": 102, "type": "resident", '
+                            '"date": "202402", "unit": "千瓦"}'
+                        ),
                         "response": "当月的居民用电量为102千瓦。",
                     },
                 ],
@@ -140,7 +151,8 @@ def test_flow_call(self, mock_server: MockServer):
             has_thoughts=True,
         )
 
-        self.check_result(resp, test_response)
+        # Test mock response type
+        self.check_result(resp, test_response)  # type: ignore[arg-type]
 
     def test_call_with_error(self, mock_server: MockServer):
         test_response = {
@@ -212,21 +224,21 @@ def check_result(resp: ApplicationResponse, test_response: Dict):
                 expected_doc_refs,
             )
 
-            for i in range(len(doc_refs)):
-                assert doc_refs[i].index_id == expected_doc_refs[i].get(
+            for i, doc_ref in enumerate(doc_refs):
+                assert doc_ref.index_id == expected_doc_refs[i].get(
                     "index_id",
                 )
-                assert doc_refs[i].doc_id == expected_doc_refs[i].get("doc_id")
-                assert doc_refs[i].doc_name == expected_doc_refs[i].get(
+                assert doc_ref.doc_id == expected_doc_refs[i].get("doc_id")
+                assert doc_ref.doc_name == expected_doc_refs[i].get(
                     "doc_name",
                 )
-                assert doc_refs[i].doc_url == expected_doc_refs[i].get(
+                assert doc_ref.doc_url == expected_doc_refs[i].get(
                     "doc_url",
                 )
-                assert doc_refs[i].title == expected_doc_refs[i].get("title")
-                assert doc_refs[i].text == expected_doc_refs[i].get("text")
-                assert doc_refs[i].biz_id == expected_doc_refs[i].get("biz_id")
-                assert json.dumps(doc_refs[i].images) == json.dumps(
+                assert doc_ref.title == expected_doc_refs[i].get("title")
+                assert doc_ref.text == expected_doc_refs[i].get("text")
+                assert doc_ref.biz_id == expected_doc_refs[i].get("biz_id")
+                assert json.dumps(doc_ref.images) == json.dumps(
                     expected_doc_refs[i].get("images"),
                 )
 
@@ -238,26 +250,26 @@ def check_result(resp: ApplicationResponse, test_response: Dict):
                 expected_thoughts,
             )
 
-            for i in range(len(thoughts)):
-                assert thoughts[i].thought == expected_thoughts[i].get(
+            for i, thought in enumerate(thoughts):
+                assert thought.thought == expected_thoughts[i].get(
                     "thought",
                 )
-                assert thoughts[i].action == expected_thoughts[i].get("action")
-                assert thoughts[i].action_name == expected_thoughts[i].get(
+                assert thought.action == expected_thoughts[i].get("action")
+                assert thought.action_name == expected_thoughts[i].get(
                     "action_name",
                 )
-                assert thoughts[i].action_type == expected_thoughts[i].get(
+                assert thought.action_type == expected_thoughts[i].get(
                     "action_type",
                 )
-                assert json.dumps(thoughts[i].action_input) == json.dumps(
+                assert json.dumps(thought.action_input) == json.dumps(
                     expected_thoughts[i].get("action_input"),
                 )
-                assert thoughts[i].action_input_stream == expected_thoughts[
-                    i
-                ].get("action_input_stream")
-                assert thoughts[i].observation == expected_thoughts[i].get(
+                assert thought.action_input_stream == (
+                    expected_thoughts[i].get("action_input_stream")
+                )
+                assert thought.observation == expected_thoughts[i].get(
                     "observation",
                 )
-                assert thoughts[i].response == expected_thoughts[i].get(
+                assert thought.response == expected_thoughts[i].get(
                     "response",
                 )

tests/legacy/test_application.py tests/unit/test_application.pytests/legacy/test_application.py renamed to tests/unit/test_application.py

@@ -40,7 +40,10 @@ def setup_class(cls):
         cls.update_phrase = {"黄鸡": 2, "红鸡": 1}
         cls.phrase_id = TEST_JOB_ID
 
-    def test_create_phrases(self, http_server):
+    def test_create_phrases(
+        self,
+        http_server,
+    ):  # pylint: disable=unused-argument
         result = AsrPhraseManager.create_phrases(
             model=self.model,
             phrases=self.phrase,
@@ -54,7 +57,10 @@ def test_create_phrases(self, http_server):
         assert len(result.output["finetuned_output"]) > 0
         self.phrase_id = result.output["finetuned_output"]
 
-    def test_update_phrases(self, http_server):
+    def test_update_phrases(
+        self,
+        http_server,
+    ):  # pylint: disable=unused-argument
         result = AsrPhraseManager.update_phrases(
             model=self.model,
             phrase_id=self.phrase_id,
@@ -66,7 +72,10 @@ def test_update_phrases(self, http_server):
         assert result.output["finetuned_output"] is not None
         assert len(result.output["finetuned_output"]) > 0
 
-    def test_query_phrases(self, http_server):
+    def test_query_phrases(
+        self,
+        http_server,
+    ):  # pylint: disable=unused-argument
         result = AsrPhraseManager.query_phrases(phrase_id=self.phrase_id)
         assert result is not None
         assert result.status_code == HTTPStatus.OK
@@ -75,46 +84,55 @@ def test_query_phrases(self, http_server):
         assert result.output["model"] is not None
         assert len(result.output["model"]) > 0
 
-    def test_list_phrases(self, http_server):
+    def test_list_phrases(
+        self,
+        http_server,
+    ):  # pylint: disable=unused-argument
         result = AsrPhraseManager.list_phrases(page=1, page_size=10)
         assert result is not None
         assert result.status_code == HTTPStatus.OK
         assert result.output["finetuned_outputs"] is not None
         assert len(result.output["finetuned_outputs"]) > 0
 
-    def test_delete_phrases(self, http_server):
+    def test_delete_phrases(
+        self,
+        http_server,
+    ):  # pylint: disable=unused-argument
         result = AsrPhraseManager.delete_phrases(phrase_id=self.phrase_id)
         assert result is not None
         assert result.status_code == HTTPStatus.OK
         assert result.output["finetuned_output"] is not None
         assert len(result.output["finetuned_output"]) > 0
 
 
-def str2bool(str):
-    return True if str.lower() == "true" else False
+def str2bool(test):  # pylint: disable=redefined-builtin
+    # Return True if test string is "true", False otherwise
+    return test.lower() == "true"
 
 
-def complete_url(url: str) -> str:
+def complete_url(url: str) -> None:
+    # Set base URLs for dashscope API
     parsed = urlparse(url)
     base_url = "".join([parsed.scheme, "://", parsed.netloc])
     dashscope.base_websocket_api_url = "/".join(
         [base_url, "api-ws", dashscope.common.env.api_version, "inference"],
     )
-    dashscope.base_http_api_url = url = "/".join(
+    dashscope.base_http_api_url = "/".join(
         [base_url, "api", dashscope.common.env.api_version],
     )
     print("Set base_websocket_api_url: ", dashscope.base_websocket_api_url)
     print("Set base_http_api_url: ", dashscope.base_http_api_url)
 
 
-def phrases(
+def phrases(  # pylint: disable=redefined-outer-name,too-many-branches
     model,
     phrase_id: str,
     phrases: dict,
     page: int,
     page_size: int,
     delete: bool,
 ):
+    # Manage ASR phrases based on provided parameters
     print("phrase_id: ", phrase_id)
     print("phrase: ", phrases)
     print("delete flag: ", delete)
@@ -126,33 +144,30 @@ def phrases(
                 phrase_id=phrase_id,
                 phrases=phrases,
             )
-        else:
-            print("Create phrases -->")
-            return AsrPhraseManager.create_phrases(
-                model=model,
-                phrases=phrases,
-            )
-    else:
-        if delete:
-            print("Delete phrases -->")
-            return AsrPhraseManager.delete_phrases(phrase_id=phrase_id)
-        else:
-            if phrase_id is not None:
-                print("Query phrases -->")
-                return AsrPhraseManager.query_phrases(phrase_id=phrase_id)
-            if page is not None and page_size is not None:
-                print(
-                    "List phrases page %d page_size %d -->"
-                    % (page, page_size),
-                )
-                return AsrPhraseManager.list_phrases(
-                    page=page,
-                    page_size=page_size,
-                )
+        print("Create phrases -->")
+        return AsrPhraseManager.create_phrases(
+            model=model,
+            phrases=phrases,
+        )
+    if delete:
+        print("Delete phrases -->")
+        return AsrPhraseManager.delete_phrases(phrase_id=phrase_id)
+    if phrase_id is not None:
+        print("Query phrases -->")
+        return AsrPhraseManager.query_phrases(phrase_id=phrase_id)
+    if page is not None and page_size is not None:
+        print(
+            f"List phrases page {page} page_size {page_size} -->",
+        )
+        return AsrPhraseManager.list_phrases(
+            page=page,
+            page_size=page_size,
+        )
+    return None
 
 
 @pytest.mark.skip
-def test_by_user():
+def test_by_user():  # pylint: disable=too-many-branches
     parser = argparse.ArgumentParser()
     parser.add_argument("--model", type=str, default="paraformer-realtime-v1")
     parser.add_argument("--phrase", type=str, default="")
@@ -184,46 +199,47 @@ def test_by_user():
         print("Response of phrases: ", resp)
         if resp is not None and resp.output is not None:
             output = resp.output
-            print("\nGet output: %s\n" % (str(output)))
+            print(f"\nGet output: {str(output)}\n")
 
             if (
                 "finetuned_output" in output
                 and output["finetuned_output"] is not None
             ):
-                print("Get phrase_id: %s" % (output["finetuned_output"]))
+                print(
+                    f"Get phrase_id: {output['finetuned_output']}",
+                )
             if "job_id" in output and output["job_id"] is not None:
-                print("Get job_id: %s" % (output["job_id"]))
+                print(f"Get job_id: {output['job_id']}")
             if "create_time" in output and output["create_time"] is not None:
-                print("Get create_time: %s" % (output["create_time"]))
+                print(f"Get create_time: {output['create_time']}")
             if "model" in output and output["model"] is not None:
-                print("Get model_id: %s" % (output["model"]))
+                print(f"Get model_id: {output['model']}")
             if "output_type" in output and output["output_type"] is not None:
-                print("Get output_type: %s" % (output["output_type"]))
+                print(f"Get output_type: {output['output_type']}")
 
             if (
                 "finetuned_outputs" in output
                 and output["finetuned_outputs"] is not None
             ):
                 outputs = output["finetuned_outputs"]
                 print(
-                    "Get %d info from page_no:%d page_size:%d total:%d ->"
-                    % (
-                        len(outputs),
-                        output["page_no"],
-                        output["page_size"],
-                        output["total"],
-                    ),
+                    f"Get {len(outputs)} info from "
+                    f"page_no:{output['page_no']} "
+                    f"page_size:{output['page_size']} "
+                    f"total:{output['total']} ->",
                 )
                 for item in outputs:
-                    print("  get phrase_id: %s" % (item["finetuned_output"]))
-                    print("  get job_id: %s" % (item["job_id"]))
-                    print("  get create_time: %s" % (item["create_time"]))
-                    print("  get model_id: %s" % (item["model"]))
-                    print("  get output_type: %s\n" % (item["output_type"]))
+                    print(
+                        f"  get phrase_id: {item['finetuned_output']}",
+                    )
+                    print(f"  get job_id: {item['job_id']}")
+                    print(f"  get create_time: {item['create_time']}")
+                    print(f"  get model_id: {item['model']}")
+                    print(f"  get output_type: {item['output_type']}\n")
     else:
         print(
-            "ERROR, status_code:%d, code_message:%s, error_message:%s"
-            % (resp.status_code, resp.code, resp.message),
+            f"ERROR, status_code:{resp.status_code}, "
+            f"code_message:{resp.code}, error_message:{resp.message}",
         )