from datetime import datetime from decimal import Decimal from decouple import config from django.contrib.auth.models import User from django.urls import reverse from api.models import Currency, Order from api.tasks import cache_market from control.models import BalanceLog from control.tasks import compute_node_balance, do_accounting from tests.test_api import BaseAPITestCase from tests.utils.node import set_up_regtest_network from tests.utils.trade import Trade def read_file(file_path): """ Read a file and return its content. """ with open(file_path, "r") as file: return file.read() class TradeTest(BaseAPITestCase): su_pass = "12345678" su_name = config("ESCROW_USERNAME", cast=str, default="admin") @classmethod def setUpTestData(cls): """ Set up initial data for the test case. """ # Create super user User.objects.create_superuser(cls.su_name, "super@user.com", cls.su_pass) # Fetch currency prices from external APIs cache_market() # Initialize bitcoin core, mine some blocks, connect nodes, open channel set_up_regtest_network() # Take the first node balances snapshot compute_node_balance() def test_login_superuser(self): """ Test the login functionality for the superuser. """ path = reverse("admin:login") data = {"username": self.su_name, "password": self.su_pass} response = self.client.post(path, data) self.assertEqual(response.status_code, 302) self.assertResponse( response ) # should skip given that /coordinator/login is not documented def test_cache_market(self): """ Test if the cache_market() call during test setup worked """ usd = Currency.objects.get(id=1) self.assertIsInstance( usd.exchange_rate, Decimal, f"Exchange rate is not a Decimal. Got {type(usd.exchange_rate)}", ) self.assertGreater( usd.exchange_rate, 0, "Exchange rate is not higher than zero" ) self.assertIsInstance( usd.timestamp, datetime, "External price timestamp is not a datetime" ) def test_initial_balance_log(self): """ Test if the initial node BalanceLog is correct. One channel should exist with 0.5BTC in local. No onchain balance should exist. """ balance_log = BalanceLog.objects.latest() self.assertIsInstance(balance_log.time, datetime) self.assertTrue(balance_log.total > 0) self.assertTrue(balance_log.ln_local > 0) self.assertTrue(balance_log.ln_local_unsettled >= 0) self.assertTrue(balance_log.ln_remote > 0) self.assertEqual(balance_log.ln_remote_unsettled, 0) self.assertTrue(balance_log.onchain_total > 0) self.assertTrue(balance_log.onchain_confirmed > 0) self.assertEqual(balance_log.onchain_unconfirmed, 0) self.assertTrue(balance_log.onchain_fraction > 0) def assert_robot(self, response, robot_index): """ Assert that the robot is created correctly. """ nickname = read_file(f"tests/robots/{robot_index}/nickname") pub_key = read_file(f"tests/robots/{robot_index}/pub_key") enc_priv_key = read_file(f"tests/robots/{robot_index}/enc_priv_key") data = response.json() self.assertEqual(response.status_code, 200) self.assertResponse(response) self.assertEqual( data["nickname"], nickname, f"Robot created nickname is not {nickname}", ) self.assertEqual( data["public_key"], pub_key, "Returned public Kky does not match" ) self.assertEqual( data["encrypted_private_key"], enc_priv_key, "Returned encrypted private key does not match", ) self.assertEqual( len(data["tg_token"]), 15, "String is not exactly 15 characters long" ) self.assertEqual( data["tg_bot_name"], config( "TELEGRAM_BOT_NAME", cast=str, default="RoboCoordinatorNotificationBot" ), "Telegram bot name is not correct", ) self.assertFalse( data["tg_enabled"], "The new robot's telegram seems to be enabled" ) self.assertEqual(data["earned_rewards"], 0, "The new robot's rewards are not 0") def test_create_robots(self): """ Test the creation of two robots to be used in the trade tests """ trade = Trade(self.client) for robot_index in [1, 2]: response = trade.create_robot(robot_index) self.assert_robot(response, robot_index) def test_make_order(self): """ Test the creation of an order. """ trade = Trade( self.client ) # init of Trade calls make_order() with the default maker form. data = trade.response.json() # Checks self.assertResponse(trade.response) self.assertIsInstance(data["id"], int, "Order ID is not an integer") self.assertEqual( data["status"], Order.Status.WFB, "Newly created order status is not 'Waiting for maker bond'", ) self.assertIsInstance( datetime.fromisoformat(data["created_at"]), datetime, "Order creation timestamp is not datetime", ) self.assertIsInstance( datetime.fromisoformat(data["expires_at"]), datetime, "Order expiry time is not datetime", ) self.assertEqual( data["type"], Order.Types.BUY, "Buy order is not of type value BUY" ) self.assertEqual(data["currency"], 1, "Order for USD is not of currency USD") self.assertIsNone( data["amount"], "Order with range has a non-null simple amount" ) self.assertTrue(data["has_range"], "Order with range has a False has_range") self.assertAlmostEqual( float(data["min_amount"]), trade.maker_form["min_amount"], "Order min amount does not match", ) self.assertAlmostEqual( float(data["max_amount"]), trade.maker_form["max_amount"], "Order max amount does not match", ) self.assertEqual( data["payment_method"], trade.maker_form["payment_method"], "Order payment method does not match", ) self.assertEqual( data["escrow_duration"], trade.maker_form["escrow_duration"], "Order escrow duration does not match", ) self.assertAlmostEqual( float(data["bond_size"]), trade.maker_form["bond_size"], "Order bond size does not match", ) self.assertAlmostEqual( float(data["latitude"]), trade.maker_form["latitude"], "Order latitude does not match", ) self.assertAlmostEqual( float(data["longitude"]), trade.maker_form["longitude"], "Order longitude does not match", ) self.assertAlmostEqual( float(data["premium"]), trade.maker_form["premium"], "Order premium does not match", ) self.assertFalse( data["is_explicit"], "Relative pricing order has True is_explicit" ) self.assertIsNone( data["satoshis"], "Relative pricing order has non-null Satoshis" ) self.assertIsNone(data["taker"], "New order's taker is not null") def test_get_order_created(self): """ Tests the creation of an order and the first request to see details, including, the creation of the maker bond invoice. """ robot_index = 1 trade = Trade( self.client, maker_index=robot_index ) # init of Trade calls make_order() with the default maker form. # Maker's first order fetch. Should trigger maker bond hold invoice generation. trade.get_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["id"], trade.order_id) self.assertIsInstance(datetime.fromisoformat(data["created_at"]), datetime) self.assertIsInstance(datetime.fromisoformat(data["expires_at"]), datetime) self.assertTrue(data["is_maker"]) self.assertTrue(data["is_participant"]) self.assertTrue(data["is_buyer"]) self.assertFalse(data["is_seller"]) self.assertEqual(data["maker_status"], "Active") self.assertEqual(data["status_message"], Order.Status(Order.Status.WFB).label) self.assertFalse(data["is_fiat_sent"]) self.assertFalse(data["is_disputed"]) self.assertEqual( data["ur_nick"], read_file(f"tests/robots/{robot_index}/nickname") ) self.assertIsInstance(data["satoshis_now"], int) self.assertFalse(data["maker_locked"]) self.assertFalse(data["taker_locked"]) self.assertFalse(data["escrow_locked"]) self.assertIsInstance(data["bond_satoshis"], int) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order() def test_publish_order(self): """ Tests a trade from order creation to published (maker bond locked). """ trade = Trade(self.client) trade.publish_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["id"], data["id"]) self.assertEqual(data["status_message"], Order.Status(Order.Status.PUB).label) self.assertTrue(data["maker_locked"]) self.assertFalse(data["taker_locked"]) self.assertFalse(data["escrow_locked"]) # Test what we can see with newly created robot 2 (only for public status) trade.get_order(robot_index=2, first_encounter=True) public_data = trade.response.json() self.assertFalse(public_data["is_participant"]) self.assertIsInstance(public_data["price_now"], float) self.assertIsInstance(data["satoshis_now"], int) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order() def test_pause_unpause_order(self): """ Tests pausing and unpausing a public order """ trade = Trade(self.client) trade.publish_order() data = trade.response.json() # PAUSE trade.pause_order() data = trade.response.json() self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.PAU).label) # UNPAUSE trade.pause_order() data = trade.response.json() self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.PUB).label) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order() def test_make_and_take_order(self): """ Tests a trade from order creation to taken. """ trade = Trade(self.client) trade.publish_order() trade.take_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.TAK).label) self.assertEqual( data["ur_nick"], read_file(f"tests/robots/{trade.taker_index}/nickname") ) self.assertEqual( data["taker_nick"], read_file(f"tests/robots/{trade.taker_index}/nickname") ) self.assertEqual( data["maker_nick"], read_file(f"tests/robots/{trade.maker_index}/nickname") ) self.assertEqual(data["maker_status"], "Active") self.assertEqual(data["taker_status"], "Active") self.assertFalse(data["is_maker"]) self.assertFalse(data["is_buyer"]) self.assertTrue(data["is_seller"]) self.assertTrue(data["is_taker"]) self.assertTrue(data["is_participant"]) self.assertTrue(data["maker_locked"]) self.assertFalse(data["taker_locked"]) self.assertFalse(data["escrow_locked"]) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order() def test_make_and_lock_contract(self): """ Tests a trade from order creation to taker bond locked. """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.WF2).label) self.assertEqual(data["maker_status"], "Active") self.assertEqual(data["taker_status"], "Active") self.assertTrue(data["is_participant"]) self.assertTrue(data["maker_locked"]) self.assertTrue(data["taker_locked"]) self.assertFalse(data["escrow_locked"]) # Maker GET trade.get_order(trade.maker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.WF2).label) self.assertTrue(data["swap_allowed"]) self.assertIsInstance(data["suggested_mining_fee_rate"], float) self.assertIsInstance(data["swap_fee_rate"], float) self.assertTrue(data["suggested_mining_fee_rate"] > 0) self.assertTrue(data["swap_fee_rate"] > 0) self.assertEqual(data["maker_status"], "Active") self.assertEqual(data["taker_status"], "Active") self.assertTrue(data["is_participant"]) self.assertTrue(data["maker_locked"]) self.assertTrue(data["taker_locked"]) self.assertFalse(data["escrow_locked"]) # Maker cancels order to avoid leaving pending HTLCs after a successful test trade.cancel_order() def test_trade_to_locked_escrow(self): """ Tests a trade from order creation until escrow locked, before invoice/address is submitted by buyer. """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.WFI).label) self.assertTrue(data["maker_locked"]) self.assertTrue(data["taker_locked"]) self.assertTrue(data["escrow_locked"]) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order(trade.taker_index) def test_trade_to_submitted_address(self): """ Tests a trade from order creation until escrow locked, before invoice/address is submitted by buyer. """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_address(trade.maker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.CHA).label) self.assertFalse(data["is_fiat_sent"]) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order(trade.maker_index) trade.cancel_order(trade.taker_index) def test_trade_to_submitted_invoice(self): """ Tests a trade from order creation until escrow locked and invoice is submitted by buyer. """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_invoice(trade.maker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.CHA).label) self.assertFalse(data["is_fiat_sent"]) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order(trade.maker_index) trade.cancel_order(trade.taker_index) def test_trade_to_confirm_fiat_sent_LN(self): """ Tests a trade from order creation until fiat sent confirmed """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_invoice(trade.maker_index) trade.confirm_fiat(trade.maker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.FSE).label) self.assertTrue(data["is_fiat_sent"]) # Cancel order to avoid leaving pending HTLCs after a successful test trade.undo_confirm_sent(trade.maker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.CHA).label) trade.cancel_order(trade.maker_index) trade.cancel_order(trade.taker_index) def test_trade_to_confirm_fiat_received_LN(self): """ Tests a trade from order creation until fiat received is confirmed by seller/taker """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_invoice(trade.maker_index) trade.confirm_fiat(trade.maker_index) trade.confirm_fiat(trade.taker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.PAY).label) self.assertTrue(data["is_fiat_sent"]) self.assertFalse(data["is_disputed"]) self.assertFalse(data["maker_locked"]) self.assertFalse(data["taker_locked"]) self.assertFalse(data["escrow_locked"]) def test_successful_LN(self): """ Tests a trade from order creation until Sats sent to buyer """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_invoice(trade.maker_index) trade.confirm_fiat(trade.maker_index) trade.confirm_fiat(trade.taker_index) trade.process_payouts() trade.get_order(trade.maker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.SUC).label) self.assertTrue(data["is_fiat_sent"]) self.assertFalse(data["is_disputed"]) self.assertIsHash(data["maker_summary"]["preimage"]) self.assertIsHash(data["maker_summary"]["payment_hash"]) def test_successful_onchain(self): """ Tests a trade from order creation until Sats sent to buyer """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_address(trade.maker_index) trade.confirm_fiat(trade.maker_index) trade.confirm_fiat(trade.taker_index) trade.process_payouts(mine_a_block=True) trade.get_order(trade.maker_index) data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual(data["status_message"], Order.Status(Order.Status.SUC).label) self.assertTrue(data["is_fiat_sent"]) self.assertFalse(data["is_disputed"]) self.assertIsInstance(data["maker_summary"]["address"], str) self.assertIsHash(data["maker_summary"]["txid"]) def test_cancel_public_order(self): """ Tests the cancellation of a public order """ trade = Trade(self.client) trade.publish_order() trade.cancel_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 400) self.assertResponse(trade.response) self.assertEqual( data["bad_request"], "This order has been cancelled by the maker" ) def test_collaborative_cancel_order_in_chat(self): """ Tests the collaborative cancellation of an order in the chat state """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_invoice(trade.maker_index) # Maker asks for cancel trade.cancel_order(trade.maker_index) self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertTrue(trade.response.json()["asked_for_cancel"]) # Taker checks order trade.get_order(trade.taker_index) self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertTrue(trade.response.json()["pending_cancel"]) # Taker accepts (ask) the cancellation trade.cancel_order(trade.taker_index) self.assertEqual(trade.response.status_code, 400) self.assertResponse(trade.response) self.assertEqual( trade.response.json()["bad_request"], "This order has been cancelled collaborativelly", ) def test_created_order_expires(self): """ Tests the expiration of a public order """ trade = Trade(self.client) # Change order expiry to now order = Order.objects.get(id=trade.response.json()["id"]) order.expires_at = datetime.now() order.save() # Make orders expire trade.clean_orders() trade.get_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual( data["status"], Order.Status.EXP, ) self.assertEqual( data["expiry_message"], Order.ExpiryReasons(Order.ExpiryReasons.NMBOND).label, ) self.assertEqual(data["expiry_reason"], Order.ExpiryReasons.NMBOND) def test_public_order_expires(self): """ Tests the expiration of a public order """ trade = Trade(self.client) trade.publish_order() # Change order expiry to now order = Order.objects.get(id=trade.response.json()["id"]) order.expires_at = datetime.now() order.save() # Make orders expire trade.clean_orders() trade.get_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual( data["status"], Order.Status.EXP, ) self.assertEqual( data["expiry_message"], Order.ExpiryReasons(Order.ExpiryReasons.NTAKEN).label, ) self.assertEqual(data["expiry_reason"], Order.ExpiryReasons.NTAKEN) def test_taken_order_expires(self): """ Tests the expiration of a public order """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() # Change order expiry to now order = Order.objects.get(id=trade.response.json()["id"]) order.expires_at = datetime.now() order.save() # Make orders expire trade.clean_orders() trade.get_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual( data["status"], Order.Status.EXP, ) self.assertEqual( data["expiry_message"], Order.ExpiryReasons(Order.ExpiryReasons.NESINV).label, ) self.assertEqual(data["expiry_reason"], Order.ExpiryReasons.NESINV) def test_escrow_locked_expires(self): """ Tests the expiration of a public order """ trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) # Change order expiry to now order = Order.objects.get(id=trade.response.json()["id"]) order.expires_at = datetime.now() order.save() # Make orders expire trade.clean_orders() trade.get_order() data = trade.response.json() self.assertEqual(trade.response.status_code, 200) self.assertResponse(trade.response) self.assertEqual( data["status"], Order.Status.EXP, ) self.assertEqual( data["expiry_message"], Order.ExpiryReasons(Order.ExpiryReasons.NINVOI).label, ) self.assertEqual(data["expiry_reason"], Order.ExpiryReasons.NINVOI) def test_ticks(self): """ Tests the historical ticks serving endpoint after creating a contract """ path = reverse("ticks") params = "?start=01-01-1970&end=01-01-2070" # Make a contract and cancel trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.cancel_order() response = self.client.get(path + params) data = response.json() self.assertEqual(response.status_code, 200) self.assertResponse(response) self.assertIsInstance(datetime.fromisoformat(data[0]["timestamp"]), datetime) self.assertIsInstance(data[0]["volume"], str) self.assertIsInstance(data[0]["price"], str) self.assertIsInstance(data[0]["premium"], str) self.assertIsInstance(data[0]["fee"], str) def test_daily_historical(self): """ Tests the daily history serving endpoint after creating a contract """ path = reverse("historical") # Run a successful trade trade = Trade(self.client) trade.publish_order() trade.take_order() trade.lock_taker_bond() trade.lock_escrow(trade.taker_index) trade.submit_payout_invoice(trade.maker_index) trade.confirm_fiat(trade.maker_index) trade.confirm_fiat(trade.taker_index) trade.process_payouts() # Do daily accounting to create the daily summary do_accounting() response = self.client.get(path) data = response.json() self.assertEqual(response.status_code, 200) # self.assertResponse(response) # Expects an array, but response is an object first_date = list(data.keys())[0] self.assertIsInstance(datetime.fromisoformat(first_date), datetime) self.assertIsInstance(data[first_date]["volume"], float) self.assertIsInstance(data[first_date]["num_contracts"], int) def test_book(self): """ Tests public book view """ path = reverse("book") trade = Trade(self.client) trade.publish_order() response = self.client.get(path) data = response.json() self.assertEqual(response.status_code, 200) self.assertResponse(response) self.assertIsInstance(datetime.fromisoformat(data[0]["created_at"]), datetime) self.assertIsInstance(datetime.fromisoformat(data[0]["expires_at"]), datetime) self.assertIsNone(data[0]["amount"]) self.assertAlmostEqual( float(data[0]["min_amount"]), trade.maker_form["min_amount"] ) self.assertAlmostEqual( float(data[0]["max_amount"]), trade.maker_form["max_amount"] ) self.assertAlmostEqual(float(data[0]["latitude"]), trade.maker_form["latitude"]) self.assertAlmostEqual( float(data[0]["longitude"]), trade.maker_form["longitude"] ) self.assertEqual( data[0]["escrow_duration"], trade.maker_form["escrow_duration"] ) self.assertFalse(data[0]["is_explicit"]) # Cancel order to avoid leaving pending HTLCs after a successful test trade.cancel_order()