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)