chia 0.5.2

Utility functions and types used by the Chia blockchain full node
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

from typing import List, Tuple, Optional, Any, Callable

import string
import sys
import time
from chia_rs import BlockRecord, ClassgroupElement
from chia.consensus.block_record import BlockRecord as PyBlockRecord
from chia.types.blockchain_format.sized_bytes import bytes32, bytes100
from random import Random
from chia.consensus.default_constants import DEFAULT_CONSTANTS
from chia.util.ints import uint32, uint64, uint8, uint128


def get_classgroup_element(rng: Random) -> ClassgroupElement:
    return ClassgroupElement(bytes100.random(rng))


def get_u4(rng: Random) -> uint8:
    return uint8(rng.randint(0, 0xF))


def get_u8(rng: Random) -> uint8:
    return uint8(rng.randint(0, 0xFF))


def get_u32(rng: Random) -> uint32:
    return uint32(rng.randint(0, 0xFFFFFFFF))


def get_ssi(rng: Random) -> uint64:
    return uint64(DEFAULT_CONSTANTS.NUM_SPS_SUB_SLOT * rng.randint(0, 0xFFFF) + rng.randint(
        0, 1
    ))


def get_u64(rng: Random) -> uint64:
    return uint64(rng.randint(0, 0xFFFFFFFFFFFFFFFF))


def get_u128(rng: Random) -> uint128:
    return uint128(rng.randint(0, 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF))


def get_optional(rng: Random, gen: Callable[[Random], Any]) -> Optional[Any]:
    if rng.randint(0, 1) == 0:
        return None
    else:
        return gen(rng)


def get_list(rng: Random, gen: Callable[[Random], Any]) -> List[Any]:
    length = rng.sample([0, 1, 5, 32, 500], 1)[0]
    ret: List[Any] = []
    for i in range(length):
        ret.append(gen(rng))
    return ret


def get_bool(rng: Random) -> bool:
    return rng.randint(0, 1) == 1


def get_hash(rng: Random) -> bytes32:
    return bytes32.random(rng)


def get_block_record(rng: Random) -> BlockRecord:
    height = get_u32(rng)
    weight = get_u128(rng)
    iters = get_u128(rng)
    sp_index = get_u4(rng)
    vdf_out = get_classgroup_element(rng)
    infused_challenge = get_optional(rng, get_classgroup_element)
    sub_slot_iters = get_ssi(rng)
    required_iters = get_u64(rng)
    deficit = get_u8(rng)
    overflow = get_bool(rng)
    prev_tx_height = get_u32(rng)
    timestamp = uint64(123456789)
    prev_tx_hash = get_optional(rng, get_hash)
    fees = get_optional(rng, get_u64)

    return BlockRecord(
        bytes32.random(rng),
        bytes32.random(rng),
        height,
        weight,
        iters,
        sp_index,
        vdf_out,
        infused_challenge,
        bytes32.random(rng),
        bytes32.random(rng),
        sub_slot_iters,
        bytes32.random(rng),
        bytes32.random(rng),
        required_iters,
        deficit,
        overflow,
        prev_tx_height,
        timestamp,
        prev_tx_hash,
        fees,
        [],
        None,
        None,
        None,
        None,
    )


def wrap_call(expr: str, br: Any) -> str:
    try:
        ret = eval(expr, None, {"br": br})
        return f"V:{ret}"
    except Exception as e:
        return f"E:{e}"


def test_block_record() -> None:
    rng = Random()
    seed = int(time.time())
    print(f"seed: {seed}")
    rng.seed(seed)

    for i in range(500000):
        br = get_block_record(rng)
        serialized = bytes(br)
        py_identity = PyBlockRecord.from_bytes(serialized)

        assert bytes(py_identity) == serialized
        assert f"{type(br)}" != f"{type(py_identity)}"

        for test_call in [
            "ip_iters",
            "sp_total_iters",
            "sp_iters",
            "ip_sub_slot_total_iters",
            "sp_sub_slot_total_iters",
        ]:
            rust_ret = wrap_call(f"br.{test_call}(DEFAULT_CONSTANTS)", br)
            py_ret = wrap_call(f"br.{test_call}(DEFAULT_CONSTANTS)", py_identity)

            assert rust_ret == py_ret

        if (i & 0x3FF) == 0:
            sys.stdout.write(f" {i}     \r")
            sys.stdout.flush()