#!/usr/bin/env python

# 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.

"""Unit tests for shop floor server."""

import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import unittest
import xmlrpclib

import shopfloor
import shopfloor_server


class ShopFloorServerTest(unittest.TestCase):

  def setUp(self):
    '''Starts shop floor server and creates client proxy.'''
    self.server_port = shopfloor_server._DEFAULT_SERVER_PORT
    self.base_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
    self.data_dir = tempfile.mkdtemp(prefix='shopfloor_data.')
    self.registration_code_log = (
        os.path.join(self.data_dir, shopfloor.REGISTRATION_CODE_LOG_CSV))
    csv_source = os.path.join(self.base_dir, 'testdata', 'shopfloor',
                              'devices.csv')
    csv_work = os.path.join(self.data_dir, 'devices.csv')
    shutil.copyfile(csv_source, csv_work)
    os.mkdir(os.path.join(self.data_dir, shopfloor.UPDATE_DIR))
    os.mkdir(os.path.join(self.data_dir, shopfloor.UPDATE_DIR, 'factory'))

    cmd = ['python', os.path.join(self.base_dir, 'shopfloor_server.py'),
           '-q', '-a', 'localhost', '-p', str(self.server_port),
           '-m', 'shopfloor.simple.ShopFloor',
           '-d', self.data_dir]
    self.process = subprocess.Popen(cmd)
    self.proxy = xmlrpclib.ServerProxy('http://localhost:%s' % self.server_port,
                                       allow_none=True)
    # Waits the server to be ready, up to 1 second.
    for i in xrange(10):
      try:
        self.proxy.Ping()
        break
      except:
        time.sleep(0.1)
        continue
    else:
      self.fail('Server never came up')

  def tearDown(self):
    '''Terminates shop floor server'''
    self.process.terminate()
    self.process.wait()
    shutil.rmtree(self.data_dir)

  def testGetHWID(self):
    # Valid HWIDs range from CR001001 to CR001025
    for i in range(25):
      serial = 'CR0010%02d' % (i + 1)
      result = self.proxy.GetHWID(serial)
      self.assertTrue(result.startswith('MAGICA '))
      self.assertEqual(len(result.split(' ')), 4)

    # Test invalid serial numbers
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetHWID, '0000')
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetHWID, 'garbage')
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetHWID, '')
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetHWID, None)
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetHWID, 'CR001000')
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetHWID, 'CR001026')

  def testGetHWIDUpdater_None(self):
    self.assertEquals(None, self.proxy.GetHWIDUpdater())

  def testGetHWIDUpdater_One(self):
    with open(os.path.join(self.data_dir, 'hwid_updater.sh'), 'w') as f:
      f.write('foobar')
    self.assertEquals('foobar', self.proxy.GetHWIDUpdater().data)

  def testGetHWIDUpdater_Two(self):
    for i in (1, 2):
      open(os.path.join(self.data_dir, 'hwid_updater_%d.sh' % i), 'w').close()
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetHWIDUpdater)

  def testGetVPD(self):
    # VPD fields defined in simple.csv
    RO_FIELDS = ('keyboard_layout', 'initial_locale', 'initial_timezone')
    RW_FIELDS_SET1 = ('wifi_mac', 'cellular_mac')
    RW_FIELDS_SET2 = ('wifi_mac', )

    vpd = self.proxy.GetVPD('CR001005')
    for field in RO_FIELDS:
      self.assertTrue(field in vpd['ro'] and vpd['ro'][field])
    for field in RW_FIELDS_SET1:
      self.assertTrue(field in vpd['rw'] and vpd['rw'][field])
    self.assertEqual(vpd['ro']['keyboard_layout'], 'xkb:us::eng')
    self.assertEqual(vpd['ro']['initial_locale'], 'en-US')
    self.assertEqual(vpd['ro']['initial_timezone'], 'America/Los_Angeles')
    self.assertEqual(vpd['rw']['wifi_mac'], '0b:ad:f0:0d:15:05')
    self.assertEqual(vpd['rw']['cellular_mac'], '70:75:65:6c:6c:65')

    vpd = self.proxy.GetVPD('CR001016')
    for field in RO_FIELDS:
      self.assertTrue(field in vpd['ro'] and vpd['ro'][field])
    for field in RW_FIELDS_SET2:
      self.assertTrue(field in vpd['rw'] and vpd['rw'][field])
    self.assertEqual(vpd['ro']['keyboard_layout'], 'xkb:us:intl:eng')
    self.assertEqual(vpd['ro']['initial_locale'], 'nl')
    self.assertEqual(vpd['ro']['initial_timezone'], 'Europe/Amsterdam')
    self.assertEqual(vpd['rw']['wifi_mac'], '0b:ad:f0:0d:15:10')
    self.assertEqual(vpd['rw']['cellular_mac'], '')

    # Checks MAC addresses
    for i in range(25):
      serial = 'CR0010%02d' % (i + 1)
      vpd = self.proxy.GetVPD(serial)
      wifi_mac = vpd['rw']['wifi_mac']
      self.assertEqual(wifi_mac, "0b:ad:f0:0d:15:%02x" % (i + 1))
      if i < 5:
        cellular_mac = vpd['rw']['cellular_mac']
        self.assertEqual(cellular_mac, "70:75:65:6c:6c:%02x" % (i + 0x61))

    # Checks invalid serial numbers
    self.assertRaises(xmlrpclib.Fault, self.proxy.GetVPD, 'MAGICA')
    return True

  def testUploadReport(self):
    # Upload simple blob
    blob = 'Simple Blob'
    report_name = 'simple_blob.rpt'
    report_path = os.path.join(self.data_dir, shopfloor.REPORTS_DIR,
                               report_name)
    self.proxy.UploadReport('CR001020', shopfloor.Binary('Simple Blob'),
                            report_name)
    self.assertTrue(os.path.exists(report_path))
    self.assertTrue(open(report_path).read(), blob)

    # Try to upload to invalid serial number
    self.assertRaises(xmlrpclib.Fault, self.proxy.UploadReport, 'CR00200', blob)

  def testFinalize(self):
    self.proxy.Finalize('CR001024')
    self.assertRaises(xmlrpclib.Fault, self.proxy.Finalize, '0999')

  def testGetTestMd5sum(self):
    md5_work = os.path.join(self.data_dir, shopfloor.UPDATE_DIR,
                            'factory', 'latest.md5sum')
    with open(md5_work, "w") as f:
      f.write('0891a16c456fcc322b656d5f91fbf060')
    self.assertEqual(self.proxy.GetTestMd5sum(),
                     '0891a16c456fcc322b656d5f91fbf060')
    os.remove(md5_work)

  def testGetTestMd5sumWithoutMd5sumFile(self):
    self.assertTrue(self.proxy.GetTestMd5sum() is None)

  def testGetRegistrationCodeMap(self):
    self.assertEquals(
        {'user': ('000000000000000000000000000000000000'
                  '0000000000000000000000000000190a55ad'),
         'group': ('010101010101010101010101010101010101'
                   '010101010101010101010101010162319fcc')},
        self.proxy.GetRegistrationCodeMap('CR001001'))

    # Make sure it was logged.
    log = open(self.registration_code_log).read()
    self.assertTrue(re.match(
        '^MAGICA MADOKA A-A 1214,'
        '000000000000000000000000000000000000'
        '0000000000000000000000000000190a55ad,'
        '010101010101010101010101010101010101'
        '010101010101010101010101010162319fcc,'
        '\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d\n', log), repr(log))

  def testUploadEvent(self):
    # Check if events dir is created.
    events_dir = os.path.join(self.data_dir, shopfloor.EVENTS_DIR)
    self.assertTrue(os.path.isdir(events_dir))

    # A new event file should be created.
    self.assertTrue(self.proxy.UploadEvent('LOG_C835C718',
                                           'PREAMBLE\n---\nEVENT_1\n'))
    event_file = os.path.join(events_dir, 'LOG_C835C718')
    self.assertTrue(os.path.isfile(event_file))

    # Additional events should be appended to existing event files.
    self.assertTrue(self.proxy.UploadEvent('LOG_C835C718',
                                           '---\nEVENT_2\n'))
    with open(event_file, 'r') as f:
      events = [event.strip() for event in f.read().split('---')]
      self.assertEqual(events[0], 'PREAMBLE')
      self.assertEqual(events[1], 'EVENT_1')
      self.assertEqual(events[2], 'EVENT_2')


if __name__ == '__main__':
  unittest.main()