test_level2.py

import unittest

from csst_dfs_client import level2

class Level2TestCase(unittest.TestCase):

    def setUp(self):
        pass

    def test_find(self):
        result = level2.find(obs_id = "1010940063886712")
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code))
        self.assertIsNotNone(result.data, "error message: " + result.message)

    def test_find_by_level2_id(self):
        result = level2.find_by_level2_id(level2_id = "1010940063886712")
        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_qc2_status(self):
        result = level2.update_qc2_status(level2_id = "1010940063886712", data_type='csst-msc-l2-mbi-cat', qc2_status=1)
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    def test_update_prc_status(self):
        result = level2.update_prc_status(level2_id = "1010940063886712", data_type='csst-msc-l2-mbi-cat', prc_status=3)
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    def test_update_qc2_status_by_file_name(self):
        result = level2.update_qc2_status_by_file_name(file_name = "CSST_MSC_MS_SCI_20310423084104_20310423084334_10109400638867_12_L2_V01_CATMIX.fits", qc2_status=1)
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    def test_update_prc_status_by_file_name(self):
        result = level2.update_prc_status_by_file_name(file_name = "CSST_MSC_MS_SCI_20310423084104_20310423084334_10109400638867_12_L2_V01_CATMIX.fits", 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_20310423084104_20310423084334_10109400638867_12_L2_V01_CATMIX.fits"
        result = level2.write(local_file = file_path, 
            module_id = "MSC",
            level0_id="1010940063886712",
            level1_id = "1010940063886712", 
            level2_id = "1010940063886712", 
            data_type='csst-msc-l2-mbi-cat', 
            file_name = "CSST_MSC_MS_SCI_20310423084104_20310423084334_10109400638867_12_L2_V01_CATMIX.fits",
            brick_id = 254,
            pipeline_id = "csst-msc-l2-mbi",
            build=1,
            run_id="20241107100248",
            version="094",
            prc_status=1)
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    def test_catalog_query(self):
        result = level2.catalog_query(sql="select * from csst_msc_l2_mbi_cat")
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)