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

from StringIO import StringIO
import json

from chromite.lib import cros_event as ce
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 = ce.EventIdGenerator()
    for expected_id in expected_ids:
      self.assertEquals(expected_id, id_gen.next())


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

    f2_msg = "This is the message for the failure"
    f2 = ce.Failure(f2_msg)
    self.assertEqual(f2.status, ce.EVENT_STATUS_FAIL)
    self.assertEqual(f2.message, 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 = ce.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.assertEquals(self.event1[ce.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[ce.EVENT_STATUS], ce.EVENT_STATUS_RUNNING)

    self.assertTrue(self.emitCalled)
    self.assertEqual(self.emitEvent[ce.EVENT_STATUS], ce.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[ce.EVENT_STATUS], ce.EVENT_STATUS_FAIL)
    self.assertEqual(self.emitEvent[ce.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[ce.EVENT_STATUS], customStatus)
    self.assertEqual(self.emitEvent[ce.EVENT_FAIL_MSG], failMsg)

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

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

    self.assertEqual(self.emitEvent[ce.EVENT_STATUS], ce.EVENT_STATUS_FAIL)
    self.assertEqual(self.emitEvent[ce.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[ce.EVENT_STATUS], ce.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 = ce.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 = ce.EventFileLogger(self.file_out, data=self.data1,
                                  encoder_func=self.encode_func)

  def testInit(self):
    self.assertTrue(isinstance(self.log, ce.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 ce.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 = ce.EventDummyLogger()

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


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

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

  def tearDown(self):
    if hasattr(self, "_last_root") and self._last_root:
      ce.setEventLogger(self._last_root)

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

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

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

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