ola-rdm-tests 0.10.3.nojsmin-2+deb9u1 / usr / lib / python2.7 / dist-packages / ola / testing / rdm / ResponderTest.py

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Library General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# ResponderTest.py
# Copyright (C) 2010 Simon Newton
#
# The test classes are broken down as follows:
#
# TestFixture - The base test class, defines common behaviour
#  ResponderTestFixture - A test which involves sending one or more RDM
#                         commands to a responder
#   OptionalParameterTestFixture - A test fixture that changes the expected
#                                  result based on the output of
#                                  SUPPORTED_PARAMETERS

import logging
import time
from ExpectedResults import (AckDiscoveryResult, AckGetResult, AckSetResult,
                             NackDiscoveryResult, NackGetResult, NackSetResult)
from TestCategory import TestCategory
from TestState import TestState
from TimingStats import TimingStats
from ola import PidStore
from ola.OlaClient import OlaClient, RDMNack

'''Automated testing for RDM responders.'''

__author__ = 'nomis52@gmail.com (Simon Newton)'


class Error(Exception):
  """The base error class."""


class UndeclaredPropertyException(Error):
  """Raised if a test attempts to get/set a property that it didn't declare."""


class TestFixture(object):
  """The base responder test class, every test inherits from this."""
  CATEGORY = TestCategory.UNCLASSIFIED
  DEPS = []
  PROVIDES = []
  REQUIRES = []

  def __init__(self, device, universe, uid, pid_store, *args, **kwargs):
    self._device_properties = device
    self._uid = uid
    self._pid_store = pid_store
    self._state = TestState.NOT_RUN
    try:
      self.pid = self.LookupPid(self.PID)
    except AttributeError:
      self.pid = None
    self._warnings = []
    self._advisories = []
    self._debug = []

  def __hash__(self):
    return hash(self.__class__.__name__)

  def __str__(self):
    return self.__class__.__name__

  def __repr__(self):
    return self.__class__.__name__

  def __cmp__(self, other):
    return cmp(self.__class__.__name__, other.__class__.__name__)

  def LookupPid(self, pid_name):
    return self._pid_store.GetName(pid_name, self._uid)

  def LookupPidValue(self, pid_value):
    return self._pid_store.GetPid(pid_value)

  def Requires(self):
    """Returns a list of the properties this test requires to run."""
    return self.REQUIRES

  @property
  def category(self):
    """The category this test belongs to."""
    return self.CATEGORY

  # Warnings are logged independently of errors.
  # They should be used to indicate deviations from the standard that will not
  # cause interoperability issues
  @property
  def warnings(self):
    """Non-fatal warnings."""
    return self._warnings

  def AddWarning(self, message):
    """Add an warning message.

    Args:
      message: The text of the warning message.
    """
    self.LogDebug('Warning: %s' % message)
    self._warnings.append(message)

  # Advisories are logged independently of errors. They should be used to
  # indicate conditions that while aren't covered by the standard, should still
  # be fixed.
  @property
  def advisories(self):
    """Non-fatal advisories message."""
    return self._advisories

  def AddAdvisory(self, message):
    """Add an advisory message.

    Args:
      message: The text of the advisory message.
    """
    self.LogDebug('Advisory: %s' % message)
    self._advisories.append(message)

  @property
  def debug(self):
    """Return a list of debug lines."""
    return self._debug

  def LogDebug(self, string):
    """Append debugging information."""
    logging.debug(string)
    self._debug.append(string)

  def Property(self, property, default=None):
    """Lookup a device property.

    Args:
      property: the name of the property.

    Returns:
      The value of the property.
    """
    if property not in self.Requires():
      raise UndeclaredPropertyException(
        '%s attempted to get %s which wasn\'t declared' %
        (self.__class__.__name__, property))
    return getattr(self._device_properties, property, default)

  def SetProperty(self, property, value):
    """Set a device property.

    Args:
      property: the name of the property.
      value: the value of the property.
    """
    if property not in self.PROVIDES:
      raise UndeclaredPropertyException(
        '%s attempted to set %s which wasn\'t declared' %
        (self.__class__.__name__, property))
    setattr(self._device_properties, property, value)

  def SetPropertyFromDict(self, dictionary, property, default=None):
    """Set a property to the value of the same name in a dictionary.

    Often it's useful to set a property to a value in a dictionary where the
    key has the same name.

    Args:
      dictionary: A dictionary that has a value for the property key
      property: the name of the property.
    """
    self.SetProperty(property, dictionary.get(property, default))

  @property
  def state(self):
    """Returns the state of the test.

    Returns:
      An instance of TestState
    """
    return self._state

  # Sub classes can override these
  def Test(self):
    self.SetBroken('test method not defined')

  def Run(self):
    """Run the test."""
    self.Test()

  def SetNotRun(self, message=None):
    """Set the state of the test to NOT_RUN and stop further processing."""
    self._state = TestState.NOT_RUN
    if message:
      self.LogDebug(' ' + message)
    self.Stop()

  def SetBroken(self, message):
    self.LogDebug(' Broken: %s' % message)
    self._state = TestState.BROKEN

  def SetFailed(self, message):
    self.LogDebug(' Failed: %s' % message)
    self._state = TestState.FAILED

  def SetPassed(self):
    self._state = TestState.PASSED


class ResponderTestFixture(TestFixture):
  """A Test that that sends one or more messages to a responder."""
  def __init__(self,
               device,
               universe,
               uid,
               pid_store,
               rdm_api,
               wrapper,
               broadcast_write_delay,
               timing_stats):
    super(ResponderTestFixture, self).__init__(device, universe, uid, pid_store)
    self._api = rdm_api
    self._expected_results = []
    self._in_reset_mode = False
    self._should_run_wrapper = True
    self._universe = universe
    self._wrapper = wrapper
    self._broadcast_write_delay_s = broadcast_write_delay / 1000.0
    self._timing_stats = timing_stats

    # This is set to the tuple of (sub_device, command_class, pid) when we sent
    # a message. It's used to identify the response if we get an ACK_TIMER and
    # use QUEUED_MESSAGEs
    self._outstanding_request = None

  @property
  def uid(self):
    return self._uid

  def SleepAfterBroadcastSet(self):
    if self._broadcast_write_delay_s:
      self.LogDebug('Sleeping after broadcast...')
    time.sleep(self._broadcast_write_delay_s)

  def Run(self):
    """Call the test method and then start running the loop wrapper."""
    # the super call invokes self.Test()
    super(ResponderTestFixture, self).Run()

    if self.state == TestState.BROKEN:
      return

    if self._should_run_wrapper:
      self._wrapper.Run()
    if self.state != TestState.NOT_RUN:
      self._in_reset_mode = True
      self.ResetState()
    self._wrapper.Reset()

  def VerifyResult(self, status, fields):
    """A hook to perform additional verification of data."""
    pass

  def ResetState(self):
    pass

  def Stop(self):
    self._should_run_wrapper = False
    self._wrapper.Stop()

  def AddExpectedResults(self, results):
    """Add a set of expected results."""
    if isinstance(results, list):
      self._expected_results = results
    else:
      self._expected_results = [results]

  def NackDiscoveryResult(self, nack_reason, **kwargs):
    """A helper method which returns a NackDiscoveryResult for the current
       PID.
    """
    return NackDiscoveryResult(self.pid.value, nack_reason, **kwargs)

  def NackGetResult(self, nack_reason, **kwargs):
    """A helper method which returns a NackGetResult for the current PID."""
    return NackGetResult(self.pid.value, nack_reason, **kwargs)

  def NackSetResult(self, nack_reason, **kwargs):
    """A helper method which returns a NackSetResult for the current PID."""
    return NackSetResult(self.pid.value, nack_reason, **kwargs)

  def AckDiscoveryResult(self, **kwargs):
    """A helper method which returns an AckDiscoveryResult for the current
      PID.
    """
    return AckDiscoveryResult(self.pid.value, **kwargs)

  def AckGetResult(self, **kwargs):
    """A helper method which returns an AckGetResult for the current PID."""
    return AckGetResult(self.pid.value, **kwargs)

  def AckSetResult(self, **kwargs):
    """A helper method which returns an AckSetResult for the current PID."""
    return AckSetResult(self.pid.value, **kwargs)

  def SendDiscovery(self, sub_device, pid, args=[]):
    """Send a raw Discovery request.

    Args:
      sub_device: The sub device
      pid: The pid value
      data: The param data
    """
    return self.SendDirectedDiscovery(self._uid, sub_device, pid, args)

  def SendDirectedDiscovery(self, uid, sub_device, pid, args=[]):
    """Send a raw Discovery request.

    Args:
      uid: The uid to send the GET to
      sub_device: The sub device
      pid: The pid value
      data: The param data
    """
    self.LogDebug(' DISCOVERY: uid: %s, pid: %s, sub device: %d, args: %s' %
                  (uid, pid, sub_device, args))
    self._MakeRequestKey(sub_device, PidStore.RDM_DISCOVERY, pid.value)
    return self._api.Discovery(self._universe,
                               uid,
                               sub_device,
                               pid,
                               self._HandleResponse,
                               args,
                               include_frames=True)

  def SendRawDiscovery(self, sub_device, pid, data=""):
    """Send a raw Discovery request.

    Args:
      sub_device: The sub device
      pid: The pid value
      data: The param data
    """
    self.LogDebug(' DISCOVERY: pid: %s, sub device: %d, data: %r' %
                  (pid, sub_device, data))
    self._MakeRequestKey(sub_device, PidStore.RDM_DISCOVERY, pid.value)
    return self._api.RawDiscovery(self._universe,
                                  self._uid,
                                  sub_device,
                                  pid,
                                  self._HandleResponse,
                                  data,
                                  include_frames=True)

  def SendGet(self, sub_device, pid, args=[]):
    """Send a GET request using the RDM API.

    Args:
      sub_device: The sub device
      pid: A PID object
      args: A list of arguments
    """
    return self.SendDirectedGet(self._uid, sub_device, pid, args)

  def SendDirectedGet(self, uid, sub_device, pid, args=[]):
    """Send a GET request using the RDM API.

    Args:
      uid: The uid to send the GET to
      sub_device: The sub device
      pid: A PID object
      args: A list of arguments
    """
    self.LogDebug(' GET: uid: %s, pid: %s, sub device: %d, args: %s' %
                  (uid, pid, sub_device, args))
    self._MakeRequestKey(sub_device, PidStore.RDM_GET, pid.value)
    ret_code = self._api.Get(self._universe,
                             uid,
                             sub_device,
                             pid,
                             self._HandleResponse,
                             args,
                             include_frames=True)
    return ret_code

  def SendRawGet(self, sub_device, pid, data=""):
    """Send a raw GET request.

    Args:
      sub_device: The sub device
      pid: The pid value
      data: The param data
    """
    self.LogDebug(' GET: uid: %s, pid: %s, sub device: %d, data: %r' %
                  (self._uid, pid, sub_device, data))
    self._MakeRequestKey(sub_device, PidStore.RDM_GET, pid.value)
    return self._api.RawGet(self._universe,
                            self._uid,
                            sub_device,
                            pid,
                            self._HandleResponse,
                            data,
                            include_frames=True)

  def SendSet(self, sub_device, pid, args=[]):
    """Send a SET request using the RDM API.

    Args:
      sub_device: The sub device
      pid: A PID object
      args: A list of arguments
    """
    return self.SendDirectedSet(self._uid, sub_device, pid, args)

  def SendDirectedSet(self, uid, sub_device, pid, args=[]):
    """Send a SET request using the RDM API.

    Args:
      uid: The uid to send the GET to
      sub_device: The sub device
      pid: A PID object
      args: A list of arguments
    """
    self.LogDebug(' SET: uid: %s, pid: %s, sub device: %d, args: %s' %
                  (uid, pid, sub_device, self._EscapeData(args)))
    self._MakeRequestKey(sub_device, PidStore.RDM_SET, pid.value)
    ret_code = self._api.Set(self._universe,
                             uid,
                             sub_device,
                             pid,
                             self._HandleResponse,
                             args,
                             include_frames=True)
    if uid.IsBroadcast():
      self.SleepAfterBroadcastSet()
    return ret_code

  def SendRawSet(self, sub_device, pid, data=""):
    """Send a raw SET request.

    Args:
      sub_device: The sub device
      pid: The pid value
      data: The param data
    """
    self.LogDebug(' SET: pid: %s, sub device: %d, data: %r' %
                  (pid, sub_device, data))
    self._MakeRequestKey(sub_device, PidStore.RDM_SET, pid.value)
    return self._api.RawSet(self._universe,
                            self._uid,
                            sub_device,
                            pid,
                            self._HandleResponse,
                            data,
                            include_frames=True)

  def _HandleResponse(self, response, unpacked_data, unpack_exception):
    """Handle a RDM response.

    Args:
      response: A RDMResponse object
      unpacked_data: The unpacked data structure for this pid
      unpack_exception: The exception if the data could not be unpacked
    """
    if not self._CheckForAckOrNack(response, unpacked_data, unpack_exception):
      return

    self._PerformMatching(response, unpacked_data, unpack_exception)

  def _MakeRequestKey(self, sub_device, command_class, pid):
    # If the request was sent to the all-subdevice, the response should come
    # from the root.
    if sub_device == PidStore.ALL_SUB_DEVICES:
      sub_device = PidStore.ROOT_DEVICE
    self._outstanding_request = (sub_device, command_class, pid)

  def _HandleQueuedResponse(self, response, unpacked_data, unpack_exception):
    """Handle a response to a get QUEUED_MESSAGE request.

    Args:
      response: A RDMResponse object
      unpacked_data: The unpacked data structure for this pid
      unpack_exception: The exception if the data could not be unpacked
    """
    if not self._CheckForAckOrNack(response, unpacked_data, unpack_exception):
      return

    if response.response_code != OlaClient.RDM_COMPLETED_OK:
      self.SetFailed('Request failed: %s' % response.ResponseCodeAsString())
      self.Stop()
      return False

    # At this stage we have NACKs and ACKs left
    request_key = (response.sub_device, response.command_class, response.pid)
    if (self._outstanding_request == request_key):
      # This is what we've been waiting for
      self._PerformMatching(response, unpacked_data, unpack_exception)
      return

    queued_message_pid = self.LookupPid('QUEUED_MESSAGE')
    status_messages_pid = self.LookupPid('STATUS_MESSAGES')
    if (response.pid == queued_message_pid.value and
        response.response_type == OlaClient.RDM_NACK_REASON):
        # A Nack here is fatal because if we get an ACK_TIMER, QUEUED_MESSAGE
        # should be supported.
        self.SetFailed(' Get Queued message returned nack: %s' %
                       response.nack_reason)
        self.Stop()
        return
    elif (response.pid == status_messages_pid.value and
          unpacked_data.get('messages', None) == []):
        # This means we've run out of messages
        if self._state == TestState.NOT_RUN:
          self.SetFailed('ACK_TIMER issued but the response was never queued')
        self.Stop()
        return

    # Otherwise fetch the next one
    self._GetQueuedMessage()

  def _CheckForAckOrNack(self, response, unpacked_data, unpack_exception):
    """Check the state of a RDM response.

    Returns:
      True if processing should continue, False if there was a transport error
      or a ACK_TIMER was received.
    """
    if not response.status.Succeeded():
      # This indicates a transport error
      self.SetBroken(' Error: %s' % response.status.message)
      self.Stop()
      return False

    if response.response_code != OlaClient.RDM_COMPLETED_OK:
      self.LogDebug(' Request status: %s' % response.ResponseCodeAsString())
      if response.response_code == OlaClient.RDM_DUB_RESPONSE:
        # track timing for DUB responses.
        self._RecordFrameTiming(response, TimingStats.DUB)
        self._LogFrameTiming(response)
      return True

    self._RecordFrameTiming(response)

    # Handle the case of an ack timer
    if response.response_type == OlaClient.RDM_ACK_TIMER:
      self.LogDebug(' Received ACK TIMER set to %d ms' % response.ack_timer)
      self._LogFrameTiming(response)
      self._wrapper.AddEvent(response.ack_timer, self._GetQueuedMessage)
      return False

    # Now log the result
    if response.WasAcked():
      if unpack_exception:
        self.LogDebug(' Response: %s, PID: 0x%04hx, TN: %d, Error: %s' %
                      (response, response.pid, response.transaction_number,
                       unpack_exception))
      else:
        escaped_string = '%s' % self._EscapeData(unpacked_data)
        self.LogDebug(' Response: %s, PID: 0x%04hx, TN: %d, PDL: %d, data: %s'
                      % (response, response.pid, response.transaction_number,
                         len(response.data), escaped_string))
    else:
      self.LogDebug(' Response: %s, PID: 0x%04hx, TN: %d' %
                    (response, response.pid, response.transaction_number))

    self._LogFrameTiming(response)
    return True

  def _EscapeData(self, data):
    if type(data) == list:
      return [self._EscapeData(i) for i in data]
    elif type(data) == dict:
      d = {}
      for k, v in data.iteritems():
        d[k] = self._EscapeData(v)
      return d
    elif type(data) == str:
      return data.encode('string-escape')
    elif type(data) == unicode:
      return data.encode('unicode-escape')
    else:
      return data

  def _PerformMatching(self, response, unpacked_data, unpack_exception):
    """Check if this response matched any of the expected values.

    Args:
      response: A RDMResponse object
      unpacked_data: The unpacked data structure for this pid
      unpack_exception: The exception if the data could not be unpacked
    """
    if response.WasAcked():
      if unpack_exception:
        self.SetFailed('Invalid Param data: %s' % unpack_exception)
        self.Stop()
        return

    if self._in_reset_mode:
      self._wrapper.Stop()
      return

    for result in self._expected_results:
      if result.Matches(response, unpacked_data):
        self._state = TestState.PASSED
        self.VerifyResult(response, unpacked_data)
        if result.warning is not None:
          self.AddWarning(result.warning)
        if result.advisory is not None:
          self.AddAdvisory(result.advisory)
        if result.action is not None:
          result.action()
        else:
          self._wrapper.Stop()

        return

    # Nothing matched
    self.SetFailed('expected one of:')
    for result in self._expected_results:
      self.LogDebug('  %s' % result)
    self.Stop()

  def _GetQueuedMessage(self):
    """Fetch queued messages."""
    queued_message_pid = self.LookupPid('QUEUED_MESSAGE')
    data = ['error']
    self.LogDebug(' GET: pid: %s, data: %s' % (queued_message_pid, data))

    self._api.Get(self._universe,
                  self._uid,
                  PidStore.ROOT_DEVICE,
                  queued_message_pid,
                  self._HandleQueuedResponse,
                  data,
                  include_frames=True)

  def _RecordFrameTiming(self, response, override_type=None):
    for frame in response.frames:
      frame_type = override_type
      if not frame_type:
        frame_type = TimingStats.FrameTypeFromCommandClass(
            response.command_class)
      self._timing_stats.RecordFrame(frame_type, frame)

  def _LogFrameTiming(self, response):
    for frame in response.frames:
      stats = []
      if frame.response_delay:
        stats.append('Response Delay: %.1fus' %
                     self._NanoSecondsToMicroSeconds(frame.response_delay))
      if frame.break_time:
        stats.append('Break: %.1fus' %
                     self._NanoSecondsToMicroSeconds(frame.break_time))
      if frame.mark_time:
        stats.append('Mark: %.1fus' %
                     self._NanoSecondsToMicroSeconds(frame.mark_time))
      if frame.data_time:
        stats.append('Data: %.1fus' %
                     self._NanoSecondsToMicroSeconds(frame.data_time))

      if stats:
        self.LogDebug('    ' + ', '.join(stats))

  def _NanoSecondsToMicroSeconds(self, nano_seconds):
    return nano_seconds / 1000.0


class OptionalParameterTestFixture(ResponderTestFixture):
  """A sub class of ResponderTestFixture that alters behaviour if the PID isn't
     supported.
  """
  def Requires(self):
    return (super(OptionalParameterTestFixture, self).Requires() +
            ['supported_parameters'])

  def PidSupported(self):
    return self.pid.value in self.Property('supported_parameters')

  def AddIfGetSupported(self, result):
    if not self.PidSupported():
      result = self.NackGetResult(RDMNack.NR_UNKNOWN_PID)
    self.AddExpectedResults(result)

  def AddIfSetSupported(self, result):
    # The lock modes means that some sets may return NR_WRITE_PROTECT. Account
    # for that here.
    if not self.PidSupported():
      expected_results = self.NackSetResult(RDMNack.NR_UNKNOWN_PID)
    else:
      expected_results = [
        self.NackSetResult(
          RDMNack.NR_WRITE_PROTECT,
          advisory='SET %s was write protected, try changing the lock mode if '
                   'enabled' % self.pid.name)
      ]
      if isinstance(result, list):
        expected_results.extend(result)
      else:
        expected_results.append(result)

    self.AddExpectedResults(expected_results)