#!/usr/bin/python -u
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012 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.


"""Unittest for event_log.py."""


import collections
import dbus
import factory_common  # pylint: disable=W0611
import logging
import os
import re
import shutil
import tempfile
import threading
import time
import unittest
import uuid
import yaml

from cros.factory.test import event_log
from cros.factory.utils import file_utils
from cros.factory.hwid.v3.common import ProbedComponentResult

MAC_RE = re.compile(r'^([a-f0-9]{2}:){5}[a-f0-9]{2}$')
UUID_RE = re.compile(r'^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-'
                     '[a-f0-9]{4}-[a-f0-9]{12}$')


def Reset():
  # Deletes state files and resets global variables.
  event_log.device_id = event_log.reimage_id = None
  shutil.rmtree(event_log.EVENT_LOG_DIR, ignore_errors=True)
  for f in [event_log.DEVICE_ID_PATH, event_log.SEQUENCE_PATH,
            event_log.BOOT_SEQUENCE_PATH, event_log.EVENTS_PATH]:
    file_utils.TryUnlink(f)


class BasicTest(unittest.TestCase):
  """Tests basic elements in event_log.py."""

  def testEventNameRE(self):
    for i in ('a', '_', 'azAZ09_', 'a0'):
      self.assertTrue(event_log.EVENT_NAME_RE.match(i))

    for i in ('', 'a.', '0', '0a'):
      self.assertFalse(event_log.EVENT_NAME_RE.match(i))


class GlobalSeqTest(unittest.TestCase):
  """Unittests for GlobalSeq."""

  def setUp(self):
    Reset()
    self.tmp = tempfile.mkdtemp(prefix='GlobalSeqTest.')

  def tearDown(self):
    shutil.rmtree(self.tmp)

  def testBasic(self):
    seq = event_log.GlobalSeq()
    for i in range(3):
      self.assertEquals(i, seq.Next())
    del seq

    # Try again with a new sequence file
    seq = event_log.GlobalSeq()
    for i in range(3, 6):
      self.assertEquals(i, seq.Next())
    del seq

  def testYamlDump(self):
    class OtherType(object):
      """A generic class."""

      def __init__(self, attr_foo):
        self.attr_foo = attr_foo

    dump = lambda x: event_log.YamlDump(x).strip()
    # FloatDigit type
    self.assertEqual('\n'.join(['0.1235', '...']),
                     dump(event_log.FloatDigit(0.12345, 4)))
    # A subclass of a dict
    self.assertEqual('\n'.join(['bar: 1', 'foo: 3']),
                     dump(collections.OrderedDict([('bar', 1), ('foo', 3)])))
    # A subclass of a list
    self.assertEqual('\n'.join(['- comp_foo', '- value_foo', '- null']),
                     dump(ProbedComponentResult('comp_foo', 'value_foo', None)))
    # Tuple type
    self.assertEqual('\n'.join(['- v1', '- v2', '- v3']),
                     dump(('v1', 'v2', 'v3')))
    # A subclass of an unicode, treating as a str
    self.assertEqual('\n'.join(['a dbus string', '...']),
                     dump(dbus.String('a dbus string')))
    # A general object
    self.assertEqual('\n'.join(['attr_foo: Foo']),
                     dump(OtherType('Foo')))
    # An object without attribute
    self.assertEqual('{}', dump(self.testYamlDump))

  def testMissingSequenceFile(self):
    # Generate a few sequence numbers.
    seq = event_log.GlobalSeq()
    self.assertEquals(0, seq.Next())
    self.assertEquals(1, seq.Next())
    # Log an event (preamble will have sequence number 2; main
    # event will have 3).
    event_log.EventLog('foo').Log('bar')
    with open(event_log.EVENTS_PATH) as f:
      assert 'SEQ: 3\n' in f.readlines()

    # Delete the sequence file to simulate corruption.
    os.unlink(event_log.SEQUENCE_PATH)
    # Sequence file should be re-created, starting with 4 plus
    # SEQ_INCREMENT_ON_BOOT.
    self.assertEquals(4 + event_log.SEQ_INCREMENT_ON_BOOT,
                      seq.Next())

    # Delete the sequence file and create a new GlobalSeq object to
    # simulate a reboot.  We'll do this a few times.
    for i in range(3):
      # Log an event to record the new sequence number for "reboot"
      event_log.EventLog('foo').Log('bar')

      del seq
      os.unlink(event_log.SEQUENCE_PATH)
      seq = event_log.GlobalSeq()
      # Sequence file should be re-created, increasing by 2 for the logged
      # event, and SEQ_INCREMENT_ON_BOOT for the reboot.
      self.assertEquals(7 + i * 3 + (i + 2) * event_log.SEQ_INCREMENT_ON_BOOT,
                        seq.Next())

  def _testThreads(self, after_read=lambda: True):
    '''Tests atomicity by doing operations in 10 threads for 1 sec.

    Args:
      after_read: See GlobalSeq._after_read.
    '''
    values = []

    start_time = time.time()
    end_time = start_time + 1

    def target():
      seq = event_log.GlobalSeq(_after_read=after_read)
      while time.time() < end_time:
        values.append(seq.Next())

    threads = [threading.Thread(target=target) for _ in xrange(10)]
    for t in threads:
      t.start()
    for t in threads:
      t.join()

    # After we sort, should be numbers [1, len(values)].
    values.sort()
    self.assertEquals(range(len(values)), values)
    return values

  def testThreadsWithSleep(self):
    values = self._testThreads(after_read=lambda: time.sleep(.05))
    # There should be about 20 to 30 values (1 every 50 ms for 1 s, plus
    # a number less than the number of threads).
    # Significantly more or less than that and something went wrong.
    self.assertTrue(len(values) > 10, values)
    self.assertTrue(len(values) < 30, values)

  def testThreadsWithoutSleep(self):
    values = self._testThreads()
    # There should be lots of values (I get over 35000 on my desktop); we'll
    # just make sure there are >1000.
    self.assertTrue(len(values) > 1000, values)


class EventLogTest(unittest.TestCase):
  """Unittests for EventLog."""

  def setUp(self):
    Reset()
    self.tmp = tempfile.mkdtemp()

  def tearDown(self):
    shutil.rmtree(self.tmp)

  def testGetBootId(self):
    assert UUID_RE.match(event_log.GetBootId())

  def testGetDeviceIdGenerateId(self):
    device_id = event_log.GetDeviceId()
    assert (MAC_RE.match(device_id) or
            UUID_RE.match(device_id)), device_id

    # Remove device_id and make sure we get the same thing
    # back again, re-reading it from disk or the wlan0 interface
    event_log.device_id = None
    self.assertEqual(device_id, event_log.GetDeviceId())

    self.assertNotEqual(device_id, event_log.GetReimageId())

  def testGetDeviceIdFromSearchPath(self):
    mock_id = 'MOCK_ID'
    device_id_search_path = os.path.join(self.tmp, '.device_id_search')
    with open(device_id_search_path, 'w') as f:
      print >> f, mock_id
    event_log.DEVICE_ID_SEARCH_PATHS = [device_id_search_path]

    # Device ID should be the same as mock_id every time it is called.
    device_id = event_log.GetDeviceId()
    self.assertEqual(mock_id, device_id)
    self.assertEqual(mock_id, event_log.GetDeviceId())

    # Ensure the mock ID remains the same even if search path is gone.
    # i.e. obtains ID from the file
    event_log.device_id = None
    event_log.DEVICE_ID_SEARCH_PATHS = []
    self.assertEqual(mock_id, event_log.GetDeviceId())

  def testGetReimageId(self):
    reimage_id = event_log.GetReimageId()
    assert UUID_RE.match(reimage_id), reimage_id

    # Remove reimage_id and make sure we get the same thing
    # back again, re-reading it from disk
    event_log.reimage_id = None
    self.assertEqual(reimage_id, event_log.GetReimageId())

    # Remove the reimage_id file; now we should get something
    # *different* back.
    event_log.reimage_id = None
    os.unlink(event_log.REIMAGE_ID_PATH)
    self.assertNotEqual(reimage_id, event_log.GetReimageId())

  def testSuppress(self):
    for suppress in [False, True]:
      Reset()
      log = event_log.EventLog('test', suppress=suppress)
      log.Log('test')
      self.assertEquals(suppress, not os.path.exists(event_log.EVENTS_PATH))

  def testEventLogDefer(self):
    self._testEventLog(True)

  def testEventLogNoDefer(self):
    self._testEventLog(False)

  def _testEventLog(self, defer):
    log = event_log.EventLog('test', defer=defer)
    self.assertEqual(os.path.exists(event_log.EVENTS_PATH), not defer)

    event0 = dict(a='A',
                  b=1,
                  c=[1, 2],
                  d={'D1': 3, 'D2': 4},
                  e=['E1', {'E2': 'E3'}],
                  f=True,
                  g=u'[[[囧]]]'.encode('utf-8'),
                  h=u'[[[囧]]]')
    log.Log('event0', **event0)

    # Open and close another logger as well
    event2 = dict(foo='bar')
    log2 = event_log.EventLog('test2', defer=defer)
    log2.Log('event2', **event2)
    log2.Close()

    log.Log('event1')
    log.Close()

    try:
      log.Log('should-fail')
      self.fail('Expected exception')
    except:  # pylint: disable=W0702
      pass

    log_data = list(yaml.load_all(open(event_log.EVENTS_PATH, 'r')))
    self.assertEqual(6, len(log_data))
    # The last one should be empty; remove it
    self.assertIsNone(None, log_data[-1])
    log_data = log_data[0:-1]

    for i in log_data:
      # Check and remove times, to make everything else easier to compare
      assert re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$',
                      i['TIME']), i['TIME']
      del i['TIME']

    self.assertEqual(
        ['EVENT', 'LOG_ID', 'PREFIX', 'SEQ',
         'boot_id', 'boot_sequence', 'device_id',
         'factory_md5sum', 'reimage_id'],
        sorted(log_data[0].keys()))
    self.assertEqual('preamble', log_data[0]['EVENT'])
    self.assertEqual('test', log_data[0]['PREFIX'])
    self.assertEqual(0, log_data[0]['SEQ'])
    self.assertEqual(event_log.GetBootId(), log_data[0]['boot_id'])
    self.assertEqual(-1, log_data[0]['boot_sequence'])
    self.assertEqual(event_log.GetDeviceId(), log_data[0]['device_id'])
    self.assertEqual(event_log.GetReimageId(), log_data[0]['reimage_id'])
    log_id = log_data[0]['LOG_ID']
    uuid.UUID(log_id)  # Make sure UUID is well-formed

    # Check all the events
    event0.update(dict(EVENT='event0', SEQ=1, LOG_ID=log_id, PREFIX='test'))
    # Yaml loader converts non-ASCII strings to unicode.
    event0['g'] = event0['h']
    self.assertEqual(event0, log_data[1])
    self.assertEqual(
        dict(EVENT='preamble',
             LOG_ID=log2.log_id,
             PREFIX='test2',
             SEQ=2,
             boot_id=event_log.GetBootId(),
             boot_sequence=-1,
             device_id=event_log.GetDeviceId(),
             factory_md5sum=None,
             reimage_id=event_log.GetReimageId()),
        log_data[2])
    # Check the preamble and event from the second logger
    self.assertEqual(
        dict(EVENT='event2',
             LOG_ID=log2.log_id,
             PREFIX='test2',
             SEQ=3,
             foo='bar'),
        log_data[3])
    self.assertEqual(dict(EVENT='event1', SEQ=4, LOG_ID=log_id, PREFIX='test'),
                     log_data[4])

  def testDeferWithoutEvents(self):
    log = event_log.EventLog('test', defer=True)
    log.Close()
    self.assertFalse(os.path.exists(event_log.EVENTS_PATH))

  def testBootSequence(self):
    try:
      os.unlink(event_log.BOOT_SEQUENCE_PATH)
    except OSError:
      pass

    for i in xrange(-1, 5):
      self.assertEqual(i, event_log.GetBootSequence())
      event_log.IncrementBootSequence()
      self.assertEqual(str(i + 1),
                       open(event_log.BOOT_SEQUENCE_PATH).read())


class GlobalEventLogTest(unittest.TestCase):
  """Unittests for GetGlobalLogger."""

  def setUp(self):
    # reset the global event logger
    event_log._global_event_logger = None  # pylint: disable=W0212
    # pylint: disable=W0212
    event_log._default_event_logger_prefix = None

    if 'CROS_FACTORY_TEST_PATH' in os.environ:
      del os.environ['CROS_FACTORY_TEST_PATH']
    if 'CROS_FACTORY_TEST_INVOCATION' in os.environ:
      del os.environ['CROS_FACTORY_TEST_INVOCATION']

  def testGlobalInstanceNoEnv(self):
    self.assertRaises(ValueError, event_log.GetGlobalLogger)

  def testGlobalInstancePrefix(self):
    event_log.SetGlobalLoggerDefaultPrefix('bar')
    log = event_log.GetGlobalLogger()
    self.assertEqual('bar', log.prefix)
    self.assertTrue(log.log_id)

  def testInvalidDefaultPrefix(self):
    self.assertRaises(ValueError,
                      event_log.SetGlobalLoggerDefaultPrefix, '---')

  def testDefaultPrefix(self):
    os.environ['CROS_FACTORY_TEST_PATH'] = 'FooTest'
    event_log.SetGlobalLoggerDefaultPrefix('bar')

    log = event_log.GetGlobalLogger()
    self.assertEqual('bar', log.prefix)
    self.assertTrue(log.log_id)

    self.assertRaises(event_log.EventLogException,
                      event_log.SetGlobalLoggerDefaultPrefix, 'bar2')

  def testGlobalInstanceWithEnv(self):
    stub_uuid = 'bfa88756-ef2b-4e58-a4a2-eda1408bc93f'
    os.environ['CROS_FACTORY_TEST_PATH'] = 'FooTest'
    os.environ['CROS_FACTORY_TEST_INVOCATION'] = stub_uuid

    log = event_log.GetGlobalLogger()
    self.assertEqual('FooTest', log.prefix)
    self.assertEqual(stub_uuid, log.log_id)

  def testSingleton(self):
    os.environ['CROS_FACTORY_TEST_PATH'] = 'FooTest'
    # pylint: disable=W0212
    self.assertEquals(None, event_log._global_event_logger)
    log1 = event_log.GetGlobalLogger()
    log2 = event_log.GetGlobalLogger()
    self.assertTrue(log1 is log2)

if __name__ == '__main__':
  logging.basicConfig(level=logging.INFO)
  unittest.main()