logos-blockchain/logos-blockchain-e2e-tests
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-e2e-tests.git synced 2026-01-02 13:13:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: disperse and retrieve with one node stopped

Browse Source
This commit is contained in:
Roman 2025-03-25 07:56:16 +00:00
parent 577de5c9fe
commit b006bda4a9
No known key found for this signature in database
GPG Key ID: 8857C6B7DBCEA320
1 changed files with 24 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
tests/data_integrity/test_data_integrity.py
View File

@ -20,7 +20,7 @@ class TestDataIntegrity(StepsDataAvailability):
self.disperse_data(DATA_TO_DISPERSE[1], to_app_id(1), to_index(0))
delay(5)
test_results = []
# Iterate through standard nodes to get blob data for 1/2 columns
# Iterate through standard nodes 1-3 to get blob data for 1/2 columns
for node in self.main_nodes[1:4]:
rcv_data = self.get_data_range(node, to_app_id(1), to_index(0), to_index(5))
rcv_data_json = json.dumps(rcv_data)
@ -34,6 +34,29 @@ class TestDataIntegrity(StepsDataAvailability):
logger.info(f"Dispersed data received by : {test_results}")
@pytest.mark.usefixtures("setup_4_node_cluster")
def test_da_disperse_retrieve_one_node_stopped(self):
# Stop nomos node 1 - dispersal and data retrieval should be still possible
self.main_nodes[1].stop()
self.disperse_data(DATA_TO_DISPERSE[1], to_app_id(1), to_index(0))
delay(5)
test_results = []
# Iterate through standard nodes 2-3 to get blob data for 1/2 columns
for node in self.main_nodes[2:4]:
rcv_data = self.get_data_range(node, to_app_id(1), to_index(0), to_index(5))
rcv_data_json = json.dumps(rcv_data)
reconstructed_data = NomosCli(command="reconstruct").run(input_values=[rcv_data_json])
if DATA_TO_DISPERSE[1] == reconstructed_data:
test_results.append(node.name())
assert len(test_results) > 0, "Dispersed data were not received by any node"
logger.info(f"Dispersed data received by : {test_results}")
@pytest.mark.usefixtures("setup_2_node_cluster")
def test_da_sampling_determines_data_presence(self):
self.disperse_data(DATA_TO_DISPERSE[1], to_app_id(1), to_index(0))