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

fix: remaining DA endpoints

Browse Source
This commit is contained in:
Roman 2025-03-28 06:18:48 +00:00
parent 4684bbc70d
commit 0c66fd1c37
No known key found for this signature in database
GPG Key ID: 8857C6B7DBCEA320
1 changed files with 32 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/api_clients/rest.py
View File

@ -1,3 +1,5 @@
from urllib.parse import quote
from src.libs.custom_logger import get_custom_logger
import json
from src.api_clients.base_client import BaseClient
@ -31,6 +33,14 @@ class REST(BaseClient):
response = self.rest_call("get", "cryptarchia/info")
return response.json()
def cryptarchia_headers(self, from_header_id, to_header_id):
response = self.rest_call("get", f"cryptarchia/headers?from={quote(from_header_id, safe='')}" f"&to={quote(to_header_id, safe='')}")
return response.json()
def da_add_share(self, data):
response = self.rest_call("post", "da/add-share", json.dumps(data))
return response
def da_disperse_data(self, data):
response = self.rest_call("post", "disperse-data", json.dumps(data))
return response
@ -38,3 +48,25 @@ class REST(BaseClient):
def da_get_range(self, query):
response = self.rest_call("post", "da/get-range", json.dumps(query))
return response.json()
def da_get_commitments(self, query):
response = self.rest_call("get", "da/get-commitments", json.dumps(query))
return response.json()
def da_get_share(self, query):
response = self.rest_call("get", "da/get-share", json.dumps(query))
return response.json()
# DA_GET_SHARES: & str = "/da/sampling/shares"; Implemented ?
def da_block_peer(self, data):
response = self.rest_call("post", "da/block-peer", json.dumps(data))
return response
def da_unblock_peer(self, data):
response = self.rest_call("post", "da/unblock-peer", json.dumps(data))
return response
def da_blacklisted_peers(self):
response = self.rest_call("get", "da/blacklisted-peers")
return response.json()