Skip to content
GitLab
An error occurred while loading the file. Please try again.
Switch branch/tag
Find file BlameHistoryPermalink
test_level1.py 3.40 KiB
Open in Web IDE
1 
import unittest
2
3 
from csst_dfs_client import level1
4
5 
class Level1TestCase(unittest.TestCase):
6
7 
    def setUp(self):
8 
        pass
9
10 
    def test_find(self):
11 
        result = level1.find(instrument='MSC', obs_time=("2024-01-01 12:12:12", "2024-06-30 12:12:12"), dataset='msc-v093')
12 
        print(result)
13 
        self.assertEqual(result.code, 200, "error code: " + str(result.code))
14 
        self.assertIsNotNone(result.data, "error message: " + result.message)
15 
    def test_find_by_brick_id(self):
16 
        result = level1.find_by_brick_id(brick_id = 401847)
17 
        print(result)
18 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
19 
        self.assertIsNotNone(result.data, "error message: " + result.message)
20 
    def test_find_by_level1_id(self):
21 
        result = level1.find_by_level1_id(level1_id = "1060940003452925")
22 
        print(result)
23 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
24 
        self.assertIsNotNone(result.data, "error message: " + result.message)
25
26 
    def test_update_qc1_status(self):
27 
        result = level1.update_qc1_status(level1_id = "1060940003452925", data_model='csst-msc-l1-mbi', qc1_status=1)
28 
        print(result)
29 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
30
31 
    def test_update_prc_status(self):
32 
        result = level1.update_prc_status(level1_id = "1060940003452925", data_model='csst-msc-l1-mbi', prc_status=3)
33 
        print(result)
34 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
35
36 
    def test_write(self):
37 
        file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L1_V01.fits"
38 
        # from io import BytesIO
39 
        # with open(file_path, "rb") as file:
40 
        #     file_path = BytesIO(file.read())
41
42 
        result = level1.write(local_file = file_path,
43 
            obs_group="10109100157991",
44 
            obs_type="WIDE",
45 
            instrument = "MSC",
46 
            level0_id="1010910015799127",
47 
            level1_id = "1010910015799127",
48 
            dag = "csst-msc-l1-mbi",
49 
            file_name= "CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L1_V01.fits",
50 
            pmapname="csst_000128.pmap",
51 
            data_model='csst-msc-l1-mbi',
52 
            build=1,
53 
            qc1_status=10,
54 
            prc_status=12)
55 
        print(result)
56 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
57
58 
    def test_find_process(self):
59 
        result = level1.find_process(level1_id="1060940003452925")
60 
        print(result)
61 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
62 
    def test_add_process(self):
63 
        result = level1.add_process(level1_id="1060940003452925", dag="csst-msc-l1-mbi", dag_run="202411071002481234", prc_time="2024-11-07 10:24:12", prc_status=1, prc_module="MSC", message="")
64 
        print(result)
65 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
66
67 
    def test_sls_find_by_qc1_status(self):
68 
        result = level1.sls_find_by_qc1_status(qc1_status=0, limit=1)
69 
        print(result)
70 
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)