t

csst_dfs_client/level0.py

 import os
-import json
-from typing import Optional, IO, Tuple, Literal
+from typing import Optional, Tuple, Literal
 from .common import request, Result
 
 DateTimeTuple = Tuple[str, str]

tests/test_level0.py

@@ -11,51 +11,54 @@ class Level0TestCase(unittest.TestCase):
         start_time = time.time()
         result = level0.find(obs_id = "10609400034529",
                             create_time = ("2021-08-30 00:00:00", "2024-12-30 23:59:59"))
+        print(f"1操作执行时间: {time.time() - start_time} 秒")
         start_time = time.time()
-        result = level0.find(ra_obj = 170,
+        result = level0.find(module_id='MSC',
+                            ra_obj = 170,
                             dec_obj = -24,
-                            radius = 2)
-        print(f"操作执行时间: {time.time() - start_time} 秒")
+                            radius = 1)
+        print(f"2操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}")
+        start_time = time.time()
         result = level0.find(file_name="CSST_MSC_MS_SCI_20240609181116_20240609181347_101091001")
-        print(f"操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}")        
-        self.assertEqual(result.code, 200, "error code: " + str(result.code))
+        print(f"3操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}")        
+        self.assertEqual(result.code, 200, "error code: " + result.message)
+        self.assertIsNotNone(result.data, "error message: " + result.message)
+
+    def test_find_by_level0_id(self):
+        result = level0.find_by_level0_id(level0_id = "1060940003452925")
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
         self.assertIsNotNone(result.data, "error message: " + result.message)
 
-    # def test_find_by_level0_id(self):
-    #     result = level0.find_by_level0_id(level0_id = "1060940003452925")
-    #     print(result)
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
-    #     self.assertIsNotNone(result.data, "error message: " + result.message)
-
-    # def test_update_qc0_status(self):
-    #     result = level0.update_qc0_status(level0_id = "1060940003452925", file_type='SCI', qc0_status=1)
-    #     print(result)
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
-
-    # def test_update_prc_status(self):
-    #     result = level0.update_prc_status(level0_id = "1060940003452925", file_type='SCI', prc_status=3)
-    #     print(result)
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
-
-    # def test_write(self):
-    #     file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits"
-    #     result = level0.write(local_file = file_path)
-    #     print(result)
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
-
-    # def test_generate_prc_message(self):
-    #     result = level0.generate_prc_msg(
-    #         module_id='MSC',
-    #         obs_id="10609400034529",
-    #         detector_no="30")
-    #     print(result)
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
-
-    # def test_process_list(self):
-    #     result = level0.process_list(level0_id="1060940003452925")
-    #     print(result)
+    def test_update_qc0_status(self):
+        result = level0.update_qc0_status(level0_id = "1060940003452925", file_type='SCI', qc0_status=1)
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
+
+    def test_update_prc_status(self):
+        result = level0.update_prc_status(level0_id = "1060940003452925", file_type='SCI', prc_status=3)
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
+
+    def test_write(self):
+        file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits"
+        result = level0.write(local_file = file_path)
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
+
+    def test_generate_prc_message(self):
+        result = level0.generate_prc_msg(
+            module_id='MSC',
+            obs_id="10609400034529",
+            detector_no="30")
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
+
+    def test_process_list(self):
+        result = level0.process_list(level0_id="1060940003452925")
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
+    def test_add_process(self):
+        result = level0.add_process(level0_id="1060940003452925", pipeline_id="csst-msc-l1-mbi", run_id="202411071002481234", prc_time="2024-11-07 10:24:12", prc_status=1, prc_module="MSC", message="")
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)           
\ No newline at end of file
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
-    # def test_add_process(self):
-    #     result = level0.add_process(level0_id="1060940003452925", pipeline_id="csst-msc-l1-mbi", run_id="202411071002481234", prc_time="2024-11-07 10:24:12", prc_status=1, prc_module="MSC", message="")
-    #     print(result)
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)           
\ No newline at end of file