Gp/v119

.github/workflows/python-app.yml → .github/workflows/ci_integration_tests.yml

@@ -1,7 +1,7 @@
 # This workflow will install Python dependencies, run tests and lint with a single version of Python
 # For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
 
-name: Python application
+name: ci_integration_tests
 
 on:
   push:

download.py

@@ -22,6 +22,7 @@
     "0.116.1",
     "0.117.0",
     "0.118.0",
+    "0.119.0-rc1",
 ]  # Replace with your versions
 
 DOWNLOAD_DIR = "download"

framework/config.py

@@ -48,6 +48,7 @@
 # contract
 ALWAYS_SUCCESS_CONTRACT_PATH = f"{get_project_root()}/source/contract/always_success"
 SPAWN_CONTRACT_PATH = f"{get_project_root()}/source/contract/test_cases/spawn_demo"
+UDT_CONTRACT_PATH = f"{get_project_root()}/source/contract/XUDTType"
 
 
 def get_tmp_path():

framework/helper/ckb_cli.py

@@ -428,6 +428,20 @@ def tx_info(tx_file_path, api_url="http://127.0.0.1:8114"):
     return run_command(cmd)
 
 
+def tx_add_output(output, out_put_data, tx_file):
+    with open(tx_file, "r") as file:
+        tx_info_str = file.read()
+
+    with open(tx_file, "w") as f:
+        tx = json.loads(tx_info_str)
+        tx["transaction"]["outputs"].append(output)
+        tx["transaction"]["outputs_data"].append(out_put_data)
+
+        tx_info_str = json.dumps(tx, indent=4)
+        f.write(tx_info_str)
+    pass
+
+
 def tx_add_type_out_put(
     code_hash, hash_type, arg, capacity_hex, out_put_data, tx_file, with_type=True
 ):

framework/helper/contract.py

@@ -10,11 +10,9 @@
 
 class CkbContract(ABC):
 
-    @abstractmethod
     def deploy(self, account_private, node: CkbNode):
         pass
 
-    @abstractmethod
     def get_deploy_hash_and_index(self) -> (str, int):
         pass
 
@@ -117,6 +115,8 @@ def invoke_ckb_contract(
     fee=1000,
     api_url="http://127.0.0.1:8114",
     cell_deps=[],
+    input_cells=[],
+    output_lock_arg="0x470dcdc5e44064909650113a274b3b36aecb6dc7",
 ):
     """
 
@@ -154,7 +154,17 @@ def invoke_ckb_contract(
     assert len(account_live_cells["live_cells"]) > 0
     input_cell_out_points = []
     input_cell_cap = 0
+    for i in range(len(input_cells)):
+        input_cell_out_points.append(input_cells[i])
+        cell = RPCClient(api_url).get_live_cell(
+            hex(input_cells[i]["index"]), input_cells[i]["tx_hash"], True
+        )
+        input_cell_cap += int(cell["cell"]["output"]["capacity"], 16)
+    input_cells_hashs = [input_cell["tx_hash"] for input_cell in input_cells]
+
     for i in range(len(account_live_cells["live_cells"])):
+        if account_live_cells["live_cells"][i]["tx_hash"] in input_cells_hashs:
+            continue
         input_cell_out_point = {
             "tx_hash": account_live_cells["live_cells"][i]["tx_hash"],
             "index": account_live_cells["live_cells"][i]["output_index"],
@@ -167,6 +177,7 @@ def invoke_ckb_contract(
             )
             * 100000000
         )
+
         input_cell_out_points.append(input_cell_out_point)
         if input_cell_cap > 10000000000:
             break
@@ -181,7 +192,7 @@ def invoke_ckb_contract(
         "lock": {
             "code_hash": "0x9bd7e06f3ecf4be0f2fcd2188b23f1b9fcc88e5d4b65a8637b17723bbda3cce8",
             "hash_type": "type",
-            "args": "0x470dcdc5e44064909650113a274b3b36aecb6dc7",
+            "args": output_lock_arg,
         },
         "type": {
             "code_hash": contract_code_hash,
@@ -218,11 +229,8 @@ def invoke_ckb_contract(
         print("add header:", head)
         tx_add_header_dep(head, tmp_tx_file)
     # add output
-    tx_add_type_out_put(
-        output_cell["type"]["code_hash"],
-        output_cell["type"]["hash_type"],
-        output_cell["type"]["args"],
-        output_cell["capacity"],
+    tx_add_output(
+        output_cell,
         data,
         tmp_tx_file,
     )
@@ -236,6 +244,11 @@ def invoke_ckb_contract(
         sign_data[0]["lock-arg"], sign_data[0]["signature"], tmp_tx_file, api_url
     )
     tx_info(tmp_tx_file, api_url)
+    for i in range(len(input_cells)):
+        cell = RPCClient(api_url).get_live_cell(
+            hex(input_cells[i]["index"]), input_cells[i]["tx_hash"], True
+        )
+
     # send tx return hash
     return tx_send(tmp_tx_file, api_url).strip()
 

framework/helper/udt_contract.py

@@ -0,0 +1,93 @@
+from framework.config import UDT_CONTRACT_PATH
+from framework.helper.contract import deploy_ckb_contract, CkbContract
+from framework.helper.miner import miner_until_tx_committed
+from framework.test_node import CkbNode
+from framework.util import (
+    ckb_hash_script,
+    to_big_uint128_le_compatible,
+    to_int_from_big_uint128_le,
+)
+from framework.helper.contract import get_ckb_contract_codehash
+
+
+class UdtContract(CkbContract):
+
+    def __init__(self, contract_hash=None, contract_tx_index=None):
+        self.contract_hash = contract_hash
+        self.contract_tx_index = contract_tx_index
+        if contract_hash is None:
+            self.deployed = False
+        self.contract_path = UDT_CONTRACT_PATH
+        self.method = {"demo": {"args": "0x", "data": "0x"}}
+
+    def deploy(self, account_private, node: CkbNode):
+        if self.deployed:
+            return
+        self.contract_path = deploy_ckb_contract(
+            account_private, self.contract_path, api_url=node.getClient().url
+        )
+        self.contract_tx_index = 0
+        miner_until_tx_committed(node, self.contract_path)
+        self.deployed = True
+
+    def get_deploy_hash_and_index(self) -> (str, int):
+        if not self.deployed:
+            raise Exception("pls deploy first")
+        return self.contract_path, self.contract_tx_index
+
+    @classmethod
+    def issue(cls, own_arg, amount) -> (str, str):
+        return ckb_hash_script(own_arg), to_big_uint128_le_compatible(amount)
+
+    @classmethod
+    def transfer(cls, own_arg, amount) -> (str, str):
+        return ckb_hash_script(own_arg), to_big_uint128_le_compatible(amount)
+
+    def balance(self, client, own_arg, query_arg):
+        pass
+
+    def list_cell(self, client, own_arg, query_arg):
+        code_hash = get_ckb_contract_codehash(
+            self.contract_path,
+            self.contract_tx_index,
+            enable_type_id=True,
+            api_url=client.url,
+        )
+        cells = client.get_cells(
+            {
+                "script": {
+                    "code_hash": "0x9bd7e06f3ecf4be0f2fcd2188b23f1b9fcc88e5d4b65a8637b17723bbda3cce8",
+                    "hash_type": "type",
+                    "args": query_arg,
+                },
+                "script_type": "lock",
+                "filter": {
+                    "script": {
+                        "code_hash": code_hash,
+                        "hash_type": "type",
+                        "args": ckb_hash_script(own_arg),
+                    }
+                },
+            },
+            "asc",
+            "0x64",
+            None,
+        )
+        info = []
+        for cell in cells["objects"]:
+            info.append(
+                {
+                    "input_cell": {
+                        "tx_hash": cell["out_point"]["tx_hash"],
+                        "index": int(cell["out_point"]["index"], 16),
+                    },
+                    "balance": to_int_from_big_uint128_le(cell["output_data"]),
+                }
+            )
+        return info
+
+    def get_arg_and_data(self, key) -> (str, str):
+        if key not in self.method.keys():
+            # return "0x0","0x0"
+            raise Exception("key not exist in method list")
+        return self.method[key]["args"], self.method[key]["data"]

framework/test_node.py

@@ -18,21 +18,34 @@ class CkbNodeConfigPath(Enum):
         "source/template/ckb/v118/ckb.toml.j2",
         "source/template/ckb/v118/ckb-miner.toml.j2",
         "source/template/ckb/v118/specs/dev.toml",
-        "download/0.118.0",
+        "download/0.119.0",
+    )
+    TESTNET = (
+        "source/template/ckb/v118/ckb.toml.j2",
+        "source/template/ckb/v118/ckb-miner.toml.j2",
+        "source/template/specs/testnet.toml.j2",
+        "download/0.119.0",
     )
 
     CURRENT_MAIN = (
         "source/template/ckb/v118/ckb.toml.j2",
         "source/template/ckb/v118/ckb-miner.toml.j2",
         "source/template/specs/mainnet.toml.j2",
-        "download/0.118.0",
+        "download/0.119.0",
     )
 
     PREVIEW_DUMMY = (
         "source/template/ckb/v118/ckb.toml.j2",
         "source/template/ckb/v118/ckb-miner.toml.j2",
         "source/template/specs/preview_dev.toml",
-        "download/0.118.0",
+        "download/0.119.0",
+    )
+
+    v119 = (
+        "source/template/ckb/v118/ckb.toml.j2",
+        "source/template/ckb/v118/ckb-miner.toml.j2",
+        "source/template/ckb/v118/specs/dev.toml",
+        "download/0.119.0",
     )
 
     v118 = (

framework/util.py

@@ -4,6 +4,14 @@
 import json
 import toml
 import os, re
+import struct
+
+import hashlib
+
+H256_ZEROS = "0x0000000000000000000000000000000000000000000000000000000000000000"
+
+U128_MIN_COMPATIBLE = 0  # Adjust according to your definition
+U128_MAX_COMPATIBLE = 2**128 - 1
 
 
 def to_json(value):
@@ -132,3 +140,58 @@ def read_toml_file(file_path):
     except Exception as e:
         print(f"Error reading TOML file: {e}")
         return None
+
+
+def to_big_uint128_le_compatible(num):
+    if num < U128_MIN_COMPATIBLE:
+        raise ValueError(f"u128 {num} too small")
+
+    if num > U128_MAX_COMPATIBLE:
+        raise ValueError(f"u128 {num} too large")
+
+    buf = bytearray(16)
+
+    for i in range(4):
+        buf[i * 4 : i * 4 + 4] = struct.pack("<I", num & 0xFFFFFFFF)
+        num >>= 32
+    return "0x" + buf.hex()
+
+
+def to_int_from_big_uint128_le(hex_str):
+    # Strip the '0x' prefix if present
+    if hex_str.startswith("0x"):
+        hex_str = hex_str[2:]
+
+    # Convert the hex string into a byte array (16 bytes for uint128)
+    buf = bytearray.fromhex(hex_str)
+
+    # Reverse the byte array to convert from little-endian to big-endian
+    buf.reverse()
+
+    # Convert the byte array into an integer
+    result = int.from_bytes(buf, byteorder="big")
+    print("to_int_from_big_uint128_le:", hex_str, " result ", result)
+
+    return result
+
+
+def ckb_hasher():
+    return hashlib.blake2b(digest_size=32, person=b"ckb-default-hash")
+
+
+def ckb_hash(message):
+    hasher = ckb_hasher()
+    hasher.update(bytes.fromhex(message.replace("0x", "")))
+    return "0x" + hasher.hexdigest()
+
+
+def ckb_hash_script(arg):
+    arg = arg.replace("0x", "")
+    pack_lock = f"0x490000001000000030000000310000009bd7e06f3ecf4be0f2fcd2188b23f1b9fcc88e5d4b65a8637b17723bbda3cce80114000000{arg}"
+    return ckb_hash(pack_lock)
+
+
+if __name__ == "__main__":
+    ret = to_big_uint128_le_compatible(100000)
+    ret1 = to_int_from_big_uint128_le(ret)
+    print(ret1)

source/template/specs/testnet.toml.j2

@@ -88,6 +88,7 @@ lock.hash_type = "type"
 
 
 [params]
+epoch_duration_target = {{ ckb_epoch_duration_target | default(8) }}
 genesis_epoch_length = {{ ckb_params_genesis_epoch_length | default(1) }}
 permanent_difficulty_in_dummy = true