Skip to content
Snippets Groups Projects
Unverified Commit 00065d64 authored by SKAJohanVenter's avatar SKAJohanVenter
Browse files

SAR-286 Tango command result now returns TaskResult where applicable.

parent 7159f3d3
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
tests/conftest.py
+ 27
− 6
View file @ 00065d64
"""This module defines elements of the pytest test harness shared by all tests."""
import logging
from queue import Empty, Queue
from unittest import mock
import pytest
from tango import EventType
from tango.test_context import DeviceTestContext
from ska_tango_base.base.task_queue_manager import TaskResult
from ska_tango_base.commands import ResultCode
@pytest.fixture(scope="class")
def device_properties():
"""
Fixture that returns device_properties to be provided to the device under test.
This is a default implementiong that provides no properties.
This is a default implementation that provides no properties.
"""
return {}
......@@ -24,6 +28,23 @@ def tango_context(device_test_config):
if component_manager_patch is not None:
device_test_config["device"].create_component_manager = component_manager_patch
def _get_command_func(dp, cmd_info, name):
"""Patch __get_command_func so that we can return a TaskResult if applicable."""
_, doc = cmd_info
def f(*args, **kwds):
result = dp.command_inout(name, *args, **kwds)
if isinstance(result, list):
if len(result) == 2:
if len(result[0]) == 1 and len(result[1]) == 1:
if result[0][0] == ResultCode.QUEUED:
return TaskResult.from_response_command(result)
return result
f.__doc__ = doc
return f
with mock.patch("tango.device_proxy.__get_command_func", _get_command_func):
tango_context = DeviceTestContext(**device_test_config)
tango_context.start()
yield tango_context
......@@ -214,7 +235,7 @@ def tango_change_event_helper(device_under_test):
self.assert_call(value)
def wait_for_lrc_id(self, unique_id: str):
"""Wait for the longRunningCommandResult unique ID to be the same as the paramater.
"""Wait for the longRunningCommandResult unique ID to be the same as the parameter.
:param unique_id: The long running command unique ID
:type unique_id: str
......
tests/test_subarray_device.py
+ 21
− 42
View file @ 00065d64
......@@ -17,7 +17,6 @@ from tango import DevState, DevFailed
# PROTECTED REGION ID(SKASubarray.test_additional_imports) ENABLED START #
from ska_tango_base import SKASubarray
from ska_tango_base.base import OpStateModel
from ska_tango_base.base.task_queue_manager import TaskResult
from ska_tango_base.commands import ResultCode
from ska_tango_base.control_model import (
AdminMode,
......@@ -92,12 +91,10 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
device_under_test.Configure('{"BAND1": 2}')
......@@ -121,11 +118,9 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
obs_state_callback = tango_change_event_helper.subscribe("obsState")
......@@ -151,9 +146,7 @@ class TestSKASubarray:
"longRunningCommandResult"
)
ver_info_tr = TaskResult.from_response_command(
device_under_test.GetVersionInfo()
)
ver_info_tr = device_under_test.GetVersionInfo()
result_callback.wait_for_lrc_id(ver_info_tr.unique_id)
versionInfo = device_under_test.longRunningCommandResult[2]
......@@ -186,16 +179,14 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
obs_state_callback = tango_change_event_helper.subscribe("obsState")
obs_state_callback.assert_call(ObsState.EMPTY)
resources_to_assign = ["BAND1", "BAND2"]
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(resources_to_assign))
)
assign_tr = device_under_test.AssignResources(json.dumps(resources_to_assign))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
obs_state_callback.assert_calls([ObsState.RESOURCING, ObsState.IDLE])
......@@ -219,12 +210,10 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
device_under_test.Configure('{"BAND1": 2}')
......@@ -249,12 +238,10 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
device_under_test.Configure('{"BAND1": 2}')
......@@ -282,12 +269,10 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1", "BAND2"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1", "BAND2"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
obs_state_callback = tango_change_event_helper.subscribe("obsState")
......@@ -308,12 +293,10 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1", "BAND2"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1", "BAND2"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
obs_state_callback = tango_change_event_helper.subscribe("obsState")
......@@ -335,12 +318,10 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
device_under_test.Configure('{"BAND1": 2}')
......@@ -367,12 +348,10 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assign_tr = TaskResult.from_response_command(
device_under_test.AssignResources(json.dumps(["BAND1"]))
)
assign_tr = device_under_test.AssignResources(json.dumps(["BAND1"]))
result_callback.wait_for_lrc_id(assign_tr.unique_id)
device_under_test.Configure('{"BAND1": 2}')
......@@ -521,7 +500,7 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assert device_under_test.assignedResources is None
......@@ -536,7 +515,7 @@ class TestSKASubarray:
"longRunningCommandResult"
)
on_tr = TaskResult.from_response_command(device_under_test.On())
on_tr = device_under_test.On()
result_callback.wait_for_lrc_id(on_tr.unique_id)
assert device_under_test.configuredCapabilities == ("BAND1:0", "BAND2:0")
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment