Skip to content
Snippets Groups Projects
Commit c21b2577 authored by Anton Joubert's avatar Anton Joubert
Browse files

Remove unused scripts and SKATestDevice 

These are no longer required.
parent 04b4891a
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
docker-compose.yml
+ 0
− 15
View file @ c21b2577
...@@ -13,7 +13,6 @@ ...@@ -13,7 +13,6 @@
# - obsdevice: Container having SKAObsDevice class # - obsdevice: Container having SKAObsDevice class
# - subarray: Container having SKASubarray class # - subarray: Container having SKASubarray class
# - telstate: Container having SKATelState class # - telstate: Container having SKATelState class
# - testdevice: Container having SKATestDevice class
# #
# Requires: # Requires:
# - None # - None
...@@ -171,17 +170,3 @@ services: ...@@ -171,17 +170,3 @@ services:
sh -c "wait-for-it.sh ${TANGO_HOST} --timeout=30 --strict -- sh -c "wait-for-it.sh ${TANGO_HOST} --timeout=30 --strict --
tango_admin --add-server SKATelState/01 SKATelState ska/telstate/01 &&\ tango_admin --add-server SKATelState/01 SKATelState ska/telstate/01 &&\
/venv/bin/python /app/skabase/SKATelState/SKATelState.py 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"
scripts/elt_ctl.py deleted 100644 → 0
+ 0
− 264
View file @ 04b4891a
#!/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))
scripts/gen_csv_info.py deleted 100644 → 0
+ 0
− 47
View file @ 04b4891a
#!/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
src/ska/base/test_device.py deleted 100644 → 0
+ 0
− 224
View file @ 04b4891a
# -*- 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()
tests/test_test_device.py deleted 100644 → 0
+ 0
− 206
View file @ 04b4891a
#########################################################################################
# -*- 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
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment