from typing import List, Optional
import tango
[docs]class Interaction:
"""Define an interaction between a consumer and provider"""
def __init__(self) -> None:
self.interaction_description = ""
self.provider_states: List[dict] = []
self.request = {}
self.response = {}
def _build_request(self, *args: List) -> dict:
request = {
"type": "command",
"name": "",
"args": args[1:],
}
if args:
request["name"] = args[0]
return request
[docs] def given(self, provider_state_description: str, *args): # pylint: disable=W0102
"""Description of what state the provider should be in
:param provider_state: Description of the provider state, defaults to ""
:type provider_state: str
:param params: A list of string paramters, defaults to []
:type params: List[str], optional
:return: self
:rtype: Provider
"""
provider_state = {
"description": provider_state_description,
"request": self._build_request(*args),
}
self.provider_states.append(provider_state)
return self
[docs] def and_given(self, provider_state_description: str, *args): # pylint: disable=W0102
"""Add a description of what state the provider should be in
:param provider_state: Description of the provider state, defaults to ""
:type provider_state: str
:param params: A list of string paramters, defaults to []
:type params: List[str], optional
:return: self
:rtype: Provider
"""
if not self.provider_states:
raise ValueError("only invoke and_given() after given()")
provider_state = {
"description": provider_state_description,
"request": self._build_request(*args),
}
self.provider_states.append(provider_state)
return self
[docs] def upon_receiving(self, scenario: str = ""):
"""Describe the interaction
:param scenario: Description of the interaction, defaults to ""
:type scenario: str, optional
"""
self.interaction_description = scenario
return self
[docs] def with_request(self, *args):
"""
Define the request from the consumer
Reference examples
- Tango attribute write
- Shortform
E.g
>>> provider_device.random_float = 5.0
Use
>>> with_request("attribute", "random_float", 5.0)
- Longform
E.g
>>> provider_device.write_attribute("random_float", 5.0)
Use
>>> with_request("method", "write_attribute", "random_float", 5.0)
OR
>>> with_request("write_attribute", "random_float", 5.0)
- Tango attribute read
- Shortform
E.g
>>> provider_device.random_float
Use
>>> with_request("attribute", "random_float")
- Longform
E.g
>>> provider_device.read_attribute("random_float")
Use
>>> with_request("read_attribute", "random_float")
- Tango commands
- Shortform
E.g
>>> provider_device.SomeCommand()
Use
>>> with_request("command", "SomeCommand")
E.g
>>> provider_device.SomeCommand(1.0)
Use
>>> with_request("command", "SomeCommand", 1.0)
- Longform
E.g
>>> provider_device.command_inout("SomeCommand")
Use
>>> with_request("method", "command_inout", "SomeCommand")
OR
>>> with_request("command_inout", "SomeCommand")
E.g
>>> provider_device.command_inout("SomeCommand", 1.0)
Use
>>> with_request("method", "command_inout", "SomeCommand", 1.0)
OR
>>> with_request("command_inout", "SomeCommand", 1.0)
:param req_type: read_attribute or command
:type req_type: string
:param name: The name of the command or attribute
:type name: string
:param arg: Any argument to be used in the command
:type arg: Any
:rtype: Interaction
"""
args = list(args)
allowed_first_arguments = [
"attribute",
"method",
"write_attribute",
"read_attribute",
"command",
"command_inout",
]
assert args[0] in allowed_first_arguments, (
f"The first argument {args[0]} is not supported. " f"Use any of {allowed_first_arguments}"
)
if args[0] in ["write_attribute", "command_inout"]:
args.insert(0, "method")
self.request = {
"type": args[0],
"name": args[1],
"args": args[2:],
}
return self
[docs] def will_respond_with(self, response_type, response=None):
"""
Define what the provider should return with.
:param response: The type of the response from the provider
:type status: Any
:param response: The provider response
:type status: Any
:rtype: Interaction
"""
self.response = {"response": response, "response_type": response_type}
return self
[docs] def to_dict(self) -> dict:
"""Return a dictionary of the Interaction
:return: A dictionary of the Interaction
:rtype: dict
"""
response = self.response.copy()
if not isinstance(response["response"], str):
response["response"] = str(response["response"])
if not isinstance(response["response_type"], str):
response["response_type"] = str(response["response_type"])
for provider_state in self.provider_states:
if isinstance(provider_state["request"]["args"], tuple):
provider_state["request"]["args"] = list(provider_state["request"]["args"])
return {
"description": self.interaction_description,
"provider_states": self.provider_states,
"request": self.request,
"response": response,
}
[docs] @classmethod
def from_dict(cls, interaction_dict: dict):
"""Returns an instance of Interaction
:param interaction_dict: Interaction in a dictionary format
:type interaction_dict: dict
:return: instance of Interaction
:rtype: Interaction
"""
interaction = Interaction()
for key in ["description", "provider_states", "request", "response"]:
assert key in interaction_dict, "Malformed dictionary for Interaction"
interaction.interaction_description = interaction_dict["description"]
interaction.response = interaction_dict["response"]
interaction.request = interaction_dict["request"]
interaction.provider_states = interaction_dict["provider_states"]
return interaction
[docs] def execute_request(self, request: dict, device_proxy: tango.DeviceProxy):
"""Execute a request against a Tango device
:param request: The request to execute
:type request: dict
:param device_proxy: The proxy to the provider device
:type device_proxy: tango.DeviceProxy
:return: The result of the request
:rtype: Any
"""
request_type = request["type"]
request_name = request["name"]
request_args = request["args"]
result = None
if request_type in ["command", "method"] and request_name:
# E.g provider_device.no_arg_command()
# provider_device.with_arg_command(1)
# provider_device.write_attribute("random_float", 5.0)
func = getattr(device_proxy, request_name)
result = func(*request_args)
elif request_type == "read_attribute":
# E.g provider_device.read_attribute("random_float")
result = device_proxy.read_attribute(*request_args)
elif request_name:
# E.g provider_device.random_float
result = getattr(device_proxy, request_name)
return result
[docs] def verify_interaction(self, device_proxy: tango.DeviceProxy):
"""Verify the request against the Provider
:param device_proxy: proxy to the provider device
:type device_proxy: tango.DeviceProxy
:raises AssertionError: If the request and response from the proxy fails
"""
result = self.execute_request(self.request, device_proxy)
if not isinstance(result, str):
result = str(result)
response = self.response["response"]
if not isinstance(response, str):
response = str(response)
if result != response:
raise AssertionError(f"Response check failed. Got {result}, expected {response}")
[docs] def setup_provider(self, device_proxy: tango.DeviceProxy):
"""Run through the provider states as defined in the Pact and execute them on the provider
:param device_proxy: The proxy to the provider
:type device_proxy: tango.DeviceProxy
"""
for provider_state in self.provider_states:
self.execute_request(provider_state["request"], device_proxy)
[docs]class Provider:
"""A Pact provider."""
def __init__(self, device_name: str) -> None:
"""Create a Provider
Provider("test/provider/1") relates to tango.DeviceProxy("test/provider/1")
:param device_name: The device name.
:type description: str
"""
self.device_name: str = device_name
self.interactions: List[Interaction] = []
[docs] def add_interaction(self, interaction: Interaction):
"""Add an Interaction
:param interaction: An interaction between the consumer and provider
:type interaction: Interaction
:return: self
:rtype: Provider
"""
assert isinstance(interaction, Interaction), "interaction should be a instance of Interaction"
self.interactions.append(interaction)
return self
[docs] def to_dict(self) -> dict:
"""Return a dictionary of the Provider
:return: A dictionary of the Provider and it's interactions
:rtype: dict
"""
return {
"provider": {
"name": self.device_name,
"interactions": [interaction.to_dict() for interaction in self.interactions],
}
}
[docs] @classmethod
def from_dict(cls, provider_dict: dict):
"""Return an instance of Provider
:param provider_dict: Provider in a dictionary format
:type provider_dict: dict
:return: instance of Provider
:rtype: Provider
"""
assert "name" in provider_dict, f"Missing name in {provider_dict}"
assert "interactions" in provider_dict, f"Missing interactions in {provider_dict}"
provider = Provider(provider_dict["name"])
for interaction in provider_dict["interactions"]:
provider.interactions.append(Interaction.from_dict(interaction))
return provider
[docs] def get_interaction_from_description(self, description) -> Optional[Interaction]:
"""Returns an interaction that matches the description
:param description: [description]
:type description: [type]
:return: [description]
:rtype: Optional[ska_pact_tango.Interaction]
"""
for interaction in self.interactions:
if interaction.interaction_description == description:
return interaction
return None