Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found
Select Git revision
  • 175_gva_migration
  • 429_rm_features
  • i18n
  • main
  • pages
  • release/0.11
  • release/0.12
  • v0.1.0
  • v0.10.0
  • v0.10.0rc0
  • v0.10.0rc1
  • v0.11.0
  • v0.11.0rc0
  • v0.11.1
  • v0.11.2
  • v0.12.0
  • v0.12.1
  • v0.2.0
  • v0.3.0
  • v0.4.0
  • v0.5.0
  • v0.6.0
  • v0.6.1
  • v0.6.2
  • v0.6.3
  • v0.6.4
  • v0.6.5
  • v0.7.0
  • v0.7.1
  • v0.7.2
  • v0.7.3
  • v0.7.4
  • v0.7.5
  • v0.7.6
  • v0.8.0
  • v0.8.1
  • v0.9.0
  • v0.9.0rc
38 results

Target

Select target project
  • elmau/silkaj
  • Mr-Djez/silkaj
  • jbar/silkaj
  • clients/python/silkaj
  • Bernard/silkaj
  • cebash/silkaj
  • jytou/silkaj
  • c-geek/silkaj
  • vincentux/silkaj
  • jeanlucdonnadieu/silkaj
  • matograine/silkaj
  • zicmama/silkaj
  • manutopik/silkaj
  • atrax/silkaj
14 results
Select Git revision
  • 72_rework_tx
  • dev
  • master
  • patch-1
  • 0.1.0
  • 0.2.0
  • v0.3.0
  • v0.4.0
  • v0.5.0
9 results
Show changes
Showing
with 3499 additions and 0 deletions
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from duniterpy.documents.block_id import BlockID
from duniterpy.documents.identity import Identity
idty1 = Identity(
currency="g1-test",
pubkey="6upqFiJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o",
uid="Claude",
block_id=BlockID.from_str(
"597334-002A45E751DCA7535D4F0A082F493E2C8EFF07612683525EB5DA92B6D17C30BD",
),
)
idty1.signature = "kFW2we2K3zx4PZODx0Wf+xdXAJTmYD+yqdyZBsPF7UwqdaCA4N\
+yHj7+09Gjsttl0i9GtWzodyJ6mBE1q7jcAw=="
lookup_one = {
"partial": False,
"results": [
{
"pubkey": "6upqFiJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o",
"uids": [
{
"uid": "Claude",
"meta": {
"timestamp": "597334-002A45E751DCA7535D4F0A08\
2F493E2C8EFF07612683525EB5DA92B6D17C30BD",
},
"revoked": False,
"revoked_on": None,
"revocation_sig": None,
"self": "kFW2we2K3zx4PZODx0Wf+xdXAJTmYD+yqdyZBsPF\
7UwqdaCA4N+yHj7+09Gjsttl0i9GtWzodyJ6mBE1q7jcAw==",
"others": [],
},
],
},
],
}
lookup_two = {
"partial": False,
"results": [
{
"pubkey": "6upqFiJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o",
"uids": [
{
"uid": "Claude",
"meta": {
"timestamp": "597334-002A45E751DCA7535D4F0A08\
2F493E2C8EFF07612683525EB5DA92B6D17C30BD",
},
"revoked": False,
"revoked_on": None,
"revocation_sig": None,
"self": "kFW2we2K3zx4PZODx0Wf+xdXAJTmYD+yqdyZBsPF\
7UwqdaCA4N+yHj7+09Gjsttl0i9GtWzodyJ6mBE1q7jcAw==",
"others": [],
},
{
"uid": "Claudia",
"meta": {
"timestamp": "597334-002A45E751DCA7535D4F0A08\
2F493E2C8EFF07612683525EB5DA92B6D17C30BD",
},
"revoked": False,
"revoked_on": None,
"revocation_sig": None,
"self": "kFW2we2K3zx4PZODx0Wf+xdXAJTmYD+yqdyZBsPF\
7UwqdaCA4N+yHj7+09Gjsttl0i9GtWzodyJ6mBE1q7jcAw==",
"others": [],
},
],
},
],
}
lookup_three = {
"partial": False,
"results": [
{
"pubkey": "6upqFiJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o",
"uids": [
{
"uid": "Claude",
"meta": {
"timestamp": "597334-002A45E751DCA7535D4F0A08\
2F493E2C8EFF07612683525EB5DA92B6D17C30BD",
},
"revoked": False,
"revoked_on": None,
"revocation_sig": None,
"self": "kFW2we2K3zx4PZODx0Wf+xdXAJTmYD+yqdyZBsPF\
7UwqdaCA4N+yHj7+09Gjsttl0i9GtWzodyJ6mBE1q7jcAw==",
"others": [],
},
],
},
{
"pubkey": "XXXXXXJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o",
"uids": [
{
"uid": "Claude",
"meta": {
"timestamp": "597334-002A45E751DCA7535D4F0A08\
2F493E2C8EFF07612683525EB5DA92B6D17C30BD",
},
"revoked": False,
"revoked_on": None,
"revocation_sig": None,
"self": "kFW2we2K3zx4PZODx0Wf+xdXAJTmYD+yqdyZBsPF\
7UwqdaCA4N+yHj7+09Gjsttl0i9GtWzodyJ6mBE1q7jcAw==",
"others": [],
},
],
},
],
}
idty2 = Identity(
currency="g1-test",
pubkey="969qRJs8KhsnkyzqarpL4RKZGMdVKNbZgu8fhsigM7Lj",
uid="aa_aa",
block_id=BlockID.from_str(
"703902-00002D6BC5E4FC540A4E188C3880A0ACCA06CD77017D26231A515312162B4070",
),
)
idty2.signature = "3RNQcKNI1VMmuCpK7wer8haOA959EQSDIR1v0Ue/7TpTCOmsU2\
zYCpC+tqgLQFxDX4A79sB61c11J5C/3Z/TCw=="
idty_block = {
"version": 12,
"nonce": 10100000000525,
"number": 597334,
"powMin": 45,
"time": 1594980185,
"medianTime": 1594978717,
"membersCount": 7,
"monetaryMass": 2157680864,
"unitbase": 3,
"issuersCount": 3,
"issuersFrame": 16,
"issuersFrameVar": 0,
"currency": "g1-test",
"issuer": "3dnbnYY9i2bHMQUGyFp5GVvJ2wBkVpus31cDJA5cfRpj",
"signature": "Eme9mi25DtUQrP3Hk6evJBQP6GRU0asJrl9G2RWUgtB71AMOWqs\
/NeraG8YBwQEGokQg1mHMUv7fEoUiEetwDw==",
"hash": "002A45E751DCA7535D4F0A082F493E2C8EFF07612683525EB5DA92B6D17C30BD",
"parameters": "",
"previousHash": "0023B87885C52CDE75694C71BED1237B5C7B686C00AB68C8D75693513E1F8765",
"previousIssuer": "39YyHCMQNmXY7NkPCXXfzpV1vYct4GBxwgfyd4d72HmB",
"inner_hash": "46D99F8431053892F230E4E07EC16A2A68B09D68EBC3F9FD796289493AFAFFB5",
"dividend": None,
"identities": [],
"joiners": [],
"actives": [],
"leavers": [],
"revoked": [],
"excluded": [],
"certifications": [],
"transactions": [],
"raw": "Version: 12\nType: Block\nCurrency: g1-test\nNumber: 597334\n\
PoWMin: 45\nTime: 1594980185\nMedianTime: 1594978717\nUnitBase: 3\n\
Issuer: 3dnbnYY9i2bHMQUGyFp5GVvJ2wBkVpus31cDJA5cfRpj\nIssuersFrame: 16\n\
IssuersFrameVar: 0\nDifferentIssuersCount: 3\n\
PreviousHash: 0023B87885C52CDE75694C71BED1237B5C7B686C00AB68C8D75693513E1F8765\n\
PreviousIssuer: 39YyHCMQNmXY7NkPCXXfzpV1vYct4GBxwgfyd4d72HmB\nMembersCount: 7\n\
Identities:\nJoiners:\nActives:\nLeavers:\nRevoked:\nExcluded:\nCertifications:\n\
Transactions:\nInnerHash: 46D99F8431053892F230E4E07EC16A2A68B09D68EBC3F9FD796289493AFAFFB5\n\
Nonce: 10100000000525\n",
}
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
# This file contains patched functions for testing purposes.
from duniterpy.documents.transaction import InputSource
from silkaj.money.tools import amount_in_current_base
from silkaj.money.transfer import MAX_INPUTS_PER_TX
from tests.patched.test_constants import mock_ud_value
def patched_get_ud_value():
return mock_ud_value
# mock get_sources()
def patched_get_sources(pubkey):
"""
Returns transaction sources.
This function doesn't cover all possibilities : only SIG() unlock condition.
Can be called many times (depending on pubkey).
If so, it will mock intermediary tx for the first 40 inputs.
Tests using this function should reset the counter at the begining of each test case.
See source_dict.py for inputs lists.
all UTXO have the same amount : 100
all UD have the same amount : 314
for pubkey CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp : 3 TX, balance = 300
for pubkey HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat : 53 TX, balance = 5300
for pubkey 2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY : 10 UD, balance = 3140
for pubkey 9cwBBgXcSVMT74xiKYygX6FM5yTdwd3NABj1CfHbbAmp : 50 UD and 20 TX, balance = 17700
else : 0 sources, balance = 0
Same hash for each TX for convenience. This may change for other testing purposes.
"""
def listinput_UD(listinput, balance, pubkey, max_ud):
a = 0
while a < max_ud:
listinput.append(
InputSource(
amount=mock_ud_value,
base=0,
source="D",
origin_id=pubkey,
index=a,
),
)
balance += amount_in_current_base(listinput[-1])
a += 1
return balance
def listinput_TX(listinput, balance, max_tx):
a = 0
while a < max_tx:
listinput.append(
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=a,
),
)
balance += amount_in_current_base(listinput[-1])
a += 1
return balance
listinput = []
balance = 0
if pubkey == "CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp":
max_ud = 0
max_tx = 3
elif pubkey == "HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat":
if Counter.counter == 0:
max_ud = 0
max_tx = 53
elif Counter.counter == 1:
listinput.append(
InputSource(
amount=100 * MAX_INPUTS_PER_TX, # 100 * 46 = 4600
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=93,
),
)
max_ud = 0
max_tx = 6
elif pubkey == "2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY":
max_ud = 10
max_tx = 0
elif pubkey == "9cwBBgXcSVMT74xiKYygX6FM5yTdwd3NABj1CfHbbAmp":
if Counter.counter == 0:
max_ud = 50
max_tx = 20
elif Counter.counter == 1:
listinput.append(
InputSource(
amount=mock_ud_value * MAX_INPUTS_PER_TX, # 46 UD = 46*314 = 1444
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=93,
),
)
max_ud = 4
max_tx = 20
elif pubkey == "BdanxHdwRRzCXZpiqvTVTX4gyyh6qFTYjeCWCkLwDifx":
listinput.append(
InputSource(
amount=9600,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=0,
),
)
max_ud = 0
max_tx = 0
else:
max_ud = 0
max_tx = 0
balance = listinput_UD(listinput, balance, pubkey, max_ud)
balance = listinput_TX(listinput, balance, max_tx)
Counter.counter += 1
return listinput, balance
class Counter:
counter = 0
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
# this file contains only constant values for testing (no function) to prevent circular dependencies
mock_ud_value = 314
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from silkaj.constants import G1_SYMBOL
# mock get_currency_symbol().symbol
def patched_get_currency_symbol():
return G1_SYMBOL
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
import sys
from duniterpy.key import SigningKey
def patched_gen_confirmation_table(
issuer_pubkey,
pubkey_amount,
tx_amounts,
outputAddresses,
OutputBackChange,
comment,
):
if not (
(
isinstance(issuer_pubkey, str)
and isinstance(pubkey_amount, int)
and isinstance(tx_amounts, list)
and isinstance(outputAddresses, list)
and isinstance(comment, str)
and isinstance(OutputBackChange, str)
)
and len(tx_amounts) == len(outputAddresses)
and sum(tx_amounts) <= pubkey_amount
):
sys.exit(
"Test error : patched_transaction_confirmation() : Parameters are not coherent",
)
def patched_handle_intermediaries_transactions(
key,
issuers,
tx_amounts,
outputAddresses,
Comment="",
OutputBackChange=None,
):
if not (
(
isinstance(key, SigningKey)
and isinstance(issuers, str)
and isinstance(tx_amounts, list)
and isinstance(outputAddresses, list)
and isinstance(Comment, str)
and (isinstance(OutputBackChange, str) or not OutputBackChange)
)
and len(tx_amounts) == len(outputAddresses)
and key.pubkey() == issuers
):
sys.exit(
"Test error : patched_handle_intermediaries_transactions() : Parameters are not coherent",
)
def patched_generate_and_send_transaction(
key,
issuers,
tx_amounts,
listinput_and_amount,
outputAddresses,
Comment,
OutputBackChange,
):
if not (
(
isinstance(key, SigningKey)
and isinstance(issuers, str)
and isinstance(tx_amounts, list)
and isinstance(listinput_and_amount, tuple)
and isinstance(outputAddresses, list)
and isinstance(Comment, str)
and isinstance(OutputBackChange, str)
)
and len(tx_amounts) == len(outputAddresses)
and sum(tx_amounts) <= listinput_and_amount[2]
and key.pubkey() == issuers
):
sys.exit(
"Test error : patched_generate_and_send_transaction() : Parameters are not coherent",
)
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from duniterpy.documents.transaction import Transaction
from tests.patched.blockchain_tools import currency
fake_received_tx_hist = [
{
"version": 10,
"locktime": 0,
"blockstamp": "150574-000003D64BB3E107EEEF90922A60DB56A6DF14FB4415A3F4A852A87F889E1C31",
"blockstampTime": 1535913486,
"issuers": ["CvrMiUhAJpNyX5sdAyZqPE6yEFfSsf6j9EpMmeKvMCWW"],
"inputs": [
"7014:0:T:00DFBACFA3434057F6B2DAA8249324C64A658E40BC85CFD40E15FADDD88BACE3:1",
],
"outputs": [
"100:0:SIG(d88fPFbDdJXJANHH7hedFMaRyGcnVZj9c5cDaE76LRN)",
"6914:0:SIG(CvrMiUhAJpNyX5sdAyZqPE6yEFfSsf6j9EpMmeKvMCWW)",
],
"unlocks": ["0:SIG(0)"],
"signatures": [
"xz/l3o9GbUclrYDNKiRaVTrBP7cppDmrjDgE2rFNLJsnpu1e/AE2bHyl\
ftU09NYEDqzCUbehv19oF6zIRVwTDw==",
],
"comment": "initialisation",
"hash": "D2271075F2308C4092B1F57B3F1BE12AB684FAFCA62BA8EFE9F7F4D7A4D8D69F",
"time": 111111114,
"block_number": 150576,
"received": None,
},
{
"version": 10,
"locktime": 0,
"blockstamp": "400498-0000000EE3E7C41160E5638B7DB3F76A82068D6D3D1CC2332EE7A39AF43A9EA6",
"blockstampTime": 1613798963,
"issuers": ["CmFKubyqbmJWbhyH2eEPVSSs4H4NeXGDfrETzEnRFtPd"],
"inputs": ["1023:0:D:CmFKubyqbmJWbhyH2eEPVSSs4H4NeXGDfrETzEnRFtPd:396949"],
"outputs": [
"100:0:SIG(d88fPFbDdJXJANHH7hedFMaRyGcnVZj9c5cDaE76LRN)",
"923:0:SIG(CmFKubyqbmJWbhyH2eEPVSSs4H4NeXGDfrETzEnRFtPd)",
],
"unlocks": ["0:SIG(0)"],
"signatures": [
"pYSOTCrl1QbsKrgjgNWnUfD3wJnpbalv9EwjAbZozTbTOSzYoj+UInzK\
S8/OiSdyVqFVDLdpewTD+FOHRENDAA==",
],
"comment": "",
"hash": "F1F2E6D6CF123AB78B98B662FE3AFDD2577B8F6CEBC245660B2E67BC9C2026F6",
"time": 111111113,
"block_number": 400500,
"received": None,
},
]
fake_sent_tx_hist = [
{
"version": 10,
"locktime": 0,
"blockstamp": "400503-0000000A7F3B6F4C5654D9CCFEA41E4726E02B08BB94EE30BD9A50908D28636D",
"blockstampTime": 1613801234,
"issuers": ["d88fPFbDdJXJANHH7hedFMaRyGcnVZj9c5cDaE76LRN"],
"inputs": [
"100:0:T:F1F2E6D6CF123AB78B98B662FE3AFDD2577B8F6CEBC245660B2E67BC9C2026F6:0",
],
"outputs": ["100:0:SIG(CvrMiUhAJpNyX5sdAyZqPE6yEFfSsf6j9EpMmeKvMCWW)"],
"unlocks": ["0:SIG(0)"],
"signatures": [
"cMNp7FF5yT/6LJT9CnNzkE08h+APEAYYwdFIROGxUZ9JGqbfPR1NRbcr\
uq5Fl9BnBcJkuMNJbOwuYV8bPCmICw==",
],
"comment": "",
"hash": "580715ECD6743590F7A99A6C97E63511BC94B0293CB0037C6A3C96482F8DC7D2",
"time": 111111112,
"block_number": 400505,
"received": None,
},
{
"version": 10,
"locktime": 0,
"blockstamp": "400503-0000000A7F3B6F4C5654D9CCFEA41E4726E02B08BB94EE30BD9A50908D28636D",
"blockstampTime": 1613801235,
"issuers": ["d88fPFbDdJXJANHH7hedFMaRyGcnVZj9c5cDaE76LRN"],
"inputs": [
"100:0:T:D2271075F2308C4092B1F57B3F1BE12AB684FAFCA62BA8EFE9F7F4D7A4D8D69F:0",
],
"outputs": ["100:0:SIG(CmFKubyqbmJWbhyH2eEPVSSs4H4NeXGDfrETzEnRFtPd)"],
"unlocks": ["0:SIG(0)"],
"signatures": [
"WL3dRX4XUenWlDYYhRmEOUgL5+Tc08LlOJWHNjmTlxqtsdHhGn7MuQ3l\
K+3Xv7PV6VFEEdc3vlJ52pWCLKN5BA==",
],
"comment": "",
"hash": "E874CDAC01D9F291DC1E03F8B0ADB6C19259DE5A11FB73A16318BA1AD59B9EDC",
"time": 111111111,
"block_number": 400505,
"received": None,
},
]
def patched_get_transactions_history(client, pubkey, received_txs, sent_txs):
for received in fake_received_tx_hist:
received_txs.append(Transaction.from_bma_history(received, currency))
for sent in fake_sent_tx_hist:
sent_txs.append(Transaction.from_bma_history(sent, currency))
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
pubkey_list = [
{"pubkey": "9cwBBgXcSVMT74xiKYygX6FM5yTdwd3NABj1CfHbbAmp", "uid": ""},
{"pubkey": "BUhLyJT17bzDVXW66xxfk1F7947vytmwJVadTaWb8sJS", "uid": ""},
{"pubkey": "CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", "uid": "riri"},
{"pubkey": "HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", "uid": "fifi"},
{"pubkey": "2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY", "uid": "loulou"},
{
"pubkey": "CvrMiUhAJpNyX5sdAyZqPE6yEFfSsf6j9EpMmeKvMCWW",
"uid": "mato",
},
]
# mock is_member
def patched_is_member(pubkey):
for account in pubkey_list:
if account["pubkey"] == pubkey and account["uid"]:
return account
return False
# patch wot requirements
def patched_wot_requirements_one_pending(requirements, search, pubkey):
return {
"identities": [
{
"uid": "toto",
"pendingMemberships": [
{
"membership": "IN",
"issuer": "5B8iMAzq1dNmFe3ZxFTBQkqhq4fsztg1gZvxHXCk1XYH",
"number": 613206,
"blockNumber": 613206,
"userid": "moul-test",
"expires_on": 1598624404,
"type": "IN",
},
],
"membershipPendingExpiresIn": 6311520,
"membershipExpiresIn": 2603791,
},
],
}
def patched_wot_requirements_no_pending(requirements, search, pubkey):
return {
"identities": [
{
"uid": "toto",
"pendingMemberships": [],
"membershipPendingExpiresIn": 0,
"membershipExpiresIn": 3724115,
},
],
}
# for history
def patched_identities_from_pubkeys(pubkeys, uids):
if not uids:
return []
uniq_pubkeys = list(filter(None, set(pubkeys)))
identities = []
for pubkey in uniq_pubkeys:
for idty in pubkey_list:
if idty.get("pubkey", False) == pubkey:
identities.append(idty)
return identities
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
import csv
from pathlib import Path
import pendulum
import pytest
from click.testing import CliRunner
from silkaj import tools
from silkaj.cli import cli
from silkaj.constants import (
ALL_DIGITAL,
CENT_MULT_TO_UNIT,
PUBKEY_MAX_LENGTH,
PUBKEY_MIN_LENGTH,
SHORT_PUBKEY_SIZE,
)
from silkaj.money import history
from silkaj.money import tools as m_tools
from silkaj.public_key import CHECKSUM_SIZE
from silkaj.wot import tools as wt
from tests.patched.blockchain_tools import currency
from tests.patched.money import patched_get_ud_value
from tests.patched.test_constants import mock_ud_value
from tests.patched.tools import patched_get_currency_symbol
from tests.patched.tx_history import patched_get_transactions_history
from tests.patched.wot import patched_identities_from_pubkeys
SHORT_PUBKEY_LENGTH_WITH_CHECKSUM = (
SHORT_PUBKEY_SIZE + CHECKSUM_SIZE + 2
) # 2 chars "…:" ==> 8 + 3 + 2 = 13
MIN_FULL_PUBKEY_LENGTH_WITH_CHECKSUM = (
PUBKEY_MIN_LENGTH + CHECKSUM_SIZE + 1
) # char `:` ==> 43 + 3 + 1 = 47
MAX_FULL_PUBKEY_LENGTH_WITH_CHECKSUM = (
PUBKEY_MAX_LENGTH + CHECKSUM_SIZE + 1
) # char `:` ==> 44 + 3 + 1 = 48
def test_history_generate_txs_list_and_pubkey_uid_display(monkeypatch):
def min_pubkey_length_with_uid(pubkey):
# uid is at least one char : "<uid> - <pubkey>" adds min 4 chars to <pubkey>
return pubkey + 4
monkeypatch.setattr(wt, "identities_from_pubkeys", patched_identities_from_pubkeys)
client = "whatever"
ud_value = 10.07
table_columns = 5
pubkey = "d88fPFbDdJXJANHH7hedFMaRyGcnVZj9c5cDaE76LRN"
received_txs, sent_txs = [], []
patched_get_transactions_history(client, pubkey, received_txs, sent_txs)
# simple table
txs_list = history.generate_txs_list(
received_txs,
sent_txs,
pubkey,
ud_value,
currency,
uids=False,
full_pubkey=False,
)
for tx_list in txs_list:
assert len(tx_list) == table_columns
assert "…:" in tx_list[1]
assert len(tx_list[1]) == SHORT_PUBKEY_LENGTH_WITH_CHECKSUM
# with uids
txs_list_uids = history.generate_txs_list(
received_txs,
sent_txs,
pubkey,
ud_value,
currency,
uids=True,
full_pubkey=False,
)
for tx_list in txs_list_uids:
assert len(tx_list) == table_columns
assert "…:" in tx_list[1]
# check all lines
assert len(txs_list_uids[0][1]) >= min_pubkey_length_with_uid(
SHORT_PUBKEY_LENGTH_WITH_CHECKSUM,
)
assert len(txs_list_uids[1][1]) == SHORT_PUBKEY_LENGTH_WITH_CHECKSUM
assert len(txs_list_uids[2][1]) >= min_pubkey_length_with_uid(
SHORT_PUBKEY_LENGTH_WITH_CHECKSUM,
)
assert len(txs_list_uids[3][1]) == SHORT_PUBKEY_LENGTH_WITH_CHECKSUM
# with full pubkeys
txs_list_full = history.generate_txs_list(
received_txs,
sent_txs,
pubkey,
ud_value,
currency,
uids=False,
full_pubkey=True,
)
for tx_list in txs_list_full:
assert len(tx_list) == table_columns
assert "…:" not in tx_list[1]
assert ":" in tx_list[1]
# this length is not true for multisig txs, which are very unlikely for now.
assert (
len(tx_list[1]) == MIN_FULL_PUBKEY_LENGTH_WITH_CHECKSUM
or len(tx_list[1]) == MAX_FULL_PUBKEY_LENGTH_WITH_CHECKSUM
)
# with full pubkeys and uids
txs_list_uids_full = history.generate_txs_list(
received_txs,
sent_txs,
pubkey,
ud_value,
currency,
uids=True,
full_pubkey=True,
)
for tx_list in txs_list_uids_full:
assert len(tx_list) == table_columns
assert "…:" not in tx_list[1]
assert ":" in tx_list[1]
# check all lines
assert len(txs_list_uids_full[0][1]) >= min_pubkey_length_with_uid(
MIN_FULL_PUBKEY_LENGTH_WITH_CHECKSUM,
)
assert (
len(txs_list_uids_full[1][1]) == MIN_FULL_PUBKEY_LENGTH_WITH_CHECKSUM
or len(txs_list_uids_full[1][1]) == MAX_FULL_PUBKEY_LENGTH_WITH_CHECKSUM
)
assert len(txs_list_uids_full[2][1]) >= min_pubkey_length_with_uid(
MIN_FULL_PUBKEY_LENGTH_WITH_CHECKSUM,
)
assert (
len(txs_list_uids_full[3][1]) == MIN_FULL_PUBKEY_LENGTH_WITH_CHECKSUM
or len(txs_list_uids_full[3][1]) == MAX_FULL_PUBKEY_LENGTH_WITH_CHECKSUM
)
@pytest.mark.parametrize(
("tx_addresses", "outputs", "occurence", "return_value"),
[
(None, None, 0, ""),
(None, None, 1, "\n"),
(None, ["output1"], 0, ""),
(None, ["output1"], 1, ""),
(None, ["output1", "output2"], 0, "\n"),
(None, ["output1", "output2"], 1, "\n"),
("pubkey", None, 0, ""),
("pubkey", None, 1, "\n"),
("pubkey", ["output1"], 0, ""),
("pubkey", ["output1"], 1, ""),
("Total", ["output1", "output2"], 0, "\n"),
("pubkey", ["output1", "output2"], 0, "\n"),
("pubkey", ["output1", "output2"], 1, "\n"),
],
)
def test_prefix(tx_addresses, outputs, occurence, return_value):
assert history.prefix(tx_addresses, outputs, occurence) == return_value
relative_amount = str(round(CENT_MULT_TO_UNIT / mock_ud_value, 2))
csv_reference = (
["Date", "Issuers/Recipients", "Amounts Ğ1", "Amounts UDĞ1", "Reference"],
[
pendulum.from_timestamp(111111114, tz="local").format(ALL_DIGITAL),
"CvrMiUhAJpNyX5sdAyZqPE6yEFfSsf6j9EpMmeKvMCWW:DNB",
"1.0",
relative_amount,
"initialisation",
],
[
pendulum.from_timestamp(111111113, tz="local").format(ALL_DIGITAL),
"CmFKubyqbmJWbhyH2eEPVSSs4H4NeXGDfrETzEnRFtPd:CQ5",
"1.0",
relative_amount,
"",
],
[
pendulum.from_timestamp(111111112, tz="local").format(ALL_DIGITAL),
"CvrMiUhAJpNyX5sdAyZqPE6yEFfSsf6j9EpMmeKvMCWW:DNB",
"-1.0",
f"-{relative_amount}",
"",
],
[
pendulum.from_timestamp(111111111, tz="local").format(ALL_DIGITAL),
"CmFKubyqbmJWbhyH2eEPVSSs4H4NeXGDfrETzEnRFtPd:CQ5",
"-1.0",
f"-{relative_amount}",
"",
],
)
def test_csv_output(monkeypatch):
monkeypatch.setattr(m_tools, "get_ud_value", patched_get_ud_value)
monkeypatch.setattr(tools, "get_currency_symbol", patched_get_currency_symbol)
monkeypatch.setattr(
history,
"get_transactions_history",
patched_get_transactions_history,
)
file = "history.csv"
pubkey = "d88fPFbDdJXJANHH7hedFMaRyGcnVZj9c5cDaE76LRN"
runner = CliRunner()
with runner.isolated_filesystem():
result = runner.invoke(
cli, ["money", "history", pubkey, "--csv-file", file], input="yes"
)
assert f"{file} file successfully saved!" in result.output
assert result.exit_code == 0
with Path(file).open() as f:
reader = csv.reader(f)
for row_file, row_ref in zip(reader, csv_reference):
assert row_file == row_ref
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
import duniterpy.api.bma.tx as bma_tx
import pytest
from click.testing import CliRunner
from silkaj.cli import cli
from silkaj.constants import FAILURE_EXIT_STATUS, G1_SYMBOL
from silkaj.money import tools as m_tools
from silkaj.public_key import gen_pubkey_checksum
from silkaj.wot import tools
from tests.patched.test_constants import mock_ud_value
from tests.patched.wot import patched_is_member
# display_amount()
@pytest.mark.parametrize(
("message", "amount", "currency_symbol"),
[("Total", 1000, G1_SYMBOL)],
)
def test_display_amount(message, amount, currency_symbol):
amount_UD = round(amount / mock_ud_value, 2)
expected = [
[
f"{message} (unit|relative)",
f"{amount / 100!s} {currency_symbol} | {amount_UD!s} UD {currency_symbol}",
],
]
tx = []
m_tools.display_amount(tx, message, amount, mock_ud_value, currency_symbol)
assert tx == expected
# display_pubkey()
@pytest.mark.parametrize(
("message", "pubkey", "uid"),
[
("From", "CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", "riri"),
("To", "DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw", ""),
],
)
def test_display_pubkey(message, pubkey, uid, monkeypatch):
monkeypatch.setattr(tools, "is_member", patched_is_member)
expected = [[f"{message} (pubkey:checksum)", gen_pubkey_checksum(pubkey)]]
if uid:
expected.append([f"{message} (id)", uid])
tx = []
m_tools.display_pubkey(tx, message, pubkey)
assert tx == expected
def test_get_sources(monkeypatch):
"""
test that get_source() will :
* only use simple SIG txs
* only use blockchain sources to compute balance
* return pending txs in first positions of the sources list
"""
def patched_tx_sources(self, pubkey):
return {
"currency": "g1-test",
"pubkey": "AhRMHUxMPXSeG7qXZrE6qCdjwK9p2bu5Eqei7xAWVEDK",
"sources": [
# this source will be returned in inputlist, and its amount used.
{
"type": "T",
"noffset": 2,
"identifier": "0310F56D22F4CEF5E41B9D5CACB6E21F224B79D9398D53A4E754866435710242",
"amount": 10,
"base": 3,
"conditions": "SIG(AhRMHUxMPXSeG7qXZrE6qCdjwK9p2bu5Eqei7xAWVEDK)",
},
# this source will not be returned (complex unlock condition)
{
"type": "T",
"noffset": 3,
"identifier": "0D6A29451E64F468C0DB19F70D0D17F65BDCC98F3A16DD55B3755BE124B3DD6C",
"amount": 30,
"base": 3,
"conditions": "(SIG(2VgEZnrGQ5hEgwoNrcXZnD9c8o5jL63LPBmJdvMyFhGe)\
|| (SIG(AhRMHUxMPXSeG7qXZrE6qCdjwK9p2bu5Eqei7xAWVEDK) && CSV(864)))",
},
],
}
def patched_tx_pending(self, pubkey):
return {
"currency": "g1-test",
"pubkey": "AhRMHUxMPXSeG7qXZrE6qCdjwK9p2bu5Eqei7xAWVEDK",
"history": {
"sending": [],
"received": [],
"receiving": [],
"sent": [],
"pending": [
{
"version": 10,
"locktime": 0,
"blockstamp": "671977-000008B6DE75715D3D83450\
A957CD75F781DA8E3E8E966D42A02F59049209533",
"blockstampTime": 1607363853,
"issuers": ["6upqFiJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o"],
"inputs": [
"2739:3:D:6upqFiJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o:664106",
],
"outputs": [
"60:3:SIG(AhRMHUxMPXSeG7qXZrE6qCdjwK9p2bu5Eqei7xAWVEDK)",
"2679:3:SIG(6upqFiJ66WV6N3bPc8x8y7rXT3syqKRmwnVyunCtEj7o)",
],
"unlocks": ["0:SIG(0)"],
"signatures": [
"lrmzr/RkecJBOczlmkp3BNCiXejBzTnHdqmNzxQJ\
yJDIx0UHON4jYkqVKeD77+nrOl8jVtonLt3ZYqd1fhi1Cw==",
],
"comment": "DEMAIN DES L_AUBE",
"hash": "D5A1A1AAA43FAA242CC2B19763619DA625092BB7FD23397AD362215375A920C8",
"time": None,
"block_number": None,
"received": None,
},
],
},
}
monkeypatch.setattr(bma_tx, "sources", patched_tx_sources)
monkeypatch.setattr(bma_tx, "pending", patched_tx_pending)
listinput, balance = m_tools.get_sources(
"AhRMHUxMPXSeG7qXZrE6qCdjwK9p2bu5Eqei7xAWVEDK",
)
assert len(listinput) == 2
# test SIG() only source is used
assert balance == 10000 # 10 in unitbase 3
assert (
listinput[0].origin_id
== "D5A1A1AAA43FAA242CC2B19763619DA625092BB7FD23397AD362215375A920C8"
)
def test_balance_errors():
"""
test balance command errors
"""
# twice the same pubkey
result = CliRunner().invoke(
cli,
[
"money",
"balance",
"BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh",
"BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh",
],
)
pubkeyCk = gen_pubkey_checksum("BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh")
assert f"Pubkey {pubkeyCk} was specified many" in result.output
assert result.exit_code >= FAILURE_EXIT_STATUS
# wrong pubkey
result = CliRunner().invoke(
cli,
[
"money",
"balance",
"B",
],
)
assert "pubkey B has a wrong format" in result.output
assert result.exit_code >= FAILURE_EXIT_STATUS
# no pubkey
result = CliRunner().invoke(cli, ["money", "balance"])
assert "You should specify one or many pubkeys" in result.output
assert result.exit_code >= FAILURE_EXIT_STATUS
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from unittest.mock import Mock
import pytest
from click import pass_context
from click.testing import CliRunner
from duniterpy.documents.transaction import (
InputSource,
OutputSource,
SIGParameter,
Transaction,
Unlock,
)
from silkaj import auth, network, tools
from silkaj.blockchain import tools as bc_tools
from silkaj.cli import cli
from silkaj.constants import CENT_MULT_TO_UNIT, G1_SYMBOL
from silkaj.money import tools as m_tools
from silkaj.money import transfer
from silkaj.wot import tools as wt
from tests.patched.auth import patched_auth_method
from tests.patched.blockchain_tools import fake_block_id, patched_get_head_block
from tests.patched.money import Counter, patched_get_sources, patched_get_ud_value
from tests.patched.test_constants import mock_ud_value
from tests.patched.tools import patched_get_currency_symbol
from tests.patched.wot import patched_is_member
# Values
# fifi: HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat : 53 TX, amount = 5300
key_fifi = patched_auth_method("fifi")
# truncBase()
@pytest.mark.parametrize(
("amount", "base", "expected"),
[(0, 0, 0), (10, 2, 0), (100, 2, 100), (306, 2, 300), (3060, 3, 3000)],
)
def test_truncBase(amount, base, expected):
assert transfer.truncBase(amount, base) == expected
# transaction_confirmation()
@pytest.mark.parametrize(
(
"issuer_pubkey",
"pubkey_balance",
"tx_amounts",
"outputAddresses",
"outputBackChange",
"reference",
),
[
# only one receiver
(
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
3000,
[1000],
["4szFkvQ5tzzhwcfUtZD32hdoG2ZzhvG3ZtfR61yjnxdw"],
"",
"",
),
# one member receiver
(
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
3000,
[1000],
["BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh"],
"",
"This is a reference",
),
# many receivers and backchange
(
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
3000,
[1000, 1000],
[
"BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh",
"4szFkvQ5tzzhwcfUtZD32hdoG2ZzhvG3ZtfR61yjnxdw",
],
"C1oAV9FX2y9iz2sdp7kZBFu3EBNAa6UkrrRG3EwouPeH",
"This is a reference",
),
# many receivers and outputs
(
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
3000,
[1000, 250],
[
"BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh",
"4szFkvQ5tzzhwcfUtZD32hdoG2ZzhvG3ZtfR61yjnxdw",
],
"",
"This is a reference",
),
],
)
def test_gen_confirmation_table(
issuer_pubkey,
pubkey_balance,
tx_amounts,
outputAddresses,
outputBackChange,
reference,
monkeypatch,
):
# patched functions
monkeypatch.setattr(wt, "is_member", patched_is_member)
monkeypatch.setattr(m_tools, "get_ud_value", patched_get_ud_value)
monkeypatch.setattr(tools, "get_currency_symbol", patched_get_currency_symbol)
# creating expected list
expected = []
total_tx_amount = sum(tx_amounts)
# display account situation
m_tools.display_amount(
expected,
"Initial balance",
pubkey_balance,
mock_ud_value,
G1_SYMBOL,
)
m_tools.display_amount(
expected,
"Total transaction amount",
total_tx_amount,
mock_ud_value,
G1_SYMBOL,
)
m_tools.display_amount(
expected,
"Balance after transaction",
(pubkey_balance - total_tx_amount),
mock_ud_value,
G1_SYMBOL,
)
m_tools.display_pubkey(expected, "From", issuer_pubkey)
# display recipients and amounts
for outputAddress, tx_amount in zip(outputAddresses, tx_amounts):
m_tools.display_pubkey(expected, "To", outputAddress)
m_tools.display_amount(expected, "Amount", tx_amount, mock_ud_value, G1_SYMBOL)
# display backchange and reference
if outputBackChange:
m_tools.display_pubkey(expected, "Backchange", outputBackChange)
expected.append(["Reference", reference])
# asserting
table_list = transfer.gen_confirmation_table(
issuer_pubkey,
pubkey_balance,
tx_amounts,
outputAddresses,
outputBackChange,
reference,
)
assert table_list == expected
# compute_amounts()
def test_compute_amounts_errors(capsys):
trials = (((0.0031, 1), 314),)
for trial in trials:
# check program exit on error
with pytest.raises(SystemExit) as pytest_exit:
# read output to check error.
transfer.compute_amounts(
trial[0],
trial[1],
)
assert pytest_exit.type == SystemExit
expected_error = f"Error: amount {trial[0][0]} is too low.\n"
assert capsys.readouterr().out == expected_error
def test_compute_amounts():
assert transfer.compute_amounts((10.0, 2.0, 0.01, 0.011, 0.019), 100) == [
1000,
200,
1,
1,
2,
]
assert transfer.compute_amounts([0.0032], mock_ud_value) == [1]
assert transfer.compute_amounts([1.00], mock_ud_value) == [314]
assert transfer.compute_amounts([1.01], mock_ud_value) == [317]
assert transfer.compute_amounts([1.99], mock_ud_value) == [625]
assert transfer.compute_amounts([1.001], mock_ud_value) == [314]
assert transfer.compute_amounts([1.009], mock_ud_value) == [317]
# This case will not happen in real use, but this particular function will allow it.
assert transfer.compute_amounts(
[0.0099],
100,
) == [1]
# generate_transaction_document()
# expected results
# result 1 : with two amounts/outputs and an outputbackchange
result1 = Transaction(
block_id=fake_block_id,
locktime=0,
issuers=["BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh"],
inputs=[
InputSource(
10000,
0,
"T",
"B37D161185A760FD81C3242D73FABD3D01F4BD9EAD98C2842061A75BAD4DFA61",
1,
),
InputSource(
257,
0,
"T",
"16F1CF9C9B89BB8C34A945F56073AB3C3ACFD858D5FA420047BA7AED1575D1FE",
1,
),
],
unlocks=[
Unlock(index=0, parameters=[SIGParameter(0)]),
Unlock(index=1, parameters=[SIGParameter(0)]),
],
outputs=[
OutputSource(
amount=1000,
base=0,
condition="SIG(DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw)",
),
OutputSource(
amount=4000,
base=0,
condition="SIG(4szFkvQ5tzzhwcfUtZD32hdoG2ZzhvG3ZtfR61yjnxdw)",
),
OutputSource(
amount=5000,
base=0,
condition="SIG(BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh)",
),
],
comment="Test reference",
)
@pytest.mark.parametrize(
(
"issuers",
"tx_amounts",
"listinput_and_amount",
"outputAddresses",
"reference",
"OutputbackChange",
"result",
),
[
# test 1 : with two amounts/outputs and an outputbackchange
(
"BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh",
(1000, 4000),
[
[
InputSource(
10000,
0,
"T",
"B37D161185A760FD81C3242D73FABD3D01F4BD9EAD98C2842061A75BAD4DFA61",
1,
),
InputSource(
257,
0,
"T",
"16F1CF9C9B89BB8C34A945F56073AB3C3ACFD858D5FA420047BA7AED1575D1FE",
1,
),
],
10000,
False,
],
(
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"4szFkvQ5tzzhwcfUtZD32hdoG2ZzhvG3ZtfR61yjnxdw",
),
"Test reference",
"BFb5yv8z1fowR6Z8mBXTALy5z7gHfMU976WtXhmRsUMh",
result1,
),
],
)
def test_generate_transaction_document(
issuers,
tx_amounts,
listinput_and_amount,
outputAddresses,
reference,
OutputbackChange,
result,
monkeypatch,
):
# patch Head_block
monkeypatch.setattr(bc_tools, "get_head_block", patched_get_head_block)
assert result == transfer.generate_transaction_document(
issuers,
tx_amounts,
listinput_and_amount,
outputAddresses,
reference,
OutputbackChange,
)
# get_list_input_for_transaction()
@pytest.mark.parametrize(
("pubkey", "TXamount", "outputs_number", "expected"),
[
# no need for interm tx, around 1 source
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 99, 2, (1, 100, False)),
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 100, 2, (1, 100, False)),
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 101, 2, (2, 200, False)),
# no need for interm.tx, arbitraty number of sources
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 199, 2, (2, 200, False)),
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 200, 2, (2, 200, False)),
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 201, 2, (3, 300, False)),
# no need for interm.tx, around last available source
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 299, 2, (3, 300, False)),
("CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp", 300, 2, (3, 300, False)),
(
"CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp",
301,
15,
"Error: you don't have enough money\n",
),
# Same tests with UD sources
# no need for interm tx, around 1 source
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
mock_ud_value - 1,
2,
(1, mock_ud_value, False),
),
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
mock_ud_value,
2,
(1, mock_ud_value, False),
),
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
mock_ud_value + 1,
2,
(2, mock_ud_value * 2, False),
),
# no need for interm.tx, arbitraty number of sources
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
5 * mock_ud_value - 1,
2,
(5, 5 * mock_ud_value, False),
),
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
5 * mock_ud_value,
2,
(5, 5 * mock_ud_value, False),
),
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
5 * mock_ud_value + 1,
2,
(6, 6 * mock_ud_value, False),
),
# no need for interm.tx, around last available source
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
10 * mock_ud_value - 1,
1,
(10, 10 * mock_ud_value, False),
),
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
10 * mock_ud_value,
2,
(10, 10 * mock_ud_value, False),
),
(
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
10 * mock_ud_value + 1,
1,
"Error: you don't have enough money\n",
),
# high limit for input number
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4600, 2, (46, 4600, False)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4601, 2, (46, 4600, True)),
# many outputs
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 3999, 15, (40, 4000, False)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4000, 15, (40, 4000, False)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4001, 15, (41, 4100, True)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4600, 15, (46, 4600, True)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4601, 15, (46, 4600, True)),
# higher limit for outputs
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 99, 93, (1, 100, False)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 100, 93, (1, 100, False)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 101, 93, (2, 200, True)),
# 1 ouput should rarely occur, but we should test it.
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4599, 1, (46, 4600, False)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4600, 1, (46, 4600, False)),
("HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat", 4601, 1, (46, 4600, True)),
# mix UD and TX source
("9cwBBgXcSVMT74xiKYygX6FM5yTdwd3NABj1CfHbbAmp", 2500, 15, (8, 2512, False)),
("9cwBBgXcSVMT74xiKYygX6FM5yTdwd3NABj1CfHbbAmp", 7800, 15, (25, 7850, False)),
# need for interm tx
("9cwBBgXcSVMT74xiKYygX6FM5yTdwd3NABj1CfHbbAmp", 14444, 15, (46, 14444, True)),
# 93 outputs, should send 1 input only
("BdanxHdwRRzCXZpiqvTVTX4gyyh6qFTYjeCWCkLwDifx", 9100, 91, (1, 9600, False)),
],
)
def test_get_list_input_for_transaction(
pubkey,
TXamount,
outputs_number,
expected,
monkeypatch,
capsys,
):
"""
expected is [len(listinput), amount, IntermediateTransaction] or "Error"
see patched_get_sources() to compute expected values.
"""
# patched functions
monkeypatch.setattr(m_tools, "get_sources", patched_get_sources)
# reset Counter.counter
Counter.counter = 0
# testing error exit
if isinstance(expected, str):
with pytest.raises(SystemExit) as pytest_exit:
result = transfer.get_list_input_for_transaction(
pubkey,
TXamount,
outputs_number,
)
assert expected == capsys.readouterr().out
assert pytest_exit.type == SystemExit
# testing good values
else:
result = transfer.get_list_input_for_transaction(
pubkey,
TXamount,
outputs_number,
)
assert (len(result[0]), result[1], result[2]) == expected
# handle_intermediaries_transactions()
@pytest.mark.parametrize(
(
"key",
"issuers",
"tx_amounts",
"outputAddresses",
"reference",
"OutputbackChange",
"expected_listinput_amount",
),
[
# test 1: with two amounts/outputs and an outputbackchange
# no need for intermediary transaction
(
key_fifi,
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
[100] * 2,
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"4szFkvQ5tzzhwcfUtZD32hdoG2ZzhvG3ZtfR61yjnxdw",
],
"Test reference",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
(
[
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=0,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=1,
),
],
200,
False,
),
),
# test 2: with 15 amounts/outputs and no outputbackchange,
# need for intermediary transaction
(
key_fifi,
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
[350] * 14, # total 4900, pubkey has 5300
["4szFkvQ5tzzhwcfUtZD32hdoG2ZzhvG3ZtfR61yjnxdw"] * 14,
"Test reference",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
(
[
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=0,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=1,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=2,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=3,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=4,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=5,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=6,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=7,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=8,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=9,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=10,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=11,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=12,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=13,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=14,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=15,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=16,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=17,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=18,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=19,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=20,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=21,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=22,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=23,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=24,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=25,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=26,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=27,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=28,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=29,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=30,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=31,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=32,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=33,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=34,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=35,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=36,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=37,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=38,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=39,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=40,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=41,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=42,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=43,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=44,
),
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=45,
),
],
4600,
True,
),
),
],
)
def test_handle_intermediaries_transactions(
key,
issuers,
tx_amounts,
outputAddresses,
reference,
OutputbackChange,
expected_listinput_amount,
monkeypatch,
):
# patched functions
patched_generate_and_send_transaction = Mock(return_value=None)
monkeypatch.setattr(m_tools, "get_sources", patched_get_sources)
monkeypatch.setattr(
transfer,
"generate_and_send_transaction",
patched_generate_and_send_transaction,
)
Counter.counter = 0
# testing
transfer.handle_intermediaries_transactions(
key,
issuers,
tx_amounts,
outputAddresses,
reference,
OutputbackChange,
)
if expected_listinput_amount[2]:
patched_generate_and_send_transaction.assert_any_call(
key,
issuers,
[
expected_listinput_amount[1],
],
expected_listinput_amount,
[issuers],
"Change operation",
)
else:
patched_generate_and_send_transaction.assert_called_once_with(
key,
issuers,
tx_amounts,
expected_listinput_amount,
outputAddresses,
reference,
issuers,
)
# test send_transaction()
@pytest.mark.parametrize(
(
"amounts",
"amountsud",
"allsources",
"recipients",
"reference",
"outputbackchange",
"yes",
"confirmation_answer",
),
[
(
[
2,
],
"",
"",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"Test",
None,
True,
"",
),
(
[2],
"",
"",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"",
"",
False,
"yes",
),
(
[2],
"",
"",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"Test reference",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
False,
"no",
),
(
[2],
"",
"",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
],
"Test reference",
None,
False,
"yes",
),
(
[2, 3],
"",
"",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
],
"Test reference",
None,
False,
"yes",
),
(
"",
[0.5, 1],
"",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
],
"Test reference",
None,
False,
"yes",
),
(
"",
"",
True,
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"Test reference",
None,
False,
"yes",
),
],
)
def test_send_transaction(
amounts,
amountsud,
allsources,
recipients,
reference,
outputbackchange,
yes,
confirmation_answer,
monkeypatch,
):
"""
This function only tests coherent values.
Errors are tested in test_transfer.py.
"""
# mocking functions
patched_gen_confirmation_table = Mock(return_value=["mock list"])
patched_handle_intermediaries_transactions = Mock(return_value=None)
# patching functions
monkeypatch.setattr(auth, "auth_method", patched_auth_method_tx)
monkeypatch.setattr(
transfer,
"gen_confirmation_table",
patched_gen_confirmation_table,
)
monkeypatch.setattr(
transfer,
"handle_intermediaries_transactions",
patched_handle_intermediaries_transactions,
)
monkeypatch.setattr(m_tools, "get_sources", patched_get_sources)
monkeypatch.setattr(m_tools, "get_ud_value", patched_get_ud_value)
# reset Counter.counter
Counter.counter = 0
# total amount for pubkey_fifi
total_amount = 5300
# compute amounts list
if allsources:
# sum of sources for account "fifi"
tx_amounts = [total_amount]
else:
if amounts:
mult = CENT_MULT_TO_UNIT
test_amounts = amounts
elif amountsud:
mult = mock_ud_value
test_amounts = amountsud
if len(recipients) != len(test_amounts) and len(test_amounts) == 1:
test_amounts = [test_amounts[0]] * len(recipients)
tx_amounts = compute_test_amounts(test_amounts, mult)
# create arguments and run cli
arguments = construct_args(
amounts,
amountsud,
allsources,
recipients,
reference,
outputbackchange,
yes,
)
CliRunner().invoke(cli, args=arguments, input=confirmation_answer)
if confirmation_answer:
patched_gen_confirmation_table.assert_called_once_with(
key_fifi.pubkey,
total_amount,
tx_amounts,
recipients,
outputbackchange,
reference,
)
if yes or confirmation_answer == "yes":
patched_handle_intermediaries_transactions.assert_called_once_with(
key_fifi,
key_fifi.pubkey,
tx_amounts,
recipients,
reference,
outputbackchange,
)
elif confirmation_answer == "no":
patched_handle_intermediaries_transactions.assert_not_called()
@pass_context
def patched_auth_method_tx(ctx):
return key_fifi
def compute_test_amounts(amounts, mult):
list_amounts = []
for amount in amounts:
list_amounts.append(round(amount * mult))
return list_amounts
# construct click arguments list
def construct_args(
amounts,
amountsud,
allsources,
recipients,
reference,
outputbackchange,
yes,
):
args_list = ["money", "transfer"]
if yes:
args_list.append("--yes")
if amounts:
for amount in amounts:
args_list.append("-a")
args_list.append(str(amount))
elif amountsud:
for amountud in amountsud:
args_list.append("-d")
args_list.append(str(amountud))
elif allsources:
args_list.append("--allSources")
for recipient in recipients:
args_list.append("-r")
args_list.append(recipient)
if reference:
args_list.append("--reference")
args_list.append(reference)
if outputbackchange is not None:
args_list.append("--outputBackChange")
args_list.append(outputbackchange)
return args_list
# generate_and_send_transaction()
@pytest.mark.parametrize(
(
"key",
"issuers",
"tx_amounts",
"listinput_and_amount",
"outputAddresses",
"reference",
"OutputbackChange",
),
[
# right tx : 1 amount for 1 receiver
(
key_fifi,
key_fifi.pubkey,
[
100,
],
(
[
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=0,
),
InputSource(
amount=200,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=1,
),
InputSource(
amount=300,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=2,
),
],
600,
False,
),
("DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",),
"",
None,
),
# right tx : 2 amounts for 2 receivers
(
key_fifi,
key_fifi.pubkey,
[
100,
250,
],
(
[
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=0,
),
InputSource(
amount=200,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=1,
),
InputSource(
amount=300,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=2,
),
],
600,
False,
),
(
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"CtM5RZHopnSRAAoWNgTWrUhDEmspcCAxn6fuCEWDWudp",
),
"Test reference",
None,
),
# Wrong tx : 3 amounts for 1 receiver
(
key_fifi,
key_fifi.pubkey,
[
100,
150,
50,
],
(
[
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=0,
),
InputSource(
amount=200,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=1,
),
InputSource(
amount=300,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=2,
),
],
600,
False,
),
("DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",),
"",
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
),
],
)
@pytest.mark.skip(
reason="network.send_document mocking fails \
test_revocation_cli{_publish{,_send_errors},_cli_revoke_{,_errors}}\
which uses this f()",
)
def test_generate_and_send_transaction(
key,
issuers,
tx_amounts,
listinput_and_amount,
outputAddresses,
reference,
OutputbackChange,
monkeypatch,
capsys,
):
# mock functions
transfer.generate_transaction_document = Mock()
network.send_document = Mock()
# patched functions
monkeypatch.setattr(bc_tools, "get_head_block", patched_get_head_block)
# monkeypatch.setattr(network, "client_instance", patched_client_instance)
transfer.generate_and_send_transaction(
key,
issuers,
tx_amounts,
listinput_and_amount,
outputAddresses,
reference,
OutputbackChange,
)
display = capsys.readouterr().out
if listinput_and_amount[2]:
assert display.find("Generate Change Transaction") != -1
else:
assert display.find("Generate Transaction:") != -1
assert display.find(f" - From: {issuers}") != -1
for tx_amount, outputAddress in zip(tx_amounts, outputAddresses):
assert display.find(
f" - To: {outputAddress}\n - Amount: {tx_amount / 100}",
)
transfer.generate_transaction_document.assert_called_once_with(
issuers,
tx_amounts,
listinput_and_amount,
outputAddresses,
reference,
OutputbackChange,
)
network.send_document.assert_called_once()
# test check_transaction_values()
@pytest.mark.parametrize(
# issuer_pubkey can be invalid. It is only used for display.
(
"reference",
"outputAddresses",
"outputBackChange",
"enough_source",
"issuer_pubkey",
"expected_outputBackchange",
),
[
(
"Test",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"",
False,
"A",
"",
),
(
"",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
],
"Hvrm4fZQWQ2M26wNczZcijce7cB8XQno2NPTwf5MovPa:5XP",
False,
"A",
"Hvrm4fZQWQ2M26wNczZcijce7cB8XQno2NPTwf5MovPa",
),
(
"Test",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
False,
"A",
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
),
(
"Test",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
]
* 92,
"",
False,
"A",
"",
),
],
)
def test_check_transaction_values(
reference,
outputAddresses,
outputBackChange,
enough_source,
issuer_pubkey,
expected_outputBackchange,
capsys,
):
result = transfer.check_transaction_values(
reference,
outputAddresses,
outputBackChange,
enough_source,
issuer_pubkey,
)
assert not capsys.readouterr().out
assert result == expected_outputBackchange
# test check_transaction_values()
@pytest.mark.parametrize(
# issuer_pubkey can be invalid. It is only used for display.
(
"reference",
"outputAddresses",
"outputBackChange",
"enough_source",
"issuer_pubkey",
),
[
(
"Wrong_Char_é",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"",
False,
"A",
),
(
"Test",
[
"Wrong_Pubkey",
],
"",
False,
"A",
),
(
"Test",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
"Wrong_Pubkey",
],
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
False,
"A",
),
(
"Test",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"Wrong_Pubkey",
False,
"A",
),
(
"Test",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
True,
"A",
),
(
"a" * 256,
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
],
"HcRgKh4LwbQVYuAc3xAdCynYXpKoiPE6qdxCMa8JeHat",
False,
"A",
),
(
"Test",
[
"DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw",
]
* 93,
"",
False,
"A",
),
],
)
def test_check_transaction_values_errors(
reference,
outputAddresses,
outputBackChange,
enough_source,
issuer_pubkey,
capsys,
):
with pytest.raises(SystemExit) as pytest_exit:
transfer.check_transaction_values(
reference,
outputAddresses,
outputBackChange,
enough_source,
issuer_pubkey,
)
assert pytest_exit.type == SystemExit
display = capsys.readouterr()
if reference.find("Wrong_Char_") != -1:
assert display.out == "Error: the reference format is invalid\n"
elif len(reference) > transfer.MAX_REFERENCE_LENGTH:
assert display.out == "Error: Transfer reference is too long\n"
elif "Wrong_Pubkey" in outputAddresses:
assert display.out.find("Error: bad format for following public key:") != -1
elif outputBackChange:
if outputBackChange == "Wrong_Pubkey":
assert display.out.find("Error: bad format for following public key:") != -1
elif enough_source is True:
assert (
display.out.find("pubkey doesn't have enough money for this transaction.")
!= -1
)
# test generate_unlocks()
@pytest.mark.parametrize(
("listinput", "expected"),
[
(
[
InputSource(
amount=100,
base=0,
source="T",
origin_id="1F3059ABF35D78DFB5AFFB3DEAB4F76878B04DB6A14757BBD6B600B1C19157E7",
index=2,
),
InputSource(
amount=mock_ud_value,
base=0,
source="D",
origin_id="2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
index=6,
),
],
[
Unlock(index=0, parameters=[SIGParameter(0)]),
Unlock(index=1, parameters=[SIGParameter(0)]),
],
),
],
)
def test_generate_unlocks(listinput, expected):
assert expected == transfer.generate_unlocks(listinput)
# test generate_output
@pytest.mark.parametrize(
("listoutput", "unitbase", "rest", "recipient_address", "expected"),
[
(
[],
0,
500,
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
[
OutputSource(
amount=500,
base=0,
condition="SIG(2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY)",
),
],
),
(
[],
2,
314,
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
[
OutputSource(
amount=3,
base=2,
condition="SIG(2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY)",
),
OutputSource(
amount=1,
base=1,
condition="SIG(2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY)",
),
OutputSource(
amount=4,
base=0,
condition="SIG(2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY)",
),
],
),
(
[
OutputSource(
amount=100,
base=0,
condition="SIG(2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY)",
),
],
0,
500,
"2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY",
[
OutputSource(
amount=100,
base=0,
condition="SIG(2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY)",
),
OutputSource(
amount=500,
base=0,
condition="SIG(2sq4w8yYVDWNxVWZqGWWDriFf5z7dn7iLahDCvEEotuY)",
),
],
),
],
)
def test_generate_output(listoutput, unitbase, rest, recipient_address, expected):
transfer.generate_output(listoutput, unitbase, rest, recipient_address)
assert len(expected) == len(listoutput)
for e, o in zip(expected, listoutput):
assert e == o
# test max_inputs_number
@pytest.mark.parametrize(
("outputs_number", "issuers_number", "expected"),
[
(1, 1, 47),
(2, 1, 46),
(93, 1, 1),
(1, 2, 46),
(2, 2, 45),
(1, 47, 1),
],
)
def test_max_inputs_number(outputs_number, issuers_number, expected):
"""
returns the maximum number of inputs.
This function does not take care of backchange line.
formula is IU <= (MAX_LINES_IN_TX_DOC - FIX_LINES - O - 2*IS)/2
"""
assert transfer.max_inputs_number(outputs_number, issuers_number) == expected
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from unittest.mock import Mock
import pytest
from click import pass_context
from click.testing import CliRunner
from silkaj import auth
from silkaj.cli import cli
from silkaj.constants import (
FAILURE_EXIT_STATUS,
MINIMAL_ABSOLUTE_TX_AMOUNT,
MINIMAL_RELATIVE_TX_AMOUNT,
PUBKEY_MIN_LENGTH,
)
from silkaj.money import tools as m_tools
from silkaj.money import transfer
from tests.patched.auth import patched_auth_method
from tests.patched.money import patched_get_sources, patched_get_ud_value
from tests.patched.test_constants import mock_ud_value
# create test auths
@pass_context
def patched_auth_method_truc(ctx):
return patched_auth_method("truc")
@pass_context
def patched_auth_method_riri(ctx):
return patched_auth_method("riri")
def test_transaction_amount(monkeypatch):
"""test passed amounts passed tx command
float ≠ 100 does not give the exact value"""
monkeypatch.setattr(m_tools, "get_ud_value", patched_get_ud_value)
trials = (
# tests for --amount (unit)
([141.89], None, ["A"], [14189]),
([141.99], None, ["A"], [14199]),
([141.01], None, ["A"], [14101]),
([141.89], None, ["A", "B"], [14189, 14189]),
([141.89, 141.99], None, ["A", "B"], [14189, 14199]),
# tests for --amount_UD
(None, [1.1], ["A"], [round(1.1 * mock_ud_value)]),
(
None,
[1.9],
[
"A",
"B",
],
[round(1.9 * mock_ud_value), round(1.9 * mock_ud_value)],
),
(None, [1.0001], ["A"], [round(1.0001 * mock_ud_value)]),
(None, [9.9999], ["A"], [round(9.9999 * mock_ud_value)]),
(
None,
[1.9, 2.3],
["A", "B"],
[round(1.9 * mock_ud_value), round(2.3 * mock_ud_value)],
),
)
for trial in trials:
assert trial[3] == transfer.transaction_amount(trial[0], trial[1], trial[2])
# transaction_amount errors()
@pytest.mark.parametrize(
("amounts", "UDs_amounts", "outputAddresses", "expected"),
[
(
None,
[0.00002],
["DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw"],
"Error: amount 0.00002 is too low.",
),
(
[10, 56],
None,
["DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw"],
"Error: The number of passed recipients is not the same as the passed amounts.",
),
(
None,
[1, 45],
["DBM6F5ChMJzpmkUdL5zD9UXKExmZGfQ1AgPDQy4MxSBw"],
"Error: The number of passed recipients is not the same as the passed amounts.",
),
],
)
def test_transaction_amount_errors(
amounts,
UDs_amounts,
outputAddresses,
expected,
capsys,
monkeypatch,
):
# patched functions
monkeypatch.setattr(m_tools, "get_ud_value", patched_get_ud_value)
# check program exit on error
with pytest.raises(SystemExit) as pytest_exit: # noqa: PT012
# read output to check error.
transfer.transaction_amount(amounts, UDs_amounts, outputAddresses)
assert expected == capsys.readouterr()
assert pytest_exit.type == SystemExit
TRANSFER = ["money", "transfer"]
MUTUALLY_EXCLUSIVE = "is mutually exclusive with arguments"
def test_tx_passed_amount_cli():
# One option
result = CliRunner().invoke(cli, [*TRANSFER, "--amount", "1"])
assert "Error: A recipient should be passed\n" in result.output
assert result.exit_code == 1
result = CliRunner().invoke(cli, [*TRANSFER, "--amountUD", "1"])
assert "Error: A recipient should be passed\n" in result.output
assert result.exit_code == 1
result = CliRunner().invoke(cli, [*TRANSFER, "--allSources"])
assert "Error: A recipient should be passed\n" in result.output
assert result.exit_code == 1
# Multiple options
result = CliRunner().invoke(cli, [*TRANSFER, "--amount", 1, "--amountUD", 1])
assert MUTUALLY_EXCLUSIVE in result.output
assert result.exit_code == 2
result = CliRunner().invoke(cli, [*TRANSFER, "--amount", 1, "--allSources"])
assert MUTUALLY_EXCLUSIVE in result.output
assert result.exit_code == 2
result = CliRunner().invoke(cli, [*TRANSFER, "--amountUD", 1, "--allSources"])
assert MUTUALLY_EXCLUSIVE in result.output
assert result.exit_code == 2
result = CliRunner().invoke(
cli,
[*TRANSFER, "--amount", 1, "--amountUD", 1, "--allSources"],
)
assert MUTUALLY_EXCLUSIVE in result.output
assert result.exit_code == 2
result = CliRunner().invoke(cli, [*TRANSFER, "-r", "A"])
assert "Error: amount, amountUD or allSources is not set." in result.output
assert result.exit_code == FAILURE_EXIT_STATUS
result = CliRunner().invoke(cli, [*TRANSFER, "-r", "A", "-r", "B", "--allSources"])
assert (
"Error: the --allSources option can only be used with one recipient."
in result.output
)
assert result.exit_code == FAILURE_EXIT_STATUS
result = CliRunner().invoke(
cli,
[*TRANSFER, "-r", "A", "-a", MINIMAL_ABSOLUTE_TX_AMOUNT - 0.001],
)
assert "Invalid value for '--amount'" in result.output
assert result.exit_code == 2
result = CliRunner().invoke(
cli,
[*TRANSFER, "-r", "A", "-d", MINIMAL_RELATIVE_TX_AMOUNT - 1e-07],
)
assert "Invalid value for '--amountUD'" in result.output
assert result.exit_code == 2
result = CliRunner().invoke(cli, [*TRANSFER, "-r", "A", "-a", 1, "-a", 2])
assert (
"Error: The number of passed recipients is not the same as the passed amounts."
in result.output
)
assert result.exit_code == FAILURE_EXIT_STATUS
result = CliRunner().invoke(
cli,
[*TRANSFER, "-r", "A", "-r", "B", "-r", "C", "-a", 1, "-a", 2],
)
assert (
"Error: The number of passed recipients is not the same as the passed amounts."
in result.output
)
assert result.exit_code == FAILURE_EXIT_STATUS
@pytest.mark.parametrize(
("arguments", "auth_method", "is_account_filled"),
[
(
[*TRANSFER, "--allSources", "-r", "A" * PUBKEY_MIN_LENGTH],
patched_auth_method_truc,
False,
),
(
[*TRANSFER, "--allSources", "-r", "A" * PUBKEY_MIN_LENGTH],
patched_auth_method_riri,
True,
),
],
)
def test_tx_passed_all_sources_empty(
arguments,
auth_method,
is_account_filled,
monkeypatch,
):
"""test that --allSources on an empty pubkey returns an error"""
# patch functions
monkeypatch.setattr(auth, "auth_method", auth_method)
monkeypatch.setattr(m_tools, "get_sources", patched_get_sources)
patched_gen_confirmation_table = Mock()
monkeypatch.setattr(
transfer,
"gen_confirmation_table",
patched_gen_confirmation_table,
)
result = CliRunner().invoke(cli, args=arguments)
# test error
if not is_account_filled:
assert (
"Error: Issuer pubkey FA4uAQ92rmxidQPgtMopaLfNNzhxu7wLgUsUkqKkSwPr:4E7 is empty. \
No transaction sent."
in result.output
)
assert result.exit_code == FAILURE_EXIT_STATUS
# test that error don't occur when issuer balance > 0
else:
patched_gen_confirmation_table.assert_called_once()
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from pathlib import Path
import pytest
from click.testing import CliRunner
from silkaj.constants import CENT_MULT_TO_UNIT
from silkaj.money import tools
from silkaj.money.transfer import parse_file_containing_amounts_recipients
from tests.patched.money import patched_get_ud_value
from tests.patched.test_constants import mock_ud_value
FILE_PATH = Path("recipients.txt")
@pytest.mark.parametrize(
("file_content", "amounts_exp", "recipients_exp"),
[
(
"ABSOLUTE\n10 pubkey1\n20 pubkey2",
[10 * CENT_MULT_TO_UNIT, 20 * CENT_MULT_TO_UNIT],
["pubkey1", "pubkey2"],
),
(
"RELATIVE\n#toto\n10 pubkey1\n#titi\n20 pubkey2",
[10 * mock_ud_value, 20 * mock_ud_value],
["pubkey1", "pubkey2"],
),
],
)
def test_parse_file_containing_amounts_recipients(
file_content,
amounts_exp,
recipients_exp,
monkeypatch,
):
monkeypatch.setattr(tools, "get_ud_value", patched_get_ud_value)
runner = CliRunner()
with runner.isolated_filesystem():
FILE_PATH.write_text(file_content, encoding="utf-8")
amounts, recipients = parse_file_containing_amounts_recipients(FILE_PATH)
assert amounts == amounts_exp
assert recipients == recipients_exp
HEADER_ERR = (
"recipients.txt must contain at first line 'ABSOLUTE' or 'RELATIVE' header\n"
)
SYNTAX_ERR = "Syntax error at line"
SPEC_ERR = "No amounts or recipients specified"
@pytest.mark.parametrize(
("file_content", "error"),
[
("ABSOLUTE\n10 pubkey1\n20", SYNTAX_ERR),
("#RELATIVE\n10 pubkey1\n20 pubkey2", HEADER_ERR),
("RELATIVE\npubkey1 10\n20 pubkey2", SYNTAX_ERR),
("ABSOLUTE", SPEC_ERR),
("", HEADER_ERR),
],
)
def test_parse_file_containing_amounts_recipients_errors(file_content, error, capsys):
runner = CliRunner()
with runner.isolated_filesystem():
FILE_PATH.write_text(file_content, encoding="utf-8")
with pytest.raises(SystemExit):
parse_file_containing_amounts_recipients(FILE_PATH)
assert error in capsys.readouterr().out
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from pathlib import Path
from unittest.mock import Mock
import pytest
from silkaj.account_storage import AccountStorage
from silkaj.blockchain import tools
from tests import helpers
@pytest.mark.parametrize(
("account_name", "currency"),
[
("test", "g1"),
("toto", "g1-test"),
],
)
def test_account_storage_account(account_name, currency, monkeypatch):
def patched_get_currency():
return currency
helpers.define_click_context(account_name=account_name)
patched_pathlib_mkdir = Mock()
monkeypatch.setattr(Path, "mkdir", patched_pathlib_mkdir)
monkeypatch.setattr(tools, "get_currency", patched_get_currency)
account_storage = AccountStorage()
assert account_storage.path == Path.home().joinpath(
account_storage.xdg_data_home,
account_storage.program_name,
currency,
account_name,
)
patched_pathlib_mkdir.assert_called_once()
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from pathlib import Path
from unittest.mock import Mock
import pytest
from click.testing import CliRunner
from silkaj import auth, cli
from silkaj.blockchain import tools
from tests.patched.auth import (
patched_auth_by_auth_file,
patched_auth_by_scrypt,
patched_auth_by_seed,
patched_auth_by_wif,
)
from tests.patched.blockchain_tools import patched_get_currency
@pytest.mark.parametrize(
("kwargs", "auth_method_called"),
[
((True, None, None), "call_auth_by_auth_file"),
((None, True, None), "call_auth_by_seed"),
((None, None, True), "call_auth_by_wif"),
((None, None, None), "call_auth_by_scrypt"),
],
)
def test_auth_options(kwargs, auth_method_called, monkeypatch):
monkeypatch.setattr(auth, "auth_by_seed", patched_auth_by_seed)
monkeypatch.setattr(auth, "auth_by_wif", patched_auth_by_wif)
monkeypatch.setattr(auth, "auth_by_auth_file", patched_auth_by_auth_file)
monkeypatch.setattr(auth, "auth_by_scrypt", patched_auth_by_scrypt)
assert auth_method_called == auth.auth_options(*kwargs)
def build_authentication_cmd(account):
command = ["--account", account] if account else []
command.append("authentication")
return command
@pytest.mark.parametrize(
("account"),
[
(None),
("test"),
],
)
def test_authentication_cmd_cli(account, monkeypatch):
patched_auth_options = Mock()
monkeypatch.setattr(auth, "auth_options", patched_auth_options)
monkeypatch.setattr(Path, "mkdir", Mock)
monkeypatch.setattr(tools, "get_currency", patched_get_currency)
command = build_authentication_cmd(account)
result = CliRunner().invoke(cli.cli, args=command)
if not account:
assert "--account general option should be specified" in result.output
patched_auth_options.assert_not_called()
else:
patched_auth_options.assert_called_once()
def test_authentication_wif(monkeypatch):
patched_auth_by_wif = Mock()
monkeypatch.setattr(auth, "auth_by_wif", patched_auth_by_wif)
command = build_authentication_cmd("test")
command.append("--auth-wif")
CliRunner().invoke(cli.cli, args=command)
patched_auth_by_wif.assert_called_once()
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
import pytest
from click.testing import CliRunner
from silkaj import auth
from silkaj.checksum import MESSAGE
from silkaj.cli import cli
from tests.patched.auth import patched_auth_method
pubkey = "DCovzCEnQm9GUWe6mr8u42JR1JAuoj3HbQUGdCkfTzSr"
checksum = "FwY"
pubkey_checksum = f"{pubkey}:{checksum}"
account = "test"
def patched_auth_method_test():
return patched_auth_method(account)
@pytest.mark.parametrize(
("command", "excepted_output"),
[
(["checksum", pubkey_checksum], "The checksum is valid"),
(["checksum", f"{pubkey}:vAK"], "The checksum is invalid"),
(["checksum", pubkey], pubkey_checksum),
(["checksum", "uid"], "Wrong public key format"),
(["checksum"], MESSAGE),
(["--account", account, "checksum"], pubkey_checksum),
(["--account", account, "checksum", "pubkey"], pubkey_checksum),
],
)
def test_checksum_command(command, excepted_output, monkeypatch):
monkeypatch.setattr(auth, "auth_method", patched_auth_method_test)
result = CliRunner().invoke(cli, args=command)
assert excepted_output in result.output
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from click.testing import CliRunner
from silkaj import constants
from silkaj.cli import cli
def test_cli_dry_run_display_options_passed_together():
# Run command with dry_run and display options
command = ["--dry-run", "--display", "wot", "membership"]
result = CliRunner().invoke(cli, args=command)
assert "Display and dry-run options can not be used together" in result.output
assert result.exit_code >= constants.FAILURE_EXIT_STATUS
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from silkaj import cli
from silkaj import g1_monetary_license as gml
from silkaj.constants import SUCCESS_EXIT_STATUS
def test_license_approval_g1_test(capsys):
gml.license_approval("g1-test")
assert not capsys.readouterr().out
# Approve is not tested
@pytest.mark.parametrize(
("display", "approve"),
[
(True, True),
(True, False),
(False, True),
(False, False),
],
)
@patch("rich_click.confirm")
@patch.object(gml.G1MonetaryLicense, "display_license")
def test_license_approval_g1(mock_display_license, mock_confirm, display, approve):
# https://stackoverflow.com/a/62939130
mock_confirm.return_value = display
gml.license_approval("g1")
if display:
mock_display_license.assert_called_once()
mock_confirm.assert_called()
@pytest.mark.parametrize(
("language", "license_sample"),
[
("en", "**Currency licensing"),
("", "**Currency licensing"),
("fr", "**Licence de la monnaie"),
("blabla", ""),
],
)
def test_language_prompt(language, license_sample):
result = CliRunner().invoke(cli.cli, args=["license"], input=language)
assert "In which language" in result.output
assert license_sample in result.output
assert result.exit_code == SUCCESS_EXIT_STATUS
def test_available_languages_and_get_license_path():
g1ml = gml.G1MonetaryLicense()
for language_code in g1ml.languages_codes:
assert g1ml.get_license_path(language_code).is_file()
def test_long_language_code_handling():
language_code = "fr-FR"
content = "Licence monétaire Ğ1"
g1ml = gml.G1MonetaryLicense()
license_path = g1ml.get_license_path(language_code)
runner = CliRunner()
with runner.isolated_filesystem():
license_path.write_text(content, encoding="utf-8")
result = runner.invoke(cli.cli, args=["license"], input=language_code)
assert language_code in result.output
assert content in result.output
license_path.unlink()
# Copyright 2016-2025 Maël Azimi <m.a@moul.re>
#
# Silkaj is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Silkaj is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
# import urllib
# from unittest.mock import patch
import pytest
# from duniterpy.api import bma
from duniterpy.api import endpoint as du_ep
from silkaj import constants, network
# from silkaj.membership import generate_membership_document
from tests import helpers
IPV6 = "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
@pytest.mark.parametrize(
("endpoint", "host", "ipv4", "ipv6", "port", "path"),
[
("127.0.0.1", "", "127.0.0.1", None, 443, None),
("127.0.0.1:80", "", "127.0.0.1", None, 80, None),
("127.0.0.1:443", "", "127.0.0.1", None, 443, None),
("127.0.0.1/path", "", "127.0.0.1", None, 443, "path"),
("127.0.0.1:80/path", "", "127.0.0.1", None, 80, "path"),
("domain.tld:80/path", "domain.tld", None, None, 80, "path"),
("localhost:80/path", "localhost", None, None, 80, "path"),
(f"[{IPV6}]", None, None, IPV6, 443, None),
(f"[{IPV6}]/path", None, None, IPV6, 443, "path"),
(f"[{IPV6}]:80/path", None, None, IPV6, 80, "path"),
],
)
def test_determine_endpoint_custom(endpoint, host, ipv4, ipv6, port, path):
helpers.define_click_context(endpoint=endpoint)
ep = network.determine_endpoint()
assert ep.host == host
assert ep.ipv4 == ipv4
assert ep.ipv6 == ipv6
assert ep.port == port
if isinstance(ep, du_ep.SecuredBMAEndpoint):
assert ep.path == path
@pytest.mark.parametrize(
("gtest", "endpoint"),
[
(True, constants.G1_TEST_DEFAULT_ENDPOINT),
(False, constants.G1_DEFAULT_ENDPOINT),
],
)
def test_determine_endpoint(gtest, endpoint):
helpers.define_click_context(gtest=gtest)
ep = network.determine_endpoint()
assert ep == du_ep.endpoint(endpoint)
# def test_send_document_success(capsys):
# display = capsys.readouterr().out
# network.send_document()
# def test_send_document_error(capsys):
# membership_doc = generate_membership_document(
# "g1", "A" * 43, "0-ahv", "test", "1-aut"
# )
#
# #with patch("urllib.request.urlopen", side_effect=urllib.error.HTTPError):
# with patch("duniterpy.api.client.API.request_url", side_effect=urllib.error.HTTPError):
# network.send_document(bma.blockchain.membership, membership_doc)
#
# display = capsys.readouterr().out
# assert "Error while publishing Membership:" in display