Update and modify files for the project

- Added new files:
  - tests/unit/chunk/test_soc.py
- Updated files:
  - pdm.lock
  - pyproject.toml
  - src/bee_py/chunk/cac.py
  - src/bee_py/chunk/soc.py
  - src/bee_py/modules/pss.py
  - tests/conftest.py
  - tests/integration/modules/conftest.py
  - tests/integration/modules/test_pss.py

pyproject.toml

@@ -8,6 +8,8 @@ dependencies = [
     "eth-ape>=0.6.26",
     "structlog>=23.2.0",
     "semver>=3.0.2",
+    "websocket-client>=1.6.4",
+    "pytest-asyncio>=0.21.1",
 ]
 requires-python = ">=3.9"
 readme = "README.md"

src/bee_py/chunk/cac.py

@@ -1,4 +1,7 @@
+from typing import Union
+
 from hexbytes import HexBytes
+from pydantic import BaseModel, Field
 
 from bee_py.chunk.bmt import bmt_hash
 from bee_py.chunk.serialize import serialize_bytes
@@ -12,7 +15,7 @@
 CAC_PAYLOAD_OFFSET = CAC_SPAN_OFFSET + SPAN_SIZE
 
 
-class Chunk:
+class Chunk(BaseModel):
     """
     * General chunk class for Swarm
 
@@ -23,15 +26,10 @@ class Chunk:
     the chunk that is required for the Chunk API.
     """
 
-    def __init__(self, data: bytes, span: bytes):
-        self.data = data
-        self.span = span
-
-    def payload(self) -> bytes:
-        return flex_bytes_at_offset(self.data, CAC_PAYLOAD_OFFSET, MIN_PAYLOAD_SIZE, MAX_PAYLOAD_SIZE)
-
-    def address(self) -> HexBytes:
-        return bmt_hash(self.data)
+    data: bytes = Field(..., description="The data of the chunk")
+    span: bytes = Field(..., description="The span of the chunk")
+    payload: bytes = Field(..., description="The payload of the chunk")
+    address: Union[HexBytes, bytes] = Field(..., description="The address of the chunk")
 
 
 def is_valid_chunk_data(data: bytes, chunk_address: bytes) -> bool:
@@ -72,5 +70,7 @@ def make_content_addressed_chunk(payload_bytes: bytes) -> Chunk:
     span = make_span(len(payload_bytes))
 
     data = serialize_bytes(span, payload_bytes)
+    payload = flex_bytes_at_offset(data, CAC_PAYLOAD_OFFSET, MIN_PAYLOAD_SIZE, MAX_PAYLOAD_SIZE)
+    address = bmt_hash(data)
 
-    return Chunk(data, span)
+    return Chunk(data=data, span=span, payload=payload, address=address)

src/bee_py/chunk/soc.py

@@ -4,7 +4,7 @@
 from eth_account import Account
 from eth_typing import ChecksumAddress as AddressType
 from hexbytes import HexBytes
-from pydantic.dataclasses import dataclass
+from pydantic import BaseModel, Field
 
 from bee_py.chunk.bmt import bmt_hash
 from bee_py.chunk.cac import (
@@ -38,8 +38,7 @@
 Identifier = NewType("Identifier", bytes)
 
 
-@dataclass
-class SingleOwnerChunkBase:
+class SingleOwnerChunkBase(BaseModel):
     """Abstract base class for Single Owner Chunks (SOCs).
 
     Represents a SOC, a type of chunk that allows a user to assign arbitrary data to an address
@@ -52,12 +51,12 @@ class SingleOwnerChunkBase:
         owner: The owner of the SOC.
     """
 
-    identifier: Identifier
-    signature: bytes
-    owner: AddressType
+    identifier: str = Field(..., description="The identifier of the SOC")
+    signature: bytes = Field(..., description="The signature of the SOC")
+    owner: str = Field(..., description="The owner of the SOC")
 
 
-class SingleOwnerChunk(SingleOwnerChunkBase, Chunk):
+class SingleOwnerChunk(Chunk, SingleOwnerChunkBase):
     """Represents a Single Owner Chunk (SOC).
 
     A concrete implementation of the SingleOwnerChunkBase class. It represents a SOC
@@ -73,21 +72,7 @@ class SingleOwnerChunk(SingleOwnerChunkBase, Chunk):
         owner: The owner of the SOC.
     """
 
-    def __init__(
-        self,
-        data: bytes,
-        identifier: Identifier,
-        signature: bytes,
-        span: bytes,
-        payload: bytes,
-        address: HexBytes,
-        owner: AddressType,
-    ):
-        SingleOwnerChunkBase.__init__(self, identifier, signature, owner)
-        self.data = data
-        self.span = span
-        self.payload = payload
-        self.address = address
+    pass
 
 
 def recover_chunk_owner(data: bytes) -> AddressType:
@@ -164,53 +149,54 @@ def make_soc_address(identifier: Identifier, address: AddressType) -> bytes:
 
 def make_single_owner_chunk(
     chunk: Chunk,
-    identifier: Identifier,
+    identifier: Union[Identifier, bytearray],
     signer: Union[AccountAPI, Account],
 ) -> SingleOwnerChunk:
     """
     Creates a single owner chunk object.
 
     Args:
         chunk: A chunk object used for the span and payload.
-        identifier: The identifier of the chunk.
+        identifier|bytearray: The identifier of the chunk.
         signer: The singer interface for signing the chunk.
             signer can be a ape account API or a eth_account object.
 
     Returns:
         SingleOwnerChunk: SingleOwnerChunk object.
     """
-    chunk_address = chunk.address()
+
+    if isinstance(identifier, bytearray):
+        # make it in raw bytes
+        identifier = bytes(identifier)
+
+    chunk_address = chunk.address
     assert_valid_chunk_data(chunk.data, chunk_address)
+    print(identifier)
 
     digest = keccak256_hash(identifier, chunk_address)
+    print("Chunk address: ", bytes_to_hex(chunk_address))
+    print("Identifier: ", bytes_to_hex(identifier))
+    print("Digest: ", digest)
+
     signature = sign(account=signer, data=digest)
+    print("Signature: ", signature.encode_vrs().hex())
 
     if isinstance(signer, AccountAPI):
         encoded_signature = signature.encode_vrs()
-        data = serialize_bytes(identifier, encoded_signature, chunk.span, chunk.payload())
+        data = serialize_bytes(identifier, encoded_signature, chunk.span, chunk.payload)
     else:
         encoded_signature = signature.signature.hex()
         encoded_signature_bytes = hex_to_bytes(encoded_signature)
-        data = serialize_bytes(identifier, encoded_signature_bytes, chunk.span, chunk.payload())
+        data = serialize_bytes(identifier, encoded_signature_bytes, chunk.span, chunk.payload)
 
     address = make_soc_address(identifier, signer.address)
 
-    # return {
-    #     "data": data,
-    #     "identifier": identifier,
-    #     "signature": signature,
-    #     "span": chunk.span,
-    #     "payload": chunk.payload(),
-    #     "address": address,
-    #     "owner": signer.address,
-    # }
-
     return SingleOwnerChunk(
         data=data,
         identifier=identifier,
         signature=encoded_signature,
         span=chunk.span,
-        payload=chunk.payload(),
+        payload=chunk.payload,
         address=address,
         owner=signer.address,
     )

src/bee_py/modules/pss.py

@@ -52,7 +52,7 @@ def send(
             logger.error(response.raise_for_status())
 
 
-def subscribe(url: str, topic: str) -> websockets.WebSocketClientProtocol:
+async def subscribe(url: str, topic: str) -> websockets.WebSocketClientProtocol:
     """
     Subscribes to messages on the given topic.
 
@@ -65,4 +65,4 @@ def subscribe(url: str, topic: str) -> websockets.WebSocketClientProtocol:
     """
     ws_url = url.replace("http", "ws")
     ws_url = f"{ws_url}/{PSS_ENDPOINT}/subscribe/{topic}"
-    return websockets.connect(ws_url)
+    return await websockets.connect(ws_url)

tests/conftest.py

@@ -135,7 +135,7 @@ def get_debug_postage(printer, bee_debug_ky_options) -> BatchId:
     stamp: BatchId
 
     printer("[*]Getting Debug Postage....")
-    return "6914152b3ccac7411220f76126e31b30fc52fd52c9e73afd5aecda06c85232e3"
+    #return "6be2b2aacd687180a109d960cb9b1da0b446f4fca3966c6b7fdd84e6d9fdad52"
 
     # if read_local_bee_stamp:
     #     printer(read_local_bee_stamp)
@@ -157,11 +157,12 @@ def get_debug_postage(printer, bee_debug_ky_options) -> BatchId:
 
 
 @pytest.fixture
-def get_peer_debug_postage(printer, read_local_bee_peer_stamp, bee_peer_debug_ky_options) -> BatchId:
+def get_peer_debug_postage(printer, bee_peer_debug_ky_options) -> BatchId:
     stamp: BatchId
 
-    if read_local_bee_peer_stamp:
-        return read_local_bee_peer_stamp
+    # if read_local_bee_peer_stamp:
+    #     return read_local_bee_peer_stamp
+    #return "6be2b2aacd687180a109d960cb9b1da0b446f4fca3966c6b7fdd84e6d9fdad52"
 
     printer("[*]Getting Debug Postage....")
     stamp = create_postage_batch(bee_peer_debug_ky_options, 100, 20)

tests/integration/modules/conftest.py

@@ -1,10 +1,10 @@
-import json
+# import json
 from pathlib import Path
 
-import pytest
+# import pytest
 
-from bee_py.modules.debug.stamps import create_postage_batch, get_postage_batch
-from bee_py.types.type import BatchId
+# from bee_py.modules.debug.stamps import create_postage_batch, get_postage_batch
+# from bee_py.types.type import BatchId
 
 # test_chunk
 

tests/integration/modules/test_pss.py

@@ -10,16 +10,14 @@ def make_test_target(target: str) -> str:
     return target[0:2]
 
 
-#! these tests only work when there is at least one peer connected
+# ! these tests only work when there is at least one peer connected
 
 
 @pytest.mark.timeout(PSS_TIMEOUT)
 def test_send_pss_message(bee_ky_options, bee_debug_ky_options, get_debug_postage):
     topic = "send-pss-message"
     message = "hello"
 
-    # get_debug_postage = "5e631cc0daab23b52d61e0ba2898034c051aced1d58aa804f269f0eb2239dbaa"
-
     peers = get_peers(bee_debug_ky_options)
     assert len(peers) > 0
 
@@ -29,15 +27,17 @@ def test_send_pss_message(bee_ky_options, bee_debug_ky_options, get_debug_postag
 
 
 @pytest.mark.timeout(PSS_TIMEOUT)
-def test_send_receive_pss_message(bee_url, bee_peer_ky_options, bee_debug_ky_options, get_peer_debug_postage):
+@pytest.mark.asyncio
+async def test_send_receive_pss_message(bee_url, bee_debug_ky_options, bee_peer_ky_options, get_peer_debug_postage):
     topic = "send-receive-pss-message"
     message = "hello"
 
-    ws = subscribe(bee_url, topic)
+    ws = await subscribe(bee_url, topic)
 
     addresses = get_node_addresses(bee_debug_ky_options)
     target = addresses.overlay
     send(bee_peer_ky_options, topic, make_test_target(target), message, get_peer_debug_postage)
 
-    received_message = ws.receive()
+    received_message = await ws.recv()
     assert received_message == message
+    ws.close()

tests/unit/chunk/test_soc.py

@@ -0,0 +1,28 @@
+from ape import accounts
+
+from bee_py.chunk.cac import make_content_addressed_chunk
+from bee_py.chunk.signer import sign
+from bee_py.chunk.soc import make_single_owner_chunk
+from bee_py.modules.soc import upload
+from bee_py.utils.hex import bytes_to_hex, hex_to_bytes
+
+
+def test_single_owner_chunk_creation():
+    payload = bytes([1, 2, 3])
+    soc_hash = "9d453ebb73b2fedaaf44ceddcf7a0aa37f3e3d6453fea5841c31f0ea6d61dc85"
+    identifier = bytearray(32)
+
+    cac = make_content_addressed_chunk(payload)
+    print(bytes_to_hex(cac.data))
+
+    signer = accounts.load("bee")
+    signer.set_autosign(True, passphrase="a")
+
+    soc = make_single_owner_chunk(cac, identifier, signer)
+    print(bytes_to_hex(soc.data))
+
+    soc_address = bytes_to_hex(soc.address)
+    owner = soc.owner
+
+    assert soc_address == soc_hash
+    assert owner == signer.address