diff --git a/tests/test_common.py b/tests/test_common.py index 97e09afd5115a12ec7b975adf9c25f88bfbed59f..f7c3b7224c060ac0c08342c4bd2f3f51557865f4 100644 --- a/tests/test_common.py +++ b/tests/test_common.py @@ -7,7 +7,7 @@ class CommonTestCase(unittest.TestCase): pass def test_download_level0_file(self): - bytes_data = download_file(file_path= "L0/MSC/SCI/60470/10109100157991/MS/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits") + bytes_data = download_file(file_path= "L2/MSC/CSST-MSC-L2-MBI-CAT/254/CSST_MSC_MS_SCI_20310423084104_20310423084334_10109400638867_12_L2_V01_CATMIX.fits") # save the downloaded file to a local directory tmp_file_dir = os.environ.get("UNIT_TEST_DATA_ROOT", "/tmp") tmp_file_path = os.path.join(tmp_file_dir, "test_downloaded_file.fits") @@ -21,7 +21,7 @@ class CommonTestCase(unittest.TestCase): os.remove(tmp_file_path) def test_read_file(self): - bytes_io = read_file(file_path= "L0/MSC/SCI/60470/10109100157991/MS/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits") + bytes_io = read_file(file_path= "L2/MSC/CSST-MSC-L2-MBI-CAT/254/CSST_MSC_MS_SCI_20310423084104_20310423084334_10109400638867_12_L2_V01_CATMIX.fits") # save the downloaded file to a local directory tmp_file_dir = os.environ.get("UNIT_TEST_DATA_ROOT", "/tmp") tmp_file_path = os.path.join(tmp_file_dir, "test_downloaded_file.fits") diff --git a/tests/test_level1.py b/tests/test_level1.py index 8be24a08f27dd78f18c5d4a40188759fe55f5c09..c68c2299d14a004b07571e48ea0f973d9b0018db 100644 --- a/tests/test_level1.py +++ b/tests/test_level1.py @@ -12,52 +12,52 @@ class Level1TestCase(unittest.TestCase): 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_qc1_status(self): - # result = level1.update_qc1_status(level1_id = "1060940003452925", file_type='SCI', qc1_status=1) - # 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", 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_L1_V01.fits" - # result = level1.write(local_file = file_path, - # module_id = "MSC", - # level0_id="1010910015799127", - # level1_id = "1010910015799127", - # pipeline_id = "csst-msc-l1-mbi", - # pmapname="csst_000128.pmap", - # file_type='SCI', - # build=1, - # prc_status=1) - # print(result) - # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) - - # def test_process_list(self): - # result = level1.process_list(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", 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) - - # def test_sls_find_by_qc1_status(self): - # result = level1.sls_find_by_qc1_status(qc1_status=0, limit=1) - # print(result) - # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", 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_qc1_status(self): + result = level1.update_qc1_status(level1_id = "1060940003452925", file_type='SCI', qc1_status=1) + 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", 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_L1_V01.fits" + result = level1.write(local_file = file_path, + module_id = "MSC", + level0_id="1010910015799127", + level1_id = "1010910015799127", + pipeline_id = "csst-msc-l1-mbi", + pmapname="csst_000128.pmap", + file_type='SCI', + build=1, + prc_status=1) + print(result) + self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) + + def test_process_list(self): + result = level1.process_list(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", 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) + + def test_sls_find_by_qc1_status(self): + result = level1.sls_find_by_qc1_status(qc1_status=0, limit=1) + print(result) + self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) \ No newline at end of file diff --git a/tests/test_level2.py b/tests/test_level2.py index 737e365e433a422a70248a41a4df1c660ecf5104..e8765c75f48cdd5801f52edbde5f934f339056fa 100644 --- a/tests/test_level2.py +++ b/tests/test_level2.py @@ -8,55 +8,55 @@ class Level2TestCase(unittest.TestCase): pass def test_find(self): - result = level2.find(obs_id = "10609400034529") + 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 = "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_qc2_status(self): - # result = level2.update_qc2_status(level2_id = "1060940003452925", data_type='csst-msc-l1-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 = "1060940003452925", data_type='csst-msc-l1-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_SCIE_20290713003437_20290713003707_10160000136_24_L1_V01_CAT.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_SCIE_20290713003437_20290713003707_10160000136_24_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/import/CSST_MSC_MS_SCIE_20290713003437_20290713003707_10160000136_24_L1_V01_CAT.fits" - # result = level2.write(local_file = file_path, - # module_id = "MSC", - # level0_id="1060940003452925", - # level1_id = "1060940003452925", - # level2_id = "1060940003452925", - # data_type='csst-msc-l1-mbi-cat', - # brick_id = 254, - # pipeline_id = "csst-msc-l1-mbi", - # build=1, - # run_id="20241107100248", - # version="093", - # prc_status=1) - # print(result) - # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", 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', + 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 tb_level2") + 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)