Added API contract test case for delivery_Service_required_capabiliti… (

#7667)

* Added API contract test case for delivery_Service_required_capabilities endpoint

* Update test_delivery_service_required_capabilities.py

* Update conftest.py

* resolved merge conflicts

traffic_control/clients/python/trafficops/tosession.py

@@ -863,6 +863,38 @@ def delete_deliveryservice_request_comment(self, query_params=None):
 		:raises: Union[LoginError, OperationError]
 		"""
 
+    #
+	# Delivery Service Required capabilities
+	#
+	@api_request('get', 'deliveryservices_required_capabilities', ('3.0', '4.0', '4.1', '5.0'))
+	def get_deliveryservices_required_capabilities(self, query_params=None):
+		"""
+		Retrieves all delivery service required capabilities.
+		:ref:`to-api-deliveryservice-required-capabilities`
+		:rtype: Tuple[Union[Dict[str, Any], List[Dict[str, Any]]], requests.Response]
+		:raises: Union[LoginError, OperationError]
+		"""
+
+	@api_request('post', 'deliveryservices_required_capabilities', ('3.0', '4.0', '4.1', '5.0'))
+	def create_deliveryservices_required_capabilities(self, data=None):
+		"""
+		Creates a new delivery service required capability.
+		:ref:`to-api-deliveryservice-required-capabilities`
+		:param data: The request data structure for the API request
+		:type data: Dict[str, Any]
+		:rtype: Tuple[Dict[str, Any], requests.Response]
+		:raises: Union[LoginError, OperationError]
+		"""
+
+	@api_request('delete', 'deliveryservices_required_capabilities', ('3.0', '4.0', '4.1', '5.0'))
+	def delete_deliveryservices_required_capabilities(self, query_params=None):
+		"""
+		Deletes a Delivery Service Required capability.
+		:ref:`to-api-deliveryservice-required-capabilities`
+		:rtype: Tuple[Dict[str, Any], requests.Response]
+		:raises: Union[LoginError, OperationError]
+		"""
+
 	#
 	# Delivery Service Health
 	#

traffic_ops/testing/api_contract/v4/conftest.py

@@ -1645,6 +1645,42 @@ def delivery_services_regex_data_post(to_session: TOSession, request_template_da
 		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],

traffic_ops/testing/api_contract/v4/data/request_template.json

@@ -396,11 +396,17 @@
 			"setNumber": 1
 		}
 	],
+	"delivery_service_required_capabilities": [
+		{
+			"deliveryServiceID": 1,
+			"requiredCapability": "disk"
+		}
+	],
 	"delivery_service_request_comments": [
 		{
 			"deliveryServiceRequestId": 2,
 			"value": "Does anyone have time to review my delivery service request?"
-     }
+		}
 	],
 	"profile_parameters": [
 		{
@@ -412,4 +418,4 @@
 			"parameterId": 3
 		}
 	]
-}
+}

traffic_ops/testing/api_contract/v4/data/response_template.json

@@ -1738,5 +1738,24 @@
                 "type": "string"
             }
         }
+    },
+    "delivery_service_required_capabilities": {
+        "type": "object",
+        "required": [
+            "deliveryServiceID",
+            "lastUpdated",
+            "requiredCapability"
+        ],
+        "properties": {
+            "deliveryServiceID": {
+                "type": "integer"
+            },
+            "lastUpdated": {
+                "type": "string"
+            },
+            "requiredCapability": {
+                "type": "string"
+            }
+        }
     }
 }

traffic_ops/testing/api_contract/v4/test_delivery_service_required_capabilities.py

@@ -0,0 +1,82 @@
+#
+# 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 delivery_service_required_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 = Union[bool, int, float, str, None]
+
+
+def test_delivery_service_required_capabilities_contract(to_session: TOSession,
+	response_template_data: dict[str, Union[Primitive, list[Union[Primitive,
+							dict[str, object], list[object]]], dict[object, object]]],
+	delivery_service_required_capabilities_post_data: dict[str, object]
+) -> None:
+	"""
+	Test step to validate keys, values and data types from delivery_service_required_capabilities 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 delivery_service_required_capabilities_post_data: Fixture to get sample data and response.
+	"""
+	# validate delivery_service_required_capabilities keys from api get response
+	logger.info("Accessing /delivery_service_required_capabilities endpoint through Traffic ops session.")
+
+	delivery_service_id = delivery_service_required_capabilities_post_data["deliveryServiceID"]
+	if not isinstance(delivery_service_id, int):
+		raise TypeError("malformed API response; 'id' property not a integer")
+
+	delivery_service_required_capabilities_get_response: tuple[
+		Union[dict[str, object], list[Union[dict[str, object], list[object], Primitive]], Primitive],
+		requests.Response
+	] = to_session.get_deliveryservices_required_capabilities(
+		query_params={"deliveryServiceID": delivery_service_id})
+	try:
+		delivery_service_required_capabilities_data = delivery_service_required_capabilities_get_response[0]
+		if not isinstance(delivery_service_required_capabilities_data, list):
+			raise TypeError("malformed API response; 'response' property not an array")
+
+		first_delivery_service_required_capabilities = delivery_service_required_capabilities_data[0]
+		if not isinstance(first_delivery_service_required_capabilities, dict):
+			raise TypeError(
+				"malformed API response; first delivery_service_required_capabilities in response is not an dict")
+		logger.info(
+			"delivery_service_required_capabilities Api response %s", first_delivery_service_required_capabilities)
+
+		delivery_service_required_capabilities_response_template = response_template_data.get(
+			"delivery_service_required_capabilities")
+		if not isinstance(delivery_service_required_capabilities_response_template, dict):
+			raise TypeError(f"delivery_service_required_capabilities response template data must be a dict, not '"
+							f"{type(delivery_service_required_capabilities_response_template)}'")
+
+		keys = ["deliveryServiceID", "requiredCapability"]
+		prereq_values = [delivery_service_required_capabilities_post_data[key] for key in keys]
+		get_values = [first_delivery_service_required_capabilities[key] for key in keys]
+
+		assert validate(instance=first_delivery_service_required_capabilities,
+		  schema=delivery_service_required_capabilities_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 delivery_service_required_capabilities endpoint:"
+	        "API response was malformed")