actions/opensearch-pyd
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[7.x] Switch to Pytest as default runner

Browse Source
This commit is contained in:
Seth Michael Larson
2020-05-19 14:18:53 -05:00
committed by Seth Michael Larson
parent f8b005f62e
commit eaff4af910
12 changed files with 274 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_elasticsearch/README.rst
+1 -2
View File
@@ -34,8 +34,7 @@ The behavior is driven by environmental variables:
Alternatively, if you wish to control what you are doing you have several additional options:
* you can just run your favorite runner in the ``test_elasticsearch`` directory
(verified to work with nose and py.test) and bypass the fetch logic entirely.
* ``run_tests.py`` will pass any parameters specified to ``pytest``
* to run a specific test, you can use ``python3 setup.py test -s <test_name>``, for example
``python3 setup.py test -s test_elasticsearch.test_helpers.TestParallelBulk.test_all_chunks_sent``
test_elasticsearch/test_client/test_utils.py
+1 -1
View File
@@ -12,7 +12,7 @@ from ..test_cases import TestCase, SkipTest
class TestQueryParams(TestCase):
def setUp(self):
def setup_method(self, _):
self.calls = []
@query_params("simple_param")
test_elasticsearch/test_helpers.py
+3 -4
View File
@@ -6,7 +6,7 @@
import mock
import time
import threading
from nose.plugins.skip import SkipTest
import pytest
from elasticsearch import helpers, Elasticsearch
from elasticsearch.serializer import JSONSerializer
@@ -41,7 +41,7 @@ class TestParallelBulk(TestCase):
self.assertEqual(50, mock_process_bulk_chunk.call_count)
@SkipTest
@pytest.mark.skip
@mock.patch(
"elasticsearch.helpers.actions._process_bulk_chunk",
# make sure we spend some time in the thread
@@ -60,8 +60,7 @@ class TestParallelBulk(TestCase):
class TestChunkActions(TestCase):
def setUp(self):
super(TestChunkActions, self).setUp()
def setup_method(self, _):
self.actions = [({"index": {}}, {"some": u"datá", "i": i}) for i in range(100)]
def test_chunks_are_chopped_by_byte_size(self):
test_elasticsearch/test_serializer.py
+1 -2
View File
@@ -144,8 +144,7 @@ class TestTextSerializer(TestCase):
class TestDeserializer(TestCase):
def setUp(self):
super(TestDeserializer, self).setUp()
def setup_method(self, _):
self.de = Deserializer(DEFAULT_SERIALIZERS)
def test_deserializes_json_by_default(self):
test_elasticsearch/test_server/__init__.py
+1 -1
View File
@@ -35,7 +35,7 @@ def get_client(**kwargs):
return new_client
def setup():
def setup_module():
get_client()
test_elasticsearch/test_server/conftest.py
+59
View File
@@ -0,0 +1,59 @@
# Licensed to Elasticsearch B.V under one or more agreements.
# Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
# See the LICENSE file in the project root for more information
import os
import time
import pytest
import elasticsearch
@pytest.fixture(scope="function")
def sync_client():
client = None
try:
kw = {
"timeout": 30,
"ca_certs": ".ci/certs/ca.pem",
"connection_class": getattr(
elasticsearch,
os.environ.get("PYTHON_CONNECTION_CLASS", "Urllib3HttpConnection"),
),
}
client = elasticsearch.Elasticsearch(
[os.environ.get("ELASTICSEARCH_HOST", {})], **kw
)
# wait for yellow status
for _ in range(100):
try:
client.cluster.health(wait_for_status="yellow")
break
except ConnectionError:
time.sleep(0.1)
else:
# timeout
pytest.skip("Elasticsearch failed to start.")
yield client
finally:
if client:
version = tuple(
[
int(x) if x.isdigit() else 999
for x in (client.info())["version"]["number"].split(".")
]
)
expand_wildcards = ["open", "closed"]
if version >= (7, 7):
expand_wildcards.append("hidden")
client.indices.delete(
index="*", ignore=404, expand_wildcards=expand_wildcards
)
client.indices.delete_template(name="*", ignore=404)
client.indices.delete_index_template(name="*", ignore=404)
client.transport.close()
test_elasticsearch/test_server/test_helpers.py
+6 -9
View File
@@ -70,7 +70,7 @@ class TestStreamingBulk(ElasticsearchTestCase):
assert False, "exception should have been raised"
def test_different_op_types(self):
if self.es_version < (0, 90, 1):
if self.es_version() < (0, 90, 1):
raise SkipTest("update supported since 0.90.1")
self.client.index(index="i", id=45, body={})
self.client.index(index="i", id=42, body={})
@@ -325,10 +325,9 @@ class TestScan(ElasticsearchTestCase):
},
]
@classmethod
def tearDownClass(cls):
cls.client.transport.perform_request("DELETE", "/_search/scroll/_all")
super(TestScan, cls).tearDownClass()
def teardown_method(self, m):
self.client.transport.perform_request("DELETE", "/_search/scroll/_all")
super(TestScan, self).teardown_method(m)
def test_order_can_be_preserved(self):
bulk = []
@@ -498,8 +497,7 @@ class TestScan(ElasticsearchTestCase):
class TestReindex(ElasticsearchTestCase):
def setUp(self):
super(TestReindex, self).setUp()
def setup_method(self, _):
bulk = []
for x in range(100):
bulk.append({"index": {"_index": "test_index", "_type": "_doc", "_id": x}})
@@ -569,8 +567,7 @@ class TestReindex(ElasticsearchTestCase):
class TestParentChildReindex(ElasticsearchTestCase):
def setUp(self):
super(TestParentChildReindex, self).setUp()
def setup_method(self, _):
body = {
"settings": {"number_of_shards": 1, "number_of_replicas": 0},
"mappings": {
test_elasticsearch/test_server/test_common.py → test_elasticsearch/test_server/test_rest_api_spec.py
+190 -202
View File
@@ -3,25 +3,22 @@
# See the LICENSE file in the project root for more information
"""
Dynamically generated set of TestCases based on set of yaml files decribing
Dynamically generated set of TestCases based on set of yaml files describing
some integration tests. These files are shared among all official Elasticsearch
clients.
"""
import sys
import re
from os import walk, environ
from os.path import exists, join, dirname, pardir
from os.path import exists, join, dirname, pardir, relpath
import yaml
from shutil import rmtree
import warnings
import pytest
from elasticsearch import TransportError, RequestError, ElasticsearchDeprecationWarning
from elasticsearch.compat import string_types
from elasticsearch.helpers.test import _get_version
from ..test_cases import SkipTest
from . import ElasticsearchTestCase
# some params had to be changed in python, keep track of them so we can rename
# those in the tests accordingly
PARAMS_RENAMES = {"type": "doc_type", "from": "from_"}
@@ -41,41 +38,57 @@ IMPLEMENTED_FEATURES = {
# broken YAML tests on some releases
SKIP_TESTS = {
"*": {
# Can't figure out the get_alias(expand_wildcards=open) failure.
"TestIndicesGetAlias10Basic",
# Disallowing expensive queries is 7.7+
"TestSearch320DisallowQueries",
# Order of overlapped templates isn't consistent
"TestIndicesSimulateIndexTemplate10Basic",
}
# can't figure out the expand_wildcards=open issue?
"indices/get_alias/10_basic[23]",
# [interval] on [date_histogram] is deprecated, use [fixed_interval] or [calendar_interval] in the future.
"search/aggregation/230_composite[6]",
"search/aggregation/250_moving_fn[1]",
# fails by not returning 'search'?
"search/320_disallow_queries[2]",
"search/40_indices_boost[1]",
# ?q= fails
"explain/30_query_string[0]",
"count/20_query_string[0]",
# index template issues
"indices/put_template/10_basic[0]",
"indices/put_template/10_basic[1]",
"indices/put_template/10_basic[2]",
"indices/put_template/10_basic[3]",
"indices/put_template/10_basic[4]",
# depends on order of response JSON which is random
"indices/simulate_index_template/10_basic[1]",
}
# Test is inconsistent due to dictionaries not being ordered.
if sys.version_info < (3, 6):
SKIP_TESTS["*"].add("TestSearchAggregation250MovingFn")
XPACK_FEATURES = None
ES_VERSION = None
class InvalidActionType(Exception):
pass
class YamlTestCase(ElasticsearchTestCase):
def setUp(self):
super(YamlTestCase, self).setUp()
if hasattr(self, "_setup_code"):
self.run_code(self._setup_code)
class YamlRunner:
def __init__(self, client):
self.client = client
self.last_response = None
self._run_code = None
self._setup_code = None
self._teardown_code = None
self._state = {}
def tearDown(self):
if hasattr(self, "_teardown_code"):
def use_spec(self, test_spec):
self._setup_code = test_spec.pop("setup", None)
self._run_code = test_spec.pop("run", None)
self._teardown_code = test_spec.pop("teardown")
def setup(self):
if self._setup_code:
self.run_code(self._setup_code)
def teardown(self):
if self._teardown_code:
self.run_code(self._teardown_code)
for repo, definition in self.client.snapshot.get_repository(
repository="_all"
for repo, definition in (
self.client.snapshot.get_repository(repository="_all")
).items():
self.client.snapshot.delete_repository(repository=repo)
if definition["type"] == "fs":
@@ -86,86 +99,45 @@ class YamlTestCase(ElasticsearchTestCase):
# stop and remove all ML stuff
if self._feature_enabled("ml"):
self.client.ml.stop_datafeed(datafeed_id="*", force=True)
for feed in self.client.ml.get_datafeeds(datafeed_id="*")["datafeeds"]:
for feed in (self.client.ml.get_datafeeds(datafeed_id="*"))["datafeeds"]:
self.client.ml.delete_datafeed(datafeed_id=feed["datafeed_id"])
self.client.ml.close_job(job_id="*", force=True)
for job in self.client.ml.get_jobs(job_id="*")["jobs"]:
for job in (self.client.ml.get_jobs(job_id="*"))["jobs"]:
self.client.ml.delete_job(
job_id=job["job_id"], wait_for_completion=True, force=True
)
# stop and remove all Rollup jobs
if self._feature_enabled("rollup"):
for rollup in self.client.rollup.get_jobs(id="*")["jobs"]:
for rollup in (self.client.rollup.get_jobs(id="*"))["jobs"]:
self.client.rollup.stop_job(
id=rollup["config"]["id"], wait_for_completion=True
)
self.client.rollup.delete_job(id=rollup["config"]["id"])
super(YamlTestCase, self).tearDown()
def es_version(self):
global ES_VERSION
if ES_VERSION is None:
version_string = (self.client.info())["version"]["number"]
if "." not in version_string:
return ()
version = version_string.strip().split(".")
ES_VERSION = tuple(int(v) if v.isdigit() else 999 for v in version)
return ES_VERSION
def _feature_enabled(self, name):
global XPACK_FEATURES, IMPLEMENTED_FEATURES
if XPACK_FEATURES is None:
try:
xinfo = self.client.xpack.info()
XPACK_FEATURES = set(
f for f in xinfo["features"] if xinfo["features"][f]["enabled"]
)
IMPLEMENTED_FEATURES.add("xpack")
except RequestError as e:
# We receive a 'RequestError' here when using 'oss' because the request
# for xpack.info() looks like this: 'GET /_xpack' which errors on the
# oss container due to having a similar route to 'GET /<index>'.
if "invalid_index_name_exception" not in e.error:
raise
XPACK_FEATURES = set()
IMPLEMENTED_FEATURES.add("no_xpack")
return name in XPACK_FEATURES
def _resolve(self, value):
# resolve variables
if isinstance(value, string_types) and value.startswith("$"):
value = value[1:]
self.assertIn(value, self._state)
value = self._state[value]
if isinstance(value, string_types):
value = value.strip()
elif isinstance(value, dict):
value = dict((k, self._resolve(v)) for (k, v) in value.items())
elif isinstance(value, list):
value = list(map(self._resolve, value))
return value
def _lookup(self, path):
# fetch the possibly nested value from last_response
value = self.last_response
if path == "$body":
return value
path = path.replace(r"\.", "\1")
for step in path.split("."):
if not step:
continue
step = step.replace("\1", ".")
step = self._resolve(step)
if step.isdigit() and step not in value:
step = int(step)
self.assertIsInstance(value, list)
self.assertGreater(len(value), step)
else:
self.assertIn(step, value)
value = value[step]
return value
def run(self):
try:
self.setup()
self.run_code(self._run_code)
finally:
self.teardown()
def run_code(self, test):
""" Execute an instruction based on it's type. """
print(test)
for action in test:
self.assertEqual(1, len(action))
assert len(action) == 1
action_type, action = list(action.items())[0]
if hasattr(self, "run_" + action_type):
@@ -174,19 +146,18 @@ class YamlTestCase(ElasticsearchTestCase):
raise InvalidActionType(action_type)
def run_do(self, action):
""" Perform an api call with given parameters. """
api = self.client
headers = action.pop("headers", None)
catch = action.pop("catch", None)
warn = action.pop("warnings", None)
self.assertEqual(1, len(action))
warn = action.pop("warnings", ())
assert len(action) == 1
method, args = list(action.items())[0]
args["headers"] = headers
# locate api endpoint
for m in method.split("."):
self.assertTrue(hasattr(api, m))
assert hasattr(api, m)
api = getattr(api, m)
# some parameters had to be renamed to not clash with python builtins,
@@ -228,32 +199,24 @@ class YamlTestCase(ElasticsearchTestCase):
% (warn, caught_warnings)
)
def _get_nodes(self):
if not hasattr(self, "_node_info"):
self._node_info = list(
self.client.nodes.info(node_id="_all", metric="clear")["nodes"].values()
)
return self._node_info
def run_catch(self, catch, exception):
if catch == "param":
assert isinstance(exception, TypeError)
return
def _get_data_nodes(self):
return len(
[
info
for info in self._get_nodes()
if info.get("attributes", {}).get("data", "true") == "true"
]
)
def _get_benchmark_nodes(self):
return len(
[
info
for info in self._get_nodes()
if info.get("attributes", {}).get("bench", "false") == "true"
]
)
assert isinstance(exception, TransportError)
if catch in CATCH_CODES:
assert CATCH_CODES[catch] == exception.status_code
elif catch[0] == "/" and catch[-1] == "/":
assert (
re.search(catch[1:-1], exception.error + " " + repr(exception.info)),
"%s not in %r" % (catch, exception.info),
) is not None
self.last_response = exception.info
def run_skip(self, skip):
global IMPLEMENTED_FEATURES
if "features" in skip:
features = skip["features"]
if not isinstance(features, (tuple, list)):
@@ -261,58 +224,37 @@ class YamlTestCase(ElasticsearchTestCase):
for feature in features:
if feature in IMPLEMENTED_FEATURES:
continue
elif feature == "requires_replica":
if self._get_data_nodes() > 1:
continue
elif feature == "benchmark":
if self._get_benchmark_nodes():
continue
raise SkipTest("Feature %s is not supported" % feature)
pytest.skip("feature '%s' is not supported" % feature)
if "version" in skip:
version, reason = skip["version"], skip["reason"]
if version == "all":
raise SkipTest(reason)
pytest.skip(reason)
min_version, max_version = version.split("-")
min_version = _get_version(min_version) or (0,)
max_version = _get_version(max_version) or (999,)
if min_version <= self.es_version <= max_version:
raise SkipTest(reason)
def run_catch(self, catch, exception):
if catch == "param":
self.assertIsInstance(exception, TypeError)
return
self.assertIsInstance(exception, TransportError)
if catch in CATCH_CODES:
self.assertEqual(CATCH_CODES[catch], exception.status_code)
elif catch[0] == "/" and catch[-1] == "/":
self.assertTrue(
re.search(catch[1:-1], exception.error + " " + repr(exception.info)),
"%s not in %r" % (catch, exception.info),
)
self.last_response = exception.info
if min_version <= (self.es_version()) <= max_version:
pytest.skip(reason)
def run_gt(self, action):
for key, value in action.items():
value = self._resolve(value)
self.assertGreater(self._lookup(key), value)
assert self._lookup(key) > value
def run_gte(self, action):
for key, value in action.items():
value = self._resolve(value)
self.assertGreaterEqual(self._lookup(key), value)
assert self._lookup(key) >= value
def run_lt(self, action):
for key, value in action.items():
value = self._resolve(value)
self.assertLess(self._lookup(key), value)
assert self._lookup(key) < value
def run_lte(self, action):
for key, value in action.items():
value = self._resolve(value)
self.assertLessEqual(self._lookup(key), value)
assert self._lookup(key) <= value
def run_set(self, action):
for key, value in action.items():
@@ -325,17 +267,17 @@ class YamlTestCase(ElasticsearchTestCase):
except AssertionError:
pass
else:
self.assertIn(value, ("", None, False, 0))
assert value in ("", None, False, 0)
def run_is_true(self, action):
value = self._lookup(action)
self.assertNotIn(value, ("", None, False, 0))
assert value not in ("", None, False, 0)
def run_length(self, action):
for path, expected in action.items():
value = self._lookup(path)
expected = self._resolve(expected)
self.assertEqual(expected, len(value))
assert expected == len(value)
def run_match(self, action):
for path, expected in action.items():
@@ -347,51 +289,65 @@ class YamlTestCase(ElasticsearchTestCase):
and expected.startswith("/")
and expected.endswith("/")
):
expected = re.compile(expected[1:-1], re.VERBOSE)
self.assertTrue(expected.search(value))
expected = re.compile(expected[1:-1], re.VERBOSE | re.MULTILINE)
assert expected.search(value), "%r does not match %r" % (
value,
expected,
)
else:
self.assertEqual(expected, value)
assert expected == value, "%r does not match %r" % (value, expected)
def _resolve(self, value):
# resolve variables
if isinstance(value, string_types) and value.startswith("$"):
value = value[1:]
assert value in self._state
value = self._state[value]
if isinstance(value, string_types):
value = value.strip()
elif isinstance(value, dict):
value = dict((k, self._resolve(v)) for (k, v) in value.items())
elif isinstance(value, list):
value = list(map(self._resolve, value))
return value
def construct_case(filename, name):
"""
Parse a definition of a test case from a yaml file and construct the
TestCase subclass dynamically.
"""
def make_test(test_name, definition, i):
def m(self):
if name in SKIP_TESTS.get(self.es_version, ()) or name in SKIP_TESTS.get(
"*", ()
):
raise SkipTest()
self.run_code(definition)
m.__doc__ = "%s:%s.test_from_yaml_%d (%s): %s" % (
__name__,
name,
i,
"/".join(filename.split("/")[-2:]),
test_name,
)
m.__name__ = "test_from_yaml_%d" % i
return m
with open(filename) as f:
tests = list(yaml.load_all(f))
attrs = {"_yaml_file": filename}
i = 0
for test in tests:
for test_name, definition in test.items():
if test_name in ("setup", "teardown"):
attrs["_%s_code" % test_name] = definition
def _lookup(self, path):
# fetch the possibly nested value from last_response
value = self.last_response
if path == "$body":
return value
path = path.replace(r"\.", "\1")
for step in path.split("."):
if not step:
continue
step = step.replace("\1", ".")
step = self._resolve(step)
if step.isdigit() and step not in value:
step = int(step)
assert isinstance(value, list)
assert len(value) > step
else:
assert step in value
value = value[step]
return value
attrs["test_from_yaml_%d" % i] = make_test(test_name, definition, i)
i += 1
def _feature_enabled(self, name):
global XPACK_FEATURES, IMPLEMENTED_FEATURES
if XPACK_FEATURES is None:
try:
xinfo = self.client.xpack.info()
XPACK_FEATURES = set(
f for f in xinfo["features"] if xinfo["features"][f]["enabled"]
)
IMPLEMENTED_FEATURES.add("xpack")
except RequestError:
XPACK_FEATURES = set()
IMPLEMENTED_FEATURES.add("no_xpack")
return name in XPACK_FEATURES
return type(name, (YamlTestCase,), attrs)
class InvalidActionType(Exception):
pass
YAML_DIR = environ.get(
@@ -412,21 +368,53 @@ YAML_DIR = environ.get(
)
YAML_TEST_SPECS = []
if exists(YAML_DIR):
# find all the test definitions in yaml files ...
for (path, dirs, files) in walk(YAML_DIR):
for path, _, files in walk(YAML_DIR):
for filename in files:
if not filename.endswith((".yaml", ".yml")):
continue
# ... parse them
name = (
(
"Test"
+ "".join(s.title() for s in path[len(YAML_DIR) + 1 :].split("/"))
+ filename.rsplit(".", 1)[0].title()
)
.replace("_", "")
.replace(".", "")
)
# and insert them into locals for test runner to find them
locals()[name] = construct_case(join(path, filename), name)
filepath = join(path, filename)
with open(filepath) as f:
tests = list(yaml.load_all(f, Loader=yaml.SafeLoader))
setup_code = None
teardown_code = None
run_codes = []
for i, test in enumerate(tests):
for test_name, definition in test.items():
if test_name == "setup":
setup_code = definition
elif test_name == "teardown":
teardown_code = definition
else:
run_codes.append((i, definition))
for i, run_code in run_codes:
src = {"setup": setup_code, "run": run_code, "teardown": teardown_code}
# Pytest already replaces '.' and '_' with '/' so we do
# it ourselves so UI and 'SKIP_TESTS' match.
pytest_param_id = (
"%s[%d]" % (relpath(filepath, YAML_DIR).rpartition(".")[0], i)
).replace(".", "/")
if pytest_param_id in SKIP_TESTS:
src["skip"] = True
YAML_TEST_SPECS.append(pytest.param(src, id=pytest_param_id))
@pytest.fixture(scope="function")
def sync_runner(sync_client):
return YamlRunner(sync_client)
@pytest.mark.parametrize("test_spec", YAML_TEST_SPECS)
def test_rest_api_spec(test_spec, sync_runner):
if test_spec.get("skip", False):
pytest.skip("Manually skipped in 'SKIP_TESTS'")
sync_runner.use_spec(test_spec)
sync_runner.run()