Skip to content
GitLab
Commit 5bdb9543 authored by Wei Shoulin's avatar Wei Shoulin
Browse files

feat(level0): add batch update QC0 status function 

- Implement a new function `update_qc0_status_by_ids` for batch updating QC0 status
- Add related unit test `test_update_qc0_status_by_ids`
- Comment out unused test cases to focus on new functionality
parent 7dd5da6b
Pipeline #9176 failed with stages
in 0 seconds
Hide whitespace changes
Inline Side-by-side
csst_dfs_client/level0.py
View file @ 5bdb9543
...@@ -125,6 +125,19 @@ def update_qc0_status(level0_id: str, qc0_status: int, dataset: str = constants. ...@@ -125,6 +125,19 @@ def update_qc0_status(level0_id: str, qc0_status: int, dataset: str = constants.
""" """
return request.put(f"/api/level0/qc0_status/{level0_id}", {'qc0_status': qc0_status, 'dataset': dataset}) return request.put(f"/api/level0/qc0_status/{level0_id}", {'qc0_status': qc0_status, 'dataset': dataset})
def update_qc0_status_by_ids(ids: List[str], qc0_status: int) -> Result:
"""
根据内部_id，批量更新0级数据的QC0状态
Args:
ids (List[str]): 内部_id列表
prc_status (int): 处理状态
Returns:
Result: 更新结果
"""
return request.put("/api/level0/qc0_status/batch/update", {'qc0_status': qc0_status, 'ids': ids})
def update_prc_status(level0_id: str, dag_run: str, prc_status: int, dataset: str = constants.DEFAULT_DATASET) -> Result: def update_prc_status(level0_id: str, dag_run: str, prc_status: int, dataset: str = constants.DEFAULT_DATASET) -> Result:
""" """
更新0级数据的处理状态 更新0级数据的处理状态
......
tests/test_level0.py
View file @ 5bdb9543
...@@ -7,59 +7,64 @@ class Level0TestCase(unittest.TestCase): ...@@ -7,59 +7,64 @@ class Level0TestCase(unittest.TestCase):
def setUp(self): def setUp(self):
pass pass
def test_find(self): # def test_find(self):
start_time = time.time() # start_time = time.time()
result = level0.find(instrument='MSC', dataset="msc-v093") # result = level0.find(instrument='MSC', dataset="msc-v093")
print(f"1操作执行时间: {time.time() - start_time} 秒") # print(f"1操作执行时间: {time.time() - start_time} 秒")
start_time = time.time() # start_time = time.time()
result = level0.find(instrument='MSC', # result = level0.find(instrument='MSC',
ra_obj = 170, # ra_obj = 170,
dec_obj = -24, # dec_obj = -24,
radius = 1) # radius = 1)
print(f"2操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}") # print(f"2操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}")
start_time = time.time() # start_time = time.time()
result = level0.find(instrument='MSC', file_name="CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits") # result = level0.find(instrument='MSC', file_name="CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits")
print(f"3操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}") # print(f"3操作执行时间: {time.time() - start_time} 秒, 数据量：{result['total_count']}")
self.assertEqual(result.code, 200, "error code: " + result.message) # self.assertEqual(result.code, 200, "error code: " + result.message)
self.assertIsNotNone(result.data, "error message: " + result.message) # self.assertIsNotNone(result.data, "error message: " + result.message)
def test_find_by_level0_id(self): # def test_find_by_level0_id(self):
result = level0.find_by_level0_id(level0_id = "1060940003452925") # result = level0.find_by_level0_id(level0_id = "1060940003452925")
print(result) # print(result)
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
self.assertIsNotNone(result.data, "error message: " + result.message) # self.assertIsNotNone(result.data, "error message: " + result.message)
def test_update_qc0_status(self): # def test_update_qc0_status(self):
result = level0.update_qc0_status(level0_id = "1010910015799127", qc0_status=1) # result = level0.update_qc0_status(level0_id = "1010910015799127", qc0_status=1)
print(result) # print(result)
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
def test_update_prc_status(self): def test_update_qc0_status_by_ids(self):
result = level0.update_prc_status(level0_id = "1010910015799127", dag_run="202411071002481234", prc_status=3) result = level0.update_qc0_status_by_ids(ids = ["676ac74a530b47ca41568858"], qc0_status=4)
print(result) print(result)
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
def test_update_prc_status_by_ids(self): # def test_update_prc_status(self):
result = level0.update_prc_status_by_ids(ids = ["676ac74a530b47ca41568858"], prc_status=4) # result = level0.update_prc_status(level0_id = "1010910015799127", dag_run="202411071002481234", prc_status=3)
print(result) # print(result)
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
def test_write(self): # def test_update_prc_status_by_ids(self):
file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits" # result = level0.update_prc_status_by_ids(ids = ["676ac74a530b47ca41568858"], prc_status=4)
result = level0.write(local_file = file_path, dataset= 'msc-v093') # print(result)
print(result) # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
def test_find_process(self): # def test_write(self):
result = level0.find_process(level0_id="1060940003452925") # file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits"
print(result) # result = level0.write(local_file = file_path, dataset= 'msc-v093')
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) # print(result)
def test_add_process(self): # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
result = level0.add_process(level0_id="1060940003452925",
dag="csst-msc-l1-mbi", # def test_find_process(self):
dag_run="202411071002481234", # result = level0.find_process(level0_id="1060940003452925")
dataset="v93", # print(result)
batch_id="v930batch", # self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
prc_time="2024-11-07 10:24:12", prc_status=1, prc_module="MSC", message="") # def test_add_process(self):
print(result) # result = level0.add_process(level0_id="1060940003452925",
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) # dag="csst-msc-l1-mbi",
\ No newline at end of file # dag_run="202411071002481234",
# dataset="v93",
# batch_id="v930batch",
# prc_time="2024-11-07 10:24:12", prc_status=1, prc_module="MSC", message="")
# print(result)
# self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
\ No newline at end of file
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment