usov_i-org/bec
1
0
Fork 0
mirror of https://github.com/ivan-usov-org/bec.git synced 2025-04-21 10:10:02 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

test(file_writer): updated test data to new xrange return values

Browse Source
This commit is contained in:
wakonig_k 2024-02-19 18:13:10 +01:00
parent 0fe0a6e119
commit d5cfbb3753
1 changed files with 22 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
file_writer/tests/test_file_writer_manager.py
View File

@ -207,7 +207,7 @@ def test_update_async_data():
def test_process_async_data_single_entry(): def test_process_async_data_single_entry():
file_manager = load_FileWriter() file_manager = load_FileWriter()
file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID") file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID")
data = [(b"0-0", {b"data": messages.DeviceMessage(signals={"data": np.zeros((10, 10))})})] data = [{"data": messages.DeviceMessage(signals={"data": np.zeros((10, 10))})}]
file_manager._process_async_data(data, "scanID", "dev1") file_manager._process_async_data(data, "scanID", "dev1")
assert np.isclose( assert np.isclose(
file_manager.scan_storage["scanID"].async_data["dev1"]["data"], np.zeros((10, 10)) file_manager.scan_storage["scanID"].async_data["dev1"]["data"], np.zeros((10, 10))
@ -218,32 +218,29 @@ def test_process_async_data_extend():
file_manager = load_FileWriter() file_manager = load_FileWriter()
file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID") file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID")
data = [ data = [
( {
b"0-0", "data": messages.DeviceMessage(
{ signals={"data": {"value": np.zeros((10, 10))}}, metadata={"async_update": "extend"}
b"data": messages.DeviceMessage( )
signals={"data": np.zeros((10, 10))}, metadata={"async_update": "extend"} }
)
},
)
for ii in range(10) for ii in range(10)
] ]
file_manager._process_async_data(data, "scanID", "dev1") file_manager._process_async_data(data, "scanID", "dev1")
assert file_manager.scan_storage["scanID"].async_data["dev1"]["data"].shape == (100, 10) assert file_manager.scan_storage["scanID"].async_data["dev1"]["data"]["value"].shape == (
100,
10,
)
def test_process_async_data_append(): def test_process_async_data_append():
file_manager = load_FileWriter() file_manager = load_FileWriter()
file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID") file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID")
data = [ data = [
( {
b"0-0", "data": messages.DeviceMessage(
{ signals={"data": {"value": np.zeros((10, 10))}}, metadata={"async_update": "append"}
b"data": messages.DeviceMessage( )
signals={"data": np.zeros((10, 10))}, metadata={"async_update": "append"} }
)
},
)
for ii in range(10) for ii in range(10)
] ]
file_manager._process_async_data(data, "scanID", "dev1") file_manager._process_async_data(data, "scanID", "dev1")
@ -254,18 +251,16 @@ def test_process_async_data_replace():
file_manager = load_FileWriter() file_manager = load_FileWriter()
file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID") file_manager.scan_storage["scanID"] = ScanStorage(10, "scanID")
data = [ data = [
( {
b"0-0", "data": messages.DeviceMessage(
{ signals={"data": {"value": np.zeros((10, 10))}},
b"data": messages.DeviceMessage( metadata={"async_update": "replace"},
signals={"data": np.zeros((10, 10))}, metadata={"async_update": "replace"} )
) }
},
)
for ii in range(10) for ii in range(10)
] ]
file_manager._process_async_data(data, "scanID", "dev1") file_manager._process_async_data(data, "scanID", "dev1")
assert file_manager.scan_storage["scanID"].async_data["dev1"]["data"].shape == (10, 10) assert file_manager.scan_storage["scanID"].async_data["dev1"]["data"]["value"].shape == (10, 10)
def test_update_scan_storage_with_status_ignores_none(): def test_update_scan_storage_with_status_ignores_none():