logos-co
/
nomos-specs
mirror of https://github.com/logos-co/nomos-specs.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DA Api Tests (#83) 

* Tests for da api full flow

* Fix test issues

* Da api ext tests (#85)

* Index store links blob to cert_id

* Tests for multiple indexes pointing to the same blob

* Test multiple indexes to the same blob in the full flow

* Update bytes_per_chunk to 31 bytes

---------

Co-authored-by: Daniel Sanchez Quiros <sanchez.quiros.daniel@gmail.com>
This commit is contained in:
gusto 2024-03-22 13:01:13 +02:00 committed by GitHub
parent 8dd2dabb7d
commit 53b8be7a05
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 174 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
da/api/common.py
View File

@ -8,10 +8,18 @@ from da.verifier import DABlob
@dataclass
class Metadata:
# index of VID certificate blob
index: int
# app identifier
app_id: bytes
# index of VID certificate blob
index: int
@dataclass
class VID:
# da certificate id
cert_id: bytes
# application + index information
metadata: Metadata
class BlobStore(ABC):

31
da/api/test_flow.py
View File

@ -22,16 +22,16 @@ class MockStore(BlobStore):
if metadata.index in self.app_id_store[metadata.app_id]:
raise ValueError("index already written")
blob = self.blob_store.pop(cert_id)
self.app_id_store[metadata.app_id][metadata.index] = blob
self.app_id_store[metadata.app_id][metadata.index] = cert_id
# Implements `get_multiple` method from BlobStore abstract class.
def get_multiple(self, app_id, indexes) -> List[Optional[DABlob]]:
return [
self.app_id_store[app_id].get(i) for i in indexes
self.blob_store.get(self.app_id_store[app_id].get(i), None) if self.app_id_store[app_id].get(i) else None for i in indexes
]
class TestFlow(TestCase):
def test_api_write_read(self):
expected_blob = "hello"
@ -70,3 +70,28 @@ class TestFlow(TestCase):
self.assertEqual([expected_blob], blobs)
def test_multiple_indexes_same_data(self):
expected_blob = "hello"
cert_id = b"11"*32
app_id = 1
idx1 = 1
idx2 = 2
mock_meta1 = Metadata(app_id, idx1)
mock_meta2 = Metadata(app_id, idx2)
mock_store = MockStore()
mock_store.populate(expected_blob, cert_id)
api = DAApi(mock_store)
api.write(cert_id, mock_meta1)
mock_store.populate(expected_blob, cert_id)
api.write(cert_id, mock_meta2)
blobs_idx1 = api.read(app_id, [idx1])
blobs_idx2 = api.read(app_id, [idx2])
self.assertEqual([expected_blob], blobs_idx1)
self.assertEqual([expected_blob], blobs_idx2)
self.assertEqual(mock_store.app_id_store[app_id][idx1], mock_store.app_id_store[app_id][idx2])

3
da/common.py
View File

@ -55,6 +55,9 @@ class Certificate:
aggregated_column_commitment: Commitment
row_commitments: List[Commitment]
def id(self) -> bytes:
return build_attestation_message(self.aggregated_column_commitment, self.row_commitments)
def verify(self, nodes_public_keys: List[BLSPublickey]) -> bool:
"""
List of nodes public keys should be a trusted list of verified proof of possession keys.

132
da/test_full_flow.py Normal file
View File

@ -0,0 +1,132 @@
from itertools import chain
from unittest import TestCase
from typing import List, Optional
from py_ecc.bls import G2ProofOfPossession as bls_pop
from da.common import NodeId, build_attestation_message
from da.api.common import DAApi, VID, Metadata
from da.verifier import DAVerifier, DABlob
from da.api.test_flow import MockStore
from da.dispersal import Dispersal, DispersalSettings
from da.test_encoder import TestEncoder
from da.encoder import DAEncoderParams, DAEncoder
class DAVerifierWApi:
def __init__(self, sk: int):
self.store = MockStore()
self.api = DAApi(self.store)
self.verifier = DAVerifier(sk)
def receive_blob(self, blob: DABlob):
if attestation := self.verifier.verify(blob):
# Warning: If aggregated col commitment and row commitment are the same,
# the build_attestation_message method will produce the same output.
cert_id = build_attestation_message(blob.aggregated_column_commitment, blob.rows_commitments)
self.store.populate(blob, cert_id)
return attestation
def receive_cert(self, vid: VID):
# Usually the certificate would be verifier here,
# but we are assuming that this it is already coming from the verified block,
# in which case all certificates had been already verified by the DA Node.
self.api.write(vid.cert_id, vid.metadata)
def read(self, app_id, indexes) -> List[Optional[DABlob]]:
return self.api.read(app_id, indexes)
class TestFullFlow(TestCase):
def setUp(self):
self.n_nodes = 16
self.nodes_ids = [NodeId(x.to_bytes(length=32, byteorder='big')) for x in range(self.n_nodes)]
self.secret_keys = list(range(1, self.n_nodes+1))
self.public_keys = [bls_pop.SkToPk(sk) for sk in self.secret_keys]
dispersal_settings = DispersalSettings(
self.nodes_ids,
self.public_keys,
self.n_nodes
)
self.dispersal = Dispersal(dispersal_settings)
self.encoder_test = TestEncoder()
self.encoder_test.setUp()
self.api_nodes = [DAVerifierWApi(k) for k in self.secret_keys]
def test_full_flow(self):
app_id = int.to_bytes(1)
index = 1
# encoder
data = self.encoder_test.data
encoding_params = DAEncoderParams(column_count=self.n_nodes // 2, bytes_per_chunk=31)
encoded_data = DAEncoder(encoding_params).encode(data)
# mock send and await method with local verifiers
def __send_and_await_response(node: int, blob: DABlob):
node = self.api_nodes[int.from_bytes(node)]
return node.receive_blob(blob)
# inject mock send and await method
self.dispersal._send_and_await_response = __send_and_await_response
certificate = self.dispersal.disperse(encoded_data)
vid = VID(
certificate.id(),
Metadata(app_id, index)
)
# verifier
for node in self.api_nodes:
node.receive_cert(vid)
# read from api and confirm its working
# notice that we need to sort the api_nodes by their public key to have the blobs sorted in the same fashion
# we do actually do dispersal.
blobs = list(chain.from_iterable(
node.read(app_id, [index])
for node in sorted(self.api_nodes, key=lambda n: bls_pop.SkToPk(n.verifier.sk))
))
original_blobs = list(self.dispersal._prepare_data(encoded_data))
self.assertEqual(blobs, original_blobs)
def test_same_blob_multiple_indexes(self):
app_id = int.to_bytes(1)
indexes = [1, 2, 3] # Different indexes to test with the same blob
# encoder
data = self.encoder_test.data
encoding_params = DAEncoderParams(column_count=self.n_nodes // 2, bytes_per_chunk=31)
encoded_data = DAEncoder(encoding_params).encode(data)
# mock send and await method with local verifiers
def __send_and_await_response(node: int, blob: DABlob):
node = self.api_nodes[int.from_bytes(node)]
return node.receive_blob(blob)
# inject mock send and await method
self.dispersal._send_and_await_response = __send_and_await_response
certificate = self.dispersal.disperse(encoded_data)
# Loop through each index and simulate dispersal with the same cert_id but different metadata
for index in indexes:
vid = VID(
certificate.id(),
Metadata(app_id, index)
)
# verifier
for node in self.api_nodes:
node.receive_cert(vid)
# Verify retrieval for each index
for index in indexes:
# Notice that we need to sort the api_nodes by their public key to have the blobs sorted in the same fashion
# as we do actually do dispersal.
blobs = list(chain.from_iterable(
node.read(app_id, [index])
for node in sorted(self.api_nodes, key=lambda n: bls_pop.SkToPk(n.verifier.sk))
))
original_blobs = list(self.dispersal._prepare_data(encoded_data))
self.assertEqual(blobs, original_blobs, f"Failed at index {index}")