Refactor lib9c models

common/_graphql.py

@@ -80,6 +80,28 @@ def _unload_from_garage(self, pubkey: bytes, nonce: int, **kwargs) -> bytes:
         result = self.execute(query)
         return bytes.fromhex(result["actionTxQuery"]["unloadFromMyGarages"])
 
+    def _claim_items(self, pubkey: bytes, nonce: int, **kwargs) -> bytes:
+        ts = kwargs.get("timestamp", (datetime.datetime.utcnow()+datetime.timedelta(days=1)).isoformat())
+        claim_data = kwargs.get("claim_data")
+        memo = kwargs.get("memo")
+
+        query = dsl_gql(
+            DSLQuery(
+                self.ds.StandaloneQuery.actionTxQuery.args(
+                    publicKey=pubkey.hex(),
+                    nonce=nonce,
+                    timestamp=ts,
+                ).select(
+                    self.ds.ActionTxQuery.claimItems.args(
+                        claimData=claim_data,
+                        memo=memo,
+                    )
+                )
+            )
+        )
+        result = self.execute(query)
+        return bytes.fromhex(result["actionTxQuery"]["unloadFromMyGarages"])
+
     def _transfer_asset(self, pubkey: bytes, nonce: int, **kwargs) -> bytes:
         ts = kwargs.get("timestamp", (datetime.datetime.utcnow()+datetime.timedelta(days=1)).isoformat())
         sender = kwargs.get("sender")

common/lib9c/models/address.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+
+class Address:
+    def __init__(self, addr: str):
+        if addr.startswith("0x"):
+            if len(addr) != 42:
+                raise ValueError("Address with 0x prefix must have exact 42 chars.")
+            self.raw = bytes.fromhex(addr[2:])
+        else:
+            if len(addr) != 40:
+                raise ValueError("Address without 0x prefix must have exact 40 chars.")
+            self.raw = bytes.fromhex(addr)
+
+    @property
+    def long_format(self):
+        return f"0x{self.raw.hex()}"
+
+    @property
+    def short_format(self):
+        return self.raw.hex()
+
+    def __eq__(self, other: Address):
+        return self.raw == other.raw

common/lib9c/models/currency.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+from typing import Dict, Union, List, Optional, Any
+
+import bencodex
+
+from common.lib9c.models.address import Address
+
+
+class Currency:
+    """
+    # Currency
+    ---
+    Lib9c Currency model which has ticker, minters, decimal_places, total_supply_trackable.
+    `minters` will be automatically sanitized to `None` if empty list provided.
+    """
+
+    def __init__(self, ticker: str, decimal_places: int, minters: Optional[List[str]] = None,
+                 total_supply_trackable: bool = False):
+        self.ticker = ticker
+        self.minters = [Address(x) for x in minters] if minters else None
+        self.decimal_places = decimal_places
+        self.total_supply_trackable = total_supply_trackable
+
+    def __eq__(self, other:Currency):
+        return (
+            self.ticker == other.ticker and
+            self.minters == other.minters and
+            self.decimal_places == other.decimal_places and
+            self.total_supply_trackable == other.total_supply_trackable
+        )
+
+    @classmethod
+    def NCG(cls):
+        return cls(
+            ticker="NCG",
+            minters=["47d082a115c63e7b58b1532d20e631538eafadde"],
+            decimal_places=2
+        )
+
+    @classmethod
+    def CRYSTAL(cls):
+        return cls(
+            ticker="CRYSTAL",
+            minters=None,
+            decimal_places=18
+        )
+
+    @property
+    def plain_value(self) -> Dict[str, Any]:
+        value = {
+            "ticker": self.ticker,
+            "decimalPlaces": chr(self.decimal_places).encode(),
+            "minters": [x.short_format for x in self.minters] if self.minters else None
+        }
+        if self.total_supply_trackable:
+            value["totalSupplyTrackable"] = True
+        return value
+
+    @property
+    def serialized_plain_value(self) -> bytes:
+        return bencodex.dumps(self.plain_value)
+
+    @staticmethod
+    def to_currency_plain_value(ticker: str) -> Dict[str, Union[str, int, None]]:
+        if ticker.lower() == "crystal":
+            return {
+                "decimalPlaces": b'\x12',
+                "minters": None,
+                "ticker": "CRYSTAL",
+            }
+        elif ticker.lower() == "garage":
+            return {
+                "decimalPlaces": b'\x12',
+                "minters": None,
+                "ticker": "GARAGE",
+                "totalSupplyTrackable": True,
+            }
+        else:
+            return {
+                "decimalPlaces": b'\x00',
+                "minters": None,
+                "ticker": ticker.upper(),
+            }
+
+    @staticmethod
+    def serialize(currency: Dict[str, Union[str, int, None]]) -> bytes:
+        return bencodex.dumps(currency)

common/lib9c/models/fungible_asset_value.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from typing import Dict, Union, List, Optional, Any
+
+import bencodex
+
+from common.lib9c.models.currency import Currency
+
+
+class FungibleAssetValue:
+    def __init__(self, currency: Currency, amount: float):
+        self.currency = currency
+        self.amount = amount
+
+    def __eq__(self, other: FungibleAssetValue):
+        return self.currency == other.currency and self.amount == other.amount
+
+    @classmethod
+    def from_raw_data(
+            cls,
+            ticker: str, decimal_places: int, minters: Optional[List[str]] = None, total_supply_trackable: bool = False,
+            amount: float = 0
+    ):
+        return cls(
+            Currency(ticker, decimal_places, minters, total_supply_trackable),
+            amount=amount
+        )
+
+    @property
+    def plain_value(self) -> List[Dict[str, Any] | float]:
+        return [self.currency.plain_value, self.amount * max(1, 10 ** self.currency.decimal_places)]
+
+    @property
+    def serialized_plain_value(self) -> bytes:
+        return bencodex.dumps(self.plain_value)
+
+    @staticmethod
+    def to_fungible_asset(ticker: str, amount: int, decimal_places: int) \
+            -> List[Union[Dict[str, Union[str, int, None]], int]]:
+        return [Currency.to_currency_plain_value(ticker), amount * max(1, 10 ** decimal_places)]
+
+    @staticmethod
+    def serialize(fungible_asset: List[Union[Dict[str, Union[str, int, None]], int]]) -> bytes:
+        return bencodex.dumps([Currency.serialize(fungible_asset[0]), fungible_asset[1]])

common/utils/actions.py

@@ -1,6 +1,6 @@
 from typing import List, Optional, Dict, Any
 
-from common.lib9c.fungible_asset import FungibleAsset
+from common.lib9c.models.fungible_asset_value import FungibleAssetValue
 
 
 def create_unload_my_garages_action_plain_value(id: str, fav_data: List, avatar_addr: str, item_data: List, memo: Optional[str]) -> Dict[str, Any]:
@@ -15,7 +15,7 @@ def create_unload_my_garages_action_plain_value(id: str, fav_data: List, avatar_
                 [
                     [
                         bytes.fromhex(x['balanceAddr'][2:] if x["balanceAddr"].startswith("0x") else x["balanceAddr"]),
-                        FungibleAsset.to_fungible_asset(x['value']['currencyTicker'], int(x['value']['value']), int(x['value']['decimalPlaces']))
+                        FungibleAssetValue.to_fungible_asset(x['value']['currencyTicker'], int(x['value']['value']), int(x['value']['decimalPlaces']))
                     ]
                     for x in fav_data
                 ],

tests/lib9c/models/test_address.py

@@ -0,0 +1,20 @@
+import pytest
+
+from common.lib9c.models.address import Address
+
+
+@pytest.mark.parametrize("addr",
+                         ["0xa5f7e0bd63AD2749D66380f36Eb33Fe0ba50A27D",
+                          "0xb3cbca0e64aeb4b5b861047fe1db5a1bec1c241f",
+                          "a5f7e0bd63AD2749D66380f36Eb33Fe0ba50A27D",
+                          "b3cbca0e64aeb4b5b861047fe1db5a1bec1c241f",
+                          ])
+def test_address(addr):
+    address = Address(addr)
+    assert len(address.raw) == 20
+    if addr.startswith("0x"):
+        assert address.raw == bytes.fromhex(addr[2:])
+        assert address.long_format == addr.lower()
+    else:
+        assert address.raw == bytes.fromhex(addr)
+        assert address.short_format == addr.lower()

tests/lib9c/models/test_currency.py

@@ -0,0 +1,58 @@
+import pytest
+
+from common.lib9c.models.address import Address
+from common.lib9c.models.currency import Currency
+
+TEST_DATASET = [
+    ("NCG", 2, ["47d082a115c63e7b58b1532d20e631538eafadde"], False,
+     b'du13:decimalPlaces1:\x02u7:minterslu40:47d082a115c63e7b58b1532d20e631538eafaddeeu6:tickeru3:NCGe'),
+    ("CRYSTAL", 18, None, False, b'du13:decimalPlaces1:\x12u7:mintersnu6:tickeru7:CRYSTALe'),
+    ("GARAGE", 18, None, True, b'du13:decimalPlaces1:\x12u7:mintersnu6:tickeru6:GARAGEu20:totalSupplyTrackablete'),
+    ("OTHER", 0, None, False, b'du13:decimalPlaces1:\x00u7:mintersnu6:tickeru5:OTHERe'),
+    ("OTHER", 0, [], False, b'du13:decimalPlaces1:\x00u7:mintersnu6:tickeru5:OTHERe'),
+    ("OTHER", 0, ["0x896cB1A849d8818BF8e1fcf4166DafD67E27Dce0"], False,
+     b'du13:decimalPlaces1:\x00u7:minterslu40:896cb1a849d8818bf8e1fcf4166dafd67e27dce0eu6:tickeru5:OTHERe'),
+    ("OTHER", 0, ["0x896cB1A849d8818BF8e1fcf4166DafD67E27Dce0", "0x3C32731b77C5D99D186572E5ce5d6AA93A8853dC"], False,
+     b'du13:decimalPlaces1:\x00u7:minterslu40:896cb1a849d8818bf8e1fcf4166dafd67e27dce0u40:3c32731b77c5d99d186572e5ce5d6aa93a8853dceu6:tickeru5:OTHERe'),
+]
+
+
+@pytest.mark.parametrize("test_data", TEST_DATASET)
+def test_currency(test_data):
+    ticker, decimal_places, minters, total_supply_trackable, _ = test_data
+    currency = Currency(ticker, decimal_places, minters, total_supply_trackable)
+    assert currency.ticker == ticker
+    assert currency.decimal_places == decimal_places
+    assert currency.minters == ([Address(x) for x in minters] if minters else None)
+    if total_supply_trackable:
+        assert currency.total_supply_trackable is total_supply_trackable
+
+
+def test_well_known_currency():
+    test_ncg = Currency.NCG()
+    expected_ncg = Currency("NCG", 2, ["47d082a115c63e7b58b1532d20e631538eafadde"], False)
+    assert test_ncg == expected_ncg
+
+    test_crystal = Currency.CRYSTAL()
+    expected_crystal = Currency("CRYSTAL", 18, None, False)
+    assert test_crystal == expected_crystal
+
+
+@pytest.mark.parametrize("test_data", TEST_DATASET)
+def test_plain_value(test_data):
+    ticker, decimal_places, minters, total_supply_trackable, _ = test_data
+    currency = Currency(ticker, decimal_places, minters, total_supply_trackable)
+    plain_value = currency.plain_value
+    assert plain_value["ticker"] == ticker
+    assert plain_value["decimalPlaces"] == chr(decimal_places).encode()
+    assert plain_value["minters"] == (
+        [x[2:].lower() if x.startswith("0x") else x.lower() for x in minters] if minters else None)
+    if total_supply_trackable:
+        assert plain_value["totalSupplyTrackable"] == total_supply_trackable
+
+
+@pytest.mark.parametrize("test_data", TEST_DATASET)
+def test_serialized_plain_value(test_data):
+    ticker, decimal_places, minters, total_supply_trackable, serialized = test_data
+    currency = Currency(ticker, decimal_places, minters, total_supply_trackable)
+    assert currency.serialized_plain_value == serialized