import argparse
import datetime
import pytest
from _pytest.doctest import DoctestItem
from _pytest.mark import Mark, MarkDecorator
from cache import patch_cache_set
from aws.client import BotocoreClient
from gcp.client import GCPClient
from gsuite.client import GsuiteClient
import custom_config
botocore_client = None
gcp_client = None
gsuite_client = None
custom_config_global = None
[docs]def pytest_addoption(parser):
frost_parser = parser.getgroup("Frost", "Frost's custom arguments")
frost_parser.addoption(
"--aws-profiles",
nargs="*",
help="Set default AWS profiles to use. Defaults to the current AWS profile i.e. [None].",
)
frost_parser.addoption(
"--aws-regions",
type=str,
help="Set AWS regions to use as a comma separate list. Defaults to all available AWS regions",
)
frost_parser.addoption(
"--gcp-project-id", type=str, help="Set GCP project to test.",
)
frost_parser.addoption(
"--gcp-folder-id",
type=str,
help="Set GCP folder to test. Will test all projects under this folder.",
)
# While only used for Heroku at the moment, GitHub tests are soon to be
# added, which will also need an "organization" option. Current plan is to
# reuse this one.
frost_parser.addoption(
"--organization",
type=str,
help="Set organization to test. Used for Heroku tests.",
)
frost_parser.addoption(
"--debug-calls", action="store_true", help="Log API calls. Requires -s"
)
frost_parser.addoption(
"--debug-cache",
action="store_true",
help="Log whether API calls hit the cache. Requires -s",
)
frost_parser.addoption(
"--offline",
action="store_true",
default=False,
help="Instruct service clients to return empty lists and not make HTTP requests.",
)
frost_parser.addoption(
"--config", type=argparse.FileType("r"), help="Path to the config file."
)
[docs]@pytest.fixture
def aws_config(pytestconfig):
return pytestconfig.custom_config.aws
[docs]@pytest.fixture
def gcp_config(pytestconfig):
return pytestconfig.custom_config.gcp
[docs]def pytest_runtest_setup(item):
"""
Add custom markers to pytest tests.
"""
if not isinstance(item, DoctestItem):
item.config.custom_config.add_markers(item)
# Reporting
[docs]def get_node_markers(node):
return [m for m in node.iter_markers()]
# METADATA_KEYS are modified by services to specify which metadata is
# relevant for the JSON output. It's unlikely that duplicate keys are
# intended, so failfast
# adapted from
# https://stackoverflow.com/questions/41281346/how-to-raise-error-if-user-tries-to-enter-duplicate-entries-in-a-set-in-python/41281734#41281734
[docs]class DuplicateKeyError(Exception):
pass
[docs]class SingleSet(set):
"""Set only allowing values to be added once
When addition of a duplicate value is detected, the `DuplicateKeyError`
exception will be raised, all non duplicate values are added to the set.
Raises:
DuplicateKeyError - when adding a value already in the set
>>> ss = SingleSet({1, 2, 3, 4})
>>> ss.add(3)
Traceback (most recent call last):
...
conftest.DuplicateKeyError: Value 3 already present
>>> ss.update({4, 5, 6, 3})
Traceback (most recent call last):
...
conftest.DuplicateKeyError: Value(s) {3, 4} already present
>>> ss
SingleSet({1, 2, 3, 4, 5, 6})
>>>
**NB:**
- duplicate values on initialization are not detected
>>> ss = SingleSet({1, 2, 3, 4, 3, 2, 1})
>>> ss
SingleSet({1, 2, 3, 4})
"""
[docs] def add(self, value):
if value in self:
raise DuplicateKeyError("Value {!r} already present".format(value))
super().add(value)
[docs] def update(self, values):
error_values = set()
for value in values:
if value in self:
error_values.add(value)
if error_values:
# we want the non-duplicate values added
super().update(values - error_values)
raise DuplicateKeyError(
"Value(s) {!r} already present".format(error_values)
)
super().update(values)
METADATA_KEYS: SingleSet = SingleSet(
[
"DBInstanceArn",
"DBInstanceIdentifier",
"GroupId",
"ImageId",
"InstanceId",
"LaunchTime",
"OwnerId",
"TagList",
"Tags",
"UserName",
"VolumeId",
"VpcId",
"__pytest_meta",
"displayName",
"id",
"kind",
"members",
"name",
"project",
"projectId",
"role",
"uniqueId",
]
)
[docs]def serialize_datetimes(obj):
"""Serializes datetimes to ISO format strings.
Used on report test_metadata since pytest-json doesn't let us pass
options to the serializer.
>>> from datetime import datetime
>>> serialize_datetimes({datetime(2000, 1, 1): -1})
{'2000-01-01T00:00:00': -1}
>>> serialize_datetimes({'foo': datetime(2000, 1, 1)})
{'foo': '2000-01-01T00:00:00'}
"""
if isinstance(obj, datetime.datetime):
return obj.isoformat()
elif isinstance(obj, list):
return [serialize_datetimes(item) for item in obj]
elif isinstance(obj, dict):
new_obj = {}
for k, v in obj.items():
new_obj[serialize_datetimes(k)] = serialize_datetimes(v)
return new_obj
return obj
[docs]def serialize_marker(marker):
if isinstance(marker, (MarkDecorator, Mark)):
args = ["...skipped..."] if marker.name == "parametrize" else marker.args
kwargs = ["...skipped..."] if marker.name == "parametrize" else marker.kwargs
return {"name": marker.name, "args": args, "kwargs": kwargs}
else:
raise NotImplementedError("Unexpected Marker type %s" % repr(marker))
[docs]def get_outcome_and_reason(report, markers, call):
xfail = "xfail" in markers
xpass = report.passed and xfail
if (
call.excinfo
and not isinstance(call.excinfo, AssertionError)
and isinstance(call.excinfo, Exception)
):
return "errored", call.excinfo
elif xpass:
return "xpassed", markers["xfail"]["kwargs"].get("reason", None)
elif xfail:
return "xfailed", markers["xfail"]["kwargs"].get("reason", None)
else:
return report.outcome, None # passed, failed, skipped
[docs]def clean_docstring(docstr):
"""
Transforms a docstring into a properly formatted single line string.
>>> clean_docstring("\\nfoo\\n bar\\n")
'foo bar'
>>> clean_docstring("foo bar")
'foo bar'
"""
return " ".join(
[word for word in docstr.replace("\n", " ").strip().split(" ") if word != ""]
)
[docs]@pytest.mark.hookwrapper
def pytest_runtest_makereport(item, call):
outcome = yield
report = outcome.get_result()
# only add this during call instead of during any stage
if report.when == "call" and not isinstance(item, DoctestItem):
metadata = get_metadata_from_funcargs(item.funcargs)
markers = {n.name: serialize_marker(n) for n in get_node_markers(item)}
severity = markers.get("severity", None) and markers.get("severity")["args"][0]
regression = (
markers.get("regression", None) and markers.get("regression")["args"][0]
)
outcome, reason = get_outcome_and_reason(report, markers, call)
rationale = markers.get("rationale", None) and clean_docstring(
markers.get("rationale")["args"][0]
)
description = item._obj.__doc__ and clean_docstring(item._obj.__doc__)
# add json metadata
report.test_metadata = serialize_datetimes(
dict(
description=description,
markers=markers,
metadata=metadata,
outcome=outcome, # 'passed', 'failed', 'skipped', 'xfailed', 'xskipped', or 'errored'
parametrized_name=item.name,
rationale=rationale,
reason=reason,
severity=severity,
regression=regression,
unparametrized_name=item.originalname,
)
)