traffic_ops/testing/api_contract/v4/conftest.py (1,237 lines of code) (raw):

# # 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. # """ This module is used to create a Traffic Ops session and to store prerequisite data for endpoints. """ import json import logging import shutil import sys import os from random import randint from typing import Any, NamedTuple, Union, Optional, TypeAlias from urllib.parse import urlparse import munch import psycopg2 import pytest import requests from trafficops.tosession import TOSession from trafficops.restapi import OperationError # Create and configure logger logger = logging.getLogger() Primitive = Union[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): """Represents an API version.""" major: int minor: int @staticmethod def from_string(ver_str: str) -> "APIVersion": """ Instantiates a new version from a string. >>> APIVersion.from_string("4.0") APIVersion(major=4, minor=0) >>> try: ... APIVersion("not a version string") ... except ValueError: ... print("whoops") ... whoops >>> >>> try: ... APIVersion("4.Q") ... except ValueError: ... print("whoops") ... whoops """ parts = ver_str.split(".", 1) if len(parts) != 2: raise ValueError("invalid version; must be of the form '{{major}}.{{minor}}'") return APIVersion(int(parts[0]), int(parts[1])) def __str__(self) -> str: """ Coalesces the version to a string. >>> print(APIVersion(4, 1)) 4.1 """ return f"{self.major}.{self.minor}" APIVersion.major.__doc__ = """The API's major version number.""" APIVersion.major.__doc__ = """The API's minor version number.""" class ArgsType(NamedTuple): """Represents the configuration needed to create Traffic Ops session.""" user: str password: str url: str port: int api_version: APIVersion def __str__(self) -> str: """ Formats the configuration as a string. Omits password and extraneous properties. >>> print(ArgsType("user", "password", "url", 420, APIVersion(4, 0))) User: 'user', URL: 'url' """ return f"User: '{self.user}', URL: '{self.url}'" ArgsType.user.__doc__ = """The username used for authentication.""" ArgsType.password.__doc__ = """The password used for authentication.""" ArgsType.url.__doc__ = """The URL of the environment.""" ArgsType.port.__doc__ = """The port number on which to connect to Traffic Ops.""" ArgsType.api_version.__doc__ = """The version number of the API to use.""" class DbArgsType(NamedTuple): """Represents the configuration needed to create Traffic Ops Database connection.""" db_name: str user: str password: str hostname: str port: int sslmode: str def __str__(self) -> str: """ Formats the configuration as a string. Omits password and extraneous properties. >>> print(ArgsType("db_name", "user", "password", "hostname", "port", "sslmode")) Dbname: 'db_name', User: 'user' """ return f"User: '{self.db_name}', : '{self.user}'" DbArgsType.db_name.__doc__ = """The DB name used for authentication.""" DbArgsType.user.__doc__ = """The DB username used for authentication.""" DbArgsType.password.__doc__ = """The DB password used for authentication.""" DbArgsType.hostname.__doc__ = """The DB hostname of the environment.""" DbArgsType.port.__doc__ = """The DB port number on which to connect to Traffic Ops.""" DbArgsType.sslmode.__doc__ = """Sslmode to use.""" @pytest.fixture(autouse=True, scope='function') def delete_pytest_cache(): """ Deletes cached data before every test case execution """ shutil.rmtree(".pytest_cache", ignore_errors=True) def pytest_addoption(parser: pytest.Parser) -> None: """ Parses the Traffic Ops arguments from command line. :param parser: Parser to parse command line arguments. """ parser.addoption( "--to-user", action="store", help="User name for Traffic Ops Session." ) parser.addoption( "--to-password", action="store", help="Password for Traffic Ops Session." ) parser.addoption( "--to-url", action="store", help="Traffic Ops URL." ) parser.addoption( "--to-db-name", action="store", help="Name for Traffic Ops Database." ) parser.addoption( "--to-db-user", action="store", help="User name for Traffic Ops Database." ) parser.addoption( "--to-db-password", action="store", help="Password for Traffic Ops Database." ) parser.addoption( "--to-db-hostname", action="store", help="Hostname for Traffic Ops Database." ) parser.addoption( "--to-db-port", action="store", help="Port for Traffic Ops Database." ) parser.addoption( "--to-db-sslmode", action="store", help="Sslmode for Traffic Ops Database." ) parser.addoption( "--config", help="Path to configuration file.", 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__), "data", "request_template.json") ) parser.addoption( "--response-template", help="Path to response prerequisites file.", default=os.path.join(os.path.dirname(__file__), "data", "response_template.json") ) def coalesce_config( arg: Optional[object], file_key: str, file_contents: Optional[dict[str, Optional[object]]], env_key: str ) -> Optional[str]: """ Coalesces configuration retrieved from different sources into a single string. This will raise a ValueError if the type of the configuration value in the parsed configuration file is not a string. In order of descending precedence this checks the command-line argument value, the configuration file value, and then the environment variable value. :param arg: The command-line argument value. :param file_key: The key under which to look in the parsed JSON configuration file data. :param file_contents: The parsed JSON configuration file (if one was used). :param env_key: The environment variable name to look for a value if one wasn't provided elsewhere. :returns: The coalesced configuration value, or 'None' if no value could be determined. """ if isinstance(arg, str): return arg if file_contents: file_value = file_contents.get(file_key) if isinstance(file_value, str): return file_value if file_value is not None: raise ValueError(f"incorrect value; want: 'str', got: '{type(file_value)}'") return os.environ.get(env_key) 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") (APIVersion(major=5, minor=270), 420) >>> parse_to_url("trafficops.example.test") (APIVersion(major=4, minor=0), 443) """ parsed = urlparse(raw) if not parsed.netloc: raise ValueError("missing network location (hostname & optional port)") if parsed.scheme and parsed.scheme.lower() != "https": raise ValueError("invalid scheme; must use HTTPS") port = 443 if ":" in parsed.netloc: port_str = parsed.netloc.split(":")[-1] try: port = int(port_str) except ValueError as e: raise ValueError(f"invalid port number: {port_str}") from e api_version = APIVersion(4, 0) if parsed.path and parsed.path != "/": ver_str = parsed.path.lstrip("/api/").split("/", 1)[0] if not ver_str: raise ValueError(f"invalid API path: {parsed.path} (should be e.g. '/api/4.0')") api_version = APIVersion.from_string(ver_str) else: logging.warning("using default API version: %s", api_version) return (api_version, port) @pytest.fixture(name="to_db_args", scope="session") def to_db_data(pytestconfig: pytest.Config) -> DbArgsType: """ PyTest fixture to store Traffic ops database arguments passed from command line. :param pytestconfig: Session-scoped fixture that returns the session's pytest.Config object. :returns: Configuration for connecting to Traffic Ops database. """ session_data: JSONData = None cfg_path = pytestconfig.getoption("--config") if isinstance(cfg_path, str): try: with open(cfg_path, encoding="utf-8", mode="r") as session_file: session_data = json.load(session_file) except (FileNotFoundError, PermissionError) as read_err: raise ValueError(f"could not read configuration file at '{cfg_path}'") from read_err if session_data is not None and not isinstance(session_data, dict): raise ValueError( f"invalid configuration file; expected top-level object, got: {type(session_data)}" ) to_db_name = coalesce_config(pytestconfig.getoption("--to-db-name"), "db_name", session_data, "TO_DB_NAME") if not to_db_name: raise ValueError( "Traffic Ops Database name is not configured - use '--to-db-name', the config file, or an " "environment variable to do so" ) to_db_user = coalesce_config(pytestconfig.getoption("--to-db-user"), "db_user", session_data, "TO_DB_USER") if not to_db_user: raise ValueError( "Traffic Ops Database Username is not configured - use '--to-db-user', the config file, or an " "environment variable to do so" ) to_db_password = coalesce_config(pytestconfig.getoption("--to-db-password"), "db_password", session_data, "TO_DB_PASSWORD") if not to_db_password: raise ValueError( "Traffic Ops Database password is not configured - use '--to-db-password', the config file," "or an environment variable to do so" ) to_db_hostname = coalesce_config(pytestconfig.getoption("--to-db-hostname"), "db_hostname", session_data, "TO_DB_HOSTNAME") if not to_db_hostname: raise ValueError( "Traffic Ops Database Hostname is not configured - use '--to-db-hostname', the config file," "or an environment variable to do so" ) to_db_port = coalesce_config(pytestconfig.getoption("--to-db-port"), "db_port", session_data, "TO_DB_PORT") if not to_db_port: raise ValueError( "Traffic Ops Database Port is not configured - use '--to-db-port', the config file, or an " "environment variable to do so" ) port = int(to_db_port) to_db_sslmode = coalesce_config(pytestconfig.getoption("--to-db-sslmode"), "db_sslmode", session_data, "TO_DB_SSLMODE") if not to_db_sslmode: raise ValueError( "Traffic Ops Database Sslmode is not configured - use '--to-db-sslmode', the config file, or an " "environment variable to do so" ) return DbArgsType( to_db_name, to_db_user, to_db_password, to_db_hostname, port, to_db_sslmode ) @pytest.fixture(name="to_args") def to_data(pytestconfig: pytest.Config) -> ArgsType: """ PyTest fixture to store Traffic ops arguments passed from command line. :param pytestconfig: Session-scoped fixture that returns the session's pytest.Config object. :returns: Configuration for connecting to Traffic Ops. """ session_data: JSONData = None cfg_path = pytestconfig.getoption("--config") if isinstance(cfg_path, str): try: with open(cfg_path, encoding="utf-8", mode="r") as session_file: session_data = json.load(session_file) except (FileNotFoundError, PermissionError) as read_err: raise ValueError(f"could not read configuration file at '{cfg_path}'") from read_err if session_data is not None and not isinstance(session_data, dict): raise ValueError( f"invalid configuration file; expected top-level object, got: {type(session_data)}" ) to_user = coalesce_config(pytestconfig.getoption("--to-user"), "user", session_data, "TO_USER") if not to_user: raise ValueError( "Traffic Ops user is not configured - use '--to-user', the config file, or an " "environment variable to do so" ) to_password = coalesce_config( pytestconfig.getoption("--to-password"), "password", session_data, "TO_PASSWORD" ) if not to_password: raise ValueError( "Traffic Ops password is not configured - use '--to-password', the config file, or an " "environment variable to do so" ) to_url = coalesce_config(pytestconfig.getoption("--to-url"), "url", session_data, "TO_USER") if not to_url: raise ValueError( "Traffic Ops URL is not configured - use '--to-url', the config file, or an " "environment variable to do so" ) try: api_version, port = parse_to_url(to_url) except ValueError as e: raise ValueError("invalid Traffic Ops URL") from e return ArgsType( to_user, to_password, to_url, port, api_version ) @pytest.fixture(name="to_session") def to_login(to_args: ArgsType) -> TOSession: """ PyTest Fixture to create a Traffic Ops session from Traffic Ops Arguments passed as command line arguments in to_args fixture in conftest. :param to_args: Fixture to get Traffic ops session arguments. :returns: An authenticated Traffic Ops session. """ # Create a Traffic Ops V4 session and login to_url = urlparse(to_args.url) to_host = to_url.hostname try: to_session = TOSession( host_ip=to_host, host_port=to_args.port, api_version=str(to_args.api_version), ssl=True, verify_cert=False ) logger.info("Established Traffic Ops Session.") except OperationError as error: logger.debug("%s", error, exc_info=True, stack_info=True) logger.error("Failure in Traffic Ops session creation. Reason: %s", error) sys.exit(-1) # Login To TO_API to_session.login(to_args.user, to_args.password) logger.info("Successfully logged into Traffic Ops.") return to_session @pytest.fixture(scope="session", name="db_connection") def to_db_connection(to_db_args: DbArgsType) -> psycopg2.connect: """ Creates new traffic ops db connection. :returns: New Traffic ops database connection """ to_db_connection = None try: to_db_connection = psycopg2.connect( user=to_db_args.user, password=to_db_args.password, host=to_db_args.hostname, port=to_db_args.port, database=to_db_args.db_name, sslmode=to_db_args.sslmode ) logger.info("Successfully connected to the Traffic Ops database.") yield to_db_connection except psycopg2.OperationalError as e: logger.error("Error connecting to the Traffic Ops database : %s", e) finally: if to_db_connection: to_db_connection.close() logger.info("Closed Traffic ops DB connection.") @pytest.fixture(name="request_template_data", scope="session") def request_prerequiste_data(pytestconfig: pytest.Config, request: pytest.FixtureRequest ) -> 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. :param request: Fixture to access information about the requesting test function and its fixtures :returns: Prerequisite request data for api endpoint. """ request_template_path = pytestconfig.getoption("--request-template") if not isinstance(request_template_path, str): # unlike the configuration file, this must be present raise ValueError("prereqisites path not configured") # Response keys for api endpoint data: Union[dict[ str, 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)}'") 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, 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. :returns: Prerequisite response data for api endpoint. """ prereq_path = pytestconfig.getoption("--response-template") if not isinstance(prereq_path, str): # unlike the configuration file, this must be present raise ValueError("prereqisites path not configured") # Response keys for api endpoint response_template: Union[dict[ str, 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): raise TypeError(f"Response template data must be an object, not '{type(response_template)}'") 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], create_check: bool) -> 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 create_check and not isinstance(api_response, munch.Munch): raise ValueError("Malformed API response; 'response' property not an munch") if not create_check and not isinstance(api_response, list): raise ValueError("Malformed API response; 'response' property not an list") else: raise ValueError("Invalid API response format") if api_response: try: api_data = api_response if not create_check: 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, create_check = False) 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, create_check = True) 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 generate_unique_data(to_session: TOSession, base_name: str, object_type: str, query_key=None)-> str: """ Generate unique data for the given endpoint. :param to_session: Fixture to get Traffic Ops session. :param object_type: api call name for get request. :param base_name: Base name for get request. :param query_key: Hitting get request using specific query key. :returns: Unique name for the corresponding api request. @param data: """ unique_name = base_name if query_key is None: query_key = "name" while True: try: response = getattr(to_session, f"get_{object_type}")(query_params={query_key:unique_name}) # Check if query params works for the corresponding api. check_data = response[0] if len(check_data) > 1: logger.info("API response returns all api data, query params is not working.") if not any(data.get(query_key) == unique_name for data in check_data): return unique_name elif len(check_data) == 0: raise ValueError("No Api response with the unique data") except ValueError: return unique_name unique_name = base_name[:4] + str(randint(0, 1000)) 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(name="cdn_post_data") def cdn_data_post(to_session: TOSession, request_template_data: list[JSONData], pytestconfig: pytest.Config) -> dict[str, object]: """ PyTest Fixture to create POST data for cdns endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get CDN request template data from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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)) try: name = cdn["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") cdn_name = name[:4] + randstr cdn["name"] = generate_unique_data(to_session=to_session, base_name=cdn_name, object_type="cdns") domain_name = cdn["domainName"] if not isinstance(domain_name, str): raise TypeError(f"domainName must be str, not '{type(domain_name)}") domainname = domain_name[:5] + randstr cdn["domainName"] = generate_unique_data(to_session=to_session, base_name=domainname, object_type="cdns", query_key="domainName") 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", cdn) # Hitting cdns POST methed response: tuple[JSONData, requests.Response] = to_session.create_cdn(data=cdn) resp_obj = check_template_data(response, "cdns") pytestconfig.cache.set("cdnDomainName",resp_obj.get("domainName")) yield resp_obj cdn_id = resp_obj.get("id") msg = to_session.delete_cdn_by_id(cdn_id=cdn_id) logger.info("Deleting cdn data... %s", msg) if msg is None: logger.error("Cdn returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="cache_group_post_data") def cache_group_data_post(to_session: TOSession, request_template_data: list[JSONData], pytestconfig: pytest.Config) -> dict[str, object]: """ PyTest Fixture to create POST data for cachegroup endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get Cachegroup data from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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 = cache_group["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") cache_group_name = name[:4] + randstr cache_group["name"] = generate_unique_data(to_session=to_session, base_name=cache_group_name, object_type="cachegroups") short_name = cache_group["shortName"] if not isinstance(short_name, str): raise TypeError(f"shortName must be str, not '{type(short_name)}") cache_group["shortName"] = short_name[:5] + randstr except KeyError as e: raise TypeError(f"missing Cache group property '{e.args[0]}'") from e # Check if type already exists, otherwise create it type_id = pytestconfig.cache.get("typeId", default=None) if type_id: cache_group["typeId"] = type_id else: 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=cache_group) resp_obj = check_template_data(response, "cachegroup") yield resp_obj cachegroup_id = resp_obj.get("id") msg = to_session.delete_cachegroups(cache_group_id=cachegroup_id) logger.info("Deleting cachegroup data... %s", msg) if msg is None: logger.error("Cachegroup returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="parameter_post_data") def parameter_data_post(to_session: TOSession, request_template_data: list[JSONData] ) -> dict[str, object]: """ 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 CDN request template data from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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: name = parameter["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") parameter_name = name[:4] + randstr parameter["name"] = generate_unique_data(to_session=to_session, base_name=parameter_name, object_type="parameters") value = parameter["value"] if not isinstance(value, str): raise TypeError(f"value must be str, not '{type(value)}") parameter_value = value[:5] + randstr parameter["value"] = generate_unique_data(to_session=to_session, base_name=parameter_value, object_type="parameters", query_key="value") 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", parameter) # Hitting cdns POST methed response: tuple[JSONData, requests.Response] = to_session.create_parameter(data=parameter) resp_obj = check_template_data(response, "parameter") yield resp_obj parameter_id = resp_obj.get("id") msg = to_session.delete_parameter(parameter_id=parameter_id) logger.info("Deleting parameter data... %s", msg) if msg is None: logger.error("Parameter returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="role_post_data") def role_data_post(to_session: TOSession, request_template_data: list[JSONData] ) -> dict[str, object]: """ PyTest Fixture to create POST data for roles endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get role data from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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)) try: name = role["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") role_name = name[:4] + randstr role["name"] = generate_unique_data(to_session=to_session, base_name=role_name, object_type="roles") 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", role) # Hitting roles POST methed response: tuple[JSONData, requests.Response] = to_session.create_role(data=role) resp_obj = check_template_data(response, "role") yield resp_obj role_name = resp_obj.get("name") msg = to_session.delete_role(query_params={"name": role_name}) logger.info("Deleting role data... %s", msg) if msg is None: logger.error("Role returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="profile_post_data") def profile_data_post(to_session: TOSession, request_template_data: list[JSONData], cdn_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for profile endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get profile data from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") profile_name = name[:4] + randstr profile["name"] = generate_unique_data(to_session=to_session, base_name=profile_name, object_type="profiles") except KeyError as e: raise TypeError(f"missing Profile property '{e.args[0]}'") from e # Check if cdn already exists, otherwise create it profile["cdn"] = cdn_post_data["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) resp_obj = check_template_data(response, "profile") yield resp_obj profile_id = resp_obj.get("id") msg = to_session.delete_profile_by_id(profile_id=profile_id) logger.info("Deleting profile data... %s", msg) if msg is None: logger.error("Profile returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="tenant_post_data") def tenant_data_post(to_session: TOSession, request_template_data: list[JSONData], pytestconfig: pytest.Config) -> dict[str, object]: """ PyTest Fixture to create POST data for tenants endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get tenant request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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)) name = pytestconfig.cache.get("tenantName", default=None) if name: tenant["name"] = name else: try: name = tenant["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") tenant_name = name[:4] + randstr tenant["name"] = generate_unique_data(to_session=to_session, base_name=tenant_name, object_type="tenants") 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", tenant) # Hitting tenants POST methed response: tuple[JSONData, requests.Response] = to_session.create_tenant(data=tenant) resp_obj = check_template_data(response, "tenant") yield resp_obj tenant_id = resp_obj.get("id") msg = to_session.delete_tenant(tenant_id=tenant_id) logger.info("Deleting tenant data... %s", msg) if msg is None: logger.error("Tenant returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="server_capabilities_post_data") def server_capabilities_data_post(to_session: TOSession, request_template_data: list[JSONData] ) -> dict[str, object]: """ PyTest Fixture to create POST data for server_capabilities endpoint. :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. """ 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)) try: name = server_capabilities["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") server_capabilities_name = name[:3] + randstr server_capabilities["name"] = generate_unique_data(to_session=to_session, base_name=server_capabilities_name, object_type="server_capabilities") except KeyError as e: raise TypeError(f"missing server_capabilities property '{e.args[0]}'") from e logger.info("New server_capabilities data to hit POST method %s", server_capabilities) # Hitting server_capabilities POST method response: tuple[ JSONData, requests.Response] = to_session.create_server_capabilities(data=server_capabilities) resp_obj = check_template_data(response, "server_capabilities") yield resp_obj server_capability_name = resp_obj.get("name") msg = to_session.delete_server_capabilities(query_params={"name": server_capability_name}) logger.info("Deleting server_capabilities data %s", msg) if msg is None: logger.error("server_capabilities returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_server_capabilities_contract") @pytest.fixture(name="division_post_data") def division_data_post(to_session: TOSession, request_template_data: list[JSONData] ) -> dict[str, object]: """ 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 data from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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)) try: name = division["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") division_name = name[:4] + randstr division["name"] = generate_unique_data(to_session=to_session, base_name=division_name, object_type="divisions") except KeyError as e: raise TypeError(f"missing Parameter property '{e.args[0]}'") from e logger.info("New division data to hit POST method %s", division) # Hitting division POST methed response: tuple[JSONData, requests.Response] = to_session.create_division(data=division) resp_obj = check_template_data(response, "divisions") yield resp_obj division_id = resp_obj.get("id") msg = to_session.delete_division(division_id=division_id) logger.info("Deleting division data... %s", msg) if msg is None: logger.error("Division returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="region_post_data") def region_data_post(to_session: TOSession, request_template_data: list[JSONData], division_post_data: dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for region endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get region data from a prerequisites file. :returns: Sample POST data and the actual API response. """ 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)) try: name = region["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") region_name = name[:4] + randstr region["name"] = generate_unique_data(to_session=to_session, base_name=region_name, object_type="regions") except KeyError as e: raise TypeError(f"missing Region property '{e.args[0]}'") from e # Check if division already exists, otherwise create it region["division"] = division_post_data["id"] region["divisionName"] = division_post_data["name"] logger.info("New region data to hit POST method %s", region) # Hitting region POST method response: tuple[JSONData, requests.Response] = to_session.create_region(data=region) resp_obj = check_template_data(response, "regions") yield resp_obj region_name = resp_obj.get("name") msg = to_session.delete_region(query_params={"name": region_name}) logger.info("Deleting region data... %s", msg) if msg is None: logger.error("Region returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(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. """ 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)) try: name = phys_locations["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") phys_locations_name = name[:4] + randstr phys_locations["name"] = generate_unique_data(to_session=to_session, base_name=phys_locations_name, object_type="physical_locations") short_name = phys_locations["shortName"] if not isinstance(short_name, str): raise TypeError(f"shortName must be str, not '{type(short_name)}'") shortname = short_name[:4] + randstr phys_locations["shortName"] = generate_unique_data(to_session=to_session, base_name=shortname, object_type="physical_locations", query_key="shortName") except KeyError as e: raise TypeError(f"missing Phys_location 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", phys_locations) # Hitting region POST method response: tuple[JSONData, requests.Response] = to_session.create_physical_locations( data=phys_locations) resp_obj = check_template_data(response, "phys_locations") yield resp_obj phys_location_id = resp_obj.get("id") msg = to_session.delete_physical_location(physical_location_id=phys_location_id) logger.info("Deleting physical locations data... %s", msg) if msg is None: logger.error("Physical location returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="server_post_data") def server_data_post(to_session: TOSession, request_template_data: list[JSONData], profile_post_data: dict[str, object], cache_group_post_data: dict[str, object], status_post_data: dict[str, object], phys_locations_post_data: dict[str, object], pytestconfig: pytest.Config)-> 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"] edge_type_id = pytestconfig.cache.get("edgeTypeId", default=None) if edge_type_id: server["typeId"] = edge_type_id else: server["typeId"] = type_id if edge_type_id: server["typeId"] = edge_type_id else: server["typeId"] = type_id pytestconfig.cache.set("typeId", type_id) server["cachegroupId"]= cache_group_post_data["id"] # Check if cdn already exists, otherwise create it server["cdnId"] = profile_post_data["cdn"] server["domainName"] = pytestconfig.cache.get("cdnDomainName", default=None) # Check if profile with cdn already exists, otherwise create it server["profileNames"] = [profile_post_data["name"]] # Check if status already exists, otherwise create it server["statusId"] = status_post_data["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") yield resp_obj server_id = resp_obj.get("id") msg = to_session.delete_server_by_id(server_id=server_id) logger.info("Deleting servers data... %s", msg) if msg is None: logger.error("Server returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="delivery_services_post_data") def delivery_services_data_post(to_session: TOSession, request_template_data: list[JSONData], tenant_post_data: dict[str, object], profile_post_data: dict[str, object], pytestconfig: pytest.Config) -> 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. :returns: Sample POST data and the actual API response. """ delivery_services = check_template_data( request_template_data["delivery_services"], "delivery_services") randstr = str(randint(0, 1000)) try: xml_id = delivery_services["xmlId"] if not isinstance(xml_id, str): raise TypeError(f"xmlId must be str, not '{type(xml_id)}'") xmlid= xml_id[:4] + randstr delivery_services["xmlId"] = generate_unique_data(to_session=to_session, base_name=xmlid, object_type="deliveryservices", query_key="xmlId") except KeyError as e: raise TypeError(f"missing delivery_services property '{e.args[0]}'") from e # Check if profile with cdn already exists, otherwise create it delivery_services["profileId"] = profile_post_data["id"] # Check if cdn already exists, otherwise create it delivery_services["cdnId"] = profile_post_data["cdn"] # Check if tenant already exists, otherwise create it pytestconfig.cache.set("tenantName", "root") delivery_services["tenantId"] = tenant_post_data["id"] # Check if type already exists, otherwise create it type_data = {"name": "HTTP", "useInTable":"deliveryservice"} type_object = create_or_get_existing(to_session, "types", "type", type_data, {"name": "HTTP", "useInTable":"deliveryservice"}) delivery_services["typeId"] = type_object["id"] delivery_services["type"] = type_object["name"] logger.info("New delivery_services data to hit POST method %s", delivery_services) # Hitting delivery_services POST method response: tuple[JSONData, requests.Response] = to_session.create_deliveryservice( data=delivery_services) resp_obj = check_template_data(response[0], "delivery_services") yield resp_obj delivery_service_id = resp_obj.get("id") msg = to_session.delete_deliveryservice_by_id(delivery_service_id=delivery_service_id) logger.info("Deleting delivery service data... %s", msg) if msg is None: logger.error("delivery service returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="origin_post_data") def origin_data_post(to_session: TOSession, request_template_data: list[JSONData], delivery_services_post_data: dict[str, object], tenant_post_data: dict[str, object] ) -> dict[str, object]: """ PyTest Fixture to create POST data for origins endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get profile data from a prerequisites file. :returns: Sample POST data and the actual API response. """ origin = check_template_data(request_template_data["origins"], "origins") randstr = str(randint(0, 1000)) try: name = origin["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") origin_name = name[:4] + randstr origin["name"] = generate_unique_data(to_session=to_session, base_name=origin_name, object_type="origins") except KeyError as e: raise TypeError(f"missing origin property '{e.args[0]}'") from e # Check if delivery_service already exists, otherwise create it delivery_services_id = delivery_services_post_data["id"] if not isinstance(delivery_services_id, int): raise TypeError("malformed API response; 'id' property not a integer") origin["deliveryServiceId"] = delivery_services_id # Check if tenant already exists, otherwise create it tenant_id = tenant_post_data["id"] if not isinstance(tenant_id, int): raise TypeError("malformed API response; 'id' property not a integer") origin["tenantId"] = tenant_id logger.info("New origin data to hit POST method %s", origin) # Hitting origins POST method response: tuple[JSONData, requests.Response] = to_session.create_origins(data=origin) resp_obj = check_template_data(response, "origins") yield resp_obj origin_id = resp_obj.get("id") msg = to_session.delete_origins(query_params={"id": origin_id}) logger.info("Deleting origin data... %s", msg) if msg is None: logger.error("Origin returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="status_post_data") def status_data_post(to_session: TOSession, request_template_data: list[JSONData] ) -> dict[str, object]: """ PyTest Fixture to create POST data for statuses endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get Status request template data from a prerequisite file. :returns: Sample POST data and the actual API response. """ status = check_template_data(request_template_data["status"], "status") # Return new post data and post response from statuses POST request randstr = str(randint(0, 1000)) try: name = status["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") status_name = name[:4] + randstr status["name"] = generate_unique_data(to_session=to_session, base_name=status_name, object_type="statuses") except KeyError as e: raise TypeError(f"missing Status property '{e.args[0]}'") from e logger.info("New status data to hit POST method %s", status) # Hitting statuses POST methed response: tuple[JSONData, requests.Response] = to_session.create_statuses(data=status) resp_obj = check_template_data(response, "statuses") yield resp_obj status_id = resp_obj.get("id") msg = to_session.delete_status_by_id(status_id=status_id) logger.info("Deleting status data... %s", msg) if msg is None: logger.error("Status returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="asn_post_data") def asn_data_post(to_session: TOSession, request_template_data: list[JSONData], cache_group_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for asn endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get asn data from a prerequisites file. :returns: Sample POST data and the actual API response. """ asn = check_template_data(request_template_data["asns"], "asns") # Return new post data and post response from asns POST request randstr = randint(0, 1000) asn["asn"] = randstr # Check if cachegroup already exists, otherwise create it asn["cachegroupId"] = cache_group_post_data["id"] logger.info("New profile data to hit POST method %s", asn) # Hitting asns POST method response: tuple[JSONData, requests.Response] = to_session.create_asn(data=asn) resp_obj = check_template_data(response, "asn") yield resp_obj asn_id = resp_obj.get("id") msg = to_session.delete_asn(query_params={"id": asn_id}) logger.info("Deleting asn data... %s", msg) if msg is None: logger.error("asn returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="job_post_data") def job_data_post(to_session: TOSession, request_template_data: list[JSONData], delivery_services_post_data: dict[str, object], ) -> dict[str, object]: """ PyTest Fixture to create POST data for jobss endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get job data from a prerequisites file. :returns: Sample POST data and the actual API response. """ job = check_template_data(request_template_data["jobs"], "jobs") # Check if delivery_service already exists, otherwise create it delivery_services_name = delivery_services_post_data["xmlId"] if not isinstance(delivery_services_name, str): raise TypeError("malformed API response; 'displayName' property not a string") job["deliveryService"] = delivery_services_name logger.info("New job data to hit POST method %s", job) # Hitting jobs POST method response: tuple[JSONData, requests.Response] = to_session.create_job(data=job) resp_obj = check_template_data(response, "jobs") yield resp_obj job_id = resp_obj.get("id") msg = to_session.delete_job(query_params={"id": job_id}) logger.info("Deleting job data... %s", msg) if msg is None: logger.error("job returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="coordinate_post_data") def coordinate_data_post(to_session: TOSession, request_template_data: list[JSONData] ) -> dict[str, object]: """ PyTest Fixture to create POST data for coordinates endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get coordinate request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ coordinate = check_template_data(request_template_data["coordinates"], "coordinates") # Return new post data and post response from coordinates POST request randstr = str(randint(0, 1000)) try: name = coordinate["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") coordinate_name = name[:4] + randstr coordinate["name"] = generate_unique_data(to_session=to_session, base_name=coordinate_name, object_type="coordinates") except KeyError as e: raise TypeError(f"missing coordinate property '{e.args[0]}'") from e logger.info("New coordinate data to hit POST method %s", coordinate) # Hitting coordinates POST methed response: tuple[JSONData, requests.Response] = to_session.create_coordinates(data=coordinate) resp_obj = check_template_data(response, "coordinate") yield resp_obj coordinate_id = resp_obj.get("id") msg = to_session.delete_coordinates(query_params={"id": coordinate_id}) logger.info("Deleting Coordinate data... %s", msg) if msg is None: logger.error("coordinate returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="user_post_data") def user_data_post(to_session: TOSession, request_template_data: list[JSONData], tenant_post_data:dict[str, object], db_connection: psycopg2.connect) -> dict[str, object]: """ PyTest Fixture to create POST data for users endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get users request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ user = check_template_data(request_template_data["users"], "users") # Return new post data and post response from users POST request randstr = str(randint(0, 1000)) try: username = user["username"] if not isinstance(username, str): raise TypeError(f"username must be str, not '{type(username)}'") unique_name = username[:4] + randstr user["username"] = generate_unique_data(to_session=to_session, base_name=unique_name, object_type="users", query_key="username") user_email = user["email"] if not isinstance(user_email, str): raise TypeError(f"user email must be str, not '{type(user_email)}'") email = randstr + user_email user["email"] = generate_unique_data(to_session=to_session, base_name=email, object_type="users", query_key="email") except KeyError as e: raise TypeError(f"missing user property '{e.args[0]}'") from e user["tenantId"] = tenant_post_data["id"] logger.info("New user data to hit POST method %s", user) # Hitting users POST methed response: tuple[JSONData, requests.Response] = to_session.create_user(data=user) resp_obj = check_template_data(response, "user") yield resp_obj user_id = resp_obj.get("id") # Create a cursor object to interact with the database cursor = db_connection.cursor() cursor.execute("DELETE FROM tm_user WHERE id = %s;", (user_id,)) # Commit the changes db_connection.commit() # Close the cursor cursor.close() @pytest.fixture(name="topology_post_data") def topology_data_post(to_session: TOSession, request_template_data: list[JSONData], server_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for topologies endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get topology request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ topology = check_template_data(request_template_data["topologies"], "topologies") # Return new post data and post response from topologies POST request randstr = str(randint(0, 1000)) try: name = topology["name"] if not isinstance(name, str): raise TypeError(f"name must be str, not '{type(name)}'") topology_name = name[:4] + randstr topology["name"] = generate_unique_data(to_session=to_session, base_name=topology_name, object_type="topologies") except KeyError as e: raise TypeError(f"missing topology property '{e.args[0]}'") from e cachegroup_name = server_post_data["cachegroup"] topology["nodes"][0]["cachegroup"] = cachegroup_name logger.info("New topology data to hit POST method %s", topology) # Hitting topology POST methed response: tuple[JSONData, requests.Response] = to_session.create_topology(data=topology) resp_obj = check_template_data(response, "topology") yield resp_obj topology_name = resp_obj.get("name") msg = to_session.delete_topology(name=topology_name) logger.info("Deleting topology data... %s", msg) if msg is None: logger.error("topology returned by Traffic Ops is missing an 'name' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="cdn_lock_post_data") def cdn_lock_data_post(to_session: TOSession, request_template_data: list[JSONData], user_post_data:dict[str, object], cdn_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for cdn_locks endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get cdn_locks request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ cdn_lock = check_template_data(request_template_data["cdn_locks"], "cdn_locks") # Return new post data and post response from cdn_locks POST request cdn_lock["cdn"] = cdn_post_data["name"] cdn_lock["sharedUserNames"][0] = user_post_data["username"] logger.info("New cdn_lock data to hit POST method %s", cdn_lock) # Hitting cdn_locks POST methed response: tuple[JSONData, requests.Response] = to_session.create_cdn_lock(data=cdn_lock) resp_obj = check_template_data(response, "cdn_lock") yield resp_obj cdn_name = resp_obj.get("cdn") msg = to_session.delete_cdn_lock(query_params={"cdn":cdn_name}) logger.info("Deleting cdn_lock data... %s", msg) if msg is None: logger.error("cdn_lock returned by Traffic Ops is missing an 'cdn' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="cdn_notification_post_data") def cdn_notification_data_post(to_session: TOSession, request_template_data: list[JSONData], cdn_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for cdn_notifications endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get cdn_notification request template. :returns: Sample POST data and the actual API response. """ cdn_notification = check_template_data( request_template_data["cdn_notifications"], "cdn_notifications") # Return new post data and post response from cdn_notifications POST request cdn_notification["cdn"] = cdn_post_data["name"] logger.info("New cdn_notification data to hit POST method %s", cdn_notification) # Hitting cdn_notification POST methed response: tuple[JSONData, requests.Response] = to_session.create_cdn_notification( data=cdn_notification) resp_obj = check_template_data(response, "cdn_notification") yield resp_obj notification_id = resp_obj.get("id") msg = to_session.delete_cdn_notification(query_params={"id":notification_id}) logger.info("Deleting cdn_notification data... %s", msg) if msg is None: logger.error("cdn_notfication returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="deliveryservice_request_post_data") def deliveryservice_request_data_post(to_session: TOSession, request_template_data: list[JSONData], delivery_services_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for deliveryservice_request endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get deliveryservice_request request template. :returns: Sample POST data and the actual API response. """ deliveryservice_request = check_template_data( request_template_data["deliveryservice_requests"], "deliveryservice_requests") # Return new post data and post response from deliveryservice_request POST request keys = ["displayName", "xmlId", "id", "cdnId", "tenantId", "type", "typeId"] for key in keys: deliveryservice_request["requested"][key] = delivery_services_post_data[key] logger.info("New deliveryservice_request data to hit POST method %s", deliveryservice_request) # Hitting deliveryservice_request POST methed response: tuple[JSONData, requests.Response] = to_session.create_deliveryservice_request( data=deliveryservice_request) resp_obj = check_template_data(response, "deliveryservice_request") yield resp_obj deliveryservice_request_id = resp_obj.get("id") msg = to_session.delete_deliveryservice_request(query_params={"id":deliveryservice_request_id}) logger.info("Deleting deliveryservice_request data... %s", msg) if msg is None: logger.error("deliveryservice_request returned by Traffic Ops is missing an 'id' property") @pytest.fixture(name="steering_post_data") def steering_data_post(to_session: TOSession, request_template_data: list[JSONData], delivery_services_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for steering endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get steering request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ steering = check_template_data(request_template_data["steering"], "steering") # Return new post data and post response from steering POST request ds_get_response = to_session.get_deliveryservices() ds_data = ds_get_response[0][0] delivery_service_id = ds_data.get("id") steering["targetId"] = delivery_services_post_data["id"] # 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": "steering_target"}) steering["typeId"]= type_object["id"] logger.info("New steering data to hit POST method %s", steering) # Hitting steering POST methed response: tuple[JSONData, requests.Response] = to_session.create_steering_targets( delivery_service_id=delivery_service_id, data=steering) resp_obj = check_template_data(response, "steering") yield resp_obj deliveryservice_id = resp_obj.get("deliveryServiceId") target_id = resp_obj.get("targetId") msg = to_session.delete_steering_targets( delivery_service_id=deliveryservice_id, target_id=target_id) logger.info("Deleting Steering data... %s", msg) if msg is None: logger.error("Steering returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="delivery_service_sslkeys_post_data") def delivery_service_sslkeys_data_post(to_session: TOSession, request_template_data: list[JSONData], cdn_post_data:dict[str, object], delivery_services_post_data:dict[str, object] ) -> dict[str, object]: """ PyTest Fixture to create POST data for delivery_service_sslkeys_post_data endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get delivery_service_sslkeys request template. :returns: Sample POST data and the actual API response. """ delivery_service_sslkeys = check_template_data( request_template_data["delivery_service_sslkeys"], "delivery_service_sslkeys") # Return new post data and post response from delivery_service_sslkeys POST request delivery_service_sslkeys["key"] = delivery_services_post_data["xmlId"] delivery_service_sslkeys["cdn"] = cdn_post_data["name"] logger.info("New delivery_service_sslkeys data to hit POST method %s", delivery_service_sslkeys) # Hitting delivery_service_sslkeys POST methed response: tuple[JSONData, requests.Response] = to_session.generate_deliveryservice_ssl_keys( data=delivery_service_sslkeys) yield delivery_service_sslkeys deliveryservice_xml_id = delivery_service_sslkeys["key"] msg = to_session.delete_deliveryservice_ssl_keys_by_xml_id(xml_id=deliveryservice_xml_id) logger.info("Deleting delivery_service_sslkeys data... %s", msg) if msg is None: logger.error("delivery_service_sslkeys returned by Traffic Ops is missing an 'xmlId' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="delivery_services_regex_post_data") def delivery_services_regex_data_post(to_session: TOSession, request_template_data: list[JSONData], delivery_services_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for delivery_services_regex endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get delivery_services_regex request template. :returns: Sample POST data and the actual API response. """ delivery_services_regex = check_template_data( request_template_data["delivery_services_regex"], "delivery_services_regex") # Return new post data and post response from delivery_services_regex POST request delivery_service_id = delivery_services_post_data["id"] # 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": "regex"}) delivery_services_regex["type"]= type_object["id"] logger.info("New delivery_services_regex data to hit POST method %s", delivery_services_regex) # Hitting delivery_services_regex POST methed response: tuple[JSONData, requests.Response] = to_session.create_deliveryservice_regexes( delivery_service_id=delivery_service_id, data=delivery_services_regex) resp_obj = check_template_data(response, "delivery_services_regex") yield [delivery_service_id,resp_obj] regex_id = resp_obj.get("id") msg = to_session.delete_deliveryservice_regex_by_regex_id( delivery_service_id=delivery_service_id, delivery_service_regex_id=regex_id) logger.info("Deleting delivery_services_regex data... %s", msg) if msg is None: logger.error("delivery_services_regex returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="cdn_federation_post_data") def cdn_federation_data_post(to_session: TOSession, request_template_data: list[JSONData], cdn_post_data:dict[str, object], user_post_data:dict[str, object], federation_resolver_post_data:dict[str, object], delivery_services_post_data: dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for cdn_name_federations endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get federations request template. :returns: Sample POST data and the actual API response. """ cdn_federation = check_template_data(request_template_data["cdn_federation"], "cdn_federation") # Return new post data and post response from cdn_federation POST request cdn_name = cdn_post_data["name"] logger.info("New federations data to hit POST method %s", cdn_federation) # Hitting cdn_federation POST methed response: tuple[JSONData, requests.Response] = to_session.create_federation_in_cdn(cdn_name=cdn_name, data= cdn_federation) cdn_federation_resp_obj = check_template_data(response, "cdn_federation") federation_id = cdn_federation_resp_obj.get("id") #Assign created federation to a user user_id = user_post_data["id"] user_federation = check_template_data(request_template_data["user_federation"], "user_federation") user_federation["userIds"][0] = user_id response: tuple[JSONData, requests.Response] = to_session.create_federation_user(federation_id=federation_id, data=user_federation) user_federation_resp_obj = check_template_data(response, "user_federation") #Assign the federation to a delivery_service delivery_service_id = delivery_services_post_data["id"] delivery_service_federation = check_template_data(request_template_data["delivery_service_federation"], "delivery_service_federation") delivery_service_federation["dsIds"][0] = delivery_service_id response: tuple[JSONData, requests.Response] = to_session.assign_delivery_services_to_federations(federation_id=federation_id, data=delivery_service_federation) delivery_service_federation_resp_obj = check_template_data(response, "delivery_service_federation") #Assign a federation resolver to created federation federation_resolver_id = federation_resolver_post_data["id"] federation_federation_resolver = check_template_data(request_template_data["federation_federation_resolver"], "federation_federation_resolver") federation_federation_resolver["fedResolverIds"][0] = federation_resolver_id response: tuple[JSONData, requests.Response] = to_session.assign_federation_resolver_to_federations(federation_id=federation_id, data=federation_federation_resolver) federation_federation_resolver_resp_obj = check_template_data(response, "federation_federation_resolver") yield [cdn_name, cdn_federation_resp_obj, cdn_federation, federation_id, delivery_service_federation_resp_obj] msg = to_session.delete_federation_in_cdn(cdn_name=cdn_name, federation_id=federation_id) logger.info("Deleting cdn_federation dara... %s", msg) if msg is None: logger.error("cdn_federation returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="delivery_service_required_capabilities_post_data") def delivery_service_required_capabilities_data_post(to_session: TOSession, request_template_data: list[JSONData], delivery_services_post_data:dict[str, object], server_capabilities_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for delivery_service_required_capabilities endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get delivery_service_required_capabilities request template. :returns: Sample POST data and the actual API response. """ delivery_service_required_capabilities = check_template_data( request_template_data["delivery_service_required_capabilities"], "delivery_service_required_capabilities") # Return new post data and post response from delivery_service_required_capabilities POST request deliveryServiceID = delivery_services_post_data["id"] requiredCapability = server_capabilities_post_data["name"] delivery_service_required_capabilities["deliveryServiceID"] = deliveryServiceID delivery_service_required_capabilities["requiredCapability"] = requiredCapability logger.info("New delivery_service_required_capabilities data to hit POST method %s", delivery_service_required_capabilities) # Hitting delivery_service_required_capabilities POST methed response: tuple[JSONData, requests.Response] = to_session.create_deliveryservices_required_capabilities( data=delivery_service_required_capabilities) resp_obj = check_template_data(response, "delivery_service_required_capabilities") yield resp_obj msg = to_session.delete_deliveryservices_required_capabilities( query_params={"deliveryServiceID":deliveryServiceID,"requiredCapability":requiredCapability}) logger.info("Deleting delivery_service_required_capabilities data... %s", msg) if msg is None: logger.error( "delivery_service_required_capabilities returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="delivery_service_request_comments_post_data") def delivery_service_request_comments_data_post(to_session: TOSession, request_template_data: list[JSONData], deliveryservice_request_post_data:dict[str, object]) -> dict[str, object]: """ PyTest Fixture to create POST data for delivery_service_request_comments endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get delivery_service_request_comments request template. :returns: Sample POST data and the actual API response. """ delivery_service_request_comments = check_template_data( request_template_data["delivery_service_request_comments"], "delivery_service_request_comments") # Return new post data and post response from delivery_service_request_comments POST request delivery_service_request_id = deliveryservice_request_post_data["id"] delivery_service_request_comments["deliveryServiceRequestId"]= delivery_service_request_id logger.info("New delivery_service_request_comments data to hit POST method %s", delivery_service_request_comments) # Hitting delivery_service_request_comments POST methed response: tuple[JSONData, requests.Response] = to_session.create_deliveryservice_request_comment( data=delivery_service_request_comments) resp_obj = check_template_data(response, "delivery_service_request_comments") yield resp_obj request_comment_id = resp_obj.get("id") msg = to_session.delete_deliveryservice_request_comment(query_params={"id":request_comment_id}) logger.info("Deleting delivery_service_request_comments data... %s", msg) if msg is None: logger.error("delivery_service_request_comments returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="profile_parameters_post_data") def profile_parameters_post_data(to_session: TOSession, request_template_data: list[JSONData], profile_post_data:dict[str, object], parameter_post_data:dict[str, object] ) -> dict[str, object]: """ PyTest Fixture to create POST data for profile parameters endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get profile parameters request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ profile_parameters = check_template_data(request_template_data["profile_parameters"], "profile_parameters") # Return new post data and post response from profile parameters POST request profile_get_response = to_session.get_profiles() profile_data = profile_get_response [0][0] profile_id = profile_data.get("id") profile_parameters["profileId"] = profile_post_data["id"] profile_parameters["parameterId"] = parameter_post_data["id"] logger.info("New profile_parameter data to hit POST method %s", profile_parameters) # Hitting profile parameters POST method response: tuple[JSONData, requests.Response] = to_session.associate_paramater_to_profile(profile_id=profile_id, data=profile_parameters) resp_obj = check_template_data(response, "profile_parameters") yield resp_obj profile_id = resp_obj.get("profileId") parameter_id = resp_obj.get("parameterId") msg = to_session.delete_profile_parameter_association_by_id(profile_id=profile_id, parameter_id=parameter_id) logger.info("Deleting Profile Parameters data... %s", msg) if msg is None: logger.error("Profile Parameter returned by Traffic Ops is missing a 'profile_id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="service_category_post_data") def service_category_data_post(to_session: TOSession, request_template_data: list[JSONData]) -> dict[str, object]: """ PyTest Fixture to create POST data for service_category endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get service_category request template. :returns: Sample POST data and the actual API response. """ service_category = check_template_data( request_template_data["service_category"], "service_category") # Return new post data and post response from service_category POST request service_category_name = service_category["name"] service_category["name"] = service_category_name + str(randint(0,1000)) logger.info("New service_category data to hit POST method %s", service_category) # Hitting service_category POST methed response: tuple[JSONData, requests.Response] = to_session.create_service_category( data=service_category) resp_obj = check_template_data(response, "service_category") yield resp_obj service_category_name = resp_obj.get("name") msg = to_session.delete_service_category(service_category_name=service_category_name) logger.info("Deleting service_category data... %s", msg) if msg is None: logger.error("service_category returned by Traffic Ops is missing an 'name' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="federation_resolver_post_data") def federation_resolver_data_post(to_session: TOSession, request_template_data: list[JSONData] ) -> dict[str, object]: """ PyTest Fixture to create POST data for federation_resolver endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get federation_resolver request template. :returns: Sample POST data and the actual API response. """ randstr = str(randint(0, 10)) federation_resolver = check_template_data( request_template_data["federation_resolver"], "federation_resolver") # 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": "federation"}) federation_resolver["typeId"] = type_object["id"] federation_resolver["ipAddress"] = ".".join(map(str, (randint(0, 255) for _ in range(4)))) logger.info("New federation_resolver data to hit POST method %s", federation_resolver) # Hitting federation_resolver POST methed response: tuple[JSONData, requests.Response] = to_session.create_federation_resolver(data=federation_resolver) resp_obj = check_template_data(response, "federation_resolver") yield resp_obj resolver_id = resp_obj.get("id") msg = to_session.delete_federation_resolver(query_params={"id":resolver_id}) logger.info("Deleting federation_resolver data... %s", msg) if msg is None: logger.error("federation_resolver returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="static_dns_entries_post_data") def static_dns_entries_data_post(to_session: TOSession, request_template_data: list[JSONData], delivery_services_post_data:dict[str, object] ) -> dict[str, object]: """ PyTest Fixture to create POST data for static_dns_entries endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get static_dns_entries request template. :returns: Sample POST data and the actual API response. """ static_dns_entries = check_template_data( request_template_data["static_dns_entries"], "static_dns_entries") # 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": "staticdnsentry"}) static_dns_entries["typeId"] = type_object["id"] static_dns_entries["deliveryServiceId"] = delivery_services_post_data["id"] logger.info("New static_dns_entries data to hit POST method %s", static_dns_entries) # Hitting static_dns_entries POST methed response: tuple[JSONData, requests.Response] = to_session.create_staticdnsentries(data=static_dns_entries) resp_obj = check_template_data(response, "static_dns_entries") yield resp_obj static_dns_entries_id = resp_obj.get("id") msg = to_session.delete_staticdnsentries(query_params={"id":static_dns_entries_id}) logger.info("Deleting static_dns_entries data... %s", msg) if msg is None: logger.error("static_dns_entries returned by Traffic Ops is missing an 'id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="edge_type_data") def edge_data_type(pytestconfig: pytest.Config, request_template_data: list[JSONData], to_session: TOSession): type_data = check_template_data(request_template_data["types"], "types") type_object = create_or_get_existing(to_session, "types", "type", type_data, {"name":"EDGE" , "useInTable":"server"}) type_id = type_object["id"] pytestconfig.cache.set("edgeTypeId", type_id) @pytest.fixture(name="server_server_capabilities_post_data") def server_server_capabilities_data_post(to_session: TOSession, edge_type_data:None, request_template_data: list[JSONData], server_post_data:dict[str, object], server_capabilities_post_data:dict[str, object], pytestconfig: pytest.Config ) -> dict[str, object]: """ PyTest Fixture to create POST data for server server capabilities endpoint. :param to_session: Fixture to get Traffic Ops session. :param request_template_data: Fixture to get Server Server Capabilities request template from a prerequisites file. :returns: Sample POST data and the actual API response. """ server_server_capabilities = check_template_data(request_template_data["server_server_capabilities"], "server_server_capabilities") # Return new post data and post response from server server capabilities POST request server_id = server_post_data.get("id") serverCapability = server_capabilities_post_data.get("name") server_server_capabilities["serverId"] = server_id server_server_capabilities["serverCapability"] = serverCapability logger.info("New server_server_capabilities data to hit POST method %s", server_server_capabilities) # Hitting server server capabilities POST method response: tuple[JSONData, requests.Response] = to_session.associate_server_capability_to_server(server_id=server_id, data=server_server_capabilities) resp_obj = check_template_data(response, "server_server_capabilities") yield resp_obj server_id = resp_obj.get("serverId") msg = to_session.delete_server_capability_association_to_server(query_params={"serverId":server_id, "serverCapability":serverCapability}) logger.info("Deleting Server Server Capability data... %s", msg) if msg is None: logger.error("Server Server Capability returned by Traffic Ops is missing a 'server_id' property") pytest.fail("Response from delete request is empty, Failing test_case") @pytest.fixture(name="logs_data") def logs_data(to_session: TOSession, request_template_data: list[JSONData], ) -> dict[str, object]: """ PyTest Fixture to retrieve log data from logs endpoint. :param to_session: Fixture to get Traffic Ops session. :returns: Log data obtained from logs endpoint. """ change_logs = check_template_data(request_template_data["logs"], "logs") # Hitting logs GET methed response: tuple[JSONData, requests.Response] = to_session.get_change_logs(data=change_logs) resp_obj = check_template_data(response[0], "change_logs") change_log_id = resp_obj.get("id") yield [change_log_id, resp_obj]