test_level1.py

import unittest

from csst_dfs_client import level1

class Level1TestCase(unittest.TestCase):

    def setUp(self):
        pass

    # def test_find(self):
    #     result = level1.find(instrument='MSC', obs_time=("2024-01-01 12:12:12", "2024-06-30 12:12:12"), dataset='msc-v093')
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code))
    #     self.assertIsNotNone(result.data, "error message: " + result.message)
    # def test_find_by_brick_id(self):
    #     result = level1.find_by_brick_id(brick_id = 401847)
    #     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_level1_id(self):
    #     result = level1.find_by_level1_id(level1_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 = level1.update_qc_status(level1_id = "1060940003452925", data_model='csst-msc-l1-mbi', qc_status=1, batch_id='inttest')
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_update_prc_status(self):
    #     result = level1.update_prc_status(level1_id = "1060940003452925", data_model='csst-msc-l1-mbi', prc_status=3, batch_id='inttest')
    #     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/L1/MSC/CSST_MSC_MS_WIDE_20230413214557_20230413214827_10100070335_13_L1_V01.fits"
    #     # from io import BytesIO
    #     # with open(file_path, "rb") as file:
    #     #     file_path = BytesIO(file.read())
            
    #     result = level1.write(local_file = file_path, 
    #         obs_group="10109100157991",
    #         obs_type="WIDE",
    #         instrument = "MSC",
    #         level0_id="1010910015799127",
    #         level1_id = "1010910015799127", 
    #         dag = "csst-msc-l1-mbi",
    #         file_name= "CSST_MSC_MS_WIDE_20230413214557_20230413214827_10100070335_13_L1_V01.fits",
    #         pmapname="csst_000128.pmap",
    #         data_model='csst-msc-l1-mbi', 
    #         build=1,
    #         qc_status=10,
    #         dataset='msc-v093',
    #         batch_id="inttest")
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_find_process(self):
    #     result = level1.find_process(level1_id="1060940003452925")
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
    # def test_add_process(self):
    #     result = level1.add_process(level1_id="1060940003452925", dag="csst-msc-l1-mbi", dag_run="202411071002481234", 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)   

    # def test_sls_find_by_qc_status(self):
    #     result = level1.sls_find_by_qc_status(qc_status=0, limit=1)
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
    
    # def test_update_qc_status_by_ids(self):
    #     result = level1.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_by_ids(self):
    #     result = level1.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_count_by_dataset(self):
        result = level1.count_by_dataset()
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)    
        self.assertIsNotNone(result.data, "error message: " + result.message)