Skip to content

Commit

Permalink
migrate fairmint.py test
Browse files Browse the repository at this point in the history
  • Loading branch information
Ouziel committed Jan 23, 2025
1 parent 71f17f7 commit ade9e62
Showing 1 changed file with 184 additions and 0 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
import logging

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.backend.bitcoind:[437:454] ==counterpartycore.lib.backend.electrs:[83:100] if len(asm) == 4: # p2pkh # catch unhexlify errs for when asm[2] isn't a pubkey (eg; for P2SH) try: pubkey = asm[3] if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=False).to_string() ): return pubkey if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=True).to_string() ): return pubkey except binascii.Error: pass for vout in tx["vout"]:. Warning

Similar lines in 2 files
==counterpartycore.lib.backend.bitcoind:[437:454]
==counterpartycore.lib.backend.electrs:[83:100]
if len(asm) == 4: # p2pkh
# catch unhexlify errs for when asm[2] isn't a pubkey (eg; for P2SH)
try:
pubkey = asm[3]
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=False).to_string()
):
return pubkey
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=True).to_string()
):
return pubkey
except binascii.Error:
pass
for vout in tx["vout"]:.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.backend.bitcoind:[458:470] ==counterpartycore.lib.backend.electrs:[87:99] if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=False).to_string() ): return pubkey if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=True).to_string() ): return pubkey except binascii.Error: pass. Warning

Similar lines in 2 files
==counterpartycore.lib.backend.bitcoind:[458:470]
==counterpartycore.lib.backend.electrs:[87:99]
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=False).to_string()
):
return pubkey
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=True).to_string()
):
return pubkey
except binascii.Error:
pass.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.backend.bitcoind:[441:453] ==counterpartycore.lib.backend.electrs:[104:116] if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=False).to_string() ): return pubkey if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=True).to_string() ): return pubkey except binascii.Error: pass. Warning

Similar lines in 2 files
==counterpartycore.lib.backend.bitcoind:[441:453]
==counterpartycore.lib.backend.electrs:[104:116]
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=False).to_string()
):
return pubkey
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=True).to_string()
):
return pubkey
except binascii.Error:
pass.

Check warning

Code scanning / pylint

Similar lines in 2 files ==burn_test:[137:148] ==dividend_test:[138:149] "status": "valid", "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], }, }, { "table": "credits", "values": { "address": defaults["addresses"][1], "asset": "XCP", "block_index": current_block_index,. Warning

Similar lines in 2 files
==burn_test:[137:148]
==dividend_test:[138:149]
"status": "valid",
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
},
},
{
"table": "credits",
"values": {
"address": defaults["addresses"][1],
"asset": "XCP",
"block_index": current_block_index,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.dispense:[182:194] ==counterpartycore.lib.messages.dispenser:[763:775] } ledger.markets.update_dispenser( db, dispenser["rowid"], set_data, { "source": dispenser["source"], "asset": dispenser["asset"], "tx_hash": dispenser["tx_hash"], }, ) # use tx_index=0 for block actions . Warning

Similar lines in 2 files
==counterpartycore.lib.messages.dispense:[182:194]
==counterpartycore.lib.messages.dispenser:[763:775]
}
ledger.markets.update_dispenser(
db,
dispenser["rowid"],
set_data,
{
"source": dispenser["source"],
"asset": dispenser["asset"],
"tx_hash": dispenser["tx_hash"],
},
) # use tx_index=0 for block actions
.

Check warning

Code scanning / pylint

Similar lines in 2 files ==0004.create_and_populate_assets_info:[24:37] ==0005.create_and_populate_events_count:[23:36] if hasattr(db, "row_factory"): db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(*) AS count FROM pragma_database_list WHERE name = ?", ("ledger_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger_db", (config.DATABASE,)) db.execute(""". Warning

Similar lines in 2 files
==0004.create_and_populate_assets_info:[24:37]
==0005.create_and_populate_events_count:[23:36]
if hasattr(db, "row_factory"):
db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(\*) AS count FROM pragma\_database\_list WHERE name = ?", ("ledger\_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger\_db", (config.DATABASE,)) db.execute(""".

Check warning

Code scanning / pylint

Similar lines in 2 files ==0002.create_and_populate_parsed_events:[23:45] ==0003.create_and_populate_all_expirations:[23:44] if hasattr(db, "row_factory"): db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(*) AS count FROM pragma_database_list WHERE name = ?", ("ledger_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger_db", (config.DATABASE,)) sqls = [ """ CREATE TABLE all_expirations( type TEXT, object_id TEXT, block_index INTEGER ); """, """. Warning

Similar lines in 2 files
==0002.create_and_populate_parsed_events:[23:45]
==0003.create_and_populate_all_expirations:[23:44]
if hasattr(db, "row_factory"):
db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(\*) AS count FROM pragma\_database\_list WHERE name = ?", ("ledger\_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger\_db", (config.DATABASE,)) sqls = [ """ CREATE TABLE all\_expirations( type TEXT, object\_id TEXT, block\_index INTEGER ); """, """.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[15:31] ==counterpartycore.lib.messages.utxo:[17:33] problems = [] if asset == config.BTC: problems.append("cannot send bitcoins") # Only for parsing. if not isinstance(quantity, int): problems.append("quantity must be in satoshis") return problems if quantity <= 0: problems.append("quantity must be greater than zero") # For SQLite3 if quantity > config.MAX_INT: problems.append("integer overflow") . Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[15:31]
==counterpartycore.lib.messages.utxo:[17:33]
problems = [] if asset == config.BTC: problems.append("cannot send bitcoins") # Only for parsing. if not isinstance(quantity, int): problems.append("quantity must be in satoshis") return problems if quantity <= 0: problems.append("quantity must be greater than zero") # For SQLite3 if quantity > config.MAX_INT: problems.append("integer overflow") .

Check warning

Code scanning / pylint

Similar lines in 2 files ==0003.create_and_populate_all_expirations:[23:35] ==0004.create_and_populate_assets_info:[24:36] if hasattr(db, "row_factory"): db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(*) AS count FROM pragma_database_list WHERE name = ?", ("ledger_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger_db", (config.DATABASE,)) . Warning

Similar lines in 2 files
==0003.create_and_populate_all_expirations:[23:35]
==0004.create_and_populate_assets_info:[24:36]
if hasattr(db, "row_factory"):
db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(\*) AS count FROM pragma\_database\_list WHERE name = ?", ("ledger\_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger\_db", (config.DATABASE,)) .

Check warning

Code scanning / pylint

Similar lines in 2 files ==0001.create_and_populate_address_events:[22:34] ==0002.create_and_populate_parsed_events:[23:35] if hasattr(db, "row_factory"): db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(*) AS count FROM pragma_database_list WHERE name = ?", ("ledger_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger_db", (config.DATABASE,)) . Warning

Similar lines in 2 files
==0001.create_and_populate_address_events:[22:34]
==0002.create_and_populate_parsed_events:[23:35]
if hasattr(db, "row_factory"):
db.row_factory = dict_factory attached = ( db.execute( "SELECT COUNT(\*) AS count FROM pragma\_database\_list WHERE name = ?", ("ledger\_db",) ).fetchone()["count"] > 0 ) if not attached: db.execute("ATTACH DATABASE ? AS ledger\_db", (config.DATABASE,)) .

Check warning

Code scanning / pylint

Similar lines in 2 files ==dispense_test:[74:86] ==dispenser_test:[225:239] "address": defaults["addresses"][0], "asset": "XCP", "block_index": current_block_index, "event": tx["tx_hash"], "quantity": 100, }, }, ], ) def test_parse_lost_found(ledger_db, blockchain_mock, defaults, test_helpers, current_block_index):. Warning

Similar lines in 2 files
==dispense_test:[74:86]
==dispenser_test:[225:239]
"address": defaults["addresses"][0],
"asset": "XCP",
"block_index": current_block_index,
"event": tx["tx_hash"],
"quantity": 100,
},
},
],
) def test_parse_lost_found(ledger_db, blockchain_mock, defaults, test_helpers, current_block_index):.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.fairmint:[117:126] ==counterpartycore.lib.messages.fairminter:[359:368] if problems: status = "invalid: " + "; ".join(problems) bindings = { "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], "block_index": tx["block_index"], "source": tx["source"], "status": status, }. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.fairmint:[117:126]
==counterpartycore.lib.messages.fairminter:[359:368]
if problems:
status = "invalid: " + "; ".join(problems)
bindings = {
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
"block_index": tx["block_index"],
"source": tx["source"],
"status": status,
}.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.detach:[113:122] ==counterpartycore.lib.messages.move:[55:65] balance["asset"], balance["quantity"], tx["tx_index"], action=action, event=tx["tx_hash"], ) bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.detach:[113:122]
==counterpartycore.lib.messages.move:[55:65]
balance["asset"],
balance["quantity"],
tx["tx_index"],
action=action,
event=tx["tx_hash"],
)
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.detach:[101:110] ==counterpartycore.lib.messages.move:[42:52] source_address = ledger.events.debit( db, source, balance["asset"], balance["quantity"], tx["tx_index"], action=action, event=tx["tx_hash"], ). Warning

Similar lines in 2 files
==counterpartycore.lib.messages.detach:[101:110]
==counterpartycore.lib.messages.move:[42:52]
source_address = ledger.events.debit(
db,
source,
balance["asset"],
balance["quantity"],
tx["tx_index"],
action=action,
event=tx["tx_hash"],
).

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[151:161] ==counterpartycore.lib.messages.utxo:[217:227] event=tx["tx_hash"], ) # destroy fee destroy_bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"], "asset": config.XCP, "quantity": fee,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[151:161]
==counterpartycore.lib.messages.utxo:[217:227]
event=tx["tx_hash"],
)
# destroy fee
destroy_bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],
"asset": config.XCP,
"quantity": fee,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[192:202] ==counterpartycore.lib.messages.detach:[65:75] status = "valid" if problems: status = "invalid: " + "; ".join(problems) # store the invalid transaction without potentially invalid parameters bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]), "block_index": tx["block_index"], "status": status,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[192:202]
==counterpartycore.lib.messages.detach:[65:75]
status = "valid"
if problems:
status = "invalid: " + "; ".join(problems)
# store the invalid transaction without potentially invalid parameters
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]),
"block_index": tx["block_index"],
"status": status,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.api.compose:[449:458] ==counterpartycore.lib.messages.fairminter:[40:49] "price": price, "quantity_by_price": quantity_by_price, "max_mint_per_tx": max_mint_per_tx, "hard_cap": hard_cap, "premint_quantity": premint_quantity, "start_block": start_block, "end_block": end_block, "soft_cap": soft_cap, "soft_cap_deadline_block": soft_cap_deadline_block,. Warning

Similar lines in 2 files
==counterpartycore.lib.api.compose:[449:458]
==counterpartycore.lib.messages.fairminter:[40:49]
"price": price,
"quantity_by_price": quantity_by_price,
"max_mint_per_tx": max_mint_per_tx,
"hard_cap": hard_cap,
"premint_quantity": premint_quantity,
"start_block": start_block,
"end_block": end_block,
"soft_cap": soft_cap,
"soft_cap_deadline_block": soft_cap_deadline_block,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==burn_test:[176:184] ==dividend_test:[138:146] "status": "valid", "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], }, }, { "table": "credits", "values": {. Warning

Similar lines in 2 files
==burn_test:[176:184]
==dividend_test:[138:146]
"status": "valid",
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
},
},
{
"table": "credits",
"values": {.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.bet:[562:570] ==counterpartycore.lib.messages.order:[779:787] bindings = { "id": helpers.make_id(tx0["tx_hash"], tx["tx_hash"]), "tx0_index": tx0["tx_index"], "tx0_hash": tx0["tx_hash"], "tx0_address": tx0["source"], "tx1_index": tx1["tx_index"], "tx1_hash": tx1["tx_hash"], "tx1_address": tx1["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.bet:[562:570]
==counterpartycore.lib.messages.order:[779:787]
bindings = {
"id": helpers.make_id(tx0["tx_hash"], tx["tx_hash"]),
"tx0_index": tx0["tx_index"],
"tx0_hash": tx0["tx_hash"],
"tx0_address": tx0["source"],
"tx1_index": tx1["tx_index"],
"tx1_hash": tx1["tx_hash"],
"tx1_address": tx1["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[225:236] ==counterpartycore.lib.messages.detach:[115:126] tx["tx_index"], action=action, event=tx["tx_hash"], ) bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]), "block_index": tx["block_index"], "status": "valid", "source": source,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[225:236]
==counterpartycore.lib.messages.detach:[115:126]
tx["tx_index"],
action=action,
event=tx["tx_hash"],
)
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]),
"block_index": tx["block_index"],
"status": "valid",
"source": source,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==attach_test:[99:106] ==detach_test:[50:57] { "table": "sends", "values": { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "status": "valid",. Warning

Similar lines in 2 files
==attach_test:[99:106]
==detach_test:[50:57]
{
"table": "sends",
"values": {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"status": "valid",.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.sweep:[95:103] ==counterpartycore.lib.messages.versions.enhancedsend:[20:29] try: # account for memo bytes memo_bytes_length = len(message) - LENGTH if memo_bytes_length < 0: raise exceptions.UnpackError("invalid message length") if memo_bytes_length > MAX_MEMO_LENGTH: raise exceptions.UnpackError("memo too long") struct_format = FORMAT + f"{memo_bytes_length}s". Warning

Similar lines in 2 files
==counterpartycore.lib.messages.sweep:[95:103]
==counterpartycore.lib.messages.versions.enhancedsend:[20:29]
try:
# account for memo bytes
memo_bytes_length = len(message) - LENGTH
if memo_bytes_length < 0:
raise exceptions.UnpackError("invalid message length")
if memo_bytes_length > MAX_MEMO_LENGTH:
raise exceptions.UnpackError("memo too long") struct_format = FORMAT + f"{memo_bytes_length}s".

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.order:[442:451] ==counterpartycore.lib.messages.versions.send1:[169:178] event=tx["tx_hash"], ) # Add parsed transaction to message-type–specific table. bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.order:[442:451]
==counterpartycore.lib.messages.versions.send1:[169:178]
event=tx["tx_hash"],
) # Add parsed transaction to message-type–specific table. bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"], "source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.fairmint:[92:102] ==counterpartycore.lib.messages.fairminter:[234:244] ] ] ).encode("utf-8") data += struct.pack(f">{len(data_content)}s", data_content) return (source, [], data) def unpack(message, return_dict=False): try: data_content = struct.unpack(f">{len(message)}s", message)[0].decode("utf-8").split("|"). Warning

Similar lines in 2 files
==counterpartycore.lib.messages.fairmint:[92:102]
==counterpartycore.lib.messages.fairminter:[234:244]
]
]
).encode("utf-8")
data += struct.pack(f">{len(data_content)}s", data_content)
return (source, [], data) def unpack(message, return_dict=False):
try:
data_content = struct.unpack(f">{len(message)}s", message)[0].decode("utf-8").split("|").

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.dividend:[278:287] ==counterpartycore.lib.messages.issuance:[808:816] ) bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"], "asset": asset,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.dividend:[278:287]
==counterpartycore.lib.messages.issuance:[808:816]
) bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"], "source": tx["source"], "asset": asset,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.btcpay:[165:174] ==counterpartycore.lib.messages.versions.send1:[170:179] ) # Add parsed transaction to message-type–specific table. bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"], "destination": tx["destination"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.btcpay:[165:174]
==counterpartycore.lib.messages.versions.send1:[170:179]
) # Add parsed transaction to message-type–specific table. bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"], "source": tx["source"], "destination": tx["destination"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.bet:[358:367] ==counterpartycore.lib.messages.dividend:[277:286] event=tx["tx_hash"], ) # Add parsed transaction to message-type–specific table. bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.bet:[358:367]
==counterpartycore.lib.messages.dividend:[277:286]
event=tx["tx_hash"],
) # Add parsed transaction to message-type–specific table. bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"], "source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[37:45] ==counterpartycore.lib.messages.utxo:[69:77] asset_balance = ledger.balances.get_balance(db, source, asset) if asset == config.XCP: # fee is always paid in XCP if asset_balance < quantity + fee: problems.append("insufficient funds for transfer and fee") else: if asset_balance < quantity: problems.append("insufficient funds for transfer"). Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[37:45]
==counterpartycore.lib.messages.utxo:[69:77]
asset_balance = ledger.balances.get_balance(db, source, asset)
if asset == config.XCP:
# fee is always paid in XCP
if asset_balance < quantity + fee:
problems.append("insufficient funds for transfer and fee")
else:
if asset_balance < quantity:
problems.append("insufficient funds for transfer").

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[225:232] ==counterpartycore.lib.messages.move:[57:65] tx["tx_index"], action=action, event=tx["tx_hash"], ) bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[225:232]
==counterpartycore.lib.messages.move:[57:65]
tx["tx_index"],
action=action,
event=tx["tx_hash"],
)
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[92:101] ==counterpartycore.lib.messages.fairmint:[83:92] data = struct.pack(config.SHORT_TXTYPE_FORMAT, ID) # to optimize the data size (avoiding fixed sizes per parameter) we use a simple # string of characters separated by |. data_content = "|".join( [ str(value) for value in [ asset, quantity,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[92:101]
==counterpartycore.lib.messages.fairmint:[83:92]
data = struct.pack(config.SHORT_TXTYPE_FORMAT, ID)
# to optimize the data size (avoiding fixed sizes per parameter) we use a simple
# string of characters separated by |.
data_content = "|".join(
[
str(value)
for value in [
asset,
quantity,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.backend.bitcoind:[455:474] ==counterpartycore.lib.backend.electrs:[101:120] if len(asm) == 3: # p2pk try: pubkey = asm[1] if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=False).to_string() ): return pubkey if ( pubkeyhash == PublicKey.from_hex(pubkey).get_address(compressed=True).to_string() ): return pubkey except binascii.Error: pass return None def search_pubkey(pubkeyhash):. Warning

Similar lines in 2 files
==counterpartycore.lib.backend.bitcoind:[455:474]
==counterpartycore.lib.backend.electrs:[101:120]
if len(asm) == 3: # p2pk
try:
pubkey = asm[1]
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=False).to_string()
):
return pubkey
if (
pubkeyhash
== PublicKey.from_hex(pubkey).get_address(compressed=True).to_string()
):
return pubkey
except binascii.Error:
pass
return None def search_pubkey(pubkeyhash):.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.api.compose:[46:53] ==counterpartycore.lib.messages.bet:[278:285] "bet_type": bet_type, "deadline": deadline, "wager_quantity": wager_quantity, "counterwager_quantity": counterwager_quantity, "target_value": target_value, "leverage": leverage, "expiration": expiration,. Warning

Similar lines in 2 files
==counterpartycore.lib.api.compose:[46:53]
==counterpartycore.lib.messages.bet:[278:285]
"bet_type": bet_type,
"deadline": deadline,
"wager_quantity": wager_quantity,
"counterwager_quantity": counterwager_quantity,
"target_value": target_value,
"leverage": leverage,
"expiration": expiration,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.versions.enhancedsend:[122:134] ==counterpartycore.lib.messages.versions.send1:[98:109] asset = ledger.issuances.resolve_subasset_longname(db, asset) # quantity must be in int satoshi (not float, string, etc) if not isinstance(quantity, int): raise exceptions.ComposeError("quantity must be an int (in satoshi)") # Only for outgoing (incoming will overburn). balance = ledger.balances.get_balance(db, source, asset) if balance < quantity and not skip_validation: raise exceptions.ComposeError("insufficient funds") . Warning

Similar lines in 2 files
==counterpartycore.lib.messages.versions.enhancedsend:[122:134]
==counterpartycore.lib.messages.versions.send1:[98:109]
asset = ledger.issuances.resolve_subasset_longname(db, asset) # quantity must be in int satoshi (not float, string, etc) if not isinstance(quantity, int): raise exceptions.ComposeError("quantity must be an int (in satoshi)") # Only for outgoing (incoming will overburn). balance = ledger.balances.get_balance(db, source, asset) if balance < quantity and not skip_validation: raise exceptions.ComposeError("insufficient funds") .

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.versions.enhancedsend:[89:95] ==counterpartycore.lib.messages.versions.send1:[59:65] results = ledger.other.get_addresses(db, address=destination) if results: result = results[0] if result and helpers.active_options( result["options"], config.ADDRESS_OPTION_REQUIRE_MEMO ):. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.versions.enhancedsend:[89:95]
==counterpartycore.lib.messages.versions.send1:[59:65]
results = ledger.other.get_addresses(db, address=destination)
if results:
result = results[0]
if result and helpers.active_options(
result["options"], config.ADDRESS_OPTION_REQUIRE_MEMO
):.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.sweep:[253:259] ==counterpartycore.lib.messages.versions.enhancedsend:[214:220] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"], "destination": destination,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.sweep:[253:259]
==counterpartycore.lib.messages.versions.enhancedsend:[214:220]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],
"destination": destination,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.issuance:[808:815] ==counterpartycore.lib.messages.order:[443:451] ) # Add parsed transaction to message-type–specific table. bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.issuance:[808:815]
==counterpartycore.lib.messages.order:[443:451]
) # Add parsed transaction to message-type–specific table. bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"], "source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.dividend:[278:286] ==counterpartycore.lib.messages.versions.enhancedsend:[211:219] ) # Add parsed transaction to message-type–specific table. bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.dividend:[278:286]
==counterpartycore.lib.messages.versions.enhancedsend:[211:219]
) # Add parsed transaction to message-type–specific table. bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"], "source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.dividend:[257:263] ==counterpartycore.lib.messages.issuance:[873:879] ledger.events.debit( db, tx["source"], config.XCP, fee, tx["tx_index"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.dividend:[257:263]
==counterpartycore.lib.messages.issuance:[873:879]
ledger.events.debit(
db,
tx["source"],
config.XCP,
fee,
tx["tx_index"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.detach:[113:119] ==counterpartycore.lib.messages.move:[45:52] balance["asset"], balance["quantity"], tx["tx_index"], action=action, event=tx["tx_hash"], ). Warning

Similar lines in 2 files
==counterpartycore.lib.messages.detach:[113:119]
==counterpartycore.lib.messages.move:[45:52]
balance["asset"],
balance["quantity"],
tx["tx_index"],
action=action,
event=tx["tx_hash"],
).

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.detach:[104:110] ==counterpartycore.lib.messages.move:[55:62] balance["asset"], balance["quantity"], tx["tx_index"], action=action, event=tx["tx_hash"], ). Warning

Similar lines in 2 files
==counterpartycore.lib.messages.detach:[104:110]
==counterpartycore.lib.messages.move:[55:62]
balance["asset"],
balance["quantity"],
tx["tx_index"],
action=action,
event=tx["tx_hash"],
).

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.destroy:[114:120] ==counterpartycore.lib.messages.dividend:[281:287] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"], "asset": asset,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.destroy:[114:120]
==counterpartycore.lib.messages.dividend:[281:287]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],
"asset": asset,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.bet:[183:190] ==counterpartycore.lib.messages.order:[273:280] if expiration < 0: problems.append("negative expiration") if expiration == 0 and not protocol.after_block_or_test_network( block_index, 317500 ): # Protocol change. problems.append("zero expiration") . Warning

Similar lines in 2 files
==counterpartycore.lib.messages.bet:[183:190]
==counterpartycore.lib.messages.order:[273:280]
if expiration < 0:
problems.append("negative expiration")
if expiration == 0 and not protocol.after_block_or_test_network(
block_index, 317500
): # Protocol change.
problems.append("zero expiration")
.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.bet:[358:366] ==counterpartycore.lib.messages.move:[59:66] event=tx["tx_hash"], ) # Add parsed transaction to message-type–specific table. bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.bet:[358:366]
==counterpartycore.lib.messages.move:[59:66]
event=tx["tx_hash"],
) # Add parsed transaction to message-type–specific table. bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.bet:[359:367] ==counterpartycore.lib.messages.btcpay:[165:173] ) # Add parsed transaction to message-type–specific table. bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.bet:[359:367]
==counterpartycore.lib.messages.btcpay:[165:173]
) # Add parsed transaction to message-type–specific table. bindings = { "tx\_index": tx["tx\_index"], "tx\_hash": tx["tx\_hash"], "block\_index": tx["block\_index"], "source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[15:24] ==counterpartycore.lib.messages.versions.send1:[35:44] problems = [] if asset == config.BTC: problems.append("cannot send bitcoins") # Only for parsing. if not isinstance(quantity, int): problems.append("quantity must be in satoshis") return problems . Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[15:24]
==counterpartycore.lib.messages.versions.send1:[35:44]
problems = [] if asset == config.BTC: problems.append("cannot send bitcoins") # Only for parsing. if not isinstance(quantity, int): problems.append("quantity must be in satoshis") return problems .

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[223:229] ==counterpartycore.lib.messages.utxo:[238:245] asset, quantity, tx["tx_index"], action=action, event=tx["tx_hash"], ). Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[223:229]
==counterpartycore.lib.messages.utxo:[238:245]
asset,
quantity,
tx["tx_index"],
action=action,
event=tx["tx_hash"],
).

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[111:118] ==counterpartycore.lib.messages.move:[24:31] value = config.DEFAULT_UTXO_VALUE if utxo_value is not None: try: value = int(utxo_value) except ValueError as e: raise exceptions.ComposeError(["utxo_value must be an integer"]) from e # else we use the source address as the destination. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[111:118]
==counterpartycore.lib.messages.move:[24:31]
value = config.DEFAULT_UTXO_VALUE
if utxo_value is not None:
try:
value = int(utxo_value)
except ValueError as e:
raise exceptions.ComposeError(["utxo_value must be an integer"]) from e
# else we use the source address as the destination.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[92:100] ==counterpartycore.lib.messages.fairminter:[209:218] data = struct.pack(config.SHORT_TXTYPE_FORMAT, ID) # to optimize the data size (avoiding fixed sizes per parameter) we use a simple # string of characters separated by |. data_content = "|".join( [ str(value) for value in [ asset,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[92:100]
==counterpartycore.lib.messages.fairminter:[209:218]
data = struct.pack(config.SHORT_TXTYPE_FORMAT, ID)
# to optimize the data size (avoiding fixed sizes per parameter) we use a simple
# string of characters separated by |.
data_content = "|".join(
[
str(value)
for value in [
asset,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.ledger.markets:[105:120] ==counterpartycore.lib.ledger.other:[186:201] WHERE expire_index = ? - 1 GROUP BY tx_hash ) WHERE status = ? ORDER BY tx_index, tx_hash """ bindings = (block_index, "open") cursor.execute(query, bindings) return cursor.fetchall() def get_open_btc_orders(db, address): cursor = db.cursor() query = """ SELECT * FROM ( SELECT *, MAX(rowid). Warning

Similar lines in 2 files
==counterpartycore.lib.ledger.markets:[105:120]
==counterpartycore.lib.ledger.other:[186:201]
WHERE expire_index = ? - 1
GROUP BY tx_hash
) WHERE status = ?
ORDER BY tx_index, tx_hash
"""
bindings = (block_index, "open")
cursor.execute(query, bindings)
return cursor.fetchall() def get_open_btc_orders(db, address):
cursor = db.cursor()
query = """
SELECT * FROM (
SELECT *, MAX(rowid).

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.api.compose:[306:312] ==counterpartycore.lib.messages.order:[367:373] "give_asset": give_asset, "give_quantity": give_quantity, "get_asset": get_asset, "get_quantity": get_quantity, "expiration": expiration, "fee_required": fee_required,. Warning

Similar lines in 2 files
==counterpartycore.lib.api.compose:[306:312]
==counterpartycore.lib.messages.order:[367:373]
"give_asset": give_asset,
"give_quantity": give_quantity,
"get_asset": get_asset,
"get_quantity": get_quantity,
"expiration": expiration,
"fee_required": fee_required,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.api.apiv1:[369:375] ==counterpartycore.lib.api.verbose:[528:534] filtered_results = [] for row in list(query_result): if "rowid" in row: del row["rowid"] if "MAX(rowid)" in row: del row["MAX(rowid)"]. Warning

Similar lines in 2 files
==counterpartycore.lib.api.apiv1:[369:375]
==counterpartycore.lib.api.verbose:[528:534]
filtered_results = []
for row in list(query_result):
if "rowid" in row:
del row["rowid"]
if "MAX(rowid)" in row:
del row["MAX(rowid)"].

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.api.apiv1:[110:116] ==counterpartycore.lib.api.queries:[91:97] "cancel", "destroy", "dispenser", "dispense", "dividend", "issuance",. Warning

Similar lines in 2 files
==counterpartycore.lib.api.apiv1:[110:116]
==counterpartycore.lib.api.queries:[91:97]
"cancel",
"destroy",
"dispenser",
"dispense",
"dividend",
"issuance",.

Check warning

Code scanning / pylint

Similar lines in 2 files ==bet_test:[454:459] ==dispenser_test:[208:213] "values": { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": defaults["addresses"][0],. Warning

Similar lines in 2 files
==bet_test:[454:459]
==dispenser_test:[208:213]
"values": {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": defaults["addresses"][0],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.move:[53:58] ==counterpartycore.lib.messages.sweep:[201:206] db, destination, balance["asset"], balance["quantity"], tx["tx_index"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.move:[53:58]
==counterpartycore.lib.messages.sweep:[201:206]
db,
destination,
balance["asset"],
balance["quantity"],
tx["tx_index"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.issuance:[810:815] ==counterpartycore.lib.messages.versions.mpma:[246:251] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.issuance:[810:815]
==counterpartycore.lib.messages.versions.mpma:[246:251]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.fairmint:[190:196] ==counterpartycore.lib.messages.fairminter:[604:609] if xcp_destination: # we credit the destination if it exists (issuer or escrow) ledger.events.credit( db, xcp_destination, config.XCP,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.fairmint:[190:196]
==counterpartycore.lib.messages.fairminter:[604:609]
if xcp_destination:
# we credit the destination if it exists (issuer or escrow)
ledger.events.credit(
db,
xcp_destination,
config.XCP,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.fairmint:[244:249] ==counterpartycore.lib.messages.fairminter:[361:366] bindings = { "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.fairmint:[244:249]
==counterpartycore.lib.messages.fairminter:[361:366]
bindings = {
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
"block_index": tx["block_index"],
"source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.fairmint:[119:124] ==counterpartycore.lib.messages.fairminter:[417:422] bindings = { "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.fairmint:[119:124]
==counterpartycore.lib.messages.fairminter:[417:422]
bindings = {
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
"block_index": tx["block_index"],
"source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.dividend:[281:286] ==counterpartycore.lib.messages.sweep:[253:258] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.dividend:[281:286]
==counterpartycore.lib.messages.sweep:[253:258]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.destroy:[114:119] ==counterpartycore.lib.messages.fairmint:[203:208] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.destroy:[114:119]
==counterpartycore.lib.messages.fairmint:[203:208]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.burn:[33:41] ==counterpartycore.lib.messages.versions.enhancedsend:[64:71] if not isinstance(quantity, int): problems.append("quantity must be in satoshis") return problems if quantity < 0: problems.append("negative quantity") . Warning

Similar lines in 2 files
==counterpartycore.lib.messages.burn:[33:41]
==counterpartycore.lib.messages.versions.enhancedsend:[64:71]
if not isinstance(quantity, int):
problems.append("quantity must be in satoshis")
return problems if quantity < 0: problems.append("negative quantity") .

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.btcpay:[168:173] ==counterpartycore.lib.messages.cancel:[100:105] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.btcpay:[168:173]
==counterpartycore.lib.messages.cancel:[100:105]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.bet:[362:367] ==counterpartycore.lib.messages.broadcast:[223:228] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.bet:[362:367]
==counterpartycore.lib.messages.broadcast:[223:228]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[155:160] ==counterpartycore.lib.messages.fairmint:[204:209] "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "block_index": tx["block_index"], "source": tx["source"], "asset": config.XCP,. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[155:160]
==counterpartycore.lib.messages.fairmint:[204:209]
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": tx["source"],
"asset": config.XCP,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[229:234] ==counterpartycore.lib.messages.detach:[69:74] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]), "block_index": tx["block_index"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[229:234]
==counterpartycore.lib.messages.detach:[69:74]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]),
"block_index": tx["block_index"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.attach:[196:201] ==counterpartycore.lib.messages.detach:[119:124] bindings = { "tx_index": tx["tx_index"], "tx_hash": tx["tx_hash"], "msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]), "block_index": tx["block_index"],. Warning

Similar lines in 2 files
==counterpartycore.lib.messages.attach:[196:201]
==counterpartycore.lib.messages.detach:[119:124]
bindings = {
"tx_index": tx["tx_index"],
"tx_hash": tx["tx_hash"],
"msg_index": ledger.other.get_send_msg_index(db, tx["tx_hash"]),
"block_index": tx["block_index"],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.ledger.markets:[122:136] ==counterpartycore.lib.ledger.other:[187:201] GROUP BY tx_hash ) WHERE status = ? ORDER BY tx_index, tx_hash """ bindings = (address, config.BTC, "open") cursor.execute(query, bindings) return cursor.fetchall() def get_matching_orders_no_cache(db, tx_hash, give_asset, get_asset): cursor = db.cursor() query = """ SELECT * FROM ( SELECT *, MAX(rowid). Warning

Similar lines in 2 files
==counterpartycore.lib.ledger.markets:[122:136]
==counterpartycore.lib.ledger.other:[187:201]
GROUP BY tx_hash
) WHERE status = ?
ORDER BY tx_index, tx_hash
"""
bindings = (address, config.BTC, "open")
cursor.execute(query, bindings)
return cursor.fetchall() def get_matching_orders_no_cache(db, tx_hash, give_asset, get_asset):
cursor = db.cursor()
query = """
SELECT * FROM (
SELECT *, MAX(rowid).

import pytest
from counterpartycore.lib import config, exceptions
from counterpartycore.lib.messages import fairmint

logger = logging.getLogger(config.LOGGER_NAME)


def test_validate(ledger_db, defaults):
assert (
fairmint.validate(
ledger_db,
defaults["addresses"][1], # source
"FREEFAIRMIN", # asset
0, # quantity
)
== []
)

assert fairmint.validate(
ledger_db,
defaults["addresses"][1], # source
"PAIDFAIRMIN", # asset
0, # quantity
) == ["Quantity must be greater than 0"]

assert fairmint.validate(
ledger_db,
defaults["addresses"][1], # source
"RAIDFAIRMIN", # asset
11, # quantity
) == ["Quantity exceeds maximum allowed per transaction"]

assert fairmint.validate(
ledger_db,
defaults["addresses"][1], # source
"QAIDFAIRMIN", # asset
35, # quantity
) == ["asset supply quantity exceeds hard cap"]


def test_compose(ledger_db, defaults):
assert fairmint.compose(
ledger_db,
defaults["addresses"][1], # source
"FREEFAIRMIN", # asset
0, # quantity
) == (
defaults["addresses"][1],
[],
b"[FREEFAIRMIN|0",
)

with pytest.raises(exceptions.ComposeError, match="asset supply quantity exceeds hard cap"):
fairmint.compose(
ledger_db,
defaults["addresses"][1], # source
"QAIDFAIRMIN", # asset
35, # quantity
)

with pytest.raises(
exceptions.ComposeError, match="quantity is not allowed for free fairminters"
):
fairmint.compose(
ledger_db,
defaults["addresses"][1], # source
"FREEFAIRMIN", # asset
35, # quantity
)


def test_unpack():
assert fairmint.unpack(b"FREEFAIRMIN|0", False) == ("FREEFAIRMIN", 0)
assert fairmint.unpack(b"FREEFAIRMIN|0", True) == {"asset": "FREEFAIRMIN", "quantity": 0}


def tes_parse_freefairmint(ledger_db, blockchain_mock, defaults, test_helpers, current_block_index):
tx = blockchain_mock.dummy_tx(ledger_db, defaults["addresses"][0])
message = b"FREEFAIRMIN|0"
fairmint.parse(ledger_db, tx, message)

test_helpers.check_records(
ledger_db,
[
{
"table": "fairmints",
"values": {
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": defaults["addresses"][0],
"asset": "FREEFAIRMIN",
"earn_quantity": 10,
"paid_quantity": 0,
"commission": 0,
"status": "valid",
},
},
{
"table": "issuances",
"values": {
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"asset": "FREEFAIRMIN",
"quantity": 10,
"divisible": True,
"source": defaults["addresses"][0],
"issuer": defaults["addresses"][0],
"transfer": False,
"callable": False,
"call_date": 0,
"call_price": 0,
"description": "",
"fee_paid": 0,
"locked": False,
"reset": False,
"status": "valid",
"asset_longname": None,
"fair_minting": True,
},
},
{
"table": "credits",
"values": {
"block_index": current_block_index,
"address": defaults["addresses"][0],
"asset": "FREEFAIRMIN",
"quantity": 10,
"calling_function": "fairmint",
"event": tx["tx_hash"],
},
},
],
)


def test_parse_escrowed_fairmint(
ledger_db, blockchain_mock, defaults, test_helpers, current_block_index
):
tx = blockchain_mock.dummy_tx(ledger_db, defaults["addresses"][0])
message = b"QAIDFAIRMIN|10"
fairmint.parse(ledger_db, tx, message)

test_helpers.check_records(
ledger_db,
[
{
"table": "fairmints",
"values": {
"tx_hash": tx["tx_hash"],
"block_index": tx["block_index"],
"source": defaults["addresses"][0],
"asset": "QAIDFAIRMIN",
"earn_quantity": 5,
"paid_quantity": 100,
"commission": 5,
"status": "valid",
},
},
{
"table": "credits",
"values": {
"block_index": current_block_index,
"address": defaults["unspendable"],
"asset": "QAIDFAIRMIN",
"quantity": 10,
"calling_function": "escrowed fairmint",
"event": tx["tx_hash"],
},
},
{
"table": "credits",
"values": {
"block_index": current_block_index,
"address": defaults["unspendable"],
"asset": "XCP",
"quantity": 100,
"calling_function": "escrowed fairmint",
"event": tx["tx_hash"],
},
},
],
)

0 comments on commit ade9e62

Please sign in to comment.