Select Git revision
test_hierarchy_device.py
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_hierarchy_device.py 22.19 KiB
# Copyright (C) 2022 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: Apache-2.0
import copy
import unittest
from unittest.mock import Mock, patch
from typing import List, Dict, Callable
from test.devices import device_base
from tango import DeviceProxy, DevState
from tango.server import command
from tango.server import device_property
from tango.test_context import DeviceTestContext
from tangostationcontrol.devices.base_device_classes import hierarchy_device
from tangostationcontrol.devices.base_device_classes import lofar_device
class TestHierarchyDevice(device_base.DeviceTestCase):
def setUp(self):
# DeviceTestCase setUp patches lofar_device DeviceProxy
super(TestHierarchyDevice, self).setUp()
self.hierarchy_mock = self.device_proxy_patch(hierarchy_device)
@unittest.mock.patch.object(hierarchy_device, "Database")
def test_get_direct_child_device(self, m_database):
"""Test whether read_attribute really returns the attribute."""
TEST_PROPERTY_CHILDREN = "hierarchy_children"
TEST_DEVICE_NAME = "stat/antennafield/lba"
TEST_CHILD_DEVICE = "stat/sdp/lba"
class MyHierarchyDevice(
lofar_device.LOFARDevice, hierarchy_device.AbstractHierarchyDevice
):
hierarchy_children = device_property(
dtype="DevVarStringArray",
mandatory=False,
default_value=[TEST_CHILD_DEVICE],
)
@command()
def has_child_proxy(self):
self.init(TEST_DEVICE_NAME, TEST_PROPERTY_CHILDREN)
if isinstance(self.child(TEST_CHILD_DEVICE), DeviceProxy):
raise Exception
with DeviceTestContext(MyHierarchyDevice, process=False, timeout=10) as proxy:
m_database.return_value.get_device_property.side_effect = [
{TEST_PROPERTY_CHILDREN: [TEST_CHILD_DEVICE]},
]
self.hierarchy_mock["object"].return_value.get_property.side_effect = [
{TEST_PROPERTY_CHILDREN: [TEST_CHILD_DEVICE]},
{TEST_PROPERTY_CHILDREN: []},
]
proxy.has_child_proxy()
class ConcreteHierarchy(hierarchy_device.AbstractHierarchyDevice):
pass
TEST_PROPERTY_NAME = "Control_Children"
def test_create_instance(self):
"""Test default values and if we can create an instance"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("MockDevice", self.TEST_PROPERTY_NAME)
self.assertEqual({}, test_hierarchy._children)
self.assertEqual(None, test_hierarchy._parent)
self.assertEqual({}, test_hierarchy.children())
self.assertRaises(hierarchy_device.NotFoundException, test_hierarchy.parent)
def test_get_or_create_proxy_cache(self):
"""Test if get_or_create_proxy caches without duplicates"""
test = TestHierarchyDevice.ConcreteHierarchy()
test.init("TestDevice", self.TEST_PROPERTY_NAME)
def test_parent_get_name(self):
"""Read the name of the parent through mocking DeviceProxy.dev_name"""
name_station = "stat/stationmanager/1"
mock_dev_station_name = Mock()
mock_dev_station_name.dev_name.return_value = name_station
self.device_proxy_mock["object"].side_effect = [mock_dev_station_name]
test = TestHierarchyDevice.ConcreteHierarchy()
test.init("TestDevice", self.TEST_PROPERTY_NAME)
test._children = None
test._parent = mock_dev_station_name
self.assertEqual(name_station, test.parent())
def test_parent_read_attribute(self):
"""Read an attribute from the parent and get the mocked data"""
attribute_value = "0.0.1"
name_station = "stat/stationmanager/1"
mock_dev_station_name = Mock()
mock_dev_station_name.dev_name.return_value = name_station
mock_dev_station_name.FPGA_firmware_version_R = attribute_value
test = TestHierarchyDevice.ConcreteHierarchy()
test.init("TestDevice", self.TEST_PROPERTY_NAME)
test._parent = mock_dev_station_name
self.assertEqual(
attribute_value,
test.read_parent_attribute("FPGA_firmware_version_R"),
)
def test_parent_read_attribute_no_parent(self):
"""Ensure that read_attribute returns None if there is no parent"""
test = TestHierarchyDevice.ConcreteHierarchy()
test.init("TestDevice", self.TEST_PROPERTY_NAME)
self.assertRaises(
hierarchy_device.NotFoundException,
test.read_parent_attribute,
"FPGA_firmware_version_R",
)
def test_parent_get_state(self):
"""Ensure that we can get parent state"""
name_station = "stat/stationmanager/1"
# Mock the state
mock_dev_station = Mock()
mock_dev_station.dev_name.return_value = name_station
mock_dev_station.state.return_value = DevState.FAULT
test = TestHierarchyDevice.ConcreteHierarchy()
test.init("TestDevice", self.TEST_PROPERTY_NAME)
test._parent = mock_dev_station
self.assertEqual(DevState.FAULT, test.parent_state())
def test_parent_get_state_no_parent(self):
"""Ensure that state returns None if no parent"""
test = TestHierarchyDevice.ConcreteHierarchy()
test.init("TestDevice", self.TEST_PROPERTY_NAME)
self.assertRaises(hierarchy_device.NotFoundException, test.parent_state)
def children_test_base(
self,
property_name: str,
direct_children: List[str],
children_properties: List[Dict[str, List[str]]],
fn: Callable[[ConcreteHierarchy], None],
):
"""Base function for testing children() method
:param fn: Callable to perform actual tests in
"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("TestDevice", property_name)
mock_children = {}
for child in direct_children:
mock_children[child] = None
test_hierarchy._children = mock_children
self.device_proxy_mock[
"object"
].return_value.get_property.side_effect = children_properties
fn(test_hierarchy)
def test_get_children_depth_1_no_filter(self):
"""Test finding all children at max depth of 1"""
depth = 1
test_children = ["stat/sdp/1", "stat/sdp/2"]
test_property_calls = [
{self.TEST_PROPERTY_NAME: ["stat/xst/1"]}, # sdp/1
{self.TEST_PROPERTY_NAME: ["stat/xstbeam/1"]}, # xst/1
{self.TEST_PROPERTY_NAME: []}, # xstbeam,/1
{self.TEST_PROPERTY_NAME: []}, # sdp/2
]
def test_fn(test_hierarchy: TestHierarchyDevice.ConcreteHierarchy):
test = test_hierarchy.children(depth=depth)
# Test if all keys from test_children are in the children() dict
self.assertSetEqual(set(test_children), set(test.keys()))
# test if for each child there are no further children
for child in test_children:
self.assertEqual(
0, len(test[child]["children"]), msg=f'{test[child]["children"]}'
)
self.children_test_base(
self.TEST_PROPERTY_NAME, test_children, test_property_calls, test_fn
)
def test_get_children_depth_2_no_filter(self):
"""Test recursively finding children limited to depth 2"""
depth = 2
test_children = ["stat/sdp/1", "stat/sdp/2"]
# Calls to get_property follow depth-first order
test_property_calls = [
{self.TEST_PROPERTY_NAME: ["stat/antennafield/1"]}, # sdp/1
{self.TEST_PROPERTY_NAME: ["stat/tilebeam/1"]}, # antennafield/1
{self.TEST_PROPERTY_NAME: []}, # tilebeam/1
{self.TEST_PROPERTY_NAME: []}, # sdp/2
]
def test_fn(test_hierarchy: TestHierarchyDevice.ConcreteHierarchy):
test = test_hierarchy.children(depth=depth)
# Test that antennafield is included
self.assertIsNotNone(
test["stat/sdp/1"]["children"].get("stat/antennafield/1")
)
# Test that antennafield child is removed due to depth limit
self.assertEqual(
0,
len(test["stat/sdp/1"]["children"]["stat/antennafield/1"]["children"]),
msg=f'{test["stat/sdp/1"]["children"]["stat/antennafield/1"]["children"]}',
)
self.children_test_base(
self.TEST_PROPERTY_NAME, test_children, test_property_calls, test_fn
)
def test_child_exists(self):
"""Test finding existing child"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("TestDevice", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["stat/child/1", "stat/child/2"]:
mock_children[child] = None
test_hierarchy._children = mock_children
mock_parent = Mock()
mock_parent.dev_name.return_value = "stat/parent/1"
test_hierarchy._parents = [mock_parent]
# try to find a child using an exact filter match
_ = test_hierarchy.child(
"stat/child/1", hierarchy_device.HierarchyMatchingFilter.EXACT
)
# try to find a child using a regex
_ = test_hierarchy.child(
"stat/child/*", hierarchy_device.HierarchyMatchingFilter.REGEX
)
def test_child_does_not_exist(self):
"""Test finding a non-existing child to raise an exception"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("TestDevice", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["stat/child/1", "stat/child/2"]:
mock_children[child] = None
test_hierarchy._children = mock_children
mock_parent = Mock()
mock_parent.dev_name.return_value = "stat/parent/1"
test_hierarchy._parents = [mock_parent]
# try to find a non-existing child using an exact filter match
with self.assertRaises(hierarchy_device.NotFoundException):
_ = test_hierarchy.child(
"stat/notexist/1", hierarchy_device.HierarchyMatchingFilter.EXACT
)
# try to find a non-existing child using a regex
with self.assertRaises(hierarchy_device.NotFoundException):
_ = test_hierarchy.child(
"stat/notexist/*", hierarchy_device.HierarchyMatchingFilter.REGEX
)
TEST_CHILDREN_ROOT = ["stat/ccd/1", "stat/psoc/1", "stat/antennafield/1"]
TEST_CHILDREN_ANTENNAFIELD = [
"stat/sdp/1",
"stat/tilebeam/1",
"stat/digitalbeam/1",
"stat/aps/1",
]
TEST_CHILDREN_SDP = ["stat/xst/1", "stat/sst/1", "stat/bst/1"]
TEST_CHILDREN_APS = [
"stat/apsct/1",
"stat/apspu/1",
"stat/unb2/1",
"stat/recvh/1",
"stat/recvl/1",
]
# Calls to get_property follow depth-first order
TEST_GET_PROPERTY_CALLS = [
{TEST_PROPERTY_NAME: []}, # cdd/1
{TEST_PROPERTY_NAME: []}, # psoc/1
{TEST_PROPERTY_NAME: TEST_CHILDREN_ANTENNAFIELD}, # antennafield/1
{TEST_PROPERTY_NAME: TEST_CHILDREN_SDP}, # sdp/1
{TEST_PROPERTY_NAME: []}, # xst/1
{TEST_PROPERTY_NAME: []}, # sst/1
{TEST_PROPERTY_NAME: []}, # bst/1
{TEST_PROPERTY_NAME: []}, # tilebeam/1
{TEST_PROPERTY_NAME: ["stat/beamlet/1"]}, # digitalbeam/1
{TEST_PROPERTY_NAME: []}, # beamlet/1
{TEST_PROPERTY_NAME: TEST_CHILDREN_APS}, # aps/1
{TEST_PROPERTY_NAME: []}, # apsct/1
{TEST_PROPERTY_NAME: []}, # apspu/1
{TEST_PROPERTY_NAME: []}, # unb2/1
{TEST_PROPERTY_NAME: []}, # recvh/1
{TEST_PROPERTY_NAME: []}, # recvl/1
]
def test_get_children_depth_elaborate_no_filter(self):
"""Create a 3 levels deep hierarchy with ~15 devices"""
def test_fn(test_hierarchy: TestHierarchyDevice.ConcreteHierarchy):
test = test_hierarchy.children(depth=-1)
# Test ccd and psoc have no children
self.assertEqual(
0,
len(test["stat/ccd/1"]["children"]),
msg=f'{test["stat/ccd/1"]["children"]}',
)
self.assertEqual(
0,
len(test["stat/psoc/1"]["children"]),
msg=f'{test["stat/psoc/1"]["children"]}',
)
# Test antennafield has 4 children
self.assertEqual(4, len(test["stat/antennafield/1"]["children"]))
# Test sdp has 3 children
self.assertEqual(
3,
len(test["stat/antennafield/1"]["children"]["stat/sdp/1"]["children"]),
)
# Test all childs of sdp have no children
for sdp_child in self.TEST_CHILDREN_SDP:
data = test["stat/antennafield/1"]["children"]["stat/sdp/1"][
"children"
][sdp_child]["children"]
self.assertEqual(0, len(data), msg=f"{data}")
# Test tilebeam has no children
data = test["stat/antennafield/1"]["children"]["stat/tilebeam/1"][
"children"
]
self.assertEqual(0, len(data), msg=f"{data}")
# Test digitalbeam has 1 child
data = test["stat/antennafield/1"]["children"]["stat/digitalbeam/1"][
"children"
]
self.assertEqual(1, len(data), msg=f"{data}")
# Test beamlet has no children
data = test["stat/antennafield/1"]["children"]["stat/digitalbeam/1"][
"children"
]["stat/beamlet/1"]["children"]
self.assertEqual(0, len(data), msg=f"{data}")
# Test aps has 5 children
self.assertEqual(
5,
len(test["stat/antennafield/1"]["children"]["stat/aps/1"]["children"]),
)
# Test all childs of aps have no children
for aps_child in self.TEST_CHILDREN_APS:
self.assertEqual(
0,
len(
test["stat/antennafield/1"]["children"]["stat/aps/1"][
"children"
][aps_child]["children"]
),
)
self.children_test_base(
self.TEST_PROPERTY_NAME,
self.TEST_CHILDREN_ROOT,
self.TEST_GET_PROPERTY_CALLS,
test_fn,
)
def test_get_child_filter(self):
"""Test we can find every device"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("TestDevice", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in self.TEST_CHILDREN_ROOT:
mock_children[child] = None
test_hierarchy._children = mock_children
test_hierarchy._parents = None
self.device_proxy_mock[
"object"
].return_value.get_property.side_effect = self.TEST_GET_PROPERTY_CALLS
all_children = copy.copy(self.TEST_CHILDREN_ROOT)
all_children.extend(self.TEST_CHILDREN_ANTENNAFIELD)
all_children.extend(self.TEST_CHILDREN_SDP)
all_children.extend(self.TEST_CHILDREN_APS)
# Find all proxies for each child and match that it is the same
# object as cached in `_proxies`
for child in all_children:
result = test_hierarchy.child(
child, matching_filter=hierarchy_device.HierarchyMatchingFilter.EXACT
)
self.assertEqual(test_hierarchy._proxies[child], result)
self.device_proxy_mock[
"object"
].return_value.get_property.side_effect = self.TEST_GET_PROPERTY_CALLS
class FakeDeviceProxy:
"""A stateful fake to return the right values
for the right device regardless of calling order."""
def __init__(self, name, timeout):
self.name = name
def dev_name(self):
return self.name
def get_property(self, prop_name):
children = {
"1": ["1.1", "1.2"],
"2": ["2.1", "2.2"],
"2.1": ["2.1.1"],
}
return {TestHierarchyDevice.TEST_PROPERTY_NAME: children.get(self.name, [])}
@patch.object(hierarchy_device, "create_device_proxy", wraps=FakeDeviceProxy)
def test_walk_down(self, m_create_device_proxy):
"""Test whether walking down the hierarchy tree (root -> leaves) works."""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("TestDevice", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["1", "2", "3"]:
mock_children[child] = None
test_hierarchy._children = mock_children
test_hierarchy._parents = None
def walker(device: DeviceProxy):
walk_order.append(device.dev_name())
# walk one level
walk_order = []
test_hierarchy.walk_down(walker, depth=1)
self.assertListEqual(["1", "2", "3"], walk_order)
# walk whole tree
walk_order = []
test_hierarchy.walk_down(walker, depth=-1)
self.assertListEqual(
["1", "1.1", "1.2", "2", "2.1", "2.1.1", "2.2", "3"],
walk_order,
)
@patch.object(hierarchy_device, "create_device_proxy", wraps=FakeDeviceProxy)
def test_walk_up(self, m_create_device_proxy):
"""Test whether walking up the hierarchy (leaves -> root) tree works."""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("TestDevice", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["1", "2", "3"]:
mock_children[child] = None
test_hierarchy._children = mock_children
test_hierarchy._parents = None
def walker(device: DeviceProxy):
walk_order.append(device.dev_name())
# walk one level
walk_order = []
test_hierarchy.walk_up(walker, depth=1)
self.assertListEqual(["3", "2", "1"], walk_order)
# walk whole tree
walk_order = []
test_hierarchy.walk_up(walker, depth=-1)
self.assertListEqual(
["3", "2.2", "2.1.1", "2.1", "2", "1.2", "1.1", "1"],
walk_order,
)
@patch.object(hierarchy_device, "create_device_proxy", wraps=FakeDeviceProxy)
def test_branch_children_names(self, m_create_device_proxy):
"""Test if branch children names are correctly retrieved and direct
parent is not included in them"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("stat/testdevice/1", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["stat/childdevice/1", "stat/childdevice/2", "stat/childdevice/3"]:
mock_children[child] = None
test_hierarchy._children = mock_children
mock_parent = Mock()
mock_parent.dev_name.return_value = "stat/parent/1"
test_hierarchy._parent = mock_parent
children_names = test_hierarchy.branch_children_names(
"*stat*", hierarchy_device.HierarchyMatchingFilter.REGEX
)
self.assertEqual(3, len(children_names))
self.assertNotIn("stat/parent/1", children_names)
children_names = test_hierarchy.branch_children_names(
"stat/parent/*",
hierarchy_device.HierarchyMatchingFilter.REGEX,
)
self.assertNotIn("stat/parent/1", children_names)
@patch.object(hierarchy_device, "create_device_proxy", wraps=FakeDeviceProxy)
def test_branch_children_names_no_parent(self, m_create_device_proxy):
"""Test branch children names with no-parent node"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("stat/testdevice/1", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["stat/childdevice/1", "stat/childdevice/2", "stat/childdevice/3"]:
mock_children[child] = None
test_hierarchy._children = mock_children
children_names = test_hierarchy.branch_children_names(
"*stat*", hierarchy_device.HierarchyMatchingFilter.REGEX
)
self.assertEqual(0, len(children_names))
@patch.object(hierarchy_device, "create_device_proxy", wraps=FakeDeviceProxy)
def test_branch_child(self, m_create_device_proxy):
"""Test if branch child is correctly retrieved and direct
parent is not included"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("stat/testdevice/1", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["stat/child/1", "stat/child/2", "stat/child/3"]:
mock_children[child] = Mock()
mock_children[child].dev_name.return_value = child
mock_children[child].name.return_value = child
test_hierarchy._children = mock_children
mock_parent = Mock()
mock_parent.dev_name.return_value = "stat/parent/1"
test_hierarchy._parent = mock_parent
child = test_hierarchy.branch_child(
"stat/child/*", hierarchy_device.HierarchyMatchingFilter.REGEX
)
self.assertEqual("stat/child/1", child.dev_name())
self.assertRaises(
hierarchy_device.NotFoundException,
test_hierarchy.branch_child,
"stat/parent/*",
hierarchy_device.HierarchyMatchingFilter.REGEX,
)
@patch.object(hierarchy_device, "create_device_proxy", wraps=FakeDeviceProxy)
def test_branch_child_no_parent(self, m_create_device_proxy):
"""Test branch child with no-parent node"""
test_hierarchy = TestHierarchyDevice.ConcreteHierarchy()
test_hierarchy.init("stat/testdevice/1", self.TEST_PROPERTY_NAME)
mock_children = {}
for child in ["stat/child/1", "stat/child/2", "stat/child/3"]:
mock_children[child] = Mock()
mock_children[child].dev_name.return_value = child
mock_children[child].name.return_value = child
test_hierarchy._children = mock_children
self.assertRaises(
hierarchy_device.NotFoundException,
test_hierarchy.branch_child,
"stat/child/*",
hierarchy_device.HierarchyMatchingFilter.REGEX,
)