c3

tests/test_facility_detector.py

+import unittest
+
+from csst_dfs_api.facility.detector import DetectorApi
+
+class DetectorApiTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = DetectorApi()
+
+    def test_find(self):
+        recs = self.api.find(module_id = 'MSC', key = 'CCD')
+        print('find:', recs)
+
+    def test_get(self):
+        rec = self.api.get(no = 'CCD01')
+        print('get:', rec)
+
+    def test_write(self):
+        rec = self.api.write(no = 'CCD02', 
+                detector_name = 'CCD02', 
+                module_id = 'MSC', 
+                filter_id='f2')
+        print('write:', rec)
+
+    def test_update(self):
+        rec = self.api.update(no = 'CCD01', filter_id = 'f1')
+        print('update:', rec)
+
+    def test_delete(self):
+        rec = self.api.delete(no = 'CCD01')
+        print('delete:', rec)
+
+    def test_find_status(self):
+        recs = self.api.find_status(detector_no = 'CCD01', 
+            status_occur_time = ('2021-06-02','2021-06-08'), 
+            limit = 0)
+        print('find status:', recs)
+
+    def test_get_status(self):
+        rec = self.api.get_status(id = 2)
+        print('get status:', rec)
+
+    def test_write_status(self):
+        rec = self.api.write_status(detector_no = 'CCD01', status = '{........}',status_time='2021-06-05 12:12:13')
+        print('write status:', rec)
\ No newline at end of file

tests/test_facility_level0.py

+import os
+import unittest
+from astropy.io import fits
+
+from csst_dfs_api.facility.level0 import Level0DataApi
+
+class Level0DataTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = Level0DataApi()
+
+    def test_find(self):
+        recs = self.api.find(obs_id = 9, obs_type = 'sci', limit = 0)
+        print('find:', recs)
+
+    def test_get(self):
+        rec = self.api.get(fits_id = 100)
+        print('get:', rec)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(fits_id = 100, status = 6)
+        print('update_proc_status:', rec)
+
+    def test_update_qc0_status(self):
+        rec = self.api.update_qc0_status(fits_id = 100, status = 7)
+        print('update_qc0_status:', rec)
+
+    def test_write(self):
+        rec = self.api.write(
+            obs_id = 13,
+            detector_no = "CCD01",
+            obs_type = "sci",            
+            obs_time = "2021-06-06 11:12:13",
+            exp_time = 150,
+            detector_status_id = 3,
+            filename = "MSC_00001234",
+            file_path = "/opt/MSC_00001234.fits")
+        print('write:', rec)  
\ No newline at end of file

tests/test_facility_level0_prc.py

+import os
+import unittest
+from astropy.io import fits
+
+from csst_dfs_api.facility.level0prc import Level0PrcApi
+
+class Level0PrcTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = Level0PrcApi()
+
+    def test_find(self):
+        recs = self.api.find(level0_id=134)
+        print('find:', recs)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(id = 8, status = 4)
+        print('update_proc_status:', rec)
+
+    def test_write(self):
+        rec = self.api.write(level0_id=134, 
+            pipeline_id = "P1",
+            prc_module = "QC0",
+            params_id = "/opt/dddasd.params",
+            prc_status = 3,
+            prc_time = '2021-06-04 11:12:13',
+            file_path = "/opt/dddasd.header")
+        print('write:', rec)
\ No newline at end of file

tests/test_facility_observation.py

+import os
+import unittest
+from astropy.io import fits
+
+from csst_dfs_api.facility.observation import ObservationApi
+
+class FacilityObservationTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = ObservationApi()
+
+    def test_find(self):
+        recs = self.api.find(module_id="MSC",limit = 0)
+        print('find:', recs)
+
+    def test_get(self):
+        rec = self.api.get(obs_id=9)
+        print('get:', rec)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(obs_id = 9, status = 3, )
+        print('update_proc_status:', rec)
+
+    def test_update_qc0_status(self):
+        rec = self.api.update_qc0_status(obs_id = 9, status = 3, )
+        print('update_qc0_status:', rec)        
+
+    def test_write(self):
+        rec = self.api.write(
+            obs_time = "2021-06-06 11:12:13",
+            exp_time = 150,
+            module_id = "MSC",
+            obs_type = "sci",
+            facility_status_id = 3,
+            module_status_id = 3)
+        print('write:', rec)        
\ No newline at end of file

tests/test_ifs_fits.py

-import os
-import unittest
-from astropy.io import fits
-
-from csst_dfs_api.ifs import FitsApi
-from csst_dfs_api.common.delegate import Delegate
-class IFSFitsTestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.api = FitsApi()
-
-    def test_find(self):
-        # recs = self.api.find(file_name='CCD1_ObsTime_300_ObsNum_7.fits')
-        # print('find:', recs)
-        # assert len(recs) == 1
-
-        recs = self.api.find(limit=1)
-        print('find:', recs)
-        # assert len(recs) > 1
-
-    # def test_read(self):
-    #     recs = self.api.find(file_name='CCD1_ObsTime_300_ObsNum_7.fits')
-    #     print("The full path: ", os.path.join(self.api.file_prefix, recs[0]['file_path']))
-
-    #     file_segments = self.api.read(file_path=recs[0]['file_path'])
-    #     file_bytes = b''.join(file_segments)
-    #     hdul = fits.HDUList.fromstring(file_bytes)
-    #     print(hdul.info())
-    #     hdr = hdul[0].header
-    #     print(repr(hdr))
-
-    # def test_update_proc_status(self):
-    #     recs = self.api.find(file_name='CCD1_ObsTime_300_ObsNum_7.fits')
-
-    #     self.api.update_proc_status(fits_id=recs[0]['id'],status=1)
-
-    #     rec = self.api.get(fits_id=recs[0]['id'])
-    #     assert rec['prc_status'] == 1
-
-    # def test_update_qc0_status(self):
-    #     recs = self.api.find(file_name='CCD1_ObsTime_300_ObsNum_7.fits')
-
-    #     self.api.update_qc0_status(fits_id=recs[0]['id'],status=1)
-
-    #     rec = self.api.get(fits_id=recs[0]['id'])
-    #     assert rec['qc0_status'] == 1
-
-    # def test_write(self):
-    #     recs = self.api.write(file_path='/opt/temp/csst_ifs/CCD2_ObsTime_1200_ObsNum_40.fits')
-
-    #     recs = self.api.find(file_name='CCD2_ObsTime_1200_ObsNum_40.fits')
-
-    #     rec = self.api.get(fits_id=recs[0]['id'])
-
-    #     print(rec)
-    
\ No newline at end of file

tests/test_ifs_level1.py

+import os
+import unittest
+from astropy.io import fits
+
+from csst_dfs_api.ifs.level1 import Level1DataApi
+
+class IFSLevel1DataTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = Level1DataApi()
+
+    def test_find(self):
+        recs = self.api.find(raw_id=1,
+            create_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13"))
+        print('find:', recs)
+
+    def test_get(self):
+        rec = self.api.get(id = 2)
+        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(raw_id=12, 
+            data_type = "sci",
+            cor_sci_id = 2,
+            prc_params = "/opt/dddasd.params",
+            flat_id = 1,
+            dark_id = 2,
+            bias_id = 3,
+            lamp_id = 4,
+            arc_id = 5,
+            sky_id = 6,            
+            prc_status = 3,
+            prc_time = '2021-06-05 11:12:13',
+            filename = "dddasd",
+            file_path = "/opt/dddasd.fits",
+            pipeline_id = "P2")
+        print('write:', rec)
\ No newline at end of file

tests/test_ifs_refs.py

-import os
-import unittest
-from astropy.io import fits
-
-from csst_dfs_api.ifs import RefFitsApi
-
-class IFSFitsTestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.api = RefFitsApi()
-
-    def test_find(self):
-        recs = self.api.find(exp_time=('2021-03-24 10:14:56', '2021-03-24 10:15:56'), ref_type=RefFitsApi.REF_FITS_FLAT)
-        print('find:', recs)
-        # assert len(recs) > 1
-
-        # recs = self.api.find()
-        # print('=' * 80)
-        # print('find:', recs)
-        # assert len(recs) > 1
-
-    # def test_read(self):
-    #     recs = self.api.find(file_name='CCD2_Flat_img.fits')
-    #     print("The full path: ", os.path.join(self.api.file_prefix, recs[0]['file_path']))
-
-    #     file_segments = self.api.read(file_path=recs[0]['file_path'])
-    #     file_bytes = b''.join(file_segments)
-    #     hdul = fits.HDUList.fromstring(file_bytes)
-    #     print(hdul.info())
-    #     hdr = hdul[0].header
-    #     print(repr(hdr))      
-
-    # def test_update_status(self):
-    #     recs = self.api.find(file_name='CCD2_Flat_img.fits')
-    #     self.api.update_status(fits_id=recs[0]['id'],status=1)
-    #     rec = self.api.get(fits_id=recs[0]['id'])
-    #     assert rec['status'] == 1
-
-    # def test_write(self):
-    #     recs = self.api.write(file_path='/opt/temp/csst_ifs/CCD3_Flat_img.fits')
-    #     recs = self.api.find(file_name='CCD3_Flat_img.fits')
-    #     rec = self.api.get(fits_id=recs[0]['id'])
-    #     print(rec)

tests/test_ifs_result0.py

-import os
-import unittest
-from astropy.io import fits
-
-from csst_dfs_api.ifs import Result0Api
-
-class IFSResult0TestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.api = Result0Api()
-
-    def test_find(self):
-        recs = self.api.find(file_name='CCD2_ObsTime_1200_ObsNum_40.fits')
-        print('find:', recs)
-
-        recs = self.api.find()
-        print('find:', recs)
-
-    def test_read(self):
-        recs = self.api.find(file_name='CCD2_ObsTime_1200_ObsNum_40.fits')
-        print("The full path: ", os.path.join(self.api.file_prefix, recs[0]['file_path']))
-
-        file_segments = self.api.read(file_path=recs[0]['file_path'])
-        file_bytes = b''.join(file_segments)
-        hdul = fits.HDUList.fromstring(file_bytes)
-        print(hdul.info())
-        hdr = hdul[0].header
-        print(repr(hdr))      
-
-    def test_write(self):
-        self.api.write(raw_id = 1, 
-                file_path='/opt/temp/csst_ifs/CCD2_ObsTime_1200_ObsNum_40.fits',
-                proc_type = 'default')
-
-        recs = self.api.find(raw_id=1)
-        print()
-        print(recs)
-        print("="*80)
-        recs = self.api.find(file_name='CCD2_ObsTime_1200_ObsNum_40.fits')
-        print(recs)

tests/test_ifs_result1.py

-import os
-import unittest
-from astropy.io import fits
-
-from csst_dfs_api.ifs import Result1Api
-
-class IFSResult1TestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.api = Result1Api()
-
-    def test_find(self):
-        recs = self.api.find(file_name='CCD2_ObsTime_1200_ObsNum_40.fits')
-        print('find:', recs)
-
-        recs = self.api.find()
-        print('find:', recs)
-
-    def test_read(self):
-        recs = self.api.find(file_name='CCD2_ObsTime_1200_ObsNum_40.fits')
-        print("The full path: ", os.path.join(self.api.file_prefix, recs[0]['file_path']))
-
-        rec, result0s = self.api.get(fits_id=recs[0]['id'])
-        print(result0s)
-        file_segments = self.api.read(file_path=rec['file_path'])
-        file_bytes = b''.join(file_segments)
-        hdul = fits.HDUList.fromstring(file_bytes)
-        print(hdul.info())
-        hdr = hdul[0].header
-        print(repr(hdr))
-
-    def test_write(self):
-        self.api.write(result0_ids = [1,2,3,4], 
-                file_path='/opt/temp/csst_ifs/CCD2_ObsTime_1200_ObsNum_40.fits',
-                proc_type = 'default')
-
-        recs = self.api.find(raw_id=1)
-        print()
-        print(recs)
-        print("="*80)
-        recs = self.api.find(file_name='CCD2_ObsTime_1200_ObsNum_40.fits')
-        print(recs)

tests/test_msc_level1.py

+import os
+import unittest
+from astropy.io import fits
+
+from csst_dfs_api.msc.level1 import Level1DataApi
+
+class MSCLevel1DataTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = Level1DataApi()
+
+    def test_find(self):
+        recs = self.api.find(raw_id=1,
+            create_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13"))
+        print('find:', recs)
+
+    def test_get(self):
+        rec = self.api.get(id = 2)
+        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(raw_id=1, 
+            data_type = "sci",
+            cor_sci_id = 1,
+            prc_params = "/opt/dddasd.params",
+            flat_id = 1,
+            dark_id = 2,
+            bias_id = 3,
+            prc_status = 3,
+            prc_time = '2021-06-04 11:12:13',
+            filename = "dddasd",
+            file_path = "/opt/dddasd.fits",
+            pipeline_id = "P1")
+        print('write:', rec)
\ No newline at end of file