From c21b257768edab5d02d7c2d238a1aa0059db053e Mon Sep 17 00:00:00 2001 From: Anton Joubert <ajoubert@ska.ac.za> Date: Fri, 13 Mar 2020 17:02:21 +0200 Subject: [PATCH] Remove unused scripts and SKATestDevice These are no longer required. --- docker-compose.yml | 15 -- scripts/elt_ctl.py | 264 ------------------------------------ scripts/gen_csv_info.py | 47 ------- src/ska/base/test_device.py | 224 ------------------------------ tests/test_test_device.py | 206 ---------------------------- 5 files changed, 756 deletions(-) delete mode 100644 scripts/elt_ctl.py delete mode 100644 scripts/gen_csv_info.py delete mode 100644 src/ska/base/test_device.py delete mode 100644 tests/test_test_device.py diff --git a/docker-compose.yml b/docker-compose.yml index 79923c01..2926da79 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,7 +13,6 @@ # - obsdevice: Container having SKAObsDevice class # - subarray: Container having SKASubarray class # - telstate: Container having SKATelState class -# - testdevice: Container having SKATestDevice class # # Requires: # - None @@ -171,17 +170,3 @@ services: sh -c "wait-for-it.sh ${TANGO_HOST} --timeout=30 --strict -- tango_admin --add-server SKATelState/01 SKATelState ska/telstate/01 &&\ /venv/bin/python /app/skabase/SKATelState/SKATelState.py 01" - - testdevice: - image: ${DOCKER_REGISTRY_HOST}/${DOCKER_REGISTRY_USER}/lmcbaseclasses:latest - restart: unless-stopped - depends_on: - - databaseds - network_mode: ${NETWORK_MODE} - container_name: ${CONTAINER_NAME_PREFIX}testdevice - environment: - - TANGO_HOST=${TANGO_HOST} - command: > - sh -c "wait-for-it.sh ${TANGO_HOST} --timeout=30 --strict -- - tango_admin --add-server SKATestDevice/01 SKATestDevice logger/test/1 &&\ - /venv/bin/python /app/skabase/SKATestDevice/SKATestDevice.py 01" diff --git a/scripts/elt_ctl.py b/scripts/elt_ctl.py deleted file mode 100644 index 20225761..00000000 --- a/scripts/elt_ctl.py +++ /dev/null @@ -1,264 +0,0 @@ -#!/usr/bin/env python - -import os -import sys -import platform -import argparse - -import PyTango -import psutil -import json -import logging -from logging import config - -from time import sleep -from PyTango import Database, DbDevInfo, DeviceProxy - -sys.path.append('..') -from eltbase.config import config, CONFIG_FILE - -LOGGING = { - 'version': 1, - 'formatters': { - 'verbose': { - 'format': '%(asctime)s [%(process)s] [%(levelname)s] [%(module)s] %(message)s', - 'datefmt': "%Y-%m-%d %H:%M:%S" - }, - 'simple': { - 'format': '%(message)s' - }, - }, - 'handlers': { - 'console': { - 'level': 'DEBUG', - 'class': 'logging.StreamHandler', - 'formatter': 'simple' - } - }, - 'loggers': { - 'aavs.ctl': { - 'handlers': ['console'], - 'level': 'DEBUG', - }, - 'aavs.none': { - 'handlers': [], - 'level': 'DEBUG', - } - } -} -logging.config.dictConfig(LOGGING) - -db = Database() - - -def setup_tango_config(loaded_config): - """ Set up tango configuration """ - devices = loaded_config["devices"] - domain = loaded_config["domain"] - default_logging_target = loaded_config["default_logging_target"] - default_logging_level = loaded_config["default_logging_level"] - - for group_name, instances in devices.iteritems(): - for device_id, config in instances.iteritems(): - - device_class = config["class"] - full_device_name = "/".join([domain, group_name, device_id]) - device_server_name = config.get("server", None) - device_server_id = config.get("server_id", domain) - - db.put_device_property(full_device_name, {"logging_level": default_logging_level, - "logging_target": default_logging_target}) - - # Set up all properties, substituting the actual value of domain for {domain} - for property_name, property_value in config["properties"].iteritems(): - if type(property_value) is list: - formatted_val = map(lambda x: x.format(domain=domain), property_value) - else: - try: - formatted_val = property_value.format(domain=domain) - except Exception as e: - formatted_val = property_value - db.put_device_property(full_device_name, {str(property_name): formatted_val}) - - if group_name == "rack" and config.get("components"): - db.put_device_property(full_device_name, {"rack_components": json.dumps(config["components"])}) - - if device_server_name is None: - print "No device server specified for %s" % device_class - exit() - - dev_info = DbDevInfo() - dev_info._class = device_class - dev_info.server = '%s/%s' % (device_server_name, device_server_id) - dev_info.name = full_device_name - db.add_device(dev_info) - - -def _get_servers(loaded_config): - """ Get list of servers from configuration """ - devices = loaded_config["devices"] - priorities = loaded_config["server_priorities"] - default_priority = max(priorities.itervalues()) + 1 - - servers = set() - server_names = set() - - for group_name, instances in devices.iteritems(): - for device_id, config in instances.iteritems(): - device_server_name = config.get("server", None) - is_python_server = config.get("python_server", True) - priority = priorities.get(device_server_name, default_priority) - - if device_server_name not in server_names: - servers.add((device_server_name, is_python_server, priority)) - server_names.add(device_server_name) - - sorted_servers = sorted(servers, key=lambda x: x[2]) - return sorted_servers - - -def status(use_json=False): - """ Check server configuration """ - hostname = platform.uname()[1] - starter = "tango/admin/" + hostname - starter_dp = DeviceProxy(starter) - log_file_location = starter_dp.get_property("LogFileHome")["LogFileHome"][0] - - running_servers = set(starter_dp.DevGetRunningServers(True)) - domain = config["domain"] - if servers.issubset(running_servers): - info = json.dumps({ - "config_file": CONFIG_FILE, - "log_location": log_file_location, - "status": "OK", - "servers_configured": list(servers), - "servers_running": list(running_servers), - }, indent=4) - if use_json: - print info - else: - log.info(info) - else: - not_running = servers.difference(running_servers) - - if use_json: - print json.dumps({ - "config_file": CONFIG_FILE, - "log_location": log_file_location, - "servers_configured": list(servers), - "servers_running": list(running_servers), - "servers_not_running": list(not_running), - "status": "FAULT" - }, indent=4) - else: - for s in not_running: - log.info("Configured server {} is not running".format(s)) - - -def kill_everything(starter_dp, servers): - """ Kill all running servers and remove from database """ - - # Get list of running servers and stop them - running_servers = starter_dp.DevGetRunningServers(True) - for server in running_servers: - starter_dp.DevStop(server) - - # Wait for servers to stop - while starter_dp.DevGetRunningServers(True): - log.info("Waiting for servers to stop") - sleep(2) - - # Get list of processes which are still running - existing_processes = [p for p in psutil.process_iter() - if len(p.cmdline()) > 1 - and p.name() in {s[0] for s in servers}] - - # Kill these processes - for p in existing_processes: - log.info("Process {} pid {} is not dead. Killing".format(" ".join(p.cmdline()), p.pid)) - try: - p.kill() - except psutil.AccessDenied as ad: - log.info("Could not kill process {}. Try running as root".format(p.pid)) - - # Get list of device servers registered in database (and remove list of default tango servers) - servers = set(db.get_server_list().value_string) - {'DataBaseds/2', 'TangoAccessControl/1', 'TangoTest/test', - 'Starter/%s' % platform.uname()[1]} - # Remove servers from database - for server in servers: - try: - db.delete_server(server) - except Exception as e: - print e.message - -# Script entry point -if __name__ == '__main__': - - parser = argparse.ArgumentParser(description='Bootstrap') - group = parser.add_mutually_exclusive_group(required=True) - group.add_argument('--run', action="store_true", default=False, help="Stop all devices, reconfigure and run") - group.add_argument('--status', action="store_true", default=False, help="Print the status of the tango servers") - group.add_argument('--config', action="store_true", default=False, help="Stop all devices and reconfigure") - group.add_argument('--stop', action="store_true", default=False, help="Print the status of the tango servers") - parser.add_argument('--use_json', action="store_true", default=False, help="Return output as json") - - args = parser.parse_args() - domain = config["domain"] - servers = _get_servers(config) - - log = logging.getLogger('aavs.ctl') - if args.use_json: - log.disabled = True - - # TODO move prints to logging, output to json for all branches - hostname = platform.uname()[1] - starter = "tango/admin/" + hostname - starter_dp = DeviceProxy(starter) - - if args.status: - status(args.use_json) - exit() - elif args.stop: - kill_everything(starter_dp, servers) - exit() - elif args.config or args.run: - kill_everything(starter_dp, servers) - # Setup all configuration - setup_tango_config(config) - - if args.use_json: - exit() - else: - log.info("Configured with {}".format(CONFIG_FILE)) - - if args.run: - db = PyTango.Database() - - # Start all required device servers - for server, _, priority in servers: - server_name = server + "/" + domain - server_info = db.get_server_info(server_name) - server_info.level = priority - server_info.mode = 1 - server_info.host = hostname - db.put_server_info(server_info) - try: - # Start device server - starter_dp.DevStart(server_name) - - # Wait for server to finish loading - class_name = db.get_device_class_list(server_name).value_string[3] - sleep(0.5) - retries = 0 - while len(db.get_device_exported_for_class(class_name).value_string) == 0 and retries < 20: - log.info("Waiting for {} to start".format(server_name)) - retries += 1 - sleep(1) - - if retries == 20: - raise Exception("Could not load {}".format(server_name)) - - except Exception as e: - print e - else: - log.info("Started {}".format(server_name)) diff --git a/scripts/gen_csv_info.py b/scripts/gen_csv_info.py deleted file mode 100644 index b7c734d1..00000000 --- a/scripts/gen_csv_info.py +++ /dev/null @@ -1,47 +0,0 @@ -#!/usr/bin/env python - -"""A utility script we can make use of to convert the xmi files to a simple csv file.""" - - -import csv -import subprocess - -from tango_simlib import sim_xmi_parser - -# Find the xmi files in the repo and store their paths -output = subprocess.check_output(["kat-search.py -f *.xmi"], shell=True) - -# Create a list of of all the file paths. -strings = output.split("\n") -# Remove the string "DEFAULT", which is always the first output of 'kat-search.py'. -strings.remove("DEFAULTS") - -# Create a csv file object. -with open("csv_file.csv", 'wb') as myfile: - wr = csv.writer(myfile, quoting=csv.QUOTE_ALL) - print csv.QUOTE_ALL - for fiel in strings: - - xmi_parser = sim_xmi_parser.XmiParser() - try: - xmi_parser.parse(fiel) - except AttributeError: - continue - - print fiel - attr_info = xmi_parser.get_reformatted_device_attr_metadata() - #cmd_info = xmi_parser.get_reformatted_cmd_metadata() - #dev_props = xmi_parser.get_reformatted_properties_metadata('deviceProperties') - #class_props = xmi_parser.get_reformatted_properties_metadata('classProperties') - for attr_name, attr_props in attr_info.items(): - print attr_props - - - break - - del xmi_parser - break - - - - diff --git a/src/ska/base/test_device.py b/src/ska/base/test_device.py deleted file mode 100644 index a30f1e32..00000000 --- a/src/ska/base/test_device.py +++ /dev/null @@ -1,224 +0,0 @@ -# -*- coding: utf-8 -*- -# -# This file is part of the SKATestDevice project -# -# -# -""" SKATestDevice - -A generic Test device for testing SKA base class functionalities. -""" -# PROTECTED REGION ID(SKATestDevice.additionnal_import) ENABLED START # -# standard imports -import os -import sys -import json - -# Tango imports -import tango -from tango import DebugIt -from tango.server import run -from tango.server import attribute, command - -# SKA specific imports -from skabase import release - -file_path = os.path.dirname(os.path.abspath(__file__)) -basedevice_path = os.path.abspath(os.path.join(file_path, os.pardir)) + "/SKABaseDevice" -sys.path.insert(0, basedevice_path) - -from SKABaseDevice import SKABaseDevice -from skabase.auxiliary.utils import (exception_manager, convert_api_value, coerce_value) - -# PROTECTED REGION END # // SKATestDevice.additionnal_imports - -__all__ = ["SKATestDevice", "main"] - - -class SKATestDevice(SKABaseDevice): - """ - A generic Test device for testing SKA base class functionalities. - """ - # PROTECTED REGION ID(SKATestDevice.class_variable) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.class_variable - - # ----------------- - # Device Properties - # ----------------- - - # ---------- - # Attributes - # ---------- - - obsState = attribute( - dtype='DevEnum', - doc="Observing State", - enum_labels=["IDLE", "CONFIGURING", "READY", "SCANNING", - "PAUSED", "ABORTED", "FAULT", ], - ) - - obsMode = attribute( - dtype='DevEnum', - doc="Observing Mode", - enum_labels=["IDLE", "IMAGING", "PULSAR-SEARCH", "PULSAR-TIMING", "DYNAMIC-SPECTRUM", - "TRANSIENT-SEARCH", "VLBI", "CALIBRATION", ], - ) - - configurationProgress = attribute( - dtype='uint16', - unit="%", - max_value=100, - min_value=0, - doc="Percentage configuration progress", - ) - - configurationDelayExpected = attribute( - dtype='uint16', - unit="seconds", - doc="Configuration delay expected in seconds", - ) - - # --------------- - # General methods - # --------------- - - def init_device(self): - SKABaseDevice.init_device(self) - # PROTECTED REGION ID(SKATestDevice.init_device) ENABLED START # - self._build_state = '{}, {}, {}'.format(release.name, release.version, - release.description) - self._version_id = release.version - # PROTECTED REGION END # // SKATestDevice.init_device - - def always_executed_hook(self): - # PROTECTED REGION ID(SKATestDevice.always_executed_hook) ENABLED START # - pass - # PROTECTED REGION END # // SKATestDevice.always_executed_hook - - def delete_device(self): - # PROTECTED REGION ID(SKATestDevice.delete_device) ENABLED START # - pass - # PROTECTED REGION END # // SKATestDevice.delete_device - - # ------------------ - # Attributes methods - # ------------------ - - def read_obsState(self): - # PROTECTED REGION ID(SKATestDevice.obsState_read) ENABLED START # - """Reads Observing State of the device""" - return 0 - # PROTECTED REGION END # // SKATestDevice.obsState_read - - def read_obsMode(self): - # PROTECTED REGION ID(SKATestDevice.obsMode_read) ENABLED START # - """Reads Observing Mode of the device""" - return 0 - # PROTECTED REGION END # // SKATestDevice.obsMode_read - - def read_configurationProgress(self): - # PROTECTED REGION ID(SKATestDevice.configurationProgress_read) ENABLED START # - """Reads percentage configuration progress""" - return 0 - # PROTECTED REGION END # // SKATestDevice.configurationProgress_read - - def read_configurationDelayExpected(self): - # PROTECTED REGION ID(SKATestDevice.configurationDelayExpected_read) ENABLED START # - """Reads configuration delay expected in seconds""" - return 0 - # PROTECTED REGION END # // SKATestDevice.configurationDelayExpected_read - - # -------- - # Commands - # -------- - - @command( - dtype_in='str', - doc_in="JSON encoded dict with this format\n{``group``: str, # name of existing group\n" - " ``command``: str, # name of command to run\n" - " ``arg_type``: str, # data type of command input argument\n" - " ``arg_value``: str, # value for command input argument\n" - " ``forward``: bool # True if command should be forwarded to " - "all subgroups (default)\n}", - dtype_out='str', - doc_out="Return value from command on the group, as a JSON encoded string.\n" - "This will be a list of dicts of the form \n[ \n{``device_name``: str, " - "# TANGO device name\n ``argout``: <value>, # return value from " - "command (type depends on command)\n ``failed``: bool # True if command failed\n}," - "\n{ ... },\n ... ]", - ) - @DebugIt() - def RunGroupCommand(self, argin): - # PROTECTED REGION ID(SKATestDevice.RunGroupCommand) ENABLED START # - with exception_manager(self): - defaults = {'arg_type': None, 'arg_value': None, 'forward': True} - required = ('group', 'command', 'arg_type', 'arg_value', 'forward') - args = self._parse_argin(argin, defaults=defaults, required=required) - group_name = args['group'] - group = self.groups.get(group_name) - if group: - group_command = args['command'] - forward = args['forward'] - if args['arg_type']: - _, param = convert_api_value({'type': args['arg_type'], - 'value': args['arg_value']}) - replies = group.command_inout(group_command, param, forward=forward) - else: - replies = group.command_inout(group_command, forward=forward) - results = [] - for reply in replies: - result = { - 'device_name': reply.dev_name(), - 'argout': coerce_value(reply.get_data()), - 'failed': reply.has_failed(), - } - results.append(result) - argout = json.dumps(results, sort_keys=True) - else: - raise RuntimeError("Invalid group requested. '{}' not in '{}'" - .format(group_name, sorted(self.groups.keys()))) - return argout - # PROTECTED REGION END # // SKATestDevice.RunGroupCommand - - @command( - ) - @DebugIt() - def On(self): - # PROTECTED REGION ID(SKATestDevice.On) ENABLED START # - """Starts the device""" - self.logger.debug("TurnOn Sending DEBUG") - self.logger.info("TurnOn Sending INFO") - self.logger.warning("TurnOn Sending WARNING") - self.logger.error("TurnOn Sending ERROR") - self.logger.critical("TurnOn Sending CRITICAL") - #TODO: Set state to ON - # PROTECTED REGION END # // SKATestDevice.On - - @command( - ) - @DebugIt() - def Stop(self): - # PROTECTED REGION ID(SKATestDevice.Stop) ENABLED START # - """Stops the device""" - self.logger.debug("TurnOFF Sending DEBUG") - self.logger.info("TurnOFF Sending INFO") - self.logger.warning("TurnOFF Sending WARNING") - self.logger.error("TurnOFF Sending ERROR") - self.logger.critical("TurnOFF Sending CRITICAL") - # TODO: Set state to OFF - # PROTECTED REGION END # // SKATestDevice.Stop - -# ---------- -# Run server -# ---------- - -def main(args=None, **kwargs): - # PROTECTED REGION ID(SKATestDevice.main) ENABLED START # - """ - Main entry point of the module. - """ - return run((SKATestDevice,), args=args, **kwargs) - # PROTECTED REGION END # // SKATestDevice.main - -if __name__ == '__main__': - main() diff --git a/tests/test_test_device.py b/tests/test_test_device.py deleted file mode 100644 index 71e85567..00000000 --- a/tests/test_test_device.py +++ /dev/null @@ -1,206 +0,0 @@ -######################################################################################### -# -*- coding: utf-8 -*- -# -# This file is part of the SKATestDevice project -# -# -# -######################################################################################### -"""Contain the tests for the SKATestDevice.""" - -# Standard imports -import sys -import os - -# Imports -import re -import pytest -from tango import DevState - -# Path -path = os.path.join(os.path.dirname(__file__), os.pardir) -sys.path.insert(0, os.path.abspath(path)) - -# PROTECTED REGION ID(SKATestDevice.test_additional_imports) ENABLED START # -from skabase.control_model import ( - AdminMode, ControlMode, HealthState, ObsMode, ObsState, SimulationMode, TestMode -) -# PROTECTED REGION END # // SKATestDevice.test_additional_imports -# Device test case -# PROTECTED REGION ID(SKATestDevice.test_SKATestDevice_decorators) ENABLED START # -@pytest.mark.usefixtures("tango_context", "initialize_device") -# PROTECTED REGION END # // SKATestDevice.test_SKATestDevice_decorators -class TestSKATestDevice(object): - """Test case for packet generation.""" - - properties = { - 'SkaLevel': '4', - 'LoggingTargetsDefault': '', - 'LoggingLevelDefault': '4', - 'GroupDefinitions': '' - } - - @classmethod - def mocking(cls): - """Mock external libraries.""" - # Example : Mock numpy - # cls.numpy = SKATestDevice.numpy = MagicMock() - # PROTECTED REGION ID(SKATestDevice.test_mocking) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_mocking - - def test_properties(self, tango_context): - # Test the properties - # PROTECTED REGION ID(SKATestDevice.test_properties) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_properties - pass - - # PROTECTED REGION ID(SKATestDevice.test_GetVersionInfo_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_GetVersionInfo_decorators - def test_GetVersionInfo(self, tango_context): - """Test for GetVersionInfo""" - # PROTECTED REGION ID(SKATestDevice.test_GetVersionInfo) ENABLED START # - versionPattern = re.compile( - r'SKATestDevice, lmcbaseclasses, [0-9].[0-9].[0-9], ' - r'A set of generic base devices for SKA Telescope.') - versionInfo = tango_context.device.GetVersionInfo() - assert (re.match(versionPattern, versionInfo[0])) is not None - # PROTECTED REGION END # // SKATestDevice.test_GetVersionInfo - - # PROTECTED REGION ID(SKATestDevice.test_State_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_State_decorators - def test_State(self, tango_context): - """Test for State""" - # PROTECTED REGION ID(SKATestDevice.test_State) ENABLED START # - assert tango_context.device.State() == DevState.UNKNOWN - # PROTECTED REGION END # // SKATestDevice.test_State - - # PROTECTED REGION ID(SKATestDevice.test_Status_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_Status_decorators - def test_Status(self, tango_context): - """Test for Status""" - # PROTECTED REGION ID(SKATestDevice.test_Status) ENABLED START # - assert tango_context.device.Status() == "The device is in UNKNOWN state." - # PROTECTED REGION END # // SKATestDevice.test_Status - - # TODO: Fix the failing test case. Commented out to make CI pipeline run successfully. - # PROTECTED REGION ID(SKATestDevice.test_RunGroupCommand_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_RunGroupCommand_decorators - # def test_RunGroupCommand(self, tango_context): - # """Test for RunGroupCommand""" - # # PROTECTED REGION ID(SKATestDevice.test_RunGroupCommand) ENABLED START # - # assert tango_context.device.RunGroupCommand("") == "" - # # PROTECTED REGION END # // SKATestDevice.test_RunGroupCommand - - # PROTECTED REGION ID(SKATestDevice.test_Reset_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_Reset_decorators - def test_Reset(self, tango_context): - """Test for Reset""" - # PROTECTED REGION ID(SKATestDevice.test_Reset) ENABLED START # - assert tango_context.device.Reset() is None - # PROTECTED REGION END # // SKATestDevice.test_Reset - - def test_On(self, tango_context): - """Test for On""" - # PROTECTED REGION ID(SKATestDevice.test_On) ENABLED START # - tango_context.device.On() - #TODO: Add assert condition - # PROTECTED REGION END # // SKATestDevice.test_On - - def test_Stop(self, tango_context): - """Test for Stop""" - # PROTECTED REGION ID(SKATestDevice.test_Stop) ENABLED START # - tango_context.device.Stop() - # TODO: Add assert condition - # PROTECTED REGION END # // SKATestDevice.test_Stop - - # PROTECTED REGION ID(SKATestDevice.test_obsState_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_obsState_decorators - def test_obsState(self, tango_context): - """Test for obsState""" - # PROTECTED REGION ID(SKATestDevice.test_obsState) ENABLED START # - assert tango_context.device.obsState == ObsState.IDLE - # PROTECTED REGION END # // SKATestDevice.test_obsState - - # PROTECTED REGION ID(SKATestDevice.test_obsMode_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_obsMode_decorators - def test_obsMode(self, tango_context): - """Test for obsMode""" - # PROTECTED REGION ID(SKATestDevice.test_obsMode) ENABLED START # - assert tango_context.device.obsMode == ObsMode.IDLE - # PROTECTED REGION END # // SKATestDevice.test_obsMode - - # PROTECTED REGION ID(SKATestDevice.test_configurationProgress_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_configurationProgress_decorators - def test_configurationProgress(self, tango_context): - """Test for configurationProgress""" - # PROTECTED REGION ID(SKATestDevice.test_configurationProgress) ENABLED START # - assert tango_context.device.configurationProgress == 0 - # PROTECTED REGION END # // SKATestDevice.test_configurationProgress - - # PROTECTED REGION ID(SKATestDevice.test_configurationDelayExpected_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_configurationDelayExpected_decorators - def test_configurationDelayExpected(self, tango_context): - """Test for configurationDelayExpected""" - # PROTECTED REGION ID(SKATestDevice.test_configurationDelayExpected) ENABLED START # - assert tango_context.device.configurationDelayExpected == 0 - # PROTECTED REGION END # // SKATestDevice.test_configurationDelayExpected - - # PROTECTED REGION ID(SKATestDevice.test_buildState_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_buildState_decorators - def test_buildState(self, tango_context): - """Test for buildState""" - # PROTECTED REGION ID(SKATestDevice.test_buildState) ENABLED START # - buildPattern = re.compile( - r'lmcbaseclasses, [0-9].[0-9].[0-9], ' - r'A set of generic base devices for SKA Telescope') - assert (re.match(buildPattern, tango_context.device.buildState)) is not None - # PROTECTED REGION END # // SKATestDevice.test_buildState - - # PROTECTED REGION ID(SKATestDevice.test_versionId_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_versionId_decorators - def test_versionId(self, tango_context): - """Test for versionId""" - # PROTECTED REGION ID(SKATestDevice.test_versionId) ENABLED START # - versionIdPattern = re.compile(r'[0-9].[0-9].[0-9]') - assert (re.match(versionIdPattern, tango_context.device.versionId)) is not None - # PROTECTED REGION END # // SKATestDevice.test_versionId - - # PROTECTED REGION ID(SKATestDevice.test_healthState_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_healthState_decorators - def test_healthState(self, tango_context): - """Test for healthState""" - # PROTECTED REGION ID(SKATestDevice.test_healthState) ENABLED START # - assert tango_context.device.healthState == HealthState.OK - # PROTECTED REGION END # // SKATestDevice.test_healthState - - # PROTECTED REGION ID(SKATestDevice.test_adminMode_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_adminMode_decorators - def test_adminMode(self, tango_context): - """Test for adminMode""" - # PROTECTED REGION ID(SKATestDevice.test_adminMode) ENABLED START # - assert tango_context.device.adminMode == AdminMode.ONLINE - # PROTECTED REGION END # // SKATestDevice.test_adminMode - - # PROTECTED REGION ID(SKATestDevice.test_controlMode_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_controlMode_decorators - def test_controlMode(self, tango_context): - """Test for controlMode""" - # PROTECTED REGION ID(SKATestDevice.test_controlMode) ENABLED START # - assert tango_context.device.controlMode == ControlMode.REMOTE - # PROTECTED REGION END # // SKATestDevice.test_controlMode - - # PROTECTED REGION ID(SKATestDevice.test_simulationMode_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_simulationMode_decorators - def test_simulationMode(self, tango_context): - """Test for simulationMode""" - # PROTECTED REGION ID(SKATestDevice.test_simulationMode) ENABLED START # - assert tango_context.device.simulationMode == SimulationMode.FALSE - # PROTECTED REGION END # // SKATestDevice.test_simulationMode - - # PROTECTED REGION ID(SKATestDevice.test_testMode_decorators) ENABLED START # - # PROTECTED REGION END # // SKATestDevice.test_testMode_decorators - def test_testMode(self, tango_context): - """Test for testMode""" - # PROTECTED REGION ID(SKATestDevice.test_testMode) ENABLED START # - assert tango_context.device.testMode == TestMode.NONE - # PROTECTED REGION END # // SKATestDevice.test_testMode -- GitLab