Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
migrate utils.py test
Browse files Browse the repository at this point in the history
Ouziel committed Jan 27, 2025
1 parent ed36e39 commit c740e0a
Showing 2 changed files with 173 additions and 0 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
import pytest
from counterpartycore.lib import exceptions
from counterpartycore.lib.utils import base58


def test_base58_functions(defaults):
assert (
base58.base58_check_encode("010966776006953d5567439e5e39f86a0d273bee", b"\x00")
== "16UwLL9Risc3QfPqBUvKofHmBQ7wMtjvM"
)
assert (
base58.base58_check_encode("010966776006953d5567439e5e39f86a0d273bee", b"\x05")
== "31nVrspaydBz8aMpxH9WkS2DuhgqS1fCuG"
)

assert (
base58.base58_check_decode("16UwLL9Risc3QfPqBUvKofHmBQ7wMtjvM", b"\x00")
== b"\x01\tfw`\x06\x95=UgC\x9e^9\xf8j\r';\xee"
)
assert (
base58.base58_check_decode("13PGb7v3nmTDugLDStRJWXw6TzsNLUKJKC", b"\x00")
== b"\x1a&jGxV\xea\xd2\x9e\xcb\xe6\xaeQ\xad:,\x8dG<\xf4"
)
assert (
base58.base58_check_decode("31nVrspaydBz8aMpxH9WkS2DuhgqS1fCuG", b"\x05")
== b"\x01\tfw`\x06\x95=UgC\x9e^9\xf8j\r';\xee"
)
assert (
base58.base58_check_decode(defaults["addresses"][0], b"\x6f")
== b"H8\xd8\xb3X\x8cL{\xa7\xc1\xd0o\x86n\x9b79\xc607"
)

with pytest.raises(exceptions.VersionByteError, match="incorrect version byte"):
base58.base58_check_decode("13PGb7v3nmTDugLDStRJWXw6TzsNLUKJKC", b"\x05")

with pytest.raises(exceptions.VersionByteError, match="incorrect version byte"):
base58.base58_check_decode("31nVrspaydBz8aMpxH9WkS2DuhgqS1fCuG", b"\x00")

with pytest.raises(exceptions.Base58Error, match="invalid base58 string"):
base58.base58_check_decode("26UwLL9Risc3QfPqBUvKofHmBQ7wMtjvM", b"\x00")

with pytest.raises(exceptions.Base58Error, match="invalid base58 string"):
base58.base58_check_decode("16UwLL9Risc3QfPqBUvKofHmBQ7wMtjvN", b"\x00")

with pytest.raises(exceptions.Base58Error, match="invalid base58 string"):
base58.base58_check_decode("16UwLL9Risc3QfPqBUvKofHmBQ7wMtjv0", b"\x00")
127 changes: 127 additions & 0 deletions counterparty-core/counterpartycore/pytest/units/utils/multisig_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
import pytest

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[1] 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[1] 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 ==dividend_test:[133:145] ==send_test:[330:342] "source": defaults["addresses"][0], "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
==dividend_test:[133:145]
==send_test:[330:342]
"source": defaults["addresses"][0],
"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.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:[176:187] ==send_test:[428:439] "status": "valid", "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], }, }, { "table": "credits", "values": { "address": defaults["p2ms_addresses"][0], "asset": "XCP", "block_index": current_block_index,. Warning

Similar lines in 2 files
==burn_test:[176:187]
==send_test:[428:439]
"status": "valid",
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
},
},
{
"table": "credits",
"values": {
"address": defaults["p2ms_addresses"][0],
"asset": "XCP",
"block_index": current_block_index,.

Check warning

Code scanning / pylint

Similar lines in 2 files ==burn_test:[137:148] ==dividend_test:[134:145] "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:[134:145]
"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 ==attach_test:[78:89] ==utxo_test:[85:96] "asset": "XCP", "quantity": 100, "event": tx["tx_hash"], "block_index": current_block_index, "tx_index": tx["tx_index"], "action": "attach to utxo", }, }, { "table": "credits", "values": {. Warning

Similar lines in 2 files
==attach_test:[78:89]
==utxo_test:[85:96]
"asset": "XCP",
"quantity": 100,
"event": tx["tx_hash"],
"block_index": current_block_index,
"tx_index": tx["tx_index"],
"action": "attach to utxo",
},
},
{
"table": "credits",
"values": {.

Check warning

Code scanning / pylint

Similar lines in 2 files ==counterpartycore.lib.messages.dispense:[182:194] ==counterpartycore.lib.messages.dispenser:[760:772] } 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:[760:772]
}
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 ==send_test:[10:23] ==versions.mpma_test:[267:278] ledger.events.insert_record( ledger_db, "addresses", { "block_index": current_block_index, "address": defaults["addresses"][6], "options": config.ADDRESS_OPTION_REQUIRE_MEMO, }, "NEW_ADDRESS_OPTIONS", ) . Warning

Similar lines in 2 files
==send_test:[10:23]
==versions.mpma_test:[267:278]
ledger.events.insert_record(
ledger_db,
"addresses",
{
"block_index": current_block_index,
"address": defaults["addresses"][6],
"options": config.ADDRESS_OPTION_REQUIRE_MEMO,
},
"NEW_ADDRESS_OPTIONS",
)
.

Check warning

Code scanning / pylint

Similar lines in 2 files ==dividend_test:[216:226] ==send_test:[569:579] "source": defaults["addresses"][0], "status": "valid", "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], }, }, { "table": "credits", "values": { "address": defaults["addresses"][1],. Warning

Similar lines in 2 files
==dividend_test:[216:226]
==send_test:[569:579]
"source": defaults["addresses"][0],
"status": "valid",
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
},
},
{
"table": "credits",
"values": {
"address": defaults["addresses"][1],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==dividend_test:[133:143] ==send_test:[379:389] "source": defaults["addresses"][0], "status": "valid", "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], }, }, { "table": "credits", "values": { "address": defaults["addresses"][1],. Warning

Similar lines in 2 files
==dividend_test:[133:143]
==send_test:[379:389]
"source": defaults["addresses"][0],
"status": "valid",
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
},
},
{
"table": "credits",
"values": {
"address": defaults["addresses"][1],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==detach_test:[51:61] ==utxo_test:[149:159] test_helpers.check_records( ledger_db, [ { "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
==detach_test:[51:61]
==utxo_test:[149:159]
test_helpers.check_records(
ledger_db,
[
{
"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.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 ==blocks_test:[41:50] ==send_test:[417:426] test_helpers.check_records( ledger_db, [ { "table": "sends", "values": { "asset": "XCP", "block_index": tx["block_index"], "destination": defaults["p2ms_addresses"][0],. Warning

Similar lines in 2 files
==blocks_test:[41:50]
==send_test:[417:426]
test_helpers.check_records(
ledger_db,
[
{
"table": "sends",
"values": {
"asset": "XCP",
"block_index": tx["block_index"],
"destination": defaults["p2ms_addresses"][0],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==blocks_test:[12:21] ==send_test:[320:329] test_helpers.check_records( ledger_db, [ { "table": "sends", "values": { "asset": "XCP", "block_index": tx["block_index"], "destination": defaults["addresses"][1],. Warning

Similar lines in 2 files
==blocks_test:[12:21]
==send_test:[320:329]
test_helpers.check_records(
ledger_db,
[
{
"table": "sends",
"values": {
"asset": "XCP",
"block_index": tx["block_index"],
"destination": defaults["addresses"][1],.

Check warning

Code scanning / pylint

Similar lines in 2 files ==move_test:[107:116] ==utxo_test:[104:113] }, }, { "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
==move_test:[107:116]
==utxo_test:[104:113]
},
},
{
"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 ==dividend_test:[133:142] ==send_test:[427:436] "source": defaults["addresses"][0], "status": "valid", "tx_hash": tx["tx_hash"], "tx_index": tx["tx_index"], }, }, { "table": "credits", "values": {. Warning

Similar lines in 2 files
==dividend_test:[133:142]
==send_test:[427:436]
"source": defaults["addresses"][0],
"status": "valid",
"tx_hash": tx["tx_hash"],
"tx_index": tx["tx_index"],
},
},
{
"table": "credits",
"values": {.
from counterpartycore.lib import exceptions
from counterpartycore.lib.utils import multisig


def test_multisig_functions(defaults):
assert not multisig.is_multisig(defaults["addresses"][0])

assert multisig.is_multisig(f"1_{defaults['addresses'][0]}_{defaults['addresses'][1]}_2")

assert (
multisig.test_array(
"1",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
2,
)
is None
)

with pytest.raises(exceptions.MultiSigAddressError, match="Signature values not integers."):
multisig.test_array(
"Q",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
2,
)

with pytest.raises(exceptions.MultiSigAddressError, match="Signature values not integers."):
multisig.test_array(
"1",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
None,
)

with pytest.raises(exceptions.MultiSigAddressError, match="Invalid signatures_required."):
multisig.test_array(
"0",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
2,
)

with pytest.raises(exceptions.MultiSigAddressError, match="Invalid signatures_required."):
multisig.test_array(
"4",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
2,
)

with pytest.raises(exceptions.MultiSigAddressError, match="Invalid signatures_possible."):
multisig.test_array(
"1",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
1,
)

with pytest.raises(exceptions.MultiSigAddressError, match="Invalid signatures_possible."):
multisig.test_array(
"2",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
4,
)

with pytest.raises(
exceptions.MultiSigAddressError, match="Invalid characters in pubkeys/pubkeyhashes."
):
multisig.test_array(
"1",
[
defaults["addresses"][1],
f"{defaults['addresses'][0]}_2",
],
2,
)

with pytest.raises(
exceptions.InputError,
match="Incorrect number of pubkeys/pubkeyhashes in multi-signature address.",
):
multisig.test_array(
"3",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
3,
)

assert (
multisig.construct_array(
"1",
[
defaults["addresses"][1],
defaults["addresses"][0],
],
2,
)
== f"1_{defaults['addresses'][0]}_{defaults['addresses'][1]}_2"
)

assert multisig.extract_array(f"1_{defaults['addresses'][0]}_{defaults['addresses'][1]}_2") == (
1,
[
defaults["addresses"][0],
defaults["addresses"][1],
],
2,
)

0 comments on commit c740e0a

Please sign in to comment.