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_qc_status(self):
        result = level2.update_qc_status(level2_id = "1010940063886712", data_model='csst-msc-l2-mbi-cat', qc_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_model='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_qc_status_by_file_name(self):
        result = level2.update_qc_status_by_file_name(file_name = "CSST_MSC_MS_WIDE_20230227011213_20230227011443_10100059495_12_L1_V01_CAT.fits", qc_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_WIDE_20230227011213_20230227011443_10100059495_12_L1_V01_CAT.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/conf-test/CSST_MSC_MS_WIDE_20230227011213_20230227011443_10100059495_12_L1_V01_CAT.fits"
        result = level2.write(local_file = file_path)
        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)