Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
set_shared_steem_instance(self.bts)
o = Witness("gtg")
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Witness("gtg", steem_instance=Steem(node="https://abc.d", autoconnect=False, num_retries=1))
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = Witness("gtg", steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Witness("gtg")
def test_witness(self, node_param):
if node_param == "instance":
set_shared_steem_instance(self.bts)
o = Witness("gtg")
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Witness("gtg", steem_instance=Steem(node="https://abc.d", autoconnect=False, num_retries=1))
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = Witness("gtg", steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Witness("gtg")
def test_witness(self, node_param):
if node_param == "instance":
set_shared_steem_instance(self.bts)
o = Witness("gtg")
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Witness("gtg", steem_instance=Steem(node="https://abc.d", autoconnect=False, num_retries=1))
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = Witness("gtg", steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Witness("gtg")
def test_update(self, node_param):
if node_param == "normal":
bts = self.bts
else:
bts = self.steemit
bts.txbuffer.clear()
w = Witness("gtg", steem_instance=bts)
props = {"account_creation_fee": "0.1 STEEM",
"maximum_block_size": 32000,
"sbd_interest_rate": 0}
tx = w.update(wif, "", props)
self.assertEqual((tx["operations"][0][0]), "witness_update")
op = tx["operations"][0][1]
self.assertIn(
"gtg",
op["owner"])
if key == "json_metadata":
value = json.dumps(json.loads(value or "{}"), indent=4)
elif key in ["posting", "witness_votes", "active", "owner"]:
value = json.dumps(value, indent=4)
elif key == "reputation" and int(value) > 0:
value = int(value)
rep = account.rep
value = "{:.2f} ({:d})".format(rep, value)
elif isinstance(value, dict) and "asset" in value:
value = str(account[key])
t.add_row([key, value])
print(t)
# witness available?
try:
witness = Witness(obj, steem_instance=stm)
witness_json = witness.json()
t = PrettyTable(["Key", "Value"])
t.align = "l"
for key in sorted(witness_json):
value = witness_json[key]
if key in ["props", "sbd_exchange_rate"]:
value = json.dumps(value, indent=4)
t.add_row([key, value])
print(t)
except exceptions.WitnessDoesNotExistsException as e:
print(str(e))
# Public Key
elif re.match("^" + stm.prefix + ".{48,55}$", obj):
account = stm.wallet.getAccountFromPublicKey(obj)
if account:
account = Account(account, steem_instance=stm)
def __init__(self, from_account="", limit=100, lazy=False, full=False, steem_instance=None):
self.steem = steem_instance or shared_steem_instance()
self.identifier = from_account
self.steem.rpc.set_next_node_on_empty_reply(False)
if self.steem.rpc.get_use_appbase():
witnessess = self.steem.rpc.list_witnesses({'start': from_account, 'limit': limit, 'order': 'by_name'}, api="database")['witnesses']
else:
witnessess = self.steem.rpc.lookup_witness_accounts(from_account, limit)
if len(witnessess) == 0:
return
super(ListWitnesses, self).__init__(
[
Witness(x, lazy=lazy, full=full, steem_instance=self.steem)
for x in witnessess
]
else:
last_account = witnessList[-1]["owner"]
if (last_limit < limit):
last_limit += 1
if self.steem.rpc.get_use_appbase() and not use_condenser:
witnessess = self.steem.rpc.list_witnesses({'start': [last_account], 'limit': last_limit, 'order': 'by_vote_name'}, api="database")['witnesses']
elif self.steem.rpc.get_use_appbase() and use_condenser:
witnessess = self.steem.rpc.get_witnesses_by_vote(last_account, last_limit, api="condenser")
else:
witnessess = self.steem.rpc.get_witnesses_by_vote(last_account, last_limit)
# self.witness_count = len(self.voted_witnessess)
if (last_limit < limit):
witnessess = witnessess[1:]
if len(witnessess) > 0:
for x in witnessess:
witnessList.append(Witness(x, lazy=lazy, full=full, steem_instance=self.steem))
if len(witnessList) == 0:
return
super(WitnessesRankedByVote, self).__init__(witnessList)
def refresh(self):
self.steem.rpc.set_next_node_on_empty_reply(False)
if self.steem.rpc.get_use_appbase():
self.active_witnessess = self.steem.rpc.get_active_witnesses(api="database")['witnesses']
self.schedule = self.steem.rpc.get_witness_schedule(api="database")
self.witness_count = self.steem.rpc.get_witness_count(api="condenser")
else:
self.active_witnessess = self.steem.rpc.get_active_witnesses()
self.schedule = self.steem.rpc.get_witness_schedule()
self.witness_count = self.steem.rpc.get_witness_count()
self.current_witness = self.steem.get_dynamic_global_properties(use_stored_data=False)["current_witness"]
self.identifier = ""
super(Witnesses, self).__init__(
[
Witness(x, lazy=self.lazy, full=self.full, steem_instance=self.steem)
for x in self.active_witnessess
]
def witnessdisable(witness):
"""Disable a witness"""
stm = shared_steem_instance()
if stm.rpc is not None:
stm.rpc.rpcconnect()
if not witness:
witness = stm.config["default_account"]
if not unlock_wallet(stm):
return
witness = Witness(witness, steem_instance=stm)
if not witness.is_active:
print("Cannot disable a disabled witness!")
return
props = witness["props"]
tx = witness.update('STM1111111111111111111111111111111114T1Anm', witness["url"], props)
if stm.unsigned and stm.nobroadcast and stm.steemconnect is not None:
tx = stm.steemconnect.url_from_tx(tx)
tx = json.dumps(tx, indent=4)
print(tx)
def witnessenable(witness, signing_key):
"""Enable a witness"""
stm = shared_steem_instance()
if stm.rpc is not None:
stm.rpc.rpcconnect()
if not witness:
witness = stm.config["default_account"]
if not unlock_wallet(stm):
return
witness = Witness(witness, steem_instance=stm)
props = witness["props"]
tx = witness.update(signing_key, witness["url"], props)
if stm.unsigned and stm.nobroadcast and stm.steemconnect is not None:
tx = stm.steemconnect.url_from_tx(tx)
tx = json.dumps(tx, indent=4)
print(tx)