import logging import pytest import os import json import requests import time import docker import http.server import sys from dotenv import load_dotenv from docker.models.containers import Container from docker.errors import APIError from contextlib import contextmanager from vega_sim.null_service import VegaServiceNull, Ports from playwright.sync_api import Browser, Page from config import console_image_name, vega_version from datetime import datetime, timedelta from fixtures.market import ( setup_simple_market, setup_opening_auction_market, setup_continuous_market, setup_perps_market, ) # Workaround for current xdist issue with displaying live logs from multiple workers # https://github.com/pytest-dev/pytest-xdist/issues/402 sys.stdout = sys.stderr docker_client = docker.from_env() logger = logging.getLogger() load_dotenv() @pytest.hookimpl(tryfirst=True) def pytest_runtest_makereport(item, call): outcome = "passed" if call.excinfo is None else "failed" item.config.cache.set(item.nodeid, outcome) def pytest_configure(config): worker_id = os.environ.get("PYTEST_XDIST_WORKER") if worker_id is not None: log_dir = os.path.join(os.getcwd(), "logs") log_name = f"tests_{worker_id}.log" if not os.path.exists(log_dir): os.makedirs(log_dir) logging.basicConfig( format=config.getini("log_file_format"), datefmt=config.getini("log_file_date_format"), filename=os.path.join(log_dir, log_name), level=config.getini("log_file_level"), ) class CustomHttpRequestHandler(http.server.SimpleHTTPRequestHandler): def do_GET(self): # Set the path to your website's directory here if self.path == "/": self.path = "dist/apps/trading/exported/index.html" return http.server.SimpleHTTPRequestHandler.do_GET(self) # Start VegaServiceNull @contextmanager def init_vega(request=None): local_server = os.getenv("LOCAL_SERVER", "false").lower() == "true" port_config = None if local_server: port_config = { Ports.DATA_NODE_REST: 8001, } default_seconds = 1 seconds_per_block = default_seconds if request and hasattr(request, "param"): seconds_per_block = request.param logger.info( "Starting VegaServiceNull", extra={"worker_id": os.environ.get("PYTEST_XDIST_WORKER")}, ) logger.info(f"Using console image: {console_image_name}") logger.info(f"Using vega version: {vega_version}") vega_service_args = { "run_with_console": False, "launch_graphql": False, "retain_log_files": True, "use_full_vega_wallet": True, "store_transactions": True, "transactions_per_block": 1000, "seconds_per_block": seconds_per_block, "genesis_time": datetime.now() - timedelta(days=1), } if port_config is not None: vega_service_args["port_config"] = port_config with VegaServiceNull(**vega_service_args) as vega: try: container = docker_client.containers.run( console_image_name, detach=True, ports={"80/tcp": vega.console_port} ) if not isinstance(container, Container): raise Exception("container instance invalid") logger.info( f"Container {container.id} started", extra={"worker_id": os.environ.get("PYTEST_XDIST_WORKER")}, ) yield vega except APIError as e: logger.info(f"Container creation failed.") logger.info(e) raise e finally: logger.info(f"Stopping container {container.id}") container.stop() logger.info(f"Removing container {container.id}") container.remove() @contextmanager def init_page(vega: VegaServiceNull, browser: Browser, request: pytest.FixtureRequest): local_server = os.getenv("LOCAL_SERVER", "false").lower() == "true" server_port = "4200" if local_server else str(vega.console_port) with browser.new_context( viewport={"width": 1920, "height": 1080}, base_url=f"http://localhost:{server_port}", ) as context, context.new_page() as page: context.tracing.start(screenshots=True, snapshots=True, sources=True) try: # Wait for the console to be up and running before any tests are run attempts = 0 while attempts < 100: try: code = requests.get(f"http://localhost:{server_port}/").status_code if code == 200: break except requests.exceptions.ConnectionError as e: attempts += 1 if attempts < 100: time.sleep(0.1) continue else: raise e # Set window._env_ so built app uses datanode from vega market sim env = json.dumps( { "VEGA_URL": f"http://localhost:{vega.data_node_rest_port}/graphql", "VEGA_WALLET_URL": f"http://localhost:{vega.wallet_port}", } ) window_env = f"window._env_ = Object.assign({{}}, window._env_, {env})" page.add_init_script(script=window_env) yield page finally: try: if not os.path.exists("apps/trading/e2e/traces"): os.makedirs("apps/trading/e2e/traces") except OSError as e: print(f"Failed to create directory '{'apps/trading/e2e/traces'}': {e}") # Check whether this test failed or passed outcome = request.config.cache.get(request.node.nodeid, None) if outcome != "passed": try: trace_path = os.path.join("traces", request.node.name + "trace.zip") context.tracing.stop(path=trace_path) except Exception as e: logger.error(f"Failed to save trace: {e}") @pytest.fixture def vega(request): with init_vega(request) as vega: yield vega @pytest.fixture def page(vega, browser, request): with init_page(vega, browser, request) as page_instance: yield page_instance # Set auth token so eager connection for MarketSim wallet is successful def auth_setup(vega: VegaServiceNull, page: Page): DEFAULT_WALLET_NAME = "MarketSim" # This is the default wallet name within VegaServiceNull and CANNOT be changed # Calling get_keypairs will internally call _load_tokens for the given wallet keypairs = vega.wallet.get_keypairs(DEFAULT_WALLET_NAME) wallet_api_token = vega.wallet.login_tokens[DEFAULT_WALLET_NAME] # Set token to localStorage so eager connect hook picks it up and immediately connects wallet_config = json.dumps( { "token": f"VWT {wallet_api_token}", "connector": "jsonRpc", "url": f"http://localhost:{vega.wallet_port}", } ) storage_javascript = [ # Store wallet config so eager connection is initiated f"localStorage.setItem('vega_wallet_config', '{wallet_config}');", # Ensure wallet ris dialog doesnt show, otherwise eager connect wont work "localStorage.setItem('vega_wallet_risk_accepted', 'true');", # Ensure initial risk dialog doesnt show "localStorage.setItem('vega_risk_accepted', 'true');", ] script = "".join(storage_javascript) page.add_init_script(script) return { "wallet": DEFAULT_WALLET_NAME, "wallet_api_token": wallet_api_token, "public_key": keypairs["Key 1"], } @pytest.fixture(scope="function") def auth(vega: VegaServiceNull, page: Page): return auth_setup(vega, page) # Set 'risk accepted' flag, so that the risk dialog doesn't show up def risk_accepted_setup(page: Page): onboarding_config = json.dumps({"state": {"dismissed": True}, "version": 0}) storage_javascript = [ "localStorage.setItem('vega_risk_accepted', 'true');", f"localStorage.setItem('vega_onboarding', '{onboarding_config}');", "localStorage.setItem('vega_telemetry_approval', 'false');", "localStorage.setItem('vega_telemetry_viewed', 'true');", ] script = "".join(storage_javascript) page.add_init_script(script) @pytest.fixture(scope="function") def risk_accepted(page: Page): risk_accepted_setup(page) @pytest.fixture(scope="function") def simple_market(vega, request): kwargs = {} if hasattr(request, "param"): kwargs.update(request.param) return setup_simple_market(vega, **kwargs) @pytest.fixture(scope="function") def opening_auction_market(vega): return setup_opening_auction_market(vega) @pytest.fixture(scope="function") def continuous_market(vega): return setup_continuous_market(vega) @pytest.fixture(scope="function") def proposed_market(vega): return setup_simple_market(vega, approve_proposal=False) @pytest.fixture(scope="function") def perps_market(vega, request): kwargs = {} if hasattr(request, "param"): kwargs.update(request.param) return setup_perps_market(vega, **kwargs)