c3

tests/test_common_db.py

@@ -9,11 +9,11 @@ class DBClientTestCase(unittest.TestCase):
     def test_db_init(self):
         db = DBClient()
         
-        r = db.select_one("select count(*) as c from ifs_rawfits")
+        r = db.select_one("select date('now')")
         if r is not None:
-            print("ifs_rawfits count:", r['c'])
+            print("now:", r)
 
-        r = db.exists("select * from ifs_rawfits where id=2323")
+        r = db.exists("select * from t_observation where id=2323")
         if r:
             print("existed")
         else:

tests/test_facility_cal_merge.py

+import os
+import unittest
+from astropy.io import fits
+
+from csst_dfs_api_local.facility.calmerge import CalMergeApi
+
+class CalMergeApiTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = CalMergeApi()
+
+    def test_find(self):
+        recs = self.api.find(detector_no='CCD01',
+            ref_type = "bias",
+            obs_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 = 3)
+        print('get:', rec)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(id = 3, status = 1)
+        print('update_proc_status:', rec)
+
+    def test_update_qc1_status(self):
+        rec = self.api.update_qc1_status(id = 3, status = 2)
+        print('update_qc1_status:', rec)    
+
+    def test_write(self):
+        rec = self.api.write(detector_no='CCD01', 
+            ref_type = "bias",
+            obs_time = "2021-06-04 11:12:13",
+            exp_time = 150,
+            filename = "/opt/dddasd.params",
+            file_path = "/opt/dddasd.fits",
+            prc_status = 3,
+            prc_time = '2021-06-04 11:12:13',
+            level0_ids = [1,2,3,4])
+        print('write:', rec)
\ No newline at end of file

tests/test_facility_detector.py

+import unittest
+
+from csst_dfs_api_local.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')
+        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 unittest
+
+from csst_dfs_api_local.facility import Level0DataApi
+
+class Level0DataApiTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = Level0DataApi()
+
+    def test_find(self):
+        recs = self.api.find(obs_id = 13, obs_type = 'sci', limit = 0)
+        print('find:', recs)
+
+    def test_get(self):
+        rec = self.api.get(fits_id = 31)
+        print('get:', rec)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(fits_id = 31, status = 6)
+        print('update_proc_status:', rec)
+
+    def test_update_qc0_status(self):
+        rec = self.api.update_qc0_status(fits_id = 31, 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)  

tests/test_facility_level0_prc.py

+import os
+import unittest
+from astropy.io import fits
+
+from csst_dfs_api_local.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 = 1, 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 unittest
+
+from csst_dfs_api_local.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=11)
+        print('get:', rec)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(obs_id = 11, status = 3, )
+        print('update_proc_status:', rec)
+
+    def test_update_qc0_status(self):
+        rec = self.api.update_qc0_status(obs_id = 11, 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_ingest.py

-import logging
-import unittest
-import os
-
-from csst_dfs_api_local.ifs import ingest
-
-log = logging.getLogger('csst')
-class IFSIngestTestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
-
-    def test_ingest(self):
-       ingest()
-

tests/test_ifs_level1.py

+import unittest
+
+from csst_dfs_api_local.ifs import Level1DataApi
+
+class IFSResult1TestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = Level1DataApi()
+
+    def test_find(self):
+        recs = self.api.find(raw_id=11,
+            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 = 1)
+        print('get:', rec)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(id = 1, status = 4)
+        print('update_proc_status:', rec)
+
+    def test_update_qc1_status(self):
+        rec = self.api.update_qc1_status(id = 1, status = 7)
+        print('update_qc1_status:', rec)
+
+    def test_write(self):
+        rec = self.api.write(raw_id=11, 
+            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 = "dddasd23",
+            file_path = "/opt/dddasd23.fits",
+            pipeline_id = "P2")
+        print('write:', rec)
\ No newline at end of file

tests/test_ifs_result0.py

-import os
-import unittest
-from astropy.io import fits
-
-from csst_dfs_api_local.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)
-        assert len(recs) == 1
-
-        recs = self.api.find()
-        print('find:', recs)
-        assert len(recs) > 1
-
-    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.root_dir, 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_local.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)
-    #     assert len(recs) == 1
-
-    #     recs = self.api.find()
-    #     print('find:', recs)
-    #     assert len(recs) > 0
-
-    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.root_dir, 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 unittest
+
+from csst_dfs_api_local.msc import Level1DataApi
+
+class IFSResult1TestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.api = Level1DataApi()
+
+    def test_find(self):
+        recs = self.api.find(raw_id=11,
+            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 = 1)
+        print('get:', rec)
+
+    def test_update_proc_status(self):
+        rec = self.api.update_proc_status(id = 1, status = 4)
+        print('update_proc_status:', rec)
+
+    def test_update_qc1_status(self):
+        rec = self.api.update_qc1_status(id = 1, status = 7)
+        print('update_qc1_status:', rec)
+
+    def test_write(self):
+        rec = self.api.write(raw_id=11, 
+            data_type = "sci",
+            cor_sci_id = 2,
+            prc_params = "/opt/dddasd.params",
+            flat_id = 1,
+            dark_id = 2,
+            bias_id = 3,
+            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