Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_add_update_node(meta_repo):
nodes_repo = NodesRepo(meta_repo.conn)
node = Node("testcurrency",
"7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ",
"""BASIC_MERKLED_API test-net.duniter.fr 13.222.11.22 9201
BASIC_MERKLED_API testnet.duniter.org 80
UNKNOWNAPI some useless information""",
BlockUID.empty(),
"doe",
"15-76543400E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67",
12376543345,
"14-AEFFCB00E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67",
0,
"duniter")
nodes_repo.insert(node)
node.previous_buid = node.current_buid
node.current_buid = "16-77543400E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67"
nodes_repo.update(node)
node2 = nodes_repo.get_one(pubkey="7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ")
assert node2.current_buid == block_uid("16-77543400E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67")
assert node2.previous_buid == block_uid("15-76543400E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67")
def test_add_get_multiple_node(meta_repo):
nodes_repo = NodesRepo(meta_repo.conn)
nodes_repo.insert(Node("testcurrency",
"7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ",
"""BASIC_MERKLED_API test-net.duniter.fr 13.222.11.22 9201
BASIC_MERKLED_API testnet.duniter.org 80
UNKNOWNAPI some useless information""",
BlockUID.empty(),
"doe",
"15-76543400E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67",
12376543345,
"14-AEFFCB00E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67",
0,
"duniter",
"0.30.17"))
nodes_repo.insert(Node("testcurrency",
"FADxcH5LmXGmGFgdixSes6nWnC4Vb4pRUBYT81zQRhjn",
"BASIC_MERKLED_API test-net.duniter.org 22.22.22.22 9201",
BlockUID.empty(),
"doe",
"18-76543400E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67",
12376543345,
"12-AEFFCB00E78B56CC21FB1DDC6CBAB24E0FACC9A798F5ED8736EA007F38617D67",
0,
async def refresh_uid(self):
"""
Refresh the node UID
"""
conn_handler = self.endpoint.conn_handler()
try:
data = await bma.wot.Lookup(conn_handler, self.pubkey).get(self._session)
self.state = Node.ONLINE
timestamp = BlockUID.empty()
uid = ""
for result in data['results']:
if result["pubkey"] == self.pubkey:
uids = result['uids']
for uid in uids:
if BlockUID.from_str(uid["meta"]["timestamp"]) >= timestamp:
timestamp = uid["meta"]["timestamp"]
uid = uid["uid"]
if self._uid != uid:
self._uid = uid
self.identity_changed.emit()
except errors.DuniterError as e:
if e.ucode == errors.NO_MATCHING_IDENTITY:
logging.debug("UID not found : {0}".format(self.pubkey[:5]))
else:
logging.debug("error in uid reply : {0}".format(self.pubkey[:5]))
def initialize_root_nodes(self, currency):
if not self.nodes(currency):
for pubkey in ROOT_SERVERS[currency]["nodes"]:
node = Node(currency=currency,
pubkey=pubkey,
endpoints=ROOT_SERVERS[currency]["nodes"][pubkey],
peer_blockstamp=BlockUID.empty(),
state=0)
self._repo.insert(node)
async def published_uid(self, community):
try:
data = await community.bma_access.future_request(bma.wot.Lookup,
req_args={'search': self.pubkey})
timestamp = BlockUID.empty()
for result in data['results']:
if result["pubkey"] == self.pubkey:
uids = result['uids']
person_uid = ""
for uid_data in uids:
if BlockUID.from_str(uid_data["meta"]["timestamp"]) >= timestamp:
timestamp = uid_data["meta"]["timestamp"]
person_uid = uid_data["uid"]
if person_uid == self.uid:
return True
except errors.DuniterError as e:
logging.debug("Lookup error : {0}".format(str(e)))
except NoPeerAvailable as e:
logging.debug(str(e))
return False
# The decimal percent of previous issuers to reach for personalized difficulty
percent_rot = attr.ib(convert=float, default=0, cmp=False, hash=False)
# The first UD time
ud_time_0 = attr.ib(convert=int, default=0, cmp=False, hash=False)
# The first UD reavallued
ud_reeval_time_0 = attr.ib(convert=int, default=0, cmp=False, hash=False)
# The dt recomputation of the ud
dt_reeval = attr.ib(convert=int, default=0, cmp=False, hash=False)
@attr.s(hash=True)
class Blockchain:
# Parameters in block 0
parameters = attr.ib(default=BlockchainParameters(), cmp=False, hash=False)
# block number and hash
current_buid = attr.ib(convert=block_uid, default=BlockUID.empty())
# Number of members
current_members_count = attr.ib(convert=int, default=0, cmp=False, hash=False)
# Current monetary mass in units
current_mass = attr.ib(convert=int, default=0, cmp=False, hash=False)
# Median time in seconds
median_time = attr.ib(convert=int, default=0, cmp=False, hash=False)
# Last members count
last_mass = attr.ib(convert=int, default=0, cmp=False, hash=False)
# Last members count
last_members_count = attr.ib(convert=int, default=0, cmp=False, hash=False)
# Last UD amount in units (multiply by 10^base)
last_ud = attr.ib(convert=int, default=1, cmp=False, hash=False)
# Last UD base
last_ud_base = attr.ib(convert=int, default=0, cmp=False, hash=False)
# Last UD base
last_ud_time = attr.ib(convert=int, default=0, cmp=False, hash=False)
def current_blockUID(self):
"""
Get the latest block considered valid
It is the most frequent last block of every known nodes
"""
blocks = [n.block for n in self.synced_nodes if n.block]
if len(blocks) > 0:
return BlockUID(blocks[0]['number'], blocks[0]['hash'])
else:
return BlockUID.empty()
def _parse_uid_lookup(data):
timestamp = BlockUID.empty()
found_uid = ""
for result in data['results']:
if result["pubkey"] == self.pubkey:
uids = result['uids']
for uid_data in uids:
if BlockUID.from_str(uid_data["meta"]["timestamp"]) >= timestamp:
timestamp = uid_data["meta"]["timestamp"]
found_uid = uid_data["uid"]
return self.name == found_uid, self.name, found_uid
@attr.s(hash=True)
class Connection:
"""
A connection represents a connection to a currency's network
It is defined by the currency name, and the key informations
used to connect to it. If the user is using an identity, it is defined here too.
"""
currency = attr.ib(convert=str)
pubkey = attr.ib(convert=str)
uid = attr.ib(convert=str, default="", cmp=False, hash=False)
scrypt_N = attr.ib(convert=int, default=4096, cmp=False, hash=False)
scrypt_r = attr.ib(convert=int, default=16, cmp=False, hash=False)
scrypt_p = attr.ib(convert=int, default=1, cmp=False, hash=False)
blockstamp = attr.ib(convert=block_uid, default=BlockUID.empty(), cmp=False, hash=False)
salt = attr.ib(convert=str, init=False, default="", cmp=False, hash=False)
password = attr.ib(init=False, convert=str, default="", cmp=False, hash=False)
def is_identity(self):
return self.uid is not ""
def is_wallet(self):
return self.uid is ""
def title(self):
return "@".join([self.uid, self.pubkey[:11]])
@property
def scrypt_params(self):
return ScryptParams(self.scrypt_N, self.scrypt_r, self.scrypt_p)