lib/cros_event_unittest.py

# -*- coding: utf-8 -*-
# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unit tests for chros_event"""

from __future__ import print_function

import json

from six.moves import StringIO

from chromite.lib import cros_event
from chromite.lib import cros_test_lib


class EventIdGeneratorTest(cros_test_lib.TestCase):
  """Test EventIdGenerator fuction"""
  def testEventIdGenerator(self):
    expected_ids = {1, 2, 3, 4, 5}
    id_gen = cros_event.EventIdGenerator()
    for expected_id in expected_ids:
      self.assertEqual(expected_id, next(id_gen))


class FailTest(cros_test_lib.TestCase):
  """Test Failure class"""
  def testInit(self):
    f1 = cros_event.Failure()
    self.assertTrue(isinstance(f1, Exception))
    self.assertEqual(f1.status, cros_event.EVENT_STATUS_FAIL)

    f2_msg = 'This is the message for the failure'
    f2 = cros_event.Failure(f2_msg)
    self.assertEqual(f2.status, cros_event.EVENT_STATUS_FAIL)
    self.assertEqual(f2.msg, f2_msg)


class EventTest(cros_test_lib.TestCase):
  """Test Event class"""
  # pylint: disable=attribute-defined-outside-init
  def setUp(self):
    self._resetEmit()

    self.id1 = 1
    self.data1 = {1: 'a', 2: 'b', 3: 'c'}
    self.event1 = cros_event.Event(eid=self.id1,
                                   data=self.data1,
                                   emit_func=self.emitHook)

  def emitHook(self, event):
    self.emitCalled = True
    self.emitEvent = event

  def _resetEmit(self):
    self.emitCalled = False
    self.emitEvent = None

  def testInit(self):
    self.assertEqual(self.event1[cros_event.EVENT_ID], self.id1)
    self.assertDictContainsSubset(self.data1, self.event1)

    self.assertTrue(isinstance(self.event1, dict))

  def testWithSuccess(self):
    """test success case"""
    with self.event1 as e:
      self.assertEqual(self.event1, e)
      self.assertEqual(e[cros_event.EVENT_STATUS],
                       cros_event.EVENT_STATUS_RUNNING)

    self.assertTrue(self.emitCalled)
    self.assertEqual(self.emitEvent[cros_event.EVENT_STATUS],
                     cros_event.EVENT_STATUS_PASS)

  def testWithFailureCall(self):
    """test with fail() call"""
    failMsg = 'failed, as it should correctly'

    with self.event1 as e:
      e.fail(message=failMsg)

    self.assertEqual(self.emitEvent[cros_event.EVENT_STATUS],
                     cros_event.EVENT_STATUS_FAIL)
    self.assertEqual(self.emitEvent[cros_event.EVENT_FAIL_MSG], failMsg)

  def testWithFailureCallWithStatus(self):
    """test with fail() and custom status call"""
    failMsg = 'failed, as it should correctly'
    customStatus = 'UnitTestFailure'

    with self.event1 as e:
      e.fail(message=failMsg, status=customStatus)

    self.assertEqual(self.emitEvent[cros_event.EVENT_STATUS], customStatus)
    self.assertEqual(self.emitEvent[cros_event.EVENT_FAIL_MSG], failMsg)

  def testWithFailure(self):
    """test with raising failure exception"""
    failMsg = 'failed, as it should correctly'

    with self.event1:
      raise cros_event.Failure(failMsg)

    self.assertEqual(self.emitEvent[cros_event.EVENT_STATUS],
                     cros_event.EVENT_STATUS_FAIL)
    self.assertEqual(self.emitEvent[cros_event.EVENT_FAIL_MSG], failMsg)

  def testWithExceptionFail(self):
    """test with raised non-Failure exception"""
    try:
      with self.event1:
        raise NameError
    except NameError:
      self.assertEqual(self.emitEvent[cros_event.EVENT_STATUS],
                       cros_event.EVENT_STATUS_FAIL)


class EventLoggerTest(cros_test_lib.TestCase):
  """Test EventLogger class"""
  # pylint: disable=attribute-defined-outside-init

  def emitHook(self, event):
    self.emitCalled = True
    self.emitEvent = event

  def _resetEmit(self):
    self.emitCalled = False
    self.emitEvent = None

  def setUp(self):
    self._resetEmit()

    self.data1 = {1:2, 3:4}

    self.events = []
    self.log1 = cros_event.EventLogger(self.emitHook, data=self.data1)

  def testEvent(self):
    e_data = {'one': 'two', 'three':'four'}

    e = self.log1.Event(data=e_data)

    self.assertDictContainsSubset(e_data, e)
    self.assertDictContainsSubset(self.data1, e)

  def testEventWithKind(self):
    kind = 'testStep'
    e = self.log1.Event(kind=kind)
    self.assertEqual(e['id'][0], kind)


class EventFileLoggerTest(cros_test_lib.TestCase):
  """Test EventFileLogger class"""
  # pylint: disable=attribute-defined-outside-init

  def encode_func(self, event):
    self.emitEvent = event
    return json.dumps(event)

  def get_event_from_file(self):
    event_str = self.file_out.getvalue()
    self.file_out.buf = '' # Clear out StringIO buffer
    return json.loads(event_str)

  def setUp(self):
    self.data1 = {'one':2, 'two':4}
    self.emitEvent = None
    self.file_out = StringIO()
    self.log = cros_event.EventFileLogger(self.file_out, data=self.data1,
                                          encoder_func=self.encode_func)

  def testInit(self):
    self.assertTrue(isinstance(self.log, cros_event.EventLogger))

  def testEvents(self):
    with self.log.Event():
      pass

    self.assertDictEqual(self.emitEvent, self.get_event_from_file())

  def testEventFail(self):
    with self.log.Event():
      raise cros_event.Failure('always fail')

    self.assertDictEqual(self.emitEvent, self.get_event_from_file())

  def testShutdown(self):
    self.log.shutdown()


class EventDummyLogger(cros_test_lib.TestCase):
  """Test EventDummyLogger class"""

  def setUp(self):
    self.log = cros_event.EventDummyLogger()

  def testInit(self):
    self.assertTrue(isinstance(self.log, cros_event.EventLogger))


class FunctionTest(cros_test_lib.TestCase):
  """Test Module Tests"""

  def setUp(self):
    self._last_root = cros_event.root

  def tearDown(self):
    if hasattr(self, '_last_root') and self._last_root:
      cros_event.setEventLogger(self._last_root)

  def SetEventLoggerTest(self):
    new_log = cros_event.EventDummyLogger()
    cros_event.setEventLogger(new_log)
    self.assertEqual(new_log, cros_event.root)

  def newEventTest(self):
    e1 = cros_event.newEvent()
    self.assertTrue(isinstance(e1, cros_event.Event))

    e2 = cros_event.newEvent(foo='bar')
    self.assertTrue(isinstance(e2, cros_event.Event))
    self.assertEqual('bar', e2['foo'])

    test_kind = 'testKind'
    e3 = cros_event.NewEvent(kind=test_kind)
    self.assertEqual(e3['id'][0], test_kind)