forked from karlfloersch/pyethereum
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_chain.py
172 lines (142 loc) · 6.35 KB
/
test_chain.py
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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import pytest
from ethereum.utils import privtoaddr
from ethereum.tools import tester
from ethereum.tests.utils import new_db
from ethereum.db import EphemDB
from ethereum.hybrid_casper import casper_utils
from ethereum.slogging import get_logger
from ethereum.tests.hybrid_casper.testing_lang import TestLangHybrid
log = get_logger('test.chain')
logger = get_logger()
_db = new_db()
# from ethereum.slogging import configure_logging
# config_string = ':info,eth.chain:debug,test.chain:info'
# configure_logging(config_string=config_string)
EPOCH_LENGTH = 25
SLASH_DELAY = 864
ALLOC = {a: {'balance': 500*10**19} for a in tester.accounts[:10]}
k0, k1, k2, k3, k4, k5, k6, k7, k8, k9 = tester.keys[:10]
a0, a1, a2, a3, a4, a5, a6, a7, a8, a9 = tester.accounts[:10]
@pytest.fixture(scope='function')
def db():
return EphemDB()
alt_db = db
def init_chain_and_casper():
genesis = casper_utils.make_casper_genesis(ALLOC, EPOCH_LENGTH, 100, 0.02, 0.002)
t = tester.Chain(genesis=genesis)
casper = tester.ABIContract(t, casper_utils.casper_abi, t.chain.config['CASPER_ADDRESS'])
return t, casper
# Mines blocks required for number_of_epochs epoch changes, plus an offset of 2 blocks
def mine_epochs(t, number_of_epochs):
distance_to_next_epoch = (EPOCH_LENGTH - t.head_state.block_number) % EPOCH_LENGTH
number_of_blocks = distance_to_next_epoch + EPOCH_LENGTH*(number_of_epochs-1) + 2
return t.mine(number_of_blocks=number_of_blocks)
def test_mining(db):
t, casper = init_chain_and_casper()
assert t.chain.state.block_number == 0
assert t.chain.state.block_difficulty == 1
for i in range(2):
t.mine()
assert t.chain.state.block_number == i + 1
def test_mining_block_rewards(db):
t, casper = init_chain_and_casper()
genesis = t.mine(coinbase=a1)
blk2 = t.mine(coinbase=a1)
blk3 = t.mine(coinbase=a1)
blk4 = t.mine(coinbase=a1)
t.mine(coinbase=a1)
assert t.chain.state.get_balance(a1) == t.chain.env.config['BLOCK_REWARD'] + t.chain.mk_poststate_of_blockhash(blk4.hash).get_balance(a1)
assert t.chain.state.get_balance(a1) == t.chain.env.config['BLOCK_REWARD'] * 2 + t.chain.mk_poststate_of_blockhash(blk3.hash).get_balance(a1)
assert t.chain.state.get_balance(a1) == t.chain.env.config['BLOCK_REWARD'] * 3 + t.chain.mk_poststate_of_blockhash(blk2.hash).get_balance(a1)
assert t.chain.state.get_balance(a1) == t.chain.env.config['BLOCK_REWARD'] * 4 + t.chain.mk_poststate_of_blockhash(genesis.hash).get_balance(a1)
assert blk2.prevhash == genesis.hash
def test_simple_chain(db):
t, casper = init_chain_and_casper()
t.tx(k0, a1, 20, gasprice=0)
blk2 = t.mine()
blk3 = t.mine()
assert blk2.hash in t.chain
assert blk3.hash in t.chain
assert t.chain.has_block(blk2.hash)
assert t.chain.has_block(blk3.hash)
assert t.chain.get_block(blk2.hash) == blk2
assert t.chain.get_block(blk3.hash) == blk3
assert t.chain.head == blk3
assert t.chain.get_children(blk2) == [blk3]
assert t.chain.get_chain() == [blk2, blk3]
assert t.chain.get_block_by_number(1) == blk2
assert t.chain.get_block_by_number(2) == blk3
assert not t.chain.get_block_by_number(3)
def test_head_change_for_longer_pow_chain(db):
"""" [L & R are blocks]
Local: L0, L1
add
Remote: R0, R1, R2
"""
t, casper = init_chain_and_casper()
t.mine()
root_hash = t.chain.head_hash
L = t.mine(2)
assert t.chain.head_hash == L.hash
t.change_head(root_hash)
R = t.mine(2)
# Test that we just need one more block before the head switches
assert t.chain.head_hash == L.hash
R = t.mine(1)
assert t.chain.head_hash == R.hash
def test_no_gas_cost_for_successful_casper_vote(db):
""" This tests that the chain is the chain is """
sender = b'\x82\xa9x\xb3\xf5\x96*[\tW\xd9\xee\x9e\xefG.\xe5[B\xf1'
test_string = 'B J0 B B'
test = TestLangHybrid(15, 100, 0.02, 0.002)
test.parse(test_string)
pre_balance = test.t.head_state.get_balance(sender)
pre_block_gas_used = test.t.head_state.gas_used
test_string = 'V0'
test.parse(test_string)
post_balance = test.t.head_state.get_balance(sender)
post_block_gas_used = test.t.head_state.gas_used
assert pre_balance == post_balance
assert pre_block_gas_used == post_block_gas_used
def test_costs_gas_for_failed_casper_vote(db):
""" This tests that the chain is the chain is """
test = TestLangHybrid(15, 100, 0.02, 0.002)
sender = b'\x82\xa9x\xb3\xf5\x96*[\tW\xd9\xee\x9e\xefG.\xe5[B\xf1'
pre_join_balance = test.t.head_state.get_balance(sender)
test_string = 'B J0 B B'
test.parse(test_string)
post_join_balance = test.t.head_state.get_balance(sender)
cost_of_joining = pre_join_balance - post_join_balance
pre_balance = test.t.head_state.get_balance(sender)
test_string = 'J1 V0 B B'
with pytest.raises(AssertionError):
test.parse(test_string)
post_balance = test.t.head_state.get_balance(sender)
# Gas is charged for failed vote
assert pre_balance > post_balance + cost_of_joining
def test_fails_if_all_casper_vote_transactions_are_not_first(db):
""" This tests that the chain is the chain is """
test_string = 'B J0 B B J1 V0 B B'
test = TestLangHybrid(15, 100, 0.02, 0.002)
with pytest.raises(AssertionError):
test.parse(test_string)
def test_no_change_for_more_work_on_non_finalized_descendant(db):
""" This tests that the chain is the chain is """
test_string = 'B J0 J1 J2 J3 B B V0 V1 V2 V3 B V0 V1 V2 V3 B S0 B V0 B1 S1 H1 R0 B B B B B H1'
test = TestLangHybrid(15, 100, 0.02, 0.002)
test.parse(test_string)
def test_change_head_for_more_votes(db):
""" This tests that the chain is the chain is """
test_string = 'B J0 J1 J2 J3 B B V0 V1 V2 V3 B S0 B V0 V1 B1 S1 R0 B B B2 V0 B1 S2 H1 V1 V2 B2 S3 H3'
test = TestLangHybrid(15, 100, 0.02, 0.002)
test.parse(test_string)
def test_double_vote_slash(db):
""" This tests that the chain is the chain is """
test_string = 'B J0 J1 J2 J3 B B S0 B V0 V1 V2 V3 B1 R0 B V0 B1 X0 B V1 V2 V3 B'
test = TestLangHybrid(15, 100, 0.02, 0.002)
test.parse(test_string)
def test_vote_surround_slash(db):
""" This tests that the chain is the chain is """
test_string = 'B J0 J1 J2 J3 B B S0 V0 V1 V2 V3 B V0 V1 V2 V3 B V0 V1 V2 V3 R0 B B B B B B B V0 B1'
test = TestLangHybrid(15, 100, 0.02, 0.002)
test.parse(test_string)