test_level0.py

import unittest
import time
from csst_dfs_client import level0

class Level0TestCase(unittest.TestCase):

    def setUp(self):
        pass

    # def test_find(self):
    #     start_time = time.time()
    #     result = level0.find(instrument='MSC', dataset="msc-v093")
    #     print(f"1操作执行时间: {time.time() - start_time} 秒")
    #     start_time = time.time()
    #     result = level0.find(instrument='MSC', dataset="msc-v093",
    #                         ra_obj = 170,
    #                         dec_obj = -24,
    #                         radius = 1)
    #     print(f"2操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}")
    #     start_time = time.time()
    #     result = level0.find(instrument='MSC', file_name="CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits")
    #     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_update_qc_status(self):
    #     result = level0.update_qc_status(level0_id = "1010910015799127", qc_status=1, dataset="msc-v093")
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_update_qc_status_by_ids(self):
    #     result = level0.update_qc_status_by_ids(ids = ["676ac74a530b47ca41568858"], qc_status=4)
    #     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 = "1010910015799127", dag_run="202411071002481234", prc_status=3)
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    def test_update_prc_status_by_ids(self):
        result = level0.update_prc_status_by_ids(ids = ["676ac74a530b47ca41568858"], prc_status=4)
        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, dataset= 'msc-v093')
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_find_process(self):
    #     result = level0.find_process(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", 
    #                             dag="csst-msc-l1-mbi", 
    #                             dag_run="202411071002481234", 
    #                             dataset="v93",
    #                             batch_id="v930batch",
    #                             prc_date="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)