Syksy/townforge
1
0
Fork 0
forked from townforge/townforge
Code Pull Requests Activity
win-msys2
townforge/tests/functional_tests/pos.py
Crypto City 775e85a856 mushrooms now grow on empty flags 
anyone can pick them
2025-08-25 14:20:20 +00:00

447 lines
17 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python3
# Copyright (c) 2023, The Monero Project
#
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification, are
# permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice, this list of
# conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice, this list
# of conditions and the following disclaimer in the documentation and/or other
# materials provided with the distribution.
#
# 3. Neither the name of the copyright holder nor the names of its contributors may be
# used to endorse or promote products derived from this software without specific
# prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
# THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
# STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
# THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
from __future__ import print_function
import time
"""Test PoS
"""
from framework.daemon import Daemon
from framework.wallet import Wallet
from framework.daemon_util import backup_daemon_blockchain
FIRST_CITY_X = 0
FIRST_CITY_Y = 0
ITEM_WOOD = 4
GAME_UPDATE_FREQUENCY = 1080
ROLE_AGRICULTURAL = 1
def gen_random_bool_vector(n, seed = 0):
v = []
r = seed
while n > 0:
r = (r * 0x100000001b3 + 0xcbf29ce484222325) & 0xffffffff
v += [(r // 256) % 2 == 0]
n -= 1
return v
class PoSTest():
def run_test(self):
self.create()
self.reset()
self.test_no_account()
self.test_no_flag()
self.create_base_chain()
res = self.daemon.get_height()
self.base_height = res.height
self.test_simple_chain([False] * 64, "PoW only", "pow")
self.test_simple_chain([True] * 64, "PoS only", "pos")
self.test_simple_chain([True, False] * 32, "alternating PoS/PoW", "alt")
self.test_simple_chain(gen_random_bool_vector(64, 11111), "random PoS/PoW", "rnd")
self.check_simple_chain_results()
self.test_reorg()
def assert_exception(self, f):
ok = False
try:
f()
except Exception as e:
ok = True
assert ok
def create(self):
print('Creating wallets')
seeds = [
[0, 'teardrop owls later width skater gadget different pegs yard ahead onslaught dynamite thorn espionage dwelt rural eels aimless shipped toaster shocking rounded maverick mystery thorn'],
[1, 'geek origin industrial payment friendly physics width putty beyond addicted rogue metro midst anvil unplugs tequila efficient feast elapse liquid degrees smuggled also bawled tequila'],
[2, 'tidy heron aching outbreak terminal inorganic nexus umpire economics auctions hope soapy hive vigilant hunter tadpoles hippo southern observant rabbits asked vector gimmick godfather heron'],
[3, 'foyer mittens gaze vocal dwindling unnoticed pimple foolish sword stacking unveil fuming husband bodies exit mugged omnibus jump whale sanity loincloth menu bite cactus vocal'],
[4, 'journal vitals technical dying dice taken evicted essential pepper dented psychic vaults oatmeal pairing wrong request damp rugged buffet lids bias pouch ladder leisure rugged'],
[6, 'velvet lymph giddy number token physics poetry unquoted nibs useful sabotage limits benches lifestyle eden nitrogen anvil fewest avoid batch vials washing fences goat unquoted']
]
self.wallet = [None] * len(seeds)
self.wallet_address = [None] * len(seeds)
self.account_id = [None] * len(seeds)
for i in range(len(seeds)):
self.wallet[i] = Wallet(idx = seeds[i][0])
# close the wallet if any, will throw if none is loaded
try: self.wallet[i].close_wallet()
except: pass
res = self.wallet[i].restore_deterministic_wallet(seed = seeds[i][1])
res = self.wallet[i].get_address()
self.wallet_address[i] = res.address
self.wallet[i].set_daemon('127.0.0.1:18182')
self.daemon = Daemon(idx = 2)
global FIRST_CITY_X
global FIRST_CITY_Y
res = self.daemon.cc_get_city(0)
FIRST_CITY_X = res.ox
FIRST_CITY_Y = res.oy
self.results = {}
def reset(self):
for idx in [2, 3, 4]:
daemon = Daemon(idx = idx)
res = daemon.get_height()
daemon.pop_blocks(res.height - 1)
daemon.flush_txpool()
def reset_base_chain(self):
daemon = self.daemon
res = daemon.get_height()
daemon.pop_blocks(res.height - self.base_height)
daemon.flush_txpool()
for w in self.wallet:
w.rescan_blockchain(hard = True)
def test_no_account(self):
print("Testing an account is needed to create PoS blocks")
daemon = self.daemon
self.reset()
try:
self.wallet[2].generate_pos_blocks(self.wallet_address[2], 1)
assert False
except Exception as e:
assert 'A game account is needed to generate PoS blocks' in str(e)
def test_no_flag(self):
print("Testing a flag is needed to create PoS blocks")
daemon = self.daemon
self.reset()
try:
self.wallet[3].generate_pos_blocks(self.wallet_address[3], 1)
assert False
except Exception as e:
assert 'No flags with nonzero PoS weight to stake' in str(e)
def create_base_chain(self):
print("Creating base chain")
daemon = self.daemon
for data in [
[self.wallet_address[0], 100], # wallet[2]
]:
daemon.generateblocks(data[0], data[1])
# create an account with an agricultural building on it
self.wallet[0].refresh()
res = self.wallet[0].get_balance()
res = self.wallet[0].cc_deposit(amount = 40 * 17 * 100000000, name = 'wallet 0', house = 'house 0')
daemon.generateblocks(self.wallet_address[0], 1)
self.wallet[0].refresh()
res = self.wallet[0].cc_get_info()
self.account_id[0] = res.account_id
x0 = FIRST_CITY_X + 200
y0 = FIRST_CITY_Y + 200
x1 = x0 + 63
y1 = y0 + 63
res = self.wallet[0].cc_buy_land(x0, y0, x1, y1)
daemon.generateblocks(self.wallet_address[0], 1)
res = daemon.get_height()
construction_height = res.height - 1
last_service_height = res.height - 1
res = daemon.cc_get_building_cost(x0, y0, x1, y1, ROLE_AGRICULTURAL, 100)
daemon.generateblocks(self.wallet_address[0], 1)
res = self.wallet[0].cc_buy_items(res.materials)
res = self.wallet[0].cc_buy_items([{'type': ITEM_WOOD, 'amount': 500}]) # for heating
daemon.generateblocks(self.wallet_address[0], 1)
res = self.wallet[0].cc_building_settings(2, ROLE_AGRICULTURAL, 100, construction_height = construction_height, last_service_height = last_service_height)
daemon.generateblocks(self.wallet_address[0], 1)
# can't while still inactive
try:
self.wallet[0].generate_pos_blocks(self.wallet_address[0], 1)
assert False
except Exception as e:
assert 'No flags with nonzero PoS weight to stake' in str(e)
# get past the next update to activate the flag
res = daemon.get_info()
blocks = (GAME_UPDATE_FREQUENCY * 8000 - res.height) % GAME_UPDATE_FREQUENCY + 1
daemon.generateblocks(self.wallet_address[0], blocks)
# check the building is active and has PoS weight
res = daemon.cc_get_account(self.account_id[0])
assert res.flags == [2]
res = daemon.cc_get_flag(2)
assert res.role == ROLE_AGRICULTURAL
assert res.active
assert res.owner == self.account_id[0]
# now add PoS blocks to match the two subchains' cumulative difficulties
res = daemon.get_info()
delta = res.pow_cumulative_difficulty - res.pos_cumulative_difficulty
assert delta >= 0
blocks_needed = int((delta / res.pos_difficulty) + 0.5)
self.wallet[0].refresh()
self.wallet[0].generate_pos_blocks(self.wallet_address[0], blocks_needed)
# we should have a chain with identical PoS and PoW cumulative difficulties
res = daemon.get_info()
assert res.pos_cumulative_difficulty == res.pow_cumulative_difficulty
def test_simple_chain(self, blocks, name, results):
assert len(blocks) >= 2
print("Testing simple chain, " + name)
daemon = self.daemon
self.reset_base_chain()
for pos in blocks:
self.wallet[0].refresh()
if pos:
self.wallet[0].generate_pos_blocks(self.wallet_address[0], 1)
else:
daemon.generateblocks(self.wallet_address[0], 1)
res = daemon.get_info()
height = res.height
assert height == self.base_height + len(blocks)
res = daemon.getblockheadersrange(self.base_height, self.base_height + len(blocks) - 1)
assert len(res.headers) == len(blocks)
for i in range(len(res.headers)):
e = res.headers[i]
assert e.height == self.base_height + i
assert e.pos == blocks[i]
if e.pos:
assert e.reward < 1800000000 * 20 // 100
else:
assert e.reward > 1700000000
assert e.num_txes == 0
assert not e.merged_mined
pow_cdiff = 0
pos_cdiff = 0
res = daemon.getblockheadersrange(0, height - 1)
for e in res.headers:
if e.pos:
pos_cdiff += e.difficulty
else:
pow_cdiff += e.difficulty
res = daemon.get_info()
height = res.height
assert res.pow_cumulative_difficulty == pow_cdiff
assert res.pos_cumulative_difficulty == pos_cdiff
assert int(res.cumulative_difficulty, 16) == (res.pow_cumulative_difficulty+1) * (res.pos_cumulative_difficulty+1)
self.results[results] = res
for start_height in [0, 16, res.height - 1]:
for count in [1, 2, 15, 100]:
end_height = start_height + count - 1
if end_height >= height:
end_height = height - 1
res1 = daemon.get_block_pos_history(start_height, end_height)
res2 = daemon.get_block_headers_range(start_height, end_height, False)
assert len(res1.pos) == len(res2.headers)
assert len(res1.pos) == end_height - start_height + 1
for i in range(len(res1.pos)):
assert res1.pos[i] == res2.headers[i].pos
def check_simple_chain_results(self):
rpow = self.results["pow"]
rpos = self.results["pos"]
ralt = self.results["alt"]
rrnd = self.results["rnd"]
assert int(ralt.cumulative_difficulty, 16) > int(rpow.cumulative_difficulty, 16)
assert int(ralt.cumulative_difficulty, 16) > int(rpos.cumulative_difficulty, 16)
assert int(rrnd.cumulative_difficulty, 16) >= int(rpow.cumulative_difficulty, 16)
assert int(rrnd.cumulative_difficulty, 16) >= int(rpos.cumulative_difficulty, 16)
assert int(ralt.cumulative_difficulty, 16) >= int(rrnd.cumulative_difficulty, 16)
def test_reorg(self):
daemon = self.daemon
second_daemon = Daemon(idx = 3)
third_daemon = Daemon(idx = 4)
print('Testing reorg to mixed block chain')
self.reset_base_chain()
res = daemon.get_info()
self.reorg_root_hash = res.top_block_hash
assert len(self.reorg_root_hash) == 64
self.reorg_root_height = res.height - 1
assert self.reorg_root_height > 0
# ensure synced
print('Syncing second and third daemons')
daemon.out_peers(8)
daemon.in_peers(8)
second_daemon.out_peers(8)
second_daemon.in_peers(8)
third_daemon.out_peers(8)
third_daemon.in_peers(8)
loops = 100
while True:
res0 = daemon.get_info()
res1 = second_daemon.get_info()
res2 = third_daemon.get_info()
if res0.top_block_hash == res1.top_block_hash and res0.top_block_hash == res2.top_block_hash:
break
time.sleep(5)
loops -= 1
assert loops >= 0
res = daemon.get_info()
cumulative_difficulty = res.cumulative_difficulty
# disconnect second and third daemon so they keep the original chain
second_daemon.out_peers(0)
second_daemon.in_peers(0)
third_daemon.out_peers(0)
third_daemon.in_peers(0)
# mine 10 pow blocks on second daemon
self.wallet[0].set_daemon('127.0.0.1:18183')
self.wallet[0].rescan_blockchain()
second_daemon.generateblocks(self.wallet_address[0], 10)
res = second_daemon.get_info()
second_top = res.top_block_hash
second_cdiff = res.cumulative_difficulty
second_cdiff_pos = res.pos_cumulative_difficulty
second_cdiff_pow = res.pow_cumulative_difficulty
# mine 10 pos blocks on first daemon
self.wallet[0].set_daemon('127.0.0.1:18182')
self.wallet[0].rescan_blockchain()
self.wallet[0].generate_pos_blocks(self.wallet_address[0], 10)
res = self.daemon.get_info()
first_top = res.top_block_hash
first_cdiff = res.cumulative_difficulty
first_cdiff_pos = res.pos_cumulative_difficulty
first_cdiff_pow = res.pow_cumulative_difficulty
# the diffs should be the same, due to the geometric mean
assert first_cdiff > cumulative_difficulty
assert first_cdiff == second_cdiff
# reconnect second daemon
second_daemon.out_peers(8)
second_daemon.in_peers(8)
# there should be no reorg
time.sleep(10)
res = second_daemon.get_info()
assert res.top_block_hash == second_top
res = self.daemon.get_info()
assert res.top_block_hash == first_top
assert first_top != second_top
# disconnect second daemon again
second_daemon.out_peers(0)
second_daemon.in_peers(0)
print("Creating other chains")
# mine 5 pos blocks and 5 pow blocks on third daemon
self.wallet[0].set_daemon('127.0.0.1:18184')
self.wallet[0].rescan_blockchain()
self.wallet[0].generate_pos_blocks(self.wallet_address[0], 5)
third_daemon.generateblocks(self.wallet_address[0], 5)
res = third_daemon.get_info()
third_top = res.top_block_hash
third_cdiff = res.cumulative_difficulty
third_cdiff_pos = res.pos_cumulative_difficulty
third_cdiff_pow = res.pow_cumulative_difficulty
# all the single chains have the same additive cumulative difficulty
assert first_cdiff_pos + first_cdiff_pow == third_cdiff_pos + third_cdiff_pow
assert second_cdiff_pos + second_cdiff_pow == third_cdiff_pos + third_cdiff_pow
# yet the mixed blocks should yield a higher chain cumulative difficulty
assert second_cdiff == first_cdiff
assert third_cdiff > second_cdiff
print('Waiting for reorg')
# reconnect second and third daemons
second_daemon.out_peers(8)
second_daemon.in_peers(8)
third_daemon.out_peers(8)
third_daemon.in_peers(8)
# wait till first and second daemons reorg to the third daemon's chain
loops = 100
while True:
res0 = daemon.get_info()
res1 = second_daemon.get_info()
res2 = third_daemon.get_info()
if res0.top_block_hash == res1.top_block_hash and res0.top_block_hash == res2.top_block_hash:
break
time.sleep(5)
loops -= 1
assert loops >= 0
# check the winning chain is the mixed one
res = daemon.get_info()
reorged_first_top_hash = res.top_block_hash
res = second_daemon.get_info()
reorged_second_top_hash = res.top_block_hash
res = third_daemon.get_info()
reorged_third_top_hash = res.top_block_hash
assert reorged_first_top_hash == third_top
assert reorged_second_top_hash == third_top
assert reorged_third_top_hash == third_top
class Guard:
def __enter__(self):
for i in [2, 3, 4]:
Daemon(idx = i).out_peers(0)
Daemon(idx = i).in_peers(0)
for i in [0, 1, 2, 3, 6]:
Wallet(idx = i).auto_refresh(False)
def __exit__(self, exc_type, exc_value, traceback):
for i in [2, 3, 4]:
Daemon(idx = i).out_peers(8)
Daemon(idx = i).in_peers(8)
for i in [0, 1, 2, 3, 6]:
Wallet(idx = i).set_daemon('127.0.0.1:18180')
Wallet(idx = i).auto_refresh(True)
if __name__ == '__main__':
with Guard() as guard:
PoSTest().run_test()
View Git Blame Copy Permalink