import os import unittest from astropy.io import fits from csst_dfs_api.facility.level1 import Level1DataApi class Level1DataTestCase(unittest.TestCase): def setUp(self): self.api = Level1DataApi() # def test_find(self): # recs = self.api.find( # level0_id='10000010101', # module_id = "MSC", # create_time = ("2024-01-01 11:12:13","2024-06-08 11:12:13")) # print('find:', recs) # def test_find_by_ids(self): # recs = self.api.find_by_ids( # module_id = "MSC", # ids=[16] # ) # print('find_by_ids:', recs) def test_find_refs_by_ids(self): recs = self.api.find_refs_by_ids( ids=[16] ) print('test_find_refs_by_ids:', recs) def test_find_by_brick_ids(self): recs = self.api.find_by_brick_ids( brick_ids=[1,2,3,4] ) print('find_by_brick_ids:', recs) def test_sls_find_by_qc1_status(self): recs = self.api.sls_find_by_qc1_status( qc1_status = 1, limit = 1 ) print('sls_find_by_qc1_status:', recs) def test_get(self): rec = self.api.get(id = 16) print('get:', rec) def test_update_proc_status(self): rec = self.api.update_proc_status(id = 2, status = 4) print('update_proc_status:', rec) def test_update_qc1_status(self): rec = self.api.update_qc1_status(id = 2, status = 7) print('update_qc1_status:', rec) def test_write(self): rec = self.api.write( level0_id = '10000010101', module_id = "MSC", file_type = "SCIE", prc_status = 3, prc_time = '2021-10-22 11:12:13', filename = "CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits", file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits", pipeline_id = "P1", build_id = 123, refs = {'flat': '3', 'arc': '2', 'bias': '1'} ) print('write:', rec)