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-07 15:43:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: use utf8 and padding separately

Browse Source
This commit is contained in:
Roman 2025-02-28 03:40:25 +00:00
parent b455b57bf9
commit bf236107bb
No known key found for this signature in database
GPG Key ID: BB3828275C58EFF1
2 changed files with 14 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/steps/da.py
View File

@ -2,8 +2,11 @@ import allure
from tenacity import retry, stop_after_delay, wait_fixed from tenacity import retry, stop_after_delay, wait_fixed
from src.env_vars import NOMOS_EXECUTOR from src.env_vars import NOMOS_EXECUTOR
from src.libs.custom_logger import get_custom_logger
from src.steps.common import StepsCommon from src.steps.common import StepsCommon
logger = get_custom_logger(__name__)
def add_padding(orig_bytes): def add_padding(orig_bytes):
""" """
@ -46,14 +49,17 @@ def remove_padding(padded_bytes):
return padded_bytes[:-padding_len] return padded_bytes[:-padding_len]
def prepare_dispersal_request(data, app_id, index, with_utf8_padding=True): def prepare_dispersal_request(data, app_id, index, utf8=True, padding=True):
if with_utf8_padding: if utf8:
data_bytes = data.encode("utf-8") data_bytes = data.encode("utf-8")
padded_bytes = add_padding(list(data_bytes))
else: else:
padded_bytes = list(data) data_bytes = bytes(data)
dispersal_data = {"data": padded_bytes, "metadata": {"app_id": app_id, "index": index}} data_list = list(data_bytes)
if padding:
data_list = add_padding(data_list)
dispersal_data = {"data": data_list, "metadata": {"app_id": app_id, "index": index}}
return dispersal_data return dispersal_data
@ -80,9 +86,9 @@ class StepsDataAvailability(StepsCommon):
@allure.step @allure.step
@retry(stop=stop_after_delay(65), wait=wait_fixed(1), reraise=True) @retry(stop=stop_after_delay(65), wait=wait_fixed(1), reraise=True)
def disperse_data(self, data, app_id, index, with_utf8_padding=True): def disperse_data(self, data, app_id, index, utf8=True, padding=True):
response = [] response = []
request = prepare_dispersal_request(data, app_id, index, with_utf8_padding) request = prepare_dispersal_request(data, app_id, index, utf8=utf8, padding=padding)
executor = self.find_executor_node() executor = self.find_executor_node()
try: try:
response = executor.send_dispersal_request(request) response = executor.send_dispersal_request(request)

2
tests/networking_privacy/test_networking_privacy.py
View File

@ -53,7 +53,7 @@ class TestNetworkingPrivacy(StepsDataAvailability):
successful_dispersals = 0 successful_dispersals = 0
for i in range(20): for i in range(20):
try: try:
self.disperse_data(data_to_disperse, to_app_id(1), to_index(0), with_utf8_padding=False) self.disperse_data(data_to_disperse, to_app_id(1), to_index(0), utf8=False, padding=False)
successful_dispersals += 1 successful_dispersals += 1
except Exception as ex: except Exception as ex:
logger.warning(f"Dispersal #{i} was not successful with error {ex}") logger.warning(f"Dispersal #{i} was not successful with error {ex}")