Skip to content
GitLab
Switch branch/tag
Find file Normal viewHistoryPermalink
test_level2.py 2.97 KB
Newer Older
Wei Shoulin's avatar
Initial commit  
Wei Shoulin committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62 
import unittest

from csst_dfs_client import level2

class Level2TestCase(unittest.TestCase):

    def setUp(self):
        pass

    def test_find(self):
        result = level2.find(obs_id = "10609400034529")
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code))
        self.assertIsNotNone(result.data, "error message: " + result.message)

    # def test_find_by_level2_id(self):
    #     result = level2.find_by_level2_id(level2_id = "1060940003452925")
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
    #     self.assertIsNotNone(result.data, "error message: " + result.message)

    # def test_update_qc2_status(self):
    #     result = level2.update_qc2_status(level2_id = "1060940003452925", data_type='csst-msc-l1-mbi-cat', qc2_status=1)
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_update_prc_status(self):
    #     result = level2.update_prc_status(level2_id = "1060940003452925", data_type='csst-msc-l1-mbi-cat', prc_status=3)
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_update_qc2_status_by_file_name(self):
    #     result = level2.update_qc2_status_by_file_name(file_name = "CSST_MSC_MS_SCIE_20290713003437_20290713003707_10160000136_24_L1_V01_CAT.fits", qc2_status=1)
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_update_prc_status_by_file_name(self):
    #     result = level2.update_prc_status_by_file_name(file_name = "CSST_MSC_MS_SCIE_20290713003437_20290713003707_10160000136_24_L1_V01_CAT.fits", prc_status=3)
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    # def test_write(self):
    #     file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCIE_20290713003437_20290713003707_10160000136_24_L1_V01_CAT.fits"
    #     result = level2.write(local_file = file_path, 
    #         module_id = "MSC",
    #         level0_id="1060940003452925",
    #         level1_id = "1060940003452925", 
    #         level2_id = "1060940003452925", 
    #         data_type='csst-msc-l1-mbi-cat', 
    #         brick_id = 254,
    #         pipeline_id = "csst-msc-l1-mbi",
    #         build=1,
    #         run_id="20241107100248",
    #         version="093",
    #         prc_status=1)
    #     print(result)
    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)

    def test_catalog_query(self):
        result = level2.catalog_query(sql="select * from tb_level2")
        print(result)
        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)