new test

csst_dfs_api/facility/level2producer.py

@@ -62,7 +62,7 @@ class Level2ProducerApi(object):
         
         :returns: csst_dfs_common.models.Result
         '''
-        return self.stub.find_nexts(**kwargs)
+        return self.stub.find_start(**kwargs)
 
     def update(self, **kwargs):
         ''' update a Level2Producer
@@ -119,7 +119,7 @@ class Level2ProducerApi(object):
         
         :returns: csst_dfs_common.models.Result
         '''  
-        return self.stub.new_job(**kwargs)
+        return self.stub.update_job(**kwargs)
 
     def new_running(self, **kwargs):
         ''' insert a Level2ProducerRuningRecord data

tests/facility/test_facility_brick.py

@@ -9,19 +9,21 @@ class FacilityBrickTestCase(unittest.TestCase):
         self.api = BrickApi()
 
     def test_find(self):
-        recs = self.api.find(module_id="MSC",limit = 0)
+        recs = self.api.find(limit = 0)
         print('find:', recs)
 
     def test_get(self):
         rec = self.api.get(id=17)
         print('get:', rec)
-        rec = self.api.get(obs_id='0000017')
-        print('get:', rec)        
 
-    def test_update_proc_status(self):
-        rec = self.api.update_proc_status(obs_id='0000017', status = 3, )
-        print('update_proc_status:', rec)
+    def test_write(self):
+        rec = self.api.write(id=1, ra = 3.2, dec = 3.3, boundingbox = '12,12')
+        print('test_write:', rec)
 
-    def test_update_qc0_status(self):
-        rec = self.api.update_qc0_status(obs_id='0000017', status = 3, )
-        print('update_qc0_status:', rec)        
+    def test_find_obs_status(self):
+        rec = self.api.find_obs_status(brick_id=1, band = 'r')
+        print('find_obs_status:', rec)
+
+    def test_find_level1_data(self):
+        rec = self.api.find_level1_data(brick_id=1, level1_id = 1, module = 'msc')
+        print('find_level1_data:', rec)          

tests/facility/test_facility_level2producer.py

-import os
 import unittest
-from astropy.io import fits
 
 from csst_dfs_api.facility.level2producer import Level2ProducerApi
 
@@ -9,20 +7,100 @@ class FacilityLevel2ProducerTestCase(unittest.TestCase):
     def setUp(self):
         self.api = Level2ProducerApi()
 
+    def test_register(self):
+        rec = self.api.register(name='Test2', 
+            gitlink='http://github.com/xxx/xxx',
+            paramfiles='/opt/csst/param1.ini',
+            priority = 3,
+            pre_producers=[1,2] )
+        print('register:', rec)
+
     def test_find(self):
-        recs = self.api.find(module_id="MSC",limit = 0)
+        recs = self.api.find(key="Te")
         print('find:', recs)
 
     def test_get(self):
-        rec = self.api.get(id=17)
+        rec = self.api.get(id=10)
         print('get:', rec)
-        rec = self.api.get(obs_id='0000017')
-        print('get:', rec)        
 
-    def test_update_proc_status(self):
-        rec = self.api.update_proc_status(obs_id='0000017', status = 3, )
-        print('update_proc_status:', rec)
+    def test_find_nexts(self):
+        recs = self.api.find_nexts(id=1)
+        print('find_nexts:', recs)
+
+    def test_find_start(self):
+        recs = self.api.find_start(key="Te")
+        print('find_start:', recs)
+
+    def test_update(self):
+        recs = self.api.update(
+            id=2,
+            name = "start2",
+            gitlink = "http://github.com/xxx/xxx",
+            paramfiles='/opt/csst/param1.ini',
+            priority = 3,
+            pre_producers=[1,3]
+        )
+        print('test_update:', recs)
+
+    def test_delete(self):
+        recs = self.api.delete(
+            id=11
+        )
+        print('test_delete:', recs)
+
+    def test_new_job(self):
+        recs = self.api.new_job(
+            dag = "start2-1-1-"
+        )
+        print('new_job:', recs)
+
+    def test_get_job(self):
+        recs = self.api.get_job(
+            id = 2
+        )
+        print('get_job:', recs)     
+
+    def test_update_job(self):
+        recs = self.api.update_job(
+            id = 2,
+            dag = "222",
+            status = 2
+        )
+        print('update_job:', recs) 
+
+    def test_new_running(self):
+        recs = self.api.new_running(
+            job_id = 2,
+            producer_id = 2,
+            brick_id = 2,
+            start_time = '2022-04-22 13:13:13'
+        )
+        print('new_running:', recs)
+
+    def test_get_running(self):
+        recs = self.api.get_running(
+            id = 2
+        )
+        print('get_running:', recs)
+
+    def test_update_running(self):
+        recs = self.api.update_running(
+            id = 3,
+            job_id = 2,
+            producer_id = 2,
+            brick_id = 2,
+            start_time = '2022-04-22 13:13:13',
+            end_time = '2022-04-22 15:13:13',
+            prc_status = 1,
+            prc_result = 'OK'
+        )
+        print('update_running:', recs)
 
-    def test_update_qc0_status(self):
-        rec = self.api.update_qc0_status(obs_id='0000017', status = 3, )
-        print('update_qc0_status:', rec)        
+    def test_find_running(self):
+        recs = self.api.find_running(
+            job_id = 2,
+            producer_id = 2,
+            brick_id = 2,
+            create_time = ('2022-04-22 10:13:13','2022-04-22 23:13:13')
+        )
+        print('find_running:', recs)                        
\ No newline at end of file

tests/ifs/test_ifs_cal_merge.py

@@ -9,15 +9,15 @@ class IFSCalMergeApiTestCase(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_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_latest_by_l0(self):
-    #     rec = self.api.get_latest_by_l0(level0_id='000001102', ref_type = "bias")
-    #     print('get_latest_by_l0:', rec)
+    def test_get_latest_by_l0(self):
+        rec = self.api.get_latest_by_l0(level0_id='000001102', ref_type = "bias")
+        print('get_latest_by_l0:', rec)
 
     def test_get(self):
         rec = self.api.get(id = 2)
@@ -25,24 +25,24 @@ class IFSCalMergeApiTestCase(unittest.TestCase):
         rec = self.api.get(cal_id = '2')
         print('get by cal_id:', 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(
-    #         cal_id = '00002',
-    #         detector_no = '01', 
-    #         ref_type = "bias",
-    #         obs_time = "2021-06-04 11:12:13",
-    #         exp_time = 150,
-    #         filename = "/opt/dddasd1.params",
-    #         file_path = "/opt/dddasd1.fits",
-    #         prc_status = 3,
-    #         prc_time = '2021-06-04 11:12:13',
-    #         level0_ids = ['1','2','3','4'])
-    #     print('write:', 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(
+            cal_id = '00002',
+            detector_no = '01', 
+            ref_type = "bias",
+            obs_time = "2021-06-04 11:12:13",
+            exp_time = 150,
+            filename = "/opt/dddasd1.params",
+            file_path = "/opt/dddasd1.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