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_appendSigner(self):
nodelist = NodeList()
stm = Steem(node=self.nodes,
keys=[self.active_key],
nobroadcast=True,
expiration=120,
num_retries=10)
tx = TransactionBuilder(use_condenser_api=True, steem_instance=stm)
tx.appendOps(Transfer(**{"from": "beem",
"to": "beem1",
"amount": Amount("1 STEEM", steem_instance=stm),
"memo": ""}))
account = Account("beem", steem_instance=stm)
with self.assertRaises(
AssertionError
):
tx.appendSigner(account, "abcdefg")
tx.appendSigner(account, "active")
self.assertTrue(len(tx.wifs) > 0)
def test_price(self, node_param):
if node_param == "instance":
set_shared_steem_instance(self.bts)
o = Price(10.0, "STEEM/SBD")
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Price(10.0, "STEEM/SBD", 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 = Price(10.0, "STEEM/SBD", steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Price(10.0, "STEEM/SBD")
def test_amount(self, node_param):
if node_param == "instance":
stm = Steem(node="https://abc.d", autoconnect=False, num_retries=1)
set_shared_steem_instance(self.bts)
o = Amount("1 SBD")
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Amount("1 SBD", steem_instance=stm)
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = Amount("1 SBD", steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Amount("1 SBD")
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(exclude_limited=False), num_retries=10))
node_list = nodelist.get_nodes(exclude_limited=True)
cls.bts = Steem(
node=node_list,
use_condenser=True,
nobroadcast=True,
unsigned=True,
keys={"active": wif},
num_retries=10
)
cls.steemit = Steem(
node="https://api.steemit.com",
nobroadcast=True,
unsigned=True,
keys={"active": wif},
num_retries=10
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(exclude_limited=False), num_retries=10))
node_list = nodelist.get_nodes(exclude_limited=True)
cls.stm = Steem(
node=node_list,
keys={"active": wif, "owner": wif2, "memo": wif3},
nobroadcast=True,
num_retries=10
)
cls.steemit = Steem(
node="https://api.steemit.com",
nobroadcast=True,
keys={"active": wif, "owner": wif2, "memo": wif3},
num_retries=10
)
set_shared_steem_instance(cls.stm)
cls.stm.set_default_account("test")
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_steem_nodes(), num_retries=10))
cls.bts = Steem(
node=nodelist.get_steem_nodes(),
nobroadcast=True,
num_retries=10
)
set_shared_steem_instance(cls.bts)
cls.asset = Asset("SBD")
cls.symbol = cls.asset["symbol"]
cls.precision = cls.asset["precision"]
cls.asset2 = Asset("STEEM")
def setUpClass(cls):
stm = shared_steem_instance()
stm.config.refreshBackup()
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(exclude_limited=False), num_retries=10))
cls.stm = Steem(
node=nodelist.get_nodes(exclude_limited=True),
nobroadcast=True,
# We want to bundle many operations into a single transaction
bundle=True,
num_retries=10
# Overwrite wallet to use this list of wifs only
)
cls.stm.set_default_account("test")
set_shared_steem_instance(cls.stm)
# self.stm.newWallet("TestingOneTwoThree")
cls.wallet = Wallet(steem_instance=cls.stm)
cls.wallet.wipe(True)
tx = TransactionBuilder(use_condenser_api=True, steem_instance=steem)
tx.appendOps(Transfer(**{"from": 'beem5',
"to": 'beem',
"amount": Amount("0.01 STEEM", steem_instance=steem),
"memo": '2 of 2 serialized/deserialized transaction'}))
tx.appendSigner("beem5", "active")
tx.addSigningInformation("beem5", "active")
tx.sign()
tx.clearWifs()
self.assertEqual(len(tx['signatures']), 1)
tx_json = tx.json()
del steem
del tx
steem = Steem(
node=self.nodes,
num_retries=10,
keys=[self.active_private_key_of_beem4],
expiration=360,
)
new_tx = TransactionBuilder(tx=tx_json, steem_instance=steem)
new_tx.appendMissingSignatures()
new_tx.sign(reconstruct_tx=False)
self.assertEqual(len(new_tx['signatures']), 2)
new_tx.broadcast()
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(exclude_limited=False), num_retries=10))
node_list = nodelist.get_nodes(exclude_limited=True)
cls.bts = Steem(
node=node_list,
use_condenser=True,
nobroadcast=True,
unsigned=True,
keys={"active": wif},
num_retries=10
)
cls.steemit = Steem(
node="https://api.steemit.com",
nobroadcast=True,
unsigned=True,
keys={"active": wif},
num_retries=10
)
acc = Account("holger80", steem_instance=cls.bts)
comment = acc.get_feed(limit=20)[-1]
manualshorttext = "Hi @{}, \n Thank you for participating in the #travelfeed curated tag. To maintain a level of quality on the project we have certain criteria that must be met for participation. Please review the following: https://travelfeed.io/@travelfeed/how-to-participate-use-travelfeed-in-your-posts \n **We require at least 250 words.** \n Thank you very much for your interest and we hope to read some great travel articles from you soon! \n If you believe that you have received this comment by mistake or have updated your post to fit our criteria, please reply to this comment with <code>!tfreview</code>. For further questions, please contact us on the [TravelFeed Discord](https://discord.gg/jWWu73H). \n Regards, @travelfeed"
# Manual comment text for posts that are not in English
manuallangtext = "Hi @{}, \n Thank you for participating in the #travelfeed curated tag. To maintain a level of quality on the project we have certain criteria that must be met for participation. Please review the following: https://travelfeed.io/@travelfeed/how-to-participate-use-travelfeed-in-your-posts \n We require at least 250 words **in English**. \n Thank you very much for your interest and we hope to read some great travel articles from you soon! \n If you believe that you have received this comment by mistake or have updated your post to fit our criteria, please reply to this comment with <code>!tfreview</code>. For further questions, please contact us on the [TravelFeed Discord](https://discord.gg/jWWu73H). \n Regards, @travelfeed"
# Copyright text
copyrighttext = "Hi @{}, \n Thank you for participating in the #travelfeed curated tag. To maintain a level of quality on the project we have certain criteria that must be met for participation. Please review the following: https://travelfeed.io/@travelfeed/how-to-participate-use-travelfeed-in-your-posts \n We require **proper sourcing** for all media and text that is not your own. \n If you have updated your post with sources, please reply to this comment with <code>!tfreview</code>. For further questions, please contact us on the [TravelFeed Discord](https://discord.gg/jWWu73H). \n Thank you very much for your interest and we hope to read some great travel articles from you soon! \n Regards, @travelfeed"
honours = {}
resteems = {}
walletpw = os.environ.get('UNLOCK') #Beem wallet passphrase must be set as environment variable
TOKEN = os.environ.get('TOKEN') #Discord secret token must be set as environment variable
logger = logging.getLogger(__name__)
logging.basicConfig(filename=logpath, format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
nl = NodeList()
weights = {'block': 0.1, 'history': 1, 'apicall': 0.1, 'config': 0.1}
node_list = nl.update_nodes(weights)
steem = Steem(nodes=node_list)
steem.set_default_nodes(node_list)
steem.wallet.unlock(walletpw)
blockchain = Blockchain()
acc = Account(curationaccount)
# blacklist = acc.get_mutings(raw_name_list=True)
blacklist = []
"""
Discord functions
"""
async def send_discord(msg, cnl):
"""Sends the message *msg* to the Discord channel *cnl*"""
await bot.wait_until_ready()
await bot.send_message(bot.get_channel(cnl), msg)
bot = commands.Bot(command_prefix="!")