Skip to content
GitLab
Commit 7736953b authored by Wei Shoulin's avatar Wei Shoulin
Browse files

new data

parent dd219ea7
Hide whitespace changes
Inline Side-by-side
tests/test_common.py
View file @ 7736953b
......@@ -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")
......
tests/test_level1.py
View file @ 7736953b
......@@ -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
tests/test_level2.py
View file @ 7736953b
......@@ -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)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment