Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy_test.conftest

# GALAXY_CONFIG_OVERRIDE_STATSD_PREFIX="galaxy" GALAXY_CONFIG_OVERRIDE_STATSD_HOST="localhost" GALAXY_CONFIG_OVERRIDE_STATSD_INFLUXDB="true" GALAXY_CONFIG_OVERRIDE_DATABASE_LOG_QUERY_COUNTS="true" GALAXY_CONFIG_OVERRIDE_ENABLE_PER_REQUEST_SQL_DEBUGGING="true" GALAXY_CONFIG_OVERRIDE_SLOW_QUERY_LOG_THRESHOLD=".05" ./run_tests.sh --structured_data_report_file 'test.json' --api  lib/galaxy_test/api/test_configuration.py
import os
import uuid
from urllib.parse import urlencode

import pytest

from galaxy.util import (
    DEFAULT_SOCKET_TIMEOUT,
    requests,
)
from galaxy.web import statsd_client as statsd
from galaxy_test.api._framework import ApiTestCase
from galaxy_test.driver.driver_util import GalaxyTestDriver

pytest_plugins = ("celery.contrib.pytest",)


[docs]@pytest.fixture(scope="session", autouse=True) def celery_includes(): yield ["galaxy.celery.tasks"]
[docs]@pytest.fixture(scope="session") def real_driver(): if not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"): driver = GalaxyTestDriver() driver.setup(ApiTestCase) try: yield driver finally: driver.tear_down() else: yield None
[docs]@pytest.fixture(scope="class") def embedded_driver(real_driver, request): request.cls._test_driver = real_driver
[docs]def get_timings(test_uuid): # timestamps didn't work - telegraf didn't sample at small enough granularity # fs = datetime.datetime.fromtimestamp(from_timestamp).strftime('%Y-%m-%dT%H:%M:%S.%fZ') # ts = datetime.datetime.fromtimestamp(to_timestamp).strftime('%Y-%m-%dT%H:%M:%S.%fZ') params = { "db": "telegraf", "q": f"select * from galaxy_ where test='{test_uuid}'", } query = urlencode(params) headers = {"content-type": "application/json"} url = f"http://localhost:8086/query?{query}" response = requests.get(url, headers=headers, timeout=DEFAULT_SOCKET_TIMEOUT) if response.ok: return response.json() else: return None
[docs]def pytest_configure(config): if config.pluginmanager.hasplugin("pytest_jsonreport"): if config.getoption("json_report"): config.pluginmanager.register(JsonReportHooks())
[docs]class JsonReportHooks:
[docs] def pytest_json_runtest_metadata(self, item, call): if call.when == "setup": statsd.CURRENT_TEST = str(uuid.uuid4()) statsd.CURRENT_TEST_METRICS = {"timing": {}, "counter": {}} return {} if call.when == "teardown": statsd.CURRENT_TEST = None statsd.CURRENT_TEST_METRICS = None return {} if call.when != "call": return {} return { "start": call.start, "stop": call.stop, "uuid": statsd.CURRENT_TEST, "local_metrics": statsd.CURRENT_TEST_METRICS, }
[docs] def pytest_json_modifyreport(self, json_report): if os.environ.get("GALAXY_TEST_COLLECT_STATSD", "0") != "0": for test in json_report["tests"]: metadata = test["metadata"] test_uuid = metadata["uuid"] metadata["statsd_metrics"] = get_timings(test_uuid)