This is an automated email from the ASF dual-hosted git repository.
mitchell852 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/trafficcontrol.git
The following commit(s) were added to refs/heads/master by this push:
new 8d364358c9 Added API Contract test case for servers endpoint and
refactored existing ones. (#7523)
8d364358c9 is described below
commit 8d364358c97f9cb4c4d99d7b4192ebd2377a4f94
Author: Gokula Krishnan <[email protected]>
AuthorDate: Thu May 18 01:57:30 2023 +0530
Added API Contract test case for servers endpoint and refactored existing
ones. (#7523)
* Refactored conftest.py and replaced request and response templates with
json-schema formats
* Refactored cachegroups, cdns, parameters, profiles, roles,
server_capabilities and tenants test cases
* Refactored divisions, regions and physical locations test cases
* Added servers test case
* Modified requirements file
* modified code
* Addressed review comments
* Re added version in tosession
---
.../clients/python/trafficops/tosession.py | 2 +-
traffic_ops/testing/api_contract/v4/conftest.py | 555 +++++++++----------
.../api_contract/v4/data/request_template.json | 138 +++++
.../api_contract/v4/data/response_template.json | 592 +++++++++++++++++++++
.../api_contract/v4/{ => data}/to_data.json | 0
.../testing/api_contract/v4/request_template.json | 86 ---
.../testing/api_contract/v4/requirements.txt | 1 +
.../testing/api_contract/v4/response_template.json | 268 ----------
.../testing/api_contract/v4/test_cachegroups.py | 97 ++--
traffic_ops/testing/api_contract/v4/test_cdns.py | 57 +-
.../testing/api_contract/v4/test_divisions.py | 55 +-
.../testing/api_contract/v4/test_parameters.py | 59 +-
.../testing/api_contract/v4/test_phys_locations.py | 75 +--
.../testing/api_contract/v4/test_profiles.py | 55 +-
.../testing/api_contract/v4/test_regions.py | 60 +--
traffic_ops/testing/api_contract/v4/test_roles.py | 54 +-
.../api_contract/v4/test_server_capabilities.py | 61 +--
.../testing/api_contract/v4/test_servers.py | 84 +++
.../testing/api_contract/v4/test_tenants.py | 58 +-
19 files changed, 1265 insertions(+), 1092 deletions(-)
diff --git a/traffic_control/clients/python/trafficops/tosession.py
b/traffic_control/clients/python/trafficops/tosession.py
index 14b17dbc9a..19ed6d8558 100644
--- a/traffic_control/clients/python/trafficops/tosession.py
+++ b/traffic_control/clients/python/trafficops/tosession.py
@@ -1396,7 +1396,7 @@ class TOSession(RestApiSession):
"""
@api_request('post', 'phys_locations', ('3.0', '3.1', '4.0', '4.1',
'5.0'))
- def create_physical_locations(self, query_params=None, data=None):
+ def create_physical_locations(self, data=None):
"""
Create a Physical Location
:ref:`to-api-phys_locations`
diff --git a/traffic_ops/testing/api_contract/v4/conftest.py
b/traffic_ops/testing/api_contract/v4/conftest.py
index ab74e6ab87..a44520e935 100644
--- a/traffic_ops/testing/api_contract/v4/conftest.py
+++ b/traffic_ops/testing/api_contract/v4/conftest.py
@@ -22,7 +22,7 @@ import logging
import sys
import os
from random import randint
-from typing import NamedTuple, Union, Optional, TypeAlias
+from typing import Any, NamedTuple, Union, Optional, TypeAlias
from urllib.parse import urlparse
import pytest
@@ -34,9 +34,9 @@ from trafficops.restapi import OperationError
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
-JSONData: TypeAlias = Union[dict[str, object], list[object], bool, int, float,
str | None]
+JSONData: TypeAlias = Union[dict[str, object], list[object], bool, int, float,
Optional[str]]
JSONData.__doc__ = """An alias for the kinds of data that JSON can encode."""
class APIVersion(NamedTuple):
@@ -124,23 +124,23 @@ def pytest_addoption(parser: pytest.Parser) -> None:
parser.addoption(
"--config",
help="Path to configuration file.",
- default=os.path.join(os.path.dirname(__file__), "to_data.json")
+ default=os.path.join(os.path.dirname(__file__), "data",
"to_data.json")
)
parser.addoption(
"--request-template",
help="Path to request prerequisites file.",
- default=os.path.join(os.path.dirname(__file__),
"request_template.json")
+ default=os.path.join(os.path.dirname(__file__), "data",
"request_template.json")
)
parser.addoption(
"--response-template",
help="Path to response prerequisites file.",
- default=os.path.join(os.path.dirname(__file__),
"response_template.json")
+ default=os.path.join(os.path.dirname(__file__), "data",
"response_template.json")
)
def coalesce_config(
- arg: object | None,
+ arg: Optional[object],
file_key: str,
- file_contents: dict[str, object | None] | None,
+ file_contents: Optional[dict[str, Optional[object]]],
env_key: str
) -> Optional[str]:
"""
@@ -176,7 +176,7 @@ def parse_to_url(raw: str) -> tuple[APIVersion, int]:
"""
Parses the API version and port number from a raw URL string.
- >>> parse_to_url("https://trafficops.example.test:420/api/5.270)
+ >>> parse_to_url("https://trafficops.example.test:420/api/5.270";)
(APIVersion(major=5, minor=270), 420)
>>> parse_to_url("trafficops.example.test")
(APIVersion(major=4, minor=0), 443)
@@ -302,7 +302,7 @@ def to_login(to_args: ArgsType) -> TOSession:
@pytest.fixture(name="request_template_data", scope="session")
def request_prerequiste_data(pytestconfig: pytest.Config, request:
pytest.FixtureRequest
- ) -> list[dict[str, object] | list[object] |
primitive]:
+ ) -> list[Union[dict[str, object], list[object],
Primitive]]:
"""
PyTest Fixture to store POST request template data for api endpoint.
:param pytestconfig: Session-scoped fixture that returns the session's
pytest.Config object.
@@ -316,27 +316,27 @@ def request_prerequiste_data(pytestconfig: pytest.Config,
request: pytest.Fixtur
raise ValueError("prereqisites path not configured")
# Response keys for api endpoint
- data: dict[
+ data: Union[dict[
str,
- list[dict[str, object] | list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ] |\
- primitive = None
+ Union[list[Union[dict[str, object], list[object], Primitive]],
dict[object, object], Primitive]
+ ], Primitive] = None
with open(request_template_path, encoding="utf-8", mode="r") as
prereq_file:
data = json.load(prereq_file)
if not isinstance(data, dict):
raise TypeError(f"request template data must be an object, not
'{type(data)}'")
- request_template = data[request.param]
- if not isinstance(request_template, list):
- raise TypeError(f"Request template data must be a list, not
'{type(request_template)}'")
-
+ try:
+ request_template = data[request.param]
+ if not isinstance(request_template, list):
+ raise TypeError(f"Request template data must be a list,
not '{type(request_template)}'")
+ except AttributeError:
+ request_template = data
return request_template
+
@pytest.fixture()
def response_template_data(pytestconfig: pytest.Config
- ) -> dict[str, primitive | list[primitive |
- dict[str, object] | list[object]] |
dict[object, object]]:
+ ) -> dict[str, Union[Primitive, list[
+ Union[Primitive, dict[str, object], list[object]]], dict[object,
object]]]:
"""
PyTest Fixture to store response template data for api endpoint.
:param pytestconfig: Session-scoped fixture that returns the session's
pytest.Config object.
@@ -348,13 +348,10 @@ def response_template_data(pytestconfig: pytest.Config
raise ValueError("prereqisites path not configured")
# Response keys for api endpoint
- response_template: dict[
+ response_template: Union[dict[
str,
- list[dict[str, object] | list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ] |\
- primitive = None
+ Union[list[Union[dict[str, object], list[object], Primitive]],
dict[object, object], Primitive]
+ ], Primitive] = None
with open(prereq_path, encoding="utf-8", mode="r") as prereq_file:
response_template = json.load(prereq_file)
if not isinstance(response_template, dict):
@@ -363,6 +360,97 @@ def response_template_data(pytestconfig: pytest.Config
return response_template
+def api_response_data(api_response: tuple[Union[Primitive, dict[str, object],
list[
+ Union[Primitive, dict[str, object], list[
+ object]]]], requests.Response]) -> dict[str, object]:
+ """
+ Checks API get/post response.
+ :param api_response: Raw api response.
+ :returns: Verified response data
+ """
+ api_data = None
+ if isinstance(api_response, tuple):
+ api_response = api_response[0]
+ if not isinstance(api_response, list):
+ raise ValueError("Malformed API response; 'response'
property not an array")
+ else:
+ raise ValueError("Invalid API response format")
+
+ if api_response:
+ try:
+ api_data = api_response[0]
+ if not isinstance(api_data, dict):
+ raise ValueError("Malformed API response;
'response' property not a dict")
+ except IndexError as e:
+ raise ValueError(f"No response data from API request
'{e.args[0]}'") from e
+
+ return api_data
+
+
+def get_existing_object(to_session: TOSession, object_type: str, query_params:
Optional[
+ dict[str, Any]])-> Union[dict[str, Any], None]:
+ """
+ Check if the given endpoint with the given query params already exists.
+ :param to_session: Fixture to get Traffic Ops session.
+ :param object_type: api call name for get request.
+ :param query_params: query params for api get request.
+ :returns: Api data for the corresponding api request.
+ """
+ api_get_response: tuple[Union[dict[str, object], list[
+ Union[dict[str, object], list[object], Primitive]], Primitive],
requests.Response] = getattr(
+ to_session, f"get_{object_type}")(query_params=query_params)
+ return api_response_data(api_get_response)
+
+
+def create_if_not_exists(to_session: TOSession, object_type: str,
+ data: dict[str, Any]) -> Union[dict[str, Any], None]:
+ """
+ Hits Post request of the given endpoint with the given data.
+ :param to_session: Fixture to get Traffic Ops session.
+ :param object_type: api call name for post request.
+ :param data: Post data for api post request.
+ :returns: Api data for the corresponding api request.
+ """
+ api_post_response: tuple[Union[dict[str, object], list[
+ Union[dict[str, object], list[object], Primitive]], Primitive],
requests.Response] = getattr(
+ to_session, f"create_{object_type}")(data=data)
+ return api_response_data(api_post_response)
+
+
+def create_or_get_existing(to_session: TOSession, get_object_type: str,
post_object_type: str, data:
+ dict[str, Any], query_params: Optional[dict[str, Any]] = None) ->
Union[dict[str, Any], None]:
+ """
+ Get Api data of the given endpoint with the given query params if it
exists. If not, create it.
+ :param to_session: Fixture to get Traffic Ops session.
+ :param get_object_type: api call name for get request.
+ :param post_object_type: api call name for post request.
+ :param query_params: query params for api get request.
+ :returns: Api data for the corresponding api request.
+ @param data:
+ """
+ existing_object = get_existing_object(to_session, get_object_type,
query_params)
+ return existing_object or create_if_not_exists(to_session,
post_object_type, data)
+
+
+def check_template_data(template_data: Union[list[JSONData], tuple[JSONData,
requests.Response]],
+ name: str) -> dict[str, object]:
+ """
+ Checks API request/response template data.
+ :param template_data: Fixture to get template data from a prerequisites
file.
+ :param name: Endpoint name
+ :returns: Verified endpoint data
+ """
+ try:
+ endpoint = template_data[0]
+ except IndexError as e:
+ raise TypeError(
+ f"malformed data; no {name} present in {name} array
property") from e
+
+ if not isinstance(endpoint, dict):
+ raise TypeError(f"malformed data; {name} must be objects, not
'{type(endpoint)}'")
+ return endpoint
+
+
@pytest.fixture()
def cdn_post_data(to_session: TOSession, request_template_data: list[JSONData]
) -> dict[str, object]:
@@ -373,14 +461,7 @@ def cdn_post_data(to_session: TOSession,
request_template_data: list[JSONData]
:param request_template_data: Fixture to get CDN request template data
from a prerequisites file.
:returns: Sample POST data and the actual API response.
"""
-
- try:
- cdn = request_template_data[0]
- except IndexError as e:
- raise TypeError("malformed prerequisite data; no CDNs present
in 'cdns' array property") from e
-
- if not isinstance(cdn, dict):
- raise TypeError(f"malformed prerequisite data; CDNs must be
objects, not '{type(cdn)}'")
+ cdn = check_template_data(request_template_data["cdns"], "cdns")
# Return new post data and post response from cdns POST request
randstr = str(randint(0, 1000))
@@ -396,21 +477,15 @@ def cdn_post_data(to_session: TOSession,
request_template_data: list[JSONData]
except KeyError as e:
raise TypeError(f"missing CDN property '{e.args[0]}'") from e
- logger.info("New cdn data to hit POST method %s", request_template_data)
+ logger.info("New cdn data to hit POST method %s", cdn)
# Hitting cdns POST methed
response: tuple[JSONData, requests.Response] =
to_session.create_cdn(data=cdn)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; cdn is not an
object")
- return resp_obj
- except IndexError:
- logger.error("No CDN response data from cdns POST request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "cdns")
+ return resp_obj
@pytest.fixture()
-def cachegroup_post_data(to_session: TOSession, request_template_data:
list[JSONData]
+def cache_group_post_data(to_session: TOSession, request_template_data:
list[JSONData]
) -> dict[str, object]:
"""
PyTest Fixture to create POST data for cachegroup endpoint.
@@ -419,58 +494,32 @@ def cachegroup_post_data(to_session: TOSession,
request_template_data: list[JSON
:param request_template_data: Fixture to get Cachegroup data from a
prerequisites file.
:returns: Sample POST data and the actual API response.
"""
- try:
- cachegroup = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no Cache group present in
'cachegroup' array property") from e
-
- if not isinstance(cachegroup, dict):
- raise TypeError(
- f"malformed prerequisite data; Cache group must be
objects, not '{type(cachegroup)}'")
-
+ cache_group = check_template_data(request_template_data["cachegroup"],
"cachegroup")
# Return new post data and post response from cachegroups POST request
randstr = str(randint(0, 1000))
try:
- name = cachegroup["name"]
+ name = cache_group["name"]
if not isinstance(name, str):
raise TypeError(f"name must be str, not '{type(name)}'")
- cachegroup["name"] = name[:4] + randstr
- short_name = cachegroup["shortName"]
+ cache_group["name"] = name[:4] + randstr
+ short_name = cache_group["shortName"]
if not isinstance(short_name, str):
raise TypeError(f"shortName must be str, not
'{type(short_name)}")
- cachegroup["shortName"] = short_name[:5] + randstr
+ cache_group["shortName"] = short_name[:5] + randstr
except KeyError as e:
raise TypeError(f"missing Cache group property '{e.args[0]}'")
from e
- # Hitting types GET method to access typeID for cachegroup POST data
- type_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
- requests.Response
- ] = to_session.get_types(query_params={"useInTable": "cachegroup"})
- try:
- type_data = type_get_response[0]
- if not isinstance(type_data, list):
- raise TypeError("malformed API response; 'response'
property not an array")
- first_type = type_data[0]
- if not isinstance(first_type, dict):
- raise TypeError("malformed API response; first Type in
response is not an object")
- cachegroup["typeId"] = first_type["id"]
- type_id = cachegroup["typeId"]
- logger.info("extracted %s from %s", type_id, type_get_response)
- except KeyError as e:
- raise TypeError(f"missing Type property '{e.args[0]}'") from e
- logger.info("New cachegroup data to hit POST method %s",
request_template_data)
+ # Check if type already exists, otherwise create it
+ type_data = check_template_data(request_template_data["types"], "types")
+ type_object = create_or_get_existing(to_session, "types", "type",
type_data,
+ {"useInTable": "cachegroup"})
+ cache_group["typeId"] = type_object["id"]
+
+ logger.info("New cachegroup data to hit POST method %s", cache_group)
# Hitting cachegroup POST method
- response: tuple[JSONData, requests.Response] =
to_session.create_cachegroups(data=cachegroup)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; cache group is
not an object")
- return resp_obj
- except IndexError:
- logger.error("No Cache group response data from cdns POST
request.")
- sys.exit(1)
+ response: tuple[JSONData, requests.Response] =
to_session.create_cachegroups(data=cache_group)
+ resp_obj = check_template_data(response, "cachegroup")
+ return resp_obj
@pytest.fixture()
@@ -480,20 +529,10 @@ def parameter_post_data(to_session: TOSession,
request_template_data: list[JSOND
PyTest Fixture to create POST data for parameters endpoint.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get parameter data from a
prerequisites file.
+ :param request_template_data: Fixture to get CDN request template data
from a prerequisites file.
:returns: Sample POST data and the actual API response.
"""
-
- try:
- parameter = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no Parameters present in
'parameters' array property") from e
-
- if not isinstance(parameter, dict):
- raise TypeError(
- f"malformed prerequisite data; Paremeters must be
objects, not '{type(parameter)}'")
-
+ parameter = check_template_data(request_template_data["parameters"],
"parameters")
# Return new post data and post response from parameters POST request
randstr = str(randint(0, 1000))
try:
@@ -508,17 +547,11 @@ def parameter_post_data(to_session: TOSession,
request_template_data: list[JSOND
except KeyError as e:
raise TypeError(f"missing Parameter property '{e.args[0]}'")
from e
- logger.info("New parameter data to hit POST method %s",
request_template_data)
+ logger.info("New parameter data to hit POST method %s", parameter)
# Hitting cdns POST methed
response: tuple[JSONData, requests.Response] =
to_session.create_parameter(data=parameter)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; parameter is
not an object")
- return resp_obj
- except IndexError:
- logger.error("No Parameter response data from parameters POST
request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "parameter")
+ return resp_obj
@pytest.fixture()
@@ -531,16 +564,7 @@ def role_post_data(to_session: TOSession,
request_template_data: list[JSONData]
:param request_template_data: Fixture to get role data from a
prerequisites file.
:returns: Sample POST data and the actual API response.
"""
-
- try:
- role = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no Roles present in
'roles' array property") from e
-
- if not isinstance(role, dict):
- raise TypeError(
- f"malformed prerequisite data; Roles must be objects,
not '{type(role)}'")
+ role = check_template_data(request_template_data["roles"], "roles")
# Return new post data and post response from roles POST request
randstr = str(randint(0, 1000))
@@ -556,23 +580,16 @@ def role_post_data(to_session: TOSession,
request_template_data: list[JSONData]
except KeyError as e:
raise TypeError(f"missing Role property '{e.args[0]}'") from e
- logger.info("New role data to hit POST method %s",
request_template_data)
+ logger.info("New role data to hit POST method %s", role)
# Hitting roles POST methed
response: tuple[JSONData, requests.Response] =
to_session.create_role(data=role)
- logger.info(response[0])
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; role is not an
object")
- return resp_obj
- except IndexError:
- logger.error("No Role response data from roles POST request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "role")
+ return resp_obj
@pytest.fixture()
def profile_post_data(to_session: TOSession, request_template_data:
list[JSONData]
- ) -> dict[str, object]:
+ ) -> dict[str, object]:
"""
PyTest Fixture to create POST data for profile endpoint.
@@ -580,17 +597,8 @@ def profile_post_data(to_session: TOSession,
request_template_data: list[JSONDat
:param request_template_data: Fixture to get profile data from a
prerequisites file.
:returns: Sample POST data and the actual API response.
"""
-
- try:
- profile = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no Profile present in
'profile' array property") from e
-
- if not isinstance(profile, dict):
- raise TypeError(f"malformed prerequisite data; profile must be
objects, not '{type(profile)}'")
-
- # Return new post data and post response from cachegroups POST request
+ profile = check_template_data(request_template_data["profiles"],
"profiles")
+ # Return new post data and post response from profiles POST request
randstr = str(randint(0, 1000))
try:
name = profile["name"]
@@ -599,35 +607,17 @@ def profile_post_data(to_session: TOSession,
request_template_data: list[JSONDat
profile["name"] = name[:4] + randstr
except KeyError as e:
raise TypeError(f"missing Profile property '{e.args[0]}'") from
e
- # Hitting types GET method to access typeID for profile POST data
- cdn_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
- requests.Response
- ] = to_session.get_cdns()
- try:
- cdn_data = cdn_get_response[0]
- if not isinstance(cdn_data, list):
- raise TypeError("malformed API response; 'response'
property not an array")
- first_cdn = cdn_data[0]
- if not isinstance(first_cdn, dict):
- raise TypeError("malformed API response; first CDN in
response is not an object")
- profile["cdn"] = first_cdn["id"]
- cdn_id = profile["cdn"]
- logger.info("extracted %s from %s", cdn_id, cdn_get_response)
- except KeyError as e:
- raise TypeError(f"missing CDN property '{e.args[0]}'") from e
- logger.info("New profile data to hit POST method %s",
request_template_data)
+ # Check if cdn already exists, otherwise create it
+ cdn_data = check_template_data(request_template_data["cdns"], "cdns")
+ cdn_object = create_or_get_existing(to_session, "cdns", "cdn", cdn_data)
+ profile["cdn"] = cdn_object["id"]
+ logger.info("New profile data to hit POST method %s", profile)
+
# Hitting profile POST method
response: tuple[JSONData, requests.Response] =
to_session.create_profile(data=profile)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; profile is not
an object")
- return resp_obj
- except IndexError:
- logger.error("No Profile response data from cdns POST request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "profile")
+ return resp_obj
@pytest.fixture()
@@ -640,15 +630,7 @@ def tenant_post_data(to_session: TOSession,
request_template_data: list[JSONData
:returns: Sample POST data and the actual API response.
"""
- try:
- tenant = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no Parameters present in
'tenants' array property") from e
-
- if not isinstance(tenant, dict):
- raise TypeError(
- f"malformed prerequisite data; tenants must be objects,
not '{type(tenant)}'")
+ tenant = check_template_data(request_template_data["tenants"],
"tenants")
# Return new post data and post response from tenants POST request
randstr = str(randint(0, 1000))
@@ -660,17 +642,12 @@ def tenant_post_data(to_session: TOSession,
request_template_data: list[JSONData
except KeyError as e:
raise TypeError(f"missing tenant property '{e.args[0]}'") from e
- logger.info("New tenant data to hit POST method %s",
request_template_data)
+ logger.info("New tenant data to hit POST method %s", tenant)
# Hitting tenants POST methed
response: tuple[JSONData, requests.Response] =
to_session.create_tenant(data=tenant)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; parameter is
not an object")
- return resp_obj
- except IndexError:
- logger.error("No Parameter response data from parameters POST
request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "tenant")
+ return resp_obj
+
@pytest.fixture()
def server_capabilities_post_data(to_session: TOSession,
request_template_data: list[JSONData]
@@ -680,17 +657,11 @@ def server_capabilities_post_data(to_session: TOSession,
request_template_data:
:param to_session: Fixture to get Traffic Ops session.
:param request_template_data: Fixture to get server_capabilities data
from a prerequisites file.
- :returns: Sample POST data and the actual API response.
+ :returns: Sample POST data and the actual API response.
"""
- try:
- server_capabilities = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no data present in
'server_capabilities' array property") from e
- if not isinstance(server_capabilities, dict):
- raise TypeError(
- f"malformed prerequisite data; data must be objects,
not '{type(server_capabilities)}'")
+ server_capabilities =
check_template_data(request_template_data["server_capabilities"],
+ "server_capabilities")
# Return new post data and post response from server_capabilities POST
request
randstr = str(randint(0, 1000))
@@ -706,14 +677,9 @@ def server_capabilities_post_data(to_session: TOSession,
request_template_data:
# Hitting server_capabilities POST method
response: tuple[
JSONData, requests.Response] =
to_session.create_server_capabilities(data=server_capabilities)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response;
server_capabilities is not an object")
- return resp_obj
- except IndexError:
- logger.error("No server_capabilities response data from
server_capabilities POST request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "server_capabilities")
+ return resp_obj
+
@pytest.fixture()
def division_post_data(to_session: TOSession, request_template_data:
list[JSONData]
@@ -722,20 +688,11 @@ def division_post_data(to_session: TOSession,
request_template_data: list[JSONDa
PyTest Fixture to create POST data for divisions endpoint.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get divisions request template
data from
- request_template file.
- :returns: Sample POST data and the actual API response.
+ :param request_template_data: Fixture to get divisions data from a
prerequisites file.
+ :returns: Sample POST data and the actual API response.
"""
- try:
- division = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no division present in
'division' array property") from e
-
- if not isinstance(division, dict):
- raise TypeError(
- f"malformed prerequisite data; divisions must be
objects, not '{type(division)}'")
+ division = check_template_data(request_template_data["divisions"],
"divisions")
# Return new post data and post response from division POST request
randstr = str(randint(0, 1000))
@@ -750,17 +707,12 @@ def division_post_data(to_session: TOSession,
request_template_data: list[JSONDa
logger.info("New division data to hit POST method %s",
request_template_data)
# Hitting division POST methed
response: tuple[JSONData, requests.Response] =
to_session.create_division(data=division)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; division is
not an object")
- return resp_obj
- except IndexError:
- logger.error("No division response data from division POST
request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "divisions")
+ return resp_obj
[email protected]()
-def region_post_data(to_session: TOSession, request_template_data:
list[JSONData]
+
[email protected](name="region_post_data")
+def region_data_post(to_session: TOSession, request_template_data:
list[JSONData]
) -> dict[str, object]:
"""
PyTest Fixture to create POST data for region endpoint.
@@ -770,14 +722,7 @@ def region_post_data(to_session: TOSession,
request_template_data: list[JSONData
:returns: Sample POST data and the actual API response.
"""
- try:
- region = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no Region present in
'regions' array property") from e
-
- if not isinstance(region, dict):
- raise TypeError(f"malformed prerequisite data; region must be
objects, not '{type(region)}'")
+ region = check_template_data(request_template_data["regions"],
"regions")
# Return new post data and post response from regions POST request
randstr = str(randint(0, 1000))
@@ -788,55 +733,33 @@ def region_post_data(to_session: TOSession,
request_template_data: list[JSONData
region["name"] = name[:4] + randstr
except KeyError as e:
raise TypeError(f"missing Region property '{e.args[0]}'") from e
- # Hitting types GET method to access typeID for region POST data
- division_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
- requests.Response
- ] = to_session.get_divisions()
- try:
- division_data = division_get_response[0]
- if not isinstance(division_data, list):
- raise TypeError("malformed API response; 'response'
property not an array")
- first_division = division_data[0]
- if not isinstance(first_division, dict):
- raise TypeError("malformed API response; first Division
in response is not an object")
- region["division"] = first_division["id"]
- division_id = region["division"]
- logger.info("extracted %s from %s", division_id,
division_get_response)
- except KeyError as e:
- raise TypeError(f"missing Regions property '{e.args[0]}'") from
e
+
+ # Check if division already exists, otherwise create it
+ division_data = check_template_data(request_template_data["divisions"],
"divisions")
+ division_object = create_or_get_existing(to_session, "divisions",
"division", division_data)
+ region["division"] = division_object["id"]
+ region["divisionName"] = division_object["name"]
logger.info("New region data to hit POST method %s",
request_template_data)
# Hitting region POST method
response: tuple[JSONData, requests.Response] =
to_session.create_region(data=region)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; region is not
an object")
- return resp_obj
- except IndexError:
- logger.error("No Region response data from divisions POST
request.")
- sys.exit(1)
+ resp_obj = check_template_data(response, "regions")
+ return resp_obj
[email protected]()
-def phys_locations_post_data(to_session: TOSession, request_template_data:
list[JSONData]
- ) -> dict[str, object]:
+
[email protected](name="phys_locations_post_data")
+def phys_locations_data_post(to_session: TOSession, request_template_data:
list[JSONData],
+ region_post_data: dict[str, object]) -> dict[str,
object]:
"""
PyTest Fixture to create POST data for phys_location endpoint.
:param to_session: Fixture to get Traffic Ops session.
:param request_template_data: Fixture to get phys_location data from a
prerequisites file.
+ :param region_post_data:
:returns: Sample POST data and the actual API response.
"""
- try:
- phys_locations = request_template_data[0]
- except IndexError as e:
- raise TypeError(
- "malformed prerequisite data; no data in
'phys_locations' array property") from e
-
- if not isinstance(phys_locations, dict):
- raise TypeError(f"malformed prerequisite data; PLs must be
objects, not '{type(phys_locations)}'")
+ phys_locations =
check_template_data(request_template_data["phys_locations"], "phys_locations")
# Return new post data and post response from phys_locations POST
request
randstr = str(randint(0, 1000))
@@ -845,39 +768,79 @@ def phys_locations_post_data(to_session: TOSession,
request_template_data: list[
if not isinstance(name, str):
raise TypeError(f"name must be str, not '{type(name)}'")
phys_locations["name"] = name[:4] + randstr
- shortName = phys_locations["shortName"]
+ short_name = phys_locations["shortName"]
if not isinstance(name, str):
- raise TypeError(f"shortName must be str, not
'{type(shortName)}'")
- phys_locations["shortName"] = shortName[:4] + randstr
+ raise TypeError(f"shortName must be str, not
'{type(short_name)}'")
+ phys_locations["shortName"] = short_name[:4] + randstr
except KeyError as e:
raise TypeError(f"missing Phys_location property
'{e.args[0]}'") from e
- # Hitting types GET method to access typeID for phys_locations POST data
- region_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
- requests.Response
- ] = to_session.get_regions()
- try:
- region_data = region_get_response[0]
- if not isinstance(region_data, list):
- raise TypeError("malformed API response; 'response'
property not an array")
- first_region = region_data[0]
- if not isinstance(first_region, dict):
- raise TypeError("malformed API response; first Region
in response is not an object")
- phys_locations["regionId"] = first_region["id"]
- region_id = phys_locations["regionId"]
- logger.info("extracted %s from %s", region_id,
region_get_response)
- except KeyError as e:
- raise TypeError(f"missing Phys_Locations property
'{e.args[0]}'") from e
+
+ # Check if region already exists, otherwise create it
+ region_id = region_post_data["id"]
+ if not isinstance(region_id, int):
+ raise TypeError("malformed API response; 'id' property not a
integer")
+ phys_locations["regionId"] = region_id
logger.info("New Phys_locations data to hit POST method %s",
request_template_data)
# Hitting region POST method
- response: tuple[JSONData, requests.Response] =
to_session.create_physical_locations(data=phys_locations)
- try:
- resp_obj = response[0]
- if not isinstance(resp_obj, dict):
- raise TypeError("malformed API response; phys_location
is not an object")
- return resp_obj
- except IndexError:
- logger.error("No Phys_Location response data from regions POST
request.")
- sys.exit(1)
+ response: tuple[JSONData, requests.Response] =
to_session.create_physical_locations(
+ data=phys_locations)
+ resp_obj = check_template_data(response, "phys_locations")
+ return resp_obj
+
[email protected]()
+def server_post_data(to_session: TOSession, request_template_data:
list[JSONData],
+ phys_locations_post_data: dict[str, object]) -> dict[str,
object]:
+ """
+ PyTest Fixture to create POST data for server endpoint.
+
+ :param to_session: Fixture to get Traffic Ops session.
+ :param request_template_data: Fixture to get profile data from a
prerequisites file.
+ :param phys_locations_post_data:
+ :returns: Sample POST data and the actual API response.
+ """
+ server = check_template_data(request_template_data["servers"],
"servers")
+
+ # Check if type already exists, otherwise create it
+ type_data = check_template_data(request_template_data["types"], "types")
+ type_object = create_or_get_existing(to_session, "types", "type",
type_data,
+ {"useInTable": "server"})
+ type_id = type_object["id"]
+ server["typeId"] = type_id
+
+ # Check if cachegroup with type already exists, otherwise create it
+ cache_group_data =
check_template_data(request_template_data["cachegroup"], "cachegroup")
+ cache_group_object = create_or_get_existing(to_session, "cachegroups",
"cachegroups",
+ cache_group_data, {"typeId":
type_id})
+ server["cachegroupId"]= cache_group_object["id"]
+
+ # Check if cdn already exists, otherwise create it
+ cdn_data = check_template_data(request_template_data["cdns"], "cdns")
+ cdn_object = create_or_get_existing(to_session, "cdns", "cdn",
cdn_data, {"name": "CDN-in-a-Box"})
+ server["cdnId"] = cdn_object["id"]
+ server["domainName"] = cdn_object["domainName"]
+
+ # Check if profile with cdn already exists, otherwise create it
+ profile_data = check_template_data(request_template_data["profiles"],
"profiles")
+ profile_object = create_or_get_existing(to_session, "profiles",
"profile", profile_data,
+ {"name": "test"})
+ server["profileNames"] = [profile_object["name"]]
+
+ # Check if status already exists, otherwise create it
+ status_data = check_template_data(request_template_data["status"],
"status")
+ status_object = create_or_get_existing(to_session, "statuses",
"statuses",
+ status_data, {"name": "REPORTED"})
+ server["statusId"] = status_object["id"]
+
+ # Check if physical location with region already exists, otherwise
create it
+ physical_location_id = phys_locations_post_data["id"]
+ if not isinstance(physical_location_id, int):
+ raise TypeError("malformed API response; 'id' property not a
integer")
+ server["physLocationId"] = physical_location_id
+
+ logger.info("New server data to hit POST method %s", server)
+ # Hitting server POST method
+ response: tuple[JSONData, requests.Response] =
to_session.create_server(data=server)
+ resp_obj = check_template_data(response, "server")
+ return resp_obj
diff --git a/traffic_ops/testing/api_contract/v4/data/request_template.json
b/traffic_ops/testing/api_contract/v4/data/request_template.json
new file mode 100644
index 0000000000..da10040eae
--- /dev/null
+++ b/traffic_ops/testing/api_contract/v4/data/request_template.json
@@ -0,0 +1,138 @@
+{
+ "cdns": [
+ {
+ "name": "test",
+ "domainName": "quest",
+ "dnssecEnabled": false
+ }
+ ],
+ "cachegroup": [
+ {
+ "name": "test",
+ "shortName": "test",
+ "latitude": 38.897663,
+ "longitude": -77.036574,
+ "fallbackToClosest": true,
+ "localizationMethods": [
+ "DEEP_CZ",
+ "CZ",
+ "GEO"
+ ],
+ "typeId": 23
+ }
+ ],
+ "parameters": [
+ {
+ "name": "test",
+ "value": "quest",
+ "configFile": "records.config",
+ "secure": false
+ }
+ ],
+ "roles": [
+ {
+ "name": "test",
+ "description": "quest"
+ }
+ ],
+ "profiles": [
+ {
+ "name": "test",
+ "description": "A test profile for API examples",
+ "cdn": 2,
+ "type": "UNK_PROFILE",
+ "routingDisabled": true
+ }
+ ],
+ "tenants": [
+ {
+ "active": true,
+ "name": "test",
+ "parentId": 1
+ }
+ ],
+ "server_capabilities": [
+ {
+ "name": "RAM"
+ }
+ ],
+ "types": [
+ {
+ "name": "MID_LOC",
+ "description": "Mid Logical Location",
+ "useInTable": "cachegroup"
+ }
+ ],
+ "status": [
+ {
+ "description": "Server is online and reported in the
health protocol.",
+ "name": "REPORTED"
+ }
+ ],
+ "divisions": [
+ {
+ "name": "test"
+ }
+ ],
+ "regions": [
+ {
+ "name": "Manchester",
+ "division": "4",
+ "divisionName": "England"
+ }
+ ],
+ "phys_locations": [
+ {
+ "address": "Buckingham Palace",
+ "city": "London",
+ "comments": "Buckingham Palace",
+ "email": "[email protected]",
+ "name": "test",
+ "phone": "0-843-816-6276",
+ "poc": "Her Majesty The Queen Elizabeth Alexandra Mary
Windsor II",
+ "regionId": 1,
+ "shortName": "test",
+ "state": "Westminster",
+ "zip": "SW1A 1AA"
+ }
+ ],
+ "servers": [
+ {
+ "cachegroupId": 79,
+ "cdnId": 2,
+ "domainName": "mycdn.ciab.test",
+ "hostName": "mid",
+ "httpsPort": 443,
+ "interfaces": [
+ {
+ "ipAddresses": [
+ {
+ "address": "127.0.0.1",
+ "gateway": "0.0.0.1",
+ "serviceAddress": true
+ }
+ ],
+ "maxBandwidth": null,
+ "monitor": true,
+ "mtu": 1500,
+ "name": "eth0"
+ }
+ ],
+ "interfaceMtu": 1500,
+ "interfaceName": "eth0",
+ "ip6Address": "::1",
+ "ip6Gateway": "::2",
+ "ipAddress": "127.0.0.1",
+ "ipGateway": "0.0.0.1",
+ "ipNetmask": "255.255.255.0",
+ "physLocationId": 3,
+ "profileNames": [
+ "MID_TIER_ATS_CACHE"
+ ],
+ "statusId": 3,
+ "tcpPort": 80,
+ "typeId": 96,
+ "updPending": false
+ }
+ ]
+}
diff --git a/traffic_ops/testing/api_contract/v4/data/response_template.json
b/traffic_ops/testing/api_contract/v4/data/response_template.json
new file mode 100644
index 0000000000..50f5abe26d
--- /dev/null
+++ b/traffic_ops/testing/api_contract/v4/data/response_template.json
@@ -0,0 +1,592 @@
+{
+ "type": "object",
+ "cdns": {
+ "type": "object",
+ "required": [
+ "dnssecEnabled",
+ "domainName",
+ "id",
+ "lastUpdated",
+ "name"
+ ],
+ "properties": {
+ "dnssecEnabled": {
+ "type": "boolean"
+ },
+ "domainName": {
+ "type": "string"
+ },
+ "id": {
+ "type": "integer"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "name": {
+ "type": "string"
+ }
+ }
+ },
+ "cachegroup": {
+ "type": "object",
+ "required": [
+ "id",
+ "name",
+ "shortName",
+ "latitude",
+ "longitude",
+ "parentCachegroupName",
+ "parentCachegroupId",
+ "secondaryParentCachegroupName",
+ "secondaryParentCachegroupId",
+ "fallbackToClosest",
+ "localizationMethods",
+ "typeName",
+ "typeId",
+ "lastUpdated",
+ "fallbacks"
+ ],
+ "properties": {
+ "id": {
+ "type": "integer"
+ },
+ "name": {
+ "type": "string"
+ },
+ "shortName": {
+ "type": "string"
+ },
+ "latitude": {
+ "type": "number"
+ },
+ "longitude": {
+ "type": "number"
+ },
+ "parentCachegroupName": {
+ "type": [
+ "string",
+ "null"
+ ]
+ },
+ "parentCachegroupId": {
+ "type": [
+ "integer",
+ "null"
+ ]
+ },
+ "secondaryParentCachegroupName": {
+ "type": [
+ "string",
+ "null"
+ ]
+ },
+ "secondaryParentCachegroupId": {
+ "type": [
+ "integer",
+ "null"
+ ]
+ },
+ "fallbackToClosest": {
+ "type": "boolean"
+ },
+ "localizationMethods": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
+ },
+ "typeName": {
+ "type": "string"
+ },
+ "typeId": {
+ "type": "integer"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "fallbacks": {
+ "type": "array",
+ "items": {}
+ }
+ }
+ },
+ "parameters": {
+ "type": "object",
+ "required": [
+ "configFile",
+ "id",
+ "lastUpdated",
+ "name",
+ "profiles",
+ "secure",
+ "value"
+ ],
+ "properties": {
+ "configFile": {
+ "type": "string"
+ },
+ "id": {
+ "type": "integer"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "name": {
+ "type": "string"
+ },
+ "profiles": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
+ },
+ "secure": {
+ "type": "boolean"
+ },
+ "value": {
+ "type": "string"
+ }
+ }
+ },
+ "roles": {
+ "type": "object",
+ "required": [
+ "name",
+ "permissions",
+ "description",
+ "lastUpdated"
+ ],
+ "properties": {
+ "name": {
+ "type": "string"
+ },
+ "permissions": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
+ },
+ "description": {
+ "type": "string"
+ },
+ "lastUpdated": {
+ "type": "string"
+ }
+ }
+ },
+ "profiles": {
+ "type": "object",
+ "required": [
+ "id",
+ "lastUpdated",
+ "name",
+ "description",
+ "cdnName",
+ "cdn",
+ "routingDisabled",
+ "type"
+ ],
+ "properties": {
+ "id": {
+ "type": "integer"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "name": {
+ "type": "string"
+ },
+ "description": {
+ "type": "string"
+ },
+ "cdnName": {
+ "type": "string"
+ },
+ "cdn": {
+ "type": "integer"
+ },
+ "routingDisabled": {
+ "type": "boolean"
+ },
+ "type": {
+ "type": "string"
+ }
+ }
+ },
+ "server_capabilities": {
+ "type": "object",
+ "required": [
+ "name",
+ "lastUpdated"
+ ],
+ "properties": {
+ "name": {
+ "type": "string"
+ },
+ "lastUpdated": {
+ "type": "string"
+ }
+ }
+ },
+ "tenants": {
+ "type": "object",
+ "required": [
+ "id",
+ "name",
+ "active",
+ "lastUpdated",
+ "parentId"
+ ],
+ "properties": {
+ "id": {
+ "type": "integer"
+ },
+ "name": {
+ "type": "string"
+ },
+ "active": {
+ "type": "boolean"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "parentId": {
+ "type": [
+ "null",
+ "integer"
+ ]
+ },
+ "parentName": {
+ "type": "string"
+ }
+ }
+ },
+ "divisions": {
+ "type": "object",
+ "required": [
+ "id",
+ "lastUpdated",
+ "name"
+ ],
+ "properties": {
+ "id": {
+ "type": "integer"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "name": {
+ "type": "string"
+ }
+ }
+ },
+ "regions": {
+ "type": "object",
+ "required": [
+ "divisionName",
+ "division",
+ "id",
+ "lastUpdated",
+ "name"
+ ],
+ "properties": {
+ "divisionName": {
+ "type": "string"
+ },
+ "division": {
+ "type": "integer"
+ },
+ "id": {
+ "type": "integer"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "name": {
+ "type": "string"
+ }
+ }
+ },
+ "phys_locations": {
+ "type": "object",
+ "required": [
+ "address",
+ "city",
+ "comments",
+ "email",
+ "id",
+ "lastUpdated",
+ "name",
+ "phone",
+ "poc",
+ "regionId",
+ "region",
+ "shortName",
+ "state",
+ "zip"
+ ],
+ "properties": {
+ "address": {
+ "type": "string"
+ },
+ "city": {
+ "type": "string"
+ },
+ "comments": {
+ "type": "string"
+ },
+ "email": {
+ "type": "string"
+ },
+ "id": {
+ "type": "integer"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "name": {
+ "type": "string"
+ },
+ "phone": {
+ "type": "string"
+ },
+ "poc": {
+ "type": "string"
+ },
+ "regionId": {
+ "type": "integer"
+ },
+ "region": {
+ "type": [
+ "null",
+ "string"
+ ]
+ },
+ "shortName": {
+ "type": "string"
+ },
+ "state": {
+ "type": "string"
+ },
+ "zip": {
+ "type": "string"
+ }
+ }
+ },
+ "servers": {
+ "type": "object",
+ "required": [
+ "cachegroup",
+ "cachegroupId",
+ "cdnId",
+ "cdnName",
+ "domainName",
+ "guid",
+ "hostName",
+ "httpsPort",
+ "id",
+ "iloIpAddress",
+ "iloIpGateway",
+ "iloIpNetmask",
+ "iloPassword",
+ "iloUsername",
+ "lastUpdated",
+ "mgmtIpAddress",
+ "mgmtIpGateway",
+ "mgmtIpNetmask",
+ "offlineReason",
+ "physLocation",
+ "physLocationId",
+ "profileNames",
+ "rack",
+ "revalPending",
+ "status",
+ "statusId",
+ "tcpPort",
+ "type",
+ "typeId",
+ "updPending",
+ "xmppId",
+ "xmppPasswd",
+ "interfaces",
+ "statusLastUpdated",
+ "configUpdateTime",
+ "configApplyTime",
+ "revalUpdateTime",
+ "revalApplyTime"
+ ],
+ "properties": {
+ "cachegroup": {
+ "type": "string"
+ },
+ "cachegroupId": {
+ "type": "integer"
+ },
+ "cdnId": {
+ "type": "integer"
+ },
+ "cdnName": {
+ "type": "string"
+ },
+ "domainName": {
+ "type": "string"
+ },
+ "guid": {
+ "type": "null"
+ },
+ "hostName": {
+ "type": "string"
+ },
+ "httpsPort": {
+ "type": "integer"
+ },
+ "id": {
+ "type": "integer"
+ },
+ "iloIpAddress": {
+ "type": "null"
+ },
+ "iloIpGateway": {
+ "type": "null"
+ },
+ "iloIpNetmask": {
+ "type": "null"
+ },
+ "iloPassword": {
+ "type": "null"
+ },
+ "iloUsername": {
+ "type": "null"
+ },
+ "lastUpdated": {
+ "type": "string"
+ },
+ "mgmtIpAddress": {
+ "type": "null"
+ },
+ "mgmtIpGateway": {
+ "type": "null"
+ },
+ "mgmtIpNetmask": {
+ "type": "null"
+ },
+ "offlineReason": {
+ "type": "null"
+ },
+ "physLocation": {
+ "type": "string"
+ },
+ "physLocationId": {
+ "type": "integer"
+ },
+ "profileNames": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
+ },
+ "rack": {
+ "type": "null"
+ },
+ "revalPending": {
+ "type": "boolean"
+ },
+ "status": {
+ "type": "string"
+ },
+ "statusId": {
+ "type": "integer"
+ },
+ "tcpPort": {
+ "type": "integer"
+ },
+ "type": {
+ "type": "string"
+ },
+ "typeId": {
+ "type": "integer"
+ },
+ "updPending": {
+ "type": "boolean"
+ },
+ "xmppId": {
+ "type": "string"
+ },
+ "xmppPasswd": {
+ "type": "null"
+ },
+ "interfaces": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "required": [
+ "ipAddresses",
+ "maxBandwidth",
+ "monitor",
+ "mtu",
+ "name",
+ "routerHostName",
+ "routerPortName"
+ ],
+ "properties": {
+ "ipAddresses": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "required": [
+ "address",
+ "gateway",
+ "serviceAddress"
+ ],
+ "properties": {
+ "address": {
+ "type": "string"
+ },
+ "gateway": {
+ "type": "string"
+ },
+ "serviceAddress": {
+ "type": "boolean"
+ }
+ }
+ }
+ },
+ "maxBandwidth": {
+ "type": "null"
+ },
+ "monitor": {
+ "type": "boolean"
+ },
+ "mtu": {
+ "type": "integer"
+ },
+ "name": {
+ "type": "string"
+ },
+ "routerHostName": {
+ "type": "string"
+ },
+ "routerPortName": {
+ "type": "string"
+ }
+ }
+ }
+ },
+ "statusLastUpdated": {
+ "type": "null"
+ },
+ "configUpdateTime": {
+ "type": "string"
+ },
+ "configApplyTime": {
+ "type": "string"
+ },
+ "revalUpdateTime": {
+ "type": "string"
+ },
+ "revalApplyTime": {
+ "type": "string"
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/traffic_ops/testing/api_contract/v4/to_data.json
b/traffic_ops/testing/api_contract/v4/data/to_data.json
similarity index 100%
rename from traffic_ops/testing/api_contract/v4/to_data.json
rename to traffic_ops/testing/api_contract/v4/data/to_data.json
diff --git a/traffic_ops/testing/api_contract/v4/request_template.json
b/traffic_ops/testing/api_contract/v4/request_template.json
deleted file mode 100644
index c67d746976..0000000000
--- a/traffic_ops/testing/api_contract/v4/request_template.json
+++ /dev/null
@@ -1,86 +0,0 @@
-{
- "cdns": [
- {
- "name": "test",
- "domainName": "quest",
- "dnssecEnabled": false
- }
- ],
- "cachegroup": [
- {
- "name": "test",
- "shortName": "test",
- "latitude": 38.897663,
- "longitude": -77.036574,
- "fallbackToClosest": true,
- "localizationMethods": [
- "DEEP_CZ",
- "CZ",
- "GEO"
- ],
- "typeId": 23
- }
- ],
- "parameters": [
- {
- "name": "test",
- "value": "quest",
- "configFile": "records.config",
- "secure": false
- }
- ],
- "roles": [
- {
- "name": "test",
- "description": "quest"
- }
- ],
- "profiles": [
- {
- "name": "test",
- "description": "A test profile for API examples",
- "cdn": 2,
- "type": "UNK_PROFILE",
- "routingDisabled": true
- }
- ],
- "tenants": [
- {
- "active": true,
- "name": "test",
- "parentId": 1
- }
- ],
- "server_capabilities": [
- {
- "name": "RAM"
- }
- ],
- "divisions":[
- {
- "name":"test"
- }
- ],
- "regions": [
- {
- "divisionName": "Quebec",
- "division": 1,
- "name": "Montreal"
- }
- ],
- "phys_locations": [
- {
- "address": "Buckingham Palace",
- "city": "London",
- "comments": "Buckingham Palace",
- "email": "[email protected]",
- "name": "test",
- "phone": "0-843-816-6276",
- "poc": "Her Majesty The Queen Elizabeth Alexandra Mary
Windsor II",
- "regionId": 3,
- "shortName": "test",
- "state": "Westminster",
- "zip": "SW1A 1AA"
- }
- ]
-}
diff --git a/traffic_ops/testing/api_contract/v4/requirements.txt
b/traffic_ops/testing/api_contract/v4/requirements.txt
index d08099f871..07c255bd51 100644
--- a/traffic_ops/testing/api_contract/v4/requirements.txt
+++ b/traffic_ops/testing/api_contract/v4/requirements.txt
@@ -13,3 +13,4 @@
#
pytest==7.2.1
+jsonschema==4.17.3
\ No newline at end of file
diff --git a/traffic_ops/testing/api_contract/v4/response_template.json
b/traffic_ops/testing/api_contract/v4/response_template.json
deleted file mode 100644
index 72508ede94..0000000000
--- a/traffic_ops/testing/api_contract/v4/response_template.json
+++ /dev/null
@@ -1,268 +0,0 @@
-{
- "type": "object",
- "cdns": {
- "type": "object",
- "properties": {
- "name": {
- "type": "str"
- },
- "domainName": {
- "type": "str"
- },
- "dnssecEnabled": {
- "type": "bool"
- },
- "id": {
- "type": "int"
- },
- "lastUpdated": {
- "type": "str"
- }
- }
- },
- "cachegroup": {
- "type": "object",
- "properties": {
- "id": {
- "type": "int"
- },
- "name": {
- "type": "str"
- },
- "shortName": {
- "type": "str"
- },
- "latitude": {
- "type": "float"
- },
- "longitude": {
- "type": "float"
- },
- "parentCachegroupName": {
- "optional": "True",
- "typeA": "str",
- "typeB": "NoneType"
- },
- "parentCachegroupId": {
- "optional": "True",
- "typeA": "int",
- "typeB": "NoneType"
- },
- "secondaryParentCachegroupName": {
- "optional": "True",
- "typeA": "str",
- "typeB": "NoneType"
- },
- "secondaryParentCachegroupId": {
- "optional": "True",
- "typeA": "int",
- "typeB": "NoneType"
- },
- "fallbackToClosest": {
- "type": "bool"
- },
- "localizationMethods": {
- "type": "list"
- },
- "typeName": {
- "type": "str"
- },
- "typeId": {
- "type": "int"
- },
- "lastUpdated": {
- "type": "str"
- },
- "fallbacks": {
- "type": "list"
- }
- }
- },
- "parameters": {
- "type": "object",
- "properties": {
- "configFile": {
- "type": "str"
- },
- "id": {
- "type": "int"
- },
- "lastUpdated": {
- "type": "str"
- },
- "name": {
- "type": "str"
- },
- "profiles": {
- "type": "list"
- },
- "secure": {
- "type": "bool"
- },
- "value": {
- "type": "str"
- }
- }
- },
- "roles": {
- "type": "object",
- "properties": {
- "name": {
- "type": "str"
- },
- "description": {
- "type": "str"
- },
- "permissions": {
- "type": "list"
- },
- "lastUpdated": {
- "type": "str"
- }
- }
- },
- "profiles": {
- "type": "object",
- "properties": {
- "id": {
- "type": "int"
- },
- "lastUpdated": {
- "type": "str"
- },
- "name": {
- "type": "str"
- },
- "description": {
- "type": "str"
- },
- "cdnName": {
- "type": "str"
- },
- "cdn": {
- "type": "int"
- },
- "routingDisabled": {
- "type": "bool"
- },
- "type": {
- "type": "str"
- }
- }
- },
- "tenants": {
- "type": "object",
- "properties": {
- "id": {
- "type": "int"
- },
- "name": {
- "type": "str"
- },
- "active": {
- "type": "bool"
- },
- "parentId": {
- "type": "int"
- },
- "lastUpdated": {
- "type": "str"
- },
- "parentName": {
- "type": "str"
- }
- }
- },
- "server_capabilities": {
- "type": "object",
- "properties": {
- "lastUpdated": {
- "type": "str"
- },
- "name": {
- "type": "str"
- }
- }
- },
- "divisions": {
- "type": "object",
- "properties": {
- "id": {
- "type": "int"
- },
- "lastUpdated": {
- "type": "str"
- },
- "name": {
- "type": "str"
- }
- }
- },
- "regions": {
- "type": "object",
- "properties": {
- "divisionName": {
- "type": "str"
- },
- "division": {
- "type": "int"
- },
- "id": {
- "type": "int"
- },
- "lastUpdated": {
- "type": "str"
- },
- "name": {
- "type": "str"
- }
- }
- },
- "phys_locations": {
- "type": "object",
- "properties": {
- "address": {
- "type": "str"
- },
- "city": {
- "type": "str"
- },
- "comments": {
- "type": "str"
- },
- "email": {
- "type": "str"
- },
- "id":{
- "type": "int"
- },
- "lastUpdated":{
- "type": "str"
- },
- "name": {
- "type": "str"
- },
- "phone": {
- "type": "str"
- },
- "poc": {
- "type": "str"
- },
- "regionId": {
- "type": "int"
- },
- "region":{
- "type": "str"
- },
- "shortName": {
- "type": "str"
- },
- "state": {
- "type": "str"
- },
- "zip": {
- "type": "str"
- }
- }
- }
-}
diff --git a/traffic_ops/testing/api_contract/v4/test_cachegroups.py
b/traffic_ops/testing/api_contract/v4/test_cachegroups.py
index 3a72583959..b65a445ae5 100644
--- a/traffic_ops/testing/api_contract/v4/test_cachegroups.py
+++ b/traffic_ops/testing/api_contract/v4/test_cachegroups.py
@@ -14,100 +14,63 @@
"""API Contract Test Case for cachegroup endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["cachegroup"],
indirect=True)
-def test_cachegroup_contract(to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
- cachegroup_post_data: dict[str, object]
- ) -> None:
+def test_cache_group_contract(to_session: TOSession,
+
response_template_data: dict[str, Union[Primitive, list[
+
Union[Primitive, dict[str, object], list[object]]], dict[
+ object,
object]]],
+
cache_group_post_data: dict[str, object]
+ ) -> None:
"""
Test step to validate keys, values and data types from cachegroup
endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
- :param cachegroup_post_data: Fixture to get sample cachegroup data and
actual cachegroup response.
+ :param cache_group_post_data: Fixture to get sample cachegroup data and
actual cachegroup response.
"""
# validate CDN keys from cdns get response
logger.info("Accessing /cachegroup endpoint through Traffic ops
session.")
- cachegroup = request_template_data[0]
- if not isinstance(cachegroup, dict):
- raise TypeError("malformed cachegroup in prerequisite data; not
an object")
-
- cachegroup_name = cachegroup.get("name")
- if not isinstance(cachegroup_name, str):
+ cache_group_name = cache_group_post_data.get("name")
+ if not isinstance(cache_group_name, str):
raise TypeError("malformed cachegroup in prerequisite data;
'name' not a string")
- cachegroup_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ cache_group_get_response: tuple[
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
- ] = to_session.get_cachegroups(query_params={"name":
str(cachegroup_name)})
+ ] = to_session.get_cachegroups(query_params={"name":
str(cache_group_name)})
try:
- cachegroup_data = cachegroup_get_response[0]
- if not isinstance(cachegroup_data, list):
+ cache_group_data = cache_group_get_response[0]
+ if not isinstance(cache_group_data, list):
raise TypeError("malformed API response; 'response'
property not an array")
- first_cachegroup = cachegroup_data[0]
- if not isinstance(first_cachegroup, dict):
+ first_cache_group = cache_group_data[0]
+ if not isinstance(first_cache_group, dict):
raise TypeError("malformed API response; first Cache
group in response is not an object")
- cachegroup_keys = set(first_cachegroup.keys())
- logger.info("Cache group Keys from cachegroup endpoint response
%s", cachegroup_keys)
-
- cachegroup_response_template =
response_template_data.get("cachegroup")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template =
cachegroup_response_template.get("properties") if isinstance(
- cachegroup_response_template, dict) else None
- if response_template is None or not
isinstance(response_template, dict):
- raise TypeError(
- f"response template data must be a dict, not
'{type(response_template)}'")
+ logger.info("Cachegroup API get response %s", first_cache_group)
+ cache_group_response_template =
response_template_data.get("cachegroup")
# validate cachegroup values from prereq data in cachegroup get
response.
- prereq_values =
[cachegroup_post_data["name"],cachegroup_post_data["shortName"],
-
cachegroup_post_data["fallbackToClosest"],cachegroup_post_data["typeId"]]
-
- get_values =
[first_cachegroup["name"],first_cachegroup["shortName"],
-
first_cachegroup["fallbackToClosest"],first_cachegroup["typeId"]]
-
- get_types = {}
- for key, value in first_cachegroup.items():
- get_types[key] = type(value).__name__
- logger.info("types from cachegroup get response %s", get_types)
-
- response_template_types= {}
- for key, value in response_template.items():
- optional = value.get("optional")
- if optional:
- optional_type = value.get("typeA") if key in
cachegroup else value.get("typeB")
- if not isinstance(optional_type, str):
- raise TypeError(f"Type data must be a
string, not '{type(optional_type)}'")
- response_template_types[key] = optional_type
- else:
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(f"Type data must be a
string, not '{type(actual_type)}'")
- response_template_types[key] = actual_type
-
- logger.info("types from cachegroup response template %s",
response_template_types)
+ keys = ["name", "shortName", "fallbackToClosest", "typeId"]
+ prereq_values = [cache_group_post_data[key] for key in keys]
+ get_values = [first_cache_group[key] for key in keys]
- # validate keys,data types for values and actual values for
cachegroup endpoint.
- assert cachegroup_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ # validate keys,data types and values for cachegroup endpoint.
+ assert validate(instance=first_cache_group,
schema=cache_group_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
@@ -115,8 +78,8 @@ def test_cachegroup_contract(to_session: TOSession,
finally:
# Delete Cache group after test execution to avoid redundancy.
try:
- cachegroup_id = cachegroup_post_data["id"]
-
to_session.delete_cachegroups(cache_group_id=cachegroup_id)
+ cache_group_id = cache_group_post_data["id"]
+
to_session.delete_cachegroups(cache_group_id=cache_group_id)
except IndexError:
logger.error("Cachegroup returned by Traffic Ops is
missing an 'id' property")
pytest.fail("Response from delete request is empty,
Failing test_cachegroup_contract")
diff --git a/traffic_ops/testing/api_contract/v4/test_cdns.py
b/traffic_ops/testing/api_contract/v4/test_cdns.py
index bf07e5f2c4..9901fab773 100644
--- a/traffic_ops/testing/api_contract/v4/test_cdns.py
+++ b/traffic_ops/testing/api_contract/v4/test_cdns.py
@@ -14,45 +14,40 @@
"""API Contract Test Case for cdns endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["cdns"], indirect=True)
def test_cdn_contract(
to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
- cdn_post_data: dict[str, object]
- ) -> None:
+ response_template_data: dict[str, Union[Primitive,
list[Union[Primitive, dict[str, object],
+ list[object]]], dict[object, object]]], cdn_post_data: dict[str,
object]) -> None:
"""
Test step to validate keys, values and data types from cdns endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
:param cdn_post_data: Fixture to get sample CDN data and actual CDN
response.
"""
# validate CDN keys from cdns get response
logger.info("Accessing /cdns endpoint through Traffic ops session.")
- cdn = request_template_data[0]
- if not isinstance(cdn, dict):
- raise TypeError("malformed cdn in prerequisite data; not an
object")
-
- cdn_name = cdn.get("name")
+ cdn_name = cdn_post_data.get("name")
if not isinstance(cdn_name, str):
raise TypeError("malformed cdn in prerequisite data; 'name' not
a string")
cdn_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_cdns(query_params={"name": cdn_name})
try:
@@ -63,40 +58,18 @@ def test_cdn_contract(
first_cdn = cdn_data[0]
if not isinstance(first_cdn, dict):
raise TypeError("malformed API response; first CDN in
response is not an object")
- cdn_keys = set(first_cdn.keys())
- logger.info("CDN Keys from cdns endpoint response %s", cdn_keys)
-
+ logger.info("CDN Api get response %s", first_cdn)
cdn_response_template = response_template_data.get("cdns")
if not isinstance(cdn_response_template, dict):
raise TypeError(
f"Cdn response template data must be a dict,
not '{type(cdn_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template = cdn_response_template.get("properties")
- if not isinstance(response_template, dict):
- raise TypeError(
- f"response template data must be a dict, not
'{type(response_template)}'")
+
# validate cdn values from prereq data in cdns get response.
- prereq_values = [cdn_post_data["name"],
cdn_post_data["domainName"],
- cdn_post_data["dnssecEnabled"]]
- get_values = [first_cdn["name"], first_cdn["domainName"],
first_cdn["dnssecEnabled"]]
- get_types = {}
- for key, value in first_cdn.items():
- get_types[key] = type(value).__name__
- logger.info("types from cdn get response %s", get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("types from cdn response template %s",
response_template_types)
+ keys = ["name", "domainName", "dnssecEnabled"]
+ prereq_values = [cdn_post_data[key] for key in keys]
+ get_values = [first_cdn[key] for key in keys]
- assert cdn_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_cdn,
schema=cdn_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
diff --git a/traffic_ops/testing/api_contract/v4/test_divisions.py
b/traffic_ops/testing/api_contract/v4/test_divisions.py
index a3da4c1bd1..7a464e7cc5 100644
--- a/traffic_ops/testing/api_contract/v4/test_divisions.py
+++ b/traffic_ops/testing/api_contract/v4/test_divisions.py
@@ -14,43 +14,39 @@
"""API Contract Test Case for divisions endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["divisions"], indirect=True)
-def test_division_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: list[dict[str, object] | list[object] |
primitive],
- division_post_data: dict[str, object]
-) -> None:
+def test_division_contract(to_session: TOSession,
+ response_template_data: list[Union[dict[str, object], list[object],
Primitive]],
+ division_post_data: dict[str, object]) -> None:
"""
Test step to validate keys, values and data types from divisions
endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param get_division_data: Fixture to get division data from a
prerequisites file.
- :param division_prereq: Fixture to get sample division data and actual
division response.
+ :param response_template_data: Fixture to get response template data
from a prerequisites file.
+ :param division_post_dat: Fixture to get sample division data and
actual division response.
"""
# validate division keys from divisions get response
logger.info("Accessing divisions endpoint through Traffic ops session.")
- division = request_template_data[0]
- if not isinstance(division, dict):
- raise TypeError("malformed division in prerequisite data; not
an object")
-
- division_name = division.get("name")
+ division_name = division_post_data.get("name")
if not isinstance(division_name, str):
- raise TypeError("malformed division in prerequisite data;
'name' not a string")
+ raise TypeError("malformed cdn in prerequisite data; 'name' not
a string")
division_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_divisions(query_params={"name": division_name})
try:
@@ -61,25 +57,18 @@ def test_division_contract(
first_division = division_data[0]
if not isinstance(first_division, dict):
raise TypeError("malformed API response; first division
in response is not an object")
- division_keys = set(first_division.keys())
- logger.info("division Keys from divisions endpoint response
%s", division_keys)
- response_template =
response_template_data.get("divisions").get("properties")
+ logger.info("Division Api get response %s", first_division)
+ division_response_template =
response_template_data.get("divisions")
+ if not isinstance(division_response_template, dict):
+ raise TypeError(
+ f"Division response template data must be a
dict, not '{type(division_response_template)}'")
+
# validate division values from prereq data in divisions get
response.
- prereq_values = [
- division_post_data["name"]]
- get_values = [first_division["name"]]
- get_types = {}
- for key, value in first_division.items():
- get_types[key] = type(value).__name__
- logger.info("types from division get response %s", get_types)
- response_template_types= {}
- for key, value in response_template.items():
- response_template_types[key] = value.get("type")
- logger.info("types from division response template %s",
response_template_types)
+ prereq_values = division_post_data["name"]
+ get_values = first_division["name"]
- assert division_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_division,
schema=division_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
diff --git a/traffic_ops/testing/api_contract/v4/test_parameters.py
b/traffic_ops/testing/api_contract/v4/test_parameters.py
index 27b257d401..62935d5e20 100644
--- a/traffic_ops/testing/api_contract/v4/test_parameters.py
+++ b/traffic_ops/testing/api_contract/v4/test_parameters.py
@@ -14,45 +14,40 @@
"""API Contract Test Case for parameters endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["parameters"],
indirect=True)
-def test_parameter_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
- parameter_post_data: dict[str, object]
-) -> None:
+def test_parameter_contract(to_session: TOSession,
+ response_template_data: dict[str, Union[Primitive, list[Union[Primitive,
+ dict[str, object],
list[object]]],
+ dict[object, object]]], parameter_post_data: dict[str, object]) -> None:
"""
Test step to validate keys, values and data types from parameters
endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
:param parameter_post_data: Fixture to get sample parameter data and
actual parameter response.
"""
# validate Parameter keys from parameters get response
logger.info("Accessing /parameters endpoint through Traffic ops
session.")
- parameter = request_template_data[0]
- if not isinstance(parameter, dict):
- raise TypeError("malformed parameter in prerequisite data; not
an object")
-
- parameter_name = parameter.get("name")
+ parameter_name = parameter_post_data.get("name")
if not isinstance(parameter_name, str):
raise TypeError("malformed parameter in prerequisite data;
'name' not a string")
parameter_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_parameters(query_params={"name": parameter_name})
try:
@@ -63,38 +58,20 @@ def test_parameter_contract(
first_parameter = parameter_data[0]
if not isinstance(first_parameter, dict):
raise TypeError("malformed API response; first
Parameter in response is not an object")
- parameter_keys = set(first_parameter.keys())
+ logger.info("Parameter Api get response %s", first_parameter)
- logger.info("Parameter Keys from parameters endpoint response
%s", parameter_keys)
parameter_response_template =
response_template_data.get("parameters")
if not isinstance(parameter_response_template, dict):
raise TypeError(
f"Parameter response template data must be a
dict, not '{type(parameter_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template =
parameter_response_template.get("properties")
+
# validate parameter values from prereq data in parameters get
response.
- prereq_values = [parameter_post_data["name"],
parameter_post_data["value"],
- parameter_post_data["configFile"],
parameter_post_data["secure"]]
- get_values = [first_parameter["name"], first_parameter["value"],
- first_parameter["configFile"], first_parameter["secure"]]
- get_types = {}
- for key, value in first_parameter.items():
- get_types[key] = type(value).__name__
- logger.info("types from parameter get response %s", get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("types from parameters response template %s",
response_template_types)
+ keys = ["name", "value", "configFile", "secure"]
+ prereq_values = [parameter_post_data[key] for key in keys]
+ get_values = [first_parameter[key] for key in keys]
+
# validate keys, data types and values from parameters get json
response.
- assert parameter_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_parameter,
schema=parameter_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
diff --git a/traffic_ops/testing/api_contract/v4/test_phys_locations.py
b/traffic_ops/testing/api_contract/v4/test_phys_locations.py
index 265c739162..f6894864dc 100644
--- a/traffic_ops/testing/api_contract/v4/test_phys_locations.py
+++ b/traffic_ops/testing/api_contract/v4/test_phys_locations.py
@@ -14,44 +14,40 @@
"""API Contract Test Case for phys_locations endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["phys_locations"],
indirect=True)
-def test_phys_locations_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: list[dict[str, object] | list[object] |
primitive],
+def test_phys_locations_contract(to_session: TOSession,
+ response_template_data: list[Union[dict[str, object], list[object],
Primitive]],
phys_locations_post_data: dict[str, object]
-) -> None:
+ ) -> None:
"""
- Test step to validate keys, values and data types from phys_locations
endpoint
- response.
+ Test step to validate keys, values and data types from phys_locations
response.
:param to_session: Fixture to get Traffic Ops session.
- :param get_phys_location_data: Fixture to get phys_location data from a
prerequisites file.
- :param phys_location_prereq: Fixture to get sample phys_location data
and actual
+ :param response_template_data: Fixture to get response template data
from a prerequisites file.
+ :param phys_location_post_data: Fixture to get sample phys_location
data and actual
phys_location response.
"""
# validate phys_location keys from phys_locations get response
logger.info("Accessing /phys_locations endpoint through Traffic ops
session.")
- phys_location = request_template_data[0]
- if not isinstance(phys_location, dict):
- raise TypeError("malformed phys_location in prerequisite data;
not an object")
-
- phys_location_name = phys_location.get("name")
+ phys_location_name = phys_locations_post_data.get("name")
if not isinstance(phys_location_name, str):
raise TypeError("malformed phys_location in prerequisite data;
'name' not a string")
phys_location_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_physical_locations(query_params={"name":
phys_location_name})
try:
@@ -62,52 +58,21 @@ def test_phys_locations_contract(
first_phys_location = phys_location_data[0]
if not isinstance(first_phys_location, dict):
raise TypeError("malformed API response; first
phys_location in response is not an object")
- phys_location_keys = set(first_phys_location.keys())
- logger.info("phys_location Keys from phys_locations endpoint
response %s", phys_location_keys)
+ logger.info("phys_location Api response %s",
first_phys_location)
phys_location_response_template =
response_template_data.get("phys_locations")
if not isinstance(phys_location_response_template, dict):
raise TypeError(
f"phys_loc response template data must be a
dict, not'{type(phys_location_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template =
phys_location_response_template.get("properties")
- # validate phys_location values from prereq data in
phys_locations get response.
- prereq_values = [
- phys_locations_post_data["name"],
- phys_locations_post_data["address"],
- phys_locations_post_data["city"],
- phys_locations_post_data["zip"],
- phys_locations_post_data["comments"],
- phys_locations_post_data["email"],
- phys_locations_post_data["phone"],
- phys_locations_post_data["poc"],
- phys_locations_post_data["regionId"],
- phys_locations_post_data["shortName"],
- phys_locations_post_data["state"]]
- get_values = [first_phys_location["name"],
first_phys_location["address"],
- first_phys_location["city"], first_phys_location["zip"],
first_phys_location["comments"],
- first_phys_location["email"], first_phys_location["phone"],
first_phys_location["poc"],
- first_phys_location["regionId"], first_phys_location["shortName"],
first_phys_location["state"]]
+ # validate phys_location values from prereq data in
phys_locations get response.
+ keys = ["name", "address", "city", "zip", "comments", "email",
"phone","poc",
+ "regionId", "shortName", "state"]
+ prereq_values = [phys_locations_post_data[key] for key in keys]
+ get_values = [first_phys_location[key] for key in keys]
- get_types = {}
- for key, value in first_phys_location.items():
- get_types[key] = type(value).__name__
- logger.info("types from phys_location get response %s",
get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("types from phys_location response template %s",
response_template_types)
#validate keys, data types and values from regions get json
response.
- assert phys_location_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_phys_location,
schema=phys_location_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
diff --git a/traffic_ops/testing/api_contract/v4/test_profiles.py
b/traffic_ops/testing/api_contract/v4/test_profiles.py
index 17164ca3ae..67e124645a 100644
--- a/traffic_ops/testing/api_contract/v4/test_profiles.py
+++ b/traffic_ops/testing/api_contract/v4/test_profiles.py
@@ -14,45 +14,39 @@
"""API Contract Test Case for profiles endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
[email protected]('request_template_data', ['profiles'], indirect=True)
-def test_profile_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
- profile_post_data: dict[str, object]
-) -> None:
+def test_profile_contract(to_session: TOSession,
+ response_template_data: dict[str, Union[Primitive,
+ list[Union[Primitive, dict[str,
object], list[object]]],
+ dict[object, object]]], profile_post_data: dict[str, object]) -> None:
"""
Test step to validate keys, values and data types from profiles endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
:param profile_post_data: Fixture to get sample Profile data and actual
Profile response.
"""
# validate Profile keys from profiles get response
logger.info("Accessing /profiles endpoint through Traffic ops session.")
- profile = request_template_data[0]
- if not isinstance(profile, dict):
- raise TypeError("malformed profile in prerequisite data; not an
object")
-
- profile_name = profile.get("name")
+ profile_name = profile_post_data.get("name")
if not isinstance(profile_name, str):
raise TypeError("malformed profile in prerequisite data; 'name'
not a string")
profile_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_profiles(query_params={"name": profile_name})
try:
@@ -63,39 +57,18 @@ def test_profile_contract(
first_profile = profile_data[0]
if not isinstance(first_profile, dict):
raise TypeError("malformed API response; first Profile
in response is not an object")
- profile_keys = set(first_profile.keys())
+ logger.info("Profile Api get response %s", first_profile)
- logger.info("Profile Keys from profiles endpoint response %s",
profile_keys)
profile_response_template =
response_template_data.get("profiles")
if not isinstance(profile_response_template, dict):
raise TypeError(
f"Profile response template data must be a
dict, not '{type(profile_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template = profile_response_template.get("properties")
+
# validate profile values from prereq data in profiles get
response.
- prereq_values = [
- profile_post_data["name"],
- profile_post_data["cdn"]
- ]
+ prereq_values = [profile_post_data["name"],
profile_post_data["cdn"]]
get_values = [first_profile["name"], first_profile["cdn"]]
- get_types = {}
- for key, value in first_profile.items():
- get_types[key] = type(value).__name__
- logger.info("Types from profile get response %s", get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("Types from profile response template %s",
response_template_types)
- assert profile_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_profile,
schema=profile_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
diff --git a/traffic_ops/testing/api_contract/v4/test_regions.py
b/traffic_ops/testing/api_contract/v4/test_regions.py
index 58466984ad..7a0a6fb795 100644
--- a/traffic_ops/testing/api_contract/v4/test_regions.py
+++ b/traffic_ops/testing/api_contract/v4/test_regions.py
@@ -14,45 +14,40 @@
"""API Contract Test Case for regions endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["regions"], indirect=True)
-def test_region_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
+def test_region_contract(to_session: TOSession,
+ response_template_data: dict[str, Union[Primitive,
list[Union[Primitive,
+ dict[str, object], list[object]]], dict[object, object]]],
region_post_data: dict[str, object]
-) -> None:
+ ) -> None:
"""
- Test step to validate keys, values and data types from regions endpoint
- response.
+ Test step to validate keys, values and data types from regions response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
:param region_post_data: Fixture to get sample region data and actual
region response.
"""
# validate region keys from regions get response
logger.info("Accessing /regions endpoint through Traffic ops session.")
- region = request_template_data[0]
- if not isinstance(region, dict):
- raise TypeError("malformed region in prerequisite data; not an
object")
-
- region_name = region.get("name")
+ region_name = region_post_data.get("name")
if not isinstance(region_name, str):
raise TypeError("malformed region in prerequisite data; 'name'
not a string")
region_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_regions(query_params={"name": region_name})
try:
@@ -63,37 +58,20 @@ def test_region_contract(
first_region = region_data[0]
if not isinstance(first_region, dict):
raise TypeError("malformed API response; first region
in response is not an object")
- region_keys = set(first_region.keys())
- logger.info("region Keys from regions endpoint response %s",
region_keys)
+ logger.info("Region Api response %s", first_region)
region_response_template = response_template_data.get("regions")
if not isinstance(region_response_template, dict):
raise TypeError(
f"region response template data must be a dict,
not '{type(region_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template = region_response_template.get("properties")
+
# validate region values from prereq data in regions get
response.
- prereq_values = [region_post_data["name"],
region_post_data["division"],
- region_post_data["divisionName"]]
- get_values = [first_region["name"], first_region["division"],
first_region["divisionName"]]
- get_types = {}
- for key, value in first_region.items():
- get_types[key] = type(value).__name__
- logger.info("types from region get response %s", get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("types from regions response template %s",
response_template_types)
+ keys = ["name", "division", "divisionName"]
+ prereq_values = [region_post_data[key] for key in keys]
+ get_values = [first_region[key] for key in keys]
+
# validate keys, data types and values from regions get json
response.
- assert region_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_region,
schema=region_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
diff --git a/traffic_ops/testing/api_contract/v4/test_roles.py
b/traffic_ops/testing/api_contract/v4/test_roles.py
index bdfd692647..79b4bda342 100644
--- a/traffic_ops/testing/api_contract/v4/test_roles.py
+++ b/traffic_ops/testing/api_contract/v4/test_roles.py
@@ -14,45 +14,40 @@
"""API Contract Test Case for roles endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["roles"], indirect=True)
-def test_role_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
- role_post_data: dict[str, object]
-) -> None:
+def test_role_contract(to_session: TOSession,
+ response_template_data: dict[str, Union[Primitive, list[Union[Primitive,
+ dict[str, object],
list[object]]],
+ dict[object, object]]], role_post_data: dict[str, object]) -> None:
"""
Test step to validate keys, values and data types from roles endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
:param role_post_data: Fixture to get sample role data and actual role
response.
"""
# validate Role keys from roles get response
logger.info("Accessing /roles endpoint through Traffic ops session.")
- role = request_template_data[0]
- if not isinstance(role, dict):
- raise TypeError("malformed role in prerequisite data; not an
object")
-
- role_name = role.get("name")
+ role_name = role_post_data.get("name")
if not isinstance(role_name, str):
raise TypeError("malformed role in prerequisite data; 'name'
not a string")
role_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_roles(query_params={"name": role_name})
try:
@@ -63,42 +58,25 @@ def test_role_contract(
first_role = role_data[0]
if not isinstance(first_role, dict):
raise TypeError("malformed API response; first role in
response is not an object")
- role_keys = set(first_role.keys())
+ logger.info("Role Api get response %s", first_role)
- logger.info("Role Keys from roles endpoint response %s",
role_keys)
role_response_template = response_template_data.get("roles")
if not isinstance(role_response_template, dict):
raise TypeError(
f"Role response template data must be a dict,
not '{type(role_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template = role_response_template.get("properties")
+
# validate roles values from prereq data in roles get response.
prereq_values = [role_post_data["name"],
role_post_data["description"]]
get_values = [first_role["name"], first_role["description"]]
- get_types = {}
- for key, value in first_role.items():
- get_types[key] = type(value).__name__
- logger.info("types from role get response %s", get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("types from role response template %s",
response_template_types)
+
# validate keys,data types for values and values from roles get
json response.
- assert role_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_role,
schema=role_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
pytest.fail("API contract test failed for roles endpoint: API
response was malformed")
finally:
- # Delete Roel after test execution to avoid redundancy.
+ # Delete Role after test execution to avoid redundancy.
try:
role_name = role_post_data["name"]
to_session.delete_role(query_params={"name": role_name})
diff --git a/traffic_ops/testing/api_contract/v4/test_server_capabilities.py
b/traffic_ops/testing/api_contract/v4/test_server_capabilities.py
index fb5723d2a4..59066d7302 100644
--- a/traffic_ops/testing/api_contract/v4/test_server_capabilities.py
+++ b/traffic_ops/testing/api_contract/v4/test_server_capabilities.py
@@ -14,29 +14,28 @@
"""API Contract Test Case for server_capabilities endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["server_capabilities"],
indirect=True)
-def test_server_capabilities_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
- server_capabilities_post_data: dict[str, object]
-) -> None:
+def test_server_capabilities_contract(to_session: TOSession,
+ response_template_data: dict[str, Union[Primitive, list[Union[Primitive,
+ dict[str, object],
list[object]]], dict[object, object]]],
+ server_capabilities_post_data: dict[str, object]) -> None:
"""
Test step to validate keys, values and data types from
server_capabilities endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
:param server_capabilities_post_data: Fixture to get sample
server_capabilities data
and actual server_capabilities response.
@@ -44,16 +43,12 @@ def test_server_capabilities_contract(
# validate server_capabilities keys from server_capabilities get
response
logger.info("Accessing /server_capabilities endpoint through Traffic
ops session.")
- server_capabilities = request_template_data[0]
- if not isinstance(server_capabilities, dict):
- raise TypeError("malformed server_capabilities in prerequisite
data; not an object")
-
- server_capabilities_name = server_capabilities.get("name")
+ server_capabilities_name = server_capabilities_post_data.get("name")
if not isinstance(server_capabilities_name, str):
raise TypeError("malformed server_capabilities in prerequisite
data; 'name' not a string")
server_capabilities_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_server_capabilities(query_params={"name":
server_capabilities_name})
try:
@@ -64,36 +59,16 @@ def test_server_capabilities_contract(
first_server_capabilities = server_capabilities_data[0]
if not isinstance(first_server_capabilities, dict):
raise TypeError("malformed API response; first
server_capabilities in response is not an object")
- server_capabilities_keys = set(first_server_capabilities.keys())
+ logger.info("Server capabilities Api get response %s",
first_server_capabilities)
+
+ response_template =
response_template_data.get("server_capabilities")
- logger.info("server_capabilities Keys from endpoint response
%s", server_capabilities_keys)
- server_capabilities_response_template =
response_template_data.get("server_capabilities")
- if not isinstance(server_capabilities_response_template, dict):
- raise TypeError(
- f"server_capabilities data must be a dict, not
'{type(server_capabilities_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template =
server_capabilities_response_template.get("properties")
# validate server_capabilities values from prereq data in api
get response.
- prereq_values = [server_capabilities_post_data["name"]]
- get_values = [first_server_capabilities["name"]]
- get_types = {}
- for key, value in first_server_capabilities.items():
- get_types[key] = type(value).__name__
- logger.info("types from server_capabilities get response %s",
get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("types from server_capabilities response template
%s", response_template_types)
+ prereq_values = server_capabilities_post_data["name"]
+ get_values = first_server_capabilities["name"]
+
# validate keys, data types and values from server_capabilities
get json response.
- assert server_capabilities_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_server_capabilities,
schema=response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
diff --git a/traffic_ops/testing/api_contract/v4/test_servers.py
b/traffic_ops/testing/api_contract/v4/test_servers.py
new file mode 100644
index 0000000000..37d7245cfd
--- /dev/null
+++ b/traffic_ops/testing/api_contract/v4/test_servers.py
@@ -0,0 +1,84 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""API Contract Test Case for servers endpoint."""
+import logging
+from typing import Union
+
+import pytest
+import requests
+from jsonschema import validate
+
+from trafficops.tosession import TOSession
+
+# Create and configure logger
+logger = logging.getLogger()
+
+Primitive = Union[bool, int, float, str, None]
+
+
+def test_server_contract(to_session: TOSession,
+ response_template_data: dict[str, Union[Primitive, list[Union[Primitive,
+ dict[str, object],
list[object]]],
+ dict[object, object]]], server_post_data: dict[str, object]) -> None:
+ """
+ Test step to validate keys, values and data types from servers endpoint
+ response.
+ :param to_session: Fixture to get Traffic Ops session.
+ :param response_template_data: Fixture to get response template data
from a prerequisites file.
+ :param server_post_data: Fixture to get sample server data and actual
server response.
+ """
+ # validate server keys from server get response
+ logger.info("Accessing /servers endpoint through Traffic ops session.")
+
+ server_id = server_post_data.get("id")
+ if not isinstance(server_id, int):
+ raise TypeError("malformed API response; 'id' property not a
integer")
+
+ server_get_response: tuple[
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
+ requests.Response
+ ] = to_session.get_servers(query_params={"id": server_id})
+ try:
+ server_data = server_get_response[0]
+ if not isinstance(server_data, list):
+ raise TypeError("malformed API response; 'response'
property not an array")
+
+ first_server = server_data[0]
+ if not isinstance(first_server, dict):
+ raise TypeError("malformed API response; first Server
in response is not an object")
+ logger.info("Server Api get response %s", first_server)
+ server_response_template = response_template_data.get("servers")
+ if not isinstance(server_response_template, dict):
+ raise TypeError(
+ f"Server response template data must be a dict,
not '{type(server_response_template)}'")
+
+ keys = ["cachegroupId", "cdnId", "domainName",
"physLocationId", "profileNames",
+ "statusId", "typeId"]
+ prereq_values = [server_post_data[key] for key in keys]
+ get_values = [first_server[key] for key in keys]
+
+ assert validate(instance=first_server,
schema=server_response_template) is None
+ assert get_values == prereq_values
+ except IndexError:
+ logger.error("Either prerequisite data or API response was
malformed")
+ pytest.fail("API contract test failed for server endpoint: API
response was malformed")
+ finally:
+ # Delete Server after test execution to avoid redundancy.
+ try:
+ server_id = server_post_data["id"]
+ to_session.delete_server_by_id(server_id=server_id)
+ except IndexError:
+ logger.error("Profile returned by Traffic Ops is
missing an 'id' property")
+ pytest.fail("Response from delete request is empty,
Failing test_server_contract")
diff --git a/traffic_ops/testing/api_contract/v4/test_tenants.py
b/traffic_ops/testing/api_contract/v4/test_tenants.py
index 9ff9cd0e57..ee643b693b 100644
--- a/traffic_ops/testing/api_contract/v4/test_tenants.py
+++ b/traffic_ops/testing/api_contract/v4/test_tenants.py
@@ -14,45 +14,40 @@
"""API Contract Test Case for tenants endpoint."""
import logging
+from typing import Union
+
import pytest
import requests
+from jsonschema import validate
from trafficops.tosession import TOSession
# Create and configure logger
logger = logging.getLogger()
-primitive = bool | int | float | str | None
+Primitive = Union[bool, int, float, str, None]
+
[email protected]('request_template_data', ["tenants"], indirect=True)
-def test_tenant_contract(
- to_session: TOSession,
- request_template_data: list[dict[str, object] | list[object] |
primitive],
- response_template_data: dict[str, primitive | list[primitive |
dict[str, object]
- | list[object]] |
dict[object, object]],
- tenant_post_data: dict[str, object]
-) -> None:
+def test_tenant_contract(to_session: TOSession,
+ response_template_data: dict[str, Union[Primitive, list[Union[Primitive,
+ dict[str, object],
list[object]]], dict[object, object]]],
+ tenant_post_data: dict[str, object]) -> None:
"""
Test step to validate keys, values and data types from tenants endpoint
response.
:param to_session: Fixture to get Traffic Ops session.
- :param request_template_data: Fixture to get request template data from
a prerequisites file.
:param response_template_data: Fixture to get response template data
from a prerequisites file.
:param tenant_post_data: Fixture to get sample tenant data and actual
tenant response.
"""
# validate tenant keys from tenants get response
logger.info("Accessing /tenants endpoint through Traffic ops session.")
- tenant = request_template_data[0]
- if not isinstance(tenant, dict):
- raise TypeError("malformed tenant in prerequisite data; not an
object")
-
- tenant_name = tenant.get("name")
+ tenant_name = tenant_post_data.get("name")
if not isinstance(tenant_name, str):
raise TypeError("malformed tenant in prerequisite data; 'name'
not a string")
tenant_get_response: tuple[
- dict[str, object] | list[dict[str, object] | list[object] |
primitive] | primitive,
+ Union[dict[str, object], list[Union[dict[str, object],
list[object], Primitive]], Primitive],
requests.Response
] = to_session.get_tenants(query_params={"name": tenant_name})
try:
@@ -63,37 +58,20 @@ def test_tenant_contract(
first_tenant = tenant_data[0]
if not isinstance(first_tenant, dict):
raise TypeError("malformed API response; first tenant
in response is not an object")
- tenant_keys = set(first_tenant.keys())
+ logger.info("Tenant Api get response %s", first_tenant)
- logger.info("tenant Keys from tenants endpoint response %s",
tenant_keys)
tenant_response_template = response_template_data.get("tenants")
if not isinstance(tenant_response_template, dict):
raise TypeError(
f"tenant response template data must be a dict,
not '{type(tenant_response_template)}'")
- response_template: dict[str, list[dict[str, object] |
list[object] | primitive] |\
- dict[object, object] |\
- primitive
- ]
- response_template = tenant_response_template.get("properties")
+
# validate tenant values from prereq data in tenants get
response.
- prereq_values = [tenant_post_data["name"],
tenant_post_data["active"],
- tenant_post_data["parentId"]]
- get_values = [first_tenant["name"], first_tenant["active"],
first_tenant["parentId"]]
- get_types = {}
- for key, value in first_tenant.items():
- get_types[key] = type(value).__name__
- logger.info("types from tenant get response %s", get_types)
- response_template_types= {}
- for key, value in response_template.items():
- actual_type = value.get("type")
- if not isinstance(actual_type, str):
- raise TypeError(
- f"Type data must be a string, not
'{type(actual_type)}'")
- response_template_types[key] = actual_type
- logger.info("types from tenants response template %s",
response_template_types)
+ keys = ["name", "active", "parentId"]
+ prereq_values = [tenant_post_data[key] for key in keys]
+ get_values = [first_tenant[key] for key in keys]
+
# validate keys, data types and values from tenants get json
response.
- assert tenant_keys == set(response_template.keys())
- assert dict(sorted(get_types.items())) ==
dict(sorted(response_template_types.items()))
+ assert validate(instance=first_tenant,
schema=tenant_response_template) is None
assert get_values == prereq_values
except IndexError:
logger.error("Either prerequisite data or API response was
malformed")
