From 058270ec7af3a21f447e9981bd93d7227e39840d Mon Sep 17 00:00:00 2001 From: tecnovert Date: Thu, 1 May 2025 00:25:39 +0200 Subject: [PATCH] tests: Add bid and active pages to test_swap_direction.py --- .../basicswap/selenium/test_swap_direction.py | 259 ++++++++++++++++-- 1 file changed, 242 insertions(+), 17 deletions(-) diff --git a/tests/basicswap/selenium/test_swap_direction.py b/tests/basicswap/selenium/test_swap_direction.py index 40f15c2..893eec3 100644 --- a/tests/basicswap/selenium/test_swap_direction.py +++ b/tests/basicswap/selenium/test_swap_direction.py @@ -7,6 +7,7 @@ # file LICENSE or http://www.opensource.org/licenses/mit-license.php. import logging +import sys import time from tests.basicswap.util import ( @@ -16,17 +17,27 @@ from util import get_driver from selenium.webdriver.common.by import By -def clear_offers(port) -> None: - logging.info(f"clear_offers {port}") - offers = read_json_api(port, "offers") +logger = logging.getLogger() +logger.level = logging.INFO +if not len(logger.handlers): + logger.addHandler(logging.StreamHandler(sys.stdout)) - for offer in offers: - read_json_api(port, "revokeoffer/{}".format(offer["offer_id"])) - for i in range(20): - time.sleep(1) +def clear_offers(port_list) -> None: + logger.info(f"clear_offers {port_list}") + + for port in port_list: offers = read_json_api(port, "offers") - if len(offers) == 0: + for offer in offers: + read_json_api(port, "revokeoffer/{}".format(offer["offer_id"])) + + for i in range(30): + time.sleep(1) + offers_sum: int = 0 + for port in port_list: + offers = read_json_api(port, "offers") + offers_sum += len(offers) + if offers_sum == 0: return raise ValueError("clear_offers failed") @@ -37,8 +48,7 @@ def test_swap_dir(driver): node1_url = f"http://localhost:{node_1_port}" node2_url = f"http://localhost:{node_2_port}" - clear_offers(node_1_port) - clear_offers(node_2_port) + clear_offers((node_1_port, node_2_port)) offer_data = { "addr_from": -1, @@ -47,6 +57,7 @@ def test_swap_dir(driver): "amt_from": 1, "amt_to": 2, "lockhrs": 24, + "automation_strat_id": 1, } rv = read_json_api(node_1_port, "offers/new", offer_data) offer_1_id = rv["offer_id"] @@ -58,31 +69,55 @@ def test_swap_dir(driver): "amt_from": 3, "amt_to": 4, "lockhrs": 24, + "automation_strat_id": 1, } rv = read_json_api(node_1_port, "offers/new", offer_data) offer_2_id = rv["offer_id"] - # Wait for offer to propagate - offers_1 = read_json_api(node_1_port, "offers") + offer_data = { + "addr_from": -1, + "coin_from": "XMR", + "coin_to": "PART", + "amt_from": 5, + "amt_to": 6, + "lockhrs": 24, + "automation_strat_id": 1, + } + rv = read_json_api(node_2_port, "offers/new", offer_data) + offer_3_id = rv["offer_id"] + + # Wait for offers to propagate + for i in range(1000): + offers_1 = read_json_api(node_1_port, "offers") + if len(offers_1) >= 3: + break + time.sleep(0.1) + assert len(offers_1) >= 3 for offer in offers_1: if offer["offer_id"] == offer_1_id: assert offer["coin_to"] == "Monero" elif offer["offer_id"] == offer_2_id: assert offer["coin_to"] == "Bitcoin" + elif offer["offer_id"] == offer_3_id: + assert offer["coin_to"] == "Particl" else: raise ValueError("Unknown offer id") - offers_2 = read_json_api(node_2_port, "offers") - while len(offers_2) < 1: + for i in range(1000): offers_2 = read_json_api(node_2_port, "offers") + if len(offers_2) >= 3: + break time.sleep(0.1) + assert len(offers_2) >= 3 for offer in offers_2: if offer["offer_id"] == offer_1_id: assert offer["coin_to"] == "Monero" elif offer["offer_id"] == offer_2_id: assert offer["coin_to"] == "Bitcoin" + elif offer["offer_id"] == offer_3_id: + assert offer["coin_to"] == "Particl" else: raise ValueError("Unknown offer id") @@ -101,11 +136,12 @@ def test_swap_dir(driver): for td in row.find_elements(By.XPATH, ".//td") ] ) - break + if len(found_rows) >= 3: + break except Exception as e: print(e) - assert len(found_rows) == 2 + assert len(found_rows) == 3 for row in found_rows: if offer_1_id in row[0]: loc_xmr = row[5].find("Monero") @@ -117,6 +153,11 @@ def test_swap_dir(driver): loc_part = row[5].find("Particl") assert loc_btc < loc_part assert "Edit" in row[9] + elif offer_3_id in row[0]: + loc_xmr = row[5].find("Monero") + loc_part = row[5].find("Particl") + assert loc_part < loc_xmr + assert "Swap" in row[9] else: raise ValueError("Unknown offer id") @@ -139,7 +180,7 @@ def test_swap_dir(driver): except Exception as e: print(e) - assert len(found_rows) == 2 + assert len(found_rows) == 3 for row in found_rows: if offer_1_id in row[0]: assert ("Monero") in row[5] @@ -152,9 +193,193 @@ def test_swap_dir(driver): loc_part = row[5].find("Particl") assert loc_btc < loc_part assert "Swap" in row[9] + elif offer_3_id in row[0]: + loc_xmr = row[5].find("Monero") + loc_part = row[5].find("Particl") + assert loc_part < loc_xmr + assert "Edit" in row[9] else: raise ValueError("Unknown offer id") + bid_data = { + "offer_id": offer_1_id, + "amount_from": 1, + } + rv = read_json_api(node_2_port, "bids/new", bid_data) + bid_1_id = rv["bid_id"] + bid_data = { + "offer_id": offer_3_id, + "amount_from": 5, + } + rv = read_json_api(node_1_port, "bids/new", bid_data) + bid_3_id = rv["bid_id"] + + bid_ids = [bid_1_id, bid_3_id] + # Wait for bids to propagate + for i in range(1000): + num_found: int = 0 + for bid_id in bid_ids: + bid = read_json_api(node_1_port, f"bids/{bid_id}") + if "error" not in bid: + num_found += 1 + if num_found >= 2: + break + time.sleep(0.5) + assert num_found >= 2 + + for i in range(1000): + num_found: int = 0 + for bid_id in bid_ids: + bid = read_json_api(node_2_port, f"bids/{bid_id}") + if "error" not in bid: + num_found += 1 + if num_found >= 2: + break + time.sleep(0.5) + assert num_found >= 2 + + driver.get(f"{node1_url}/bid/{bid_1_id}") + td_sent = driver.find_element( + "xpath", "//td[contains(text(), 'Sent')]/following-sibling::td" + ) + assert "False" in td_sent.get_attribute("innerHTML") + td_ys = driver.find_element( + "xpath", "//td[contains(text(), 'You Send')]/following-sibling::td" + ) + assert "Particl" in td_ys.get_attribute("innerHTML") + td_yg = driver.find_element( + "xpath", "//td[contains(text(), 'You Get')]/following-sibling::td" + ) + assert "Monero" in td_yg.get_attribute("innerHTML") + + driver.get(f"{node2_url}/bid/{bid_1_id}") + td_sent = driver.find_element( + "xpath", "//td[contains(text(), 'Sent')]/following-sibling::td" + ) + assert "True" in td_sent.get_attribute("innerHTML") + td_ys = driver.find_element( + "xpath", "//td[contains(text(), 'You Send')]/following-sibling::td" + ) + assert "Monero" in td_ys.get_attribute("innerHTML") + td_yg = driver.find_element( + "xpath", "//td[contains(text(), 'You Get')]/following-sibling::td" + ) + assert "Particl" in td_yg.get_attribute("innerHTML") + + driver.get(f"{node1_url}/bid/{bid_3_id}") + td_sent = driver.find_element( + "xpath", "//td[contains(text(), 'Sent')]/following-sibling::td" + ) + assert "True" in td_sent.get_attribute("innerHTML") + td_ys = driver.find_element( + "xpath", "//td[contains(text(), 'You Send')]/following-sibling::td" + ) + assert "Particl" in td_ys.get_attribute("innerHTML") + td_yg = driver.find_element( + "xpath", "//td[contains(text(), 'You Get')]/following-sibling::td" + ) + assert "Monero" in td_yg.get_attribute("innerHTML") + + driver.get(f"{node2_url}/bid/{bid_3_id}") + td_sent = driver.find_element( + "xpath", "//td[contains(text(), 'Sent')]/following-sibling::td" + ) + assert "False" in td_sent.get_attribute("innerHTML") + td_ys = driver.find_element( + "xpath", "//td[contains(text(), 'You Send')]/following-sibling::td" + ) + assert "Monero" in td_ys.get_attribute("innerHTML") + td_yg = driver.find_element( + "xpath", "//td[contains(text(), 'You Get')]/following-sibling::td" + ) + assert "Particl" in td_yg.get_attribute("innerHTML") + + logger.info(f"Waiting for {node1_url}/active") + driver.get(f"{node1_url}/active") + bid_rows = dict() + for i in range(120): + try: + bid_rows = dict() + table = driver.find_element(By.XPATH, "//tbody[@id='active-swaps-body']") + for row in table.find_elements(By.XPATH, ".//tr"): + tds = row.find_elements(By.XPATH, ".//td") + td_details = tds[2] + td_send = tds[5] + td_recv = tds[3] + td_send_amount = td_send.find_element( + By.XPATH, ".//div[contains(@class, 'font-semibold')]" + ) + td_recv_amount = td_recv.find_element( + By.XPATH, ".//div[contains(@class, 'font-semibold')]" + ) + row_data = ( + td_send.get_attribute("innerHTML"), + td_send_amount.get_attribute("innerHTML"), + td_recv.get_attribute("innerHTML"), + td_recv_amount.get_attribute("innerHTML"), + ) + if bid_1_id in td_details.get_attribute("innerHTML"): + bid_rows[bid_1_id] = row_data + elif bid_3_id in td_details.get_attribute("innerHTML"): + bid_rows[bid_3_id] = row_data + if len(bid_rows) >= 2: + break + except Exception as e: + print(e) + time.sleep(2) + assert "PART" in bid_rows[bid_1_id][0] + assert float(bid_rows[bid_1_id][1]) == 1.0 + assert "XMR" in bid_rows[bid_1_id][2] + assert float(bid_rows[bid_1_id][3]) == 2.0 + + assert "PART" in bid_rows[bid_3_id][0] + assert float(bid_rows[bid_3_id][1]) == 6.0 + assert "XMR" in bid_rows[bid_3_id][2] + assert float(bid_rows[bid_3_id][3]) == 5.0 + + logger.info(f"Waiting for {node2_url}/active") + driver.get(f"{node2_url}/active") + bid_rows = dict() + for i in range(120): + try: + bid_rows = dict() + table = driver.find_element(By.XPATH, "//tbody[@id='active-swaps-body']") + for row in table.find_elements(By.XPATH, ".//tr"): + tds = row.find_elements(By.XPATH, ".//td") + td_details = tds[2] + td_send = tds[5] + td_recv = tds[3] + td_send_amount = td_send.find_element( + By.XPATH, ".//div[contains(@class, 'font-semibold')]" + ) + td_recv_amount = td_recv.find_element( + By.XPATH, ".//div[contains(@class, 'font-semibold')]" + ) + row_data = ( + td_send.get_attribute("innerHTML"), + td_send_amount.get_attribute("innerHTML"), + td_recv.get_attribute("innerHTML"), + td_recv_amount.get_attribute("innerHTML"), + ) + if bid_1_id in td_details.get_attribute("innerHTML"): + bid_rows[bid_1_id] = row_data + elif bid_3_id in td_details.get_attribute("innerHTML"): + bid_rows[bid_3_id] = row_data + if len(bid_rows) >= 2: + break + except Exception as e: + print(e) + time.sleep(2) + assert "XMR" in bid_rows[bid_1_id][0] + assert float(bid_rows[bid_1_id][1]) == 2.0 + assert "PART" in bid_rows[bid_1_id][2] + assert float(bid_rows[bid_1_id][3]) == 1.0 + + assert "XMR" in bid_rows[bid_3_id][0] + assert float(bid_rows[bid_3_id][1]) == 5.0 + assert "PART" in bid_rows[bid_3_id][2] + assert float(bid_rows[bid_3_id][3]) == 6.0 + print("Test Passed!")