Added Api contract test case for delivery_service_server test case

traffic_ops/testing/api_contract/v4/conftest.py

@@ -1969,3 +1969,39 @@ def logs_data(to_session: TOSession, request_template_data: list[JSONData],
 	change_log_id = resp_obj.get("id")
 
 	yield [change_log_id, resp_obj]
+
+
+@pytest.fixture(name="delivery_service_server_post_data")
+def delivery_service_server_data_post(to_session: TOSession,
+		request_template_data: list[JSONData], delivery_services_post_data:dict[str, object],
+		server_post_data:dict[str, object]) -> dict[str, object]:
+	"""
+	PyTest Fixture to create POST data for delivery_service_server endpoint.
+	:param to_session: Fixture to get Traffic Ops session.
+	:param request_template_data: Fixture to get delivery_service_server request template.
+	:returns: Sample POST data and the actual API response.
+	"""
+
+	delivery_service_server = check_template_data(
+		request_template_data["delivery_service_server"], "delivery_service_server")
+
+	# Return new post data and post response from delivery_service_server POST request
+	deliveryServiceID = delivery_services_post_data["id"]
+	serverID = server_post_data["id"]
+	delivery_service_server["dsId"] = deliveryServiceID
+	delivery_service_server["servers"] = [serverID]
+
+	logger.info("New delivery_service_server data to hit POST method %s",
+	     delivery_service_server)
+	# Hitting delivery_service_server POST methed
+	response: tuple[JSONData, requests.Response] = to_session.assign_deliveryservice_servers_by_ids(
+		data=delivery_service_server)
+	resp_obj = check_template_data(response, "delivery_service_server")
+	yield resp_obj
+	msg = to_session.delete_deliveryservice_servers_by_id(delivery_service_id=deliveryServiceID,
+						       server_id=serverID)
+	logger.info("Deleting delivery_service_server data... %s", msg)
+	if msg is None:
+		logger.error(
+		"delivery_service_server returned by Traffic Ops is missing an 'id' property")
+		pytest.fail("Response from delete request is empty, Failing test_case")

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

@@ -419,9 +419,11 @@
 			"replace": true
 		}
 	],
-	"federation_federation_resolver":[
+	"federation_federation_resolver": [
 		{
-			"fedResolverIds": [39],
+			"fedResolverIds": [
+				39
+			],
 			"replace": true
 		}
 	],
@@ -450,7 +452,7 @@
 	"service_category": [
 		{
 			"name": "SERVICE-CATEGORY-NAME"
-      }
+		}
 	],
 	"federation_resolver": [
 		{
@@ -467,19 +469,28 @@
 			"typeId": 41
 		}
 	],
-  "server_server_capabilities":[
+	"server_server_capabilities": [
 		{
 			"serverId": 1,
 			"serverCapability": "disk"
-    	}
-  	],
-	  "logs":[
+		}
+	],
+	"logs": [
 		{
 			"ticketNum": null,
 			"level": "APICHANGE",
 			"user": "admin",
 			"id": 444,
 			"message": "log for test purposes"
 		}
+	],
+	"delivery_service_server": [
+		{
+			"dsId": 41,
+			"replace": true,
+			"servers": [
+				71
+			]
+		}
 	]
 }

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

@@ -2253,7 +2253,7 @@
             }
         }
     },
-    "logs":{
+    "logs": {
         "type": "object",
         "required": [
             "ticketNum",
@@ -2301,5 +2301,24 @@
                 ]
             }
         }
+    },
+    "delivery_service_server": {
+        "type": "object",
+        "required": [
+            "server",
+            "deliveryService",
+            "lastUpdated"
+        ],
+        "properties": {
+            "server": {
+                "type": "integer"
+            },
+            "deliveryService": {
+                "type": "integer"
+            },
+            "lastUpdated": {
+                "type": "string"
+            }
+        }
     }
 }

traffic_ops/testing/api_contract/v4/test_delivery_service_server.py

@@ -0,0 +1,80 @@
+#
+# 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_server 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_server_contract(to_session: TOSession,
+	response_template_data: dict[str, Union[Primitive, list[Union[Primitive,
+							dict[str, object], list[object]]], dict[object, object]]],
+	delivery_service_server_post_data: dict[str, object]
+) -> None:
+	"""
+	Test step to validate keys, values and data types from delivery_service_server 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_server_post_data: Fixture to get delivery services server data.
+	"""
+	# validate delivery_service_server keys from api get response
+	logger.info("Accessing /delivery_service_server endpoint through Traffic ops session.")
+
+	delivery_service_id = delivery_service_server_post_data["dsId"]
+	if not isinstance(delivery_service_id, int):
+		raise TypeError("malformed API response; 'dsId' property not a integer")
+	server_id = delivery_service_server_post_data["servers"][0]
+	if not isinstance(server_id, int):
+		raise TypeError("malformed API response; 'servers' property not a integer")
+
+
+	delivery_service_server_get_response: tuple[
+		Union[dict[str, object], list[Union[dict[str, object], list[object], Primitive]], Primitive],
+		requests.Response
+	] = to_session.get_deliveryserviceserver(query_params={"page":1,"limit":2,"orderby":"lastUpdated"})
+	try:
+		first_delivery_service_server = delivery_service_server_get_response[0][0]
+		if not isinstance(first_delivery_service_server, dict):
+			raise TypeError(
+				"malformed API response; first delivery_service_server in response is not an dict")
+		logger.info("delivery_service_server Api get response %s", first_delivery_service_server)
+
+		delivery_service_server_response_template = response_template_data.get(
+			"delivery_service_server")
+		if not isinstance(delivery_service_server_response_template, dict):
+			raise TypeError(f"delivery_service_server response template data must be a dict, not '"
+							f"{type(delivery_service_server_response_template)}'")
+
+		prereq_values = [delivery_service_server_post_data["dsId"], delivery_service_server_post_data["servers"][0]]
+		get_keys = ["deliveryService", "server"]
+		get_values = [first_delivery_service_server[key] for key in get_keys]
+
+		assert validate(instance=first_delivery_service_server,
+		  schema=delivery_service_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 delivery_service_server endpoint: API response was malformed")