catatester/devicemap/devicemap_e2e_test.py

"""E2E Tests for the devicemap.py script."""

import ast
import logging
import os
import subprocess
import sys
import unittest

from catatester.devicemap import devicemap_lib
import yaml


_DEVICEMAP_DIR = os.path.dirname(os.path.abspath(__file__))
_DEVICEMAP_SCRIPT = os.path.join(_DEVICEMAP_DIR, 'devicemap.py')
# Maximum allowed bots per host
_MAX_BOTS = 60
_YAML_FILE = os.path.join(_DEVICEMAP_DIR, 'devicemap.yaml')


class DeviceMapE2ETest(unittest.TestCase):
  """E2E Tests for the devicemap.py script."""

  def setUp(self):
    super(DeviceMapE2ETest, self).setUp()
    self.configs = None
    with open(_YAML_FILE, 'r') as f:
      self.configs = devicemap_lib.Configs(yaml.safe_load(f))

  def testGetBotConfig_fromStdout(self):
    p = subprocess.Popen(
        [sys.executable, _DEVICEMAP_SCRIPT, 'home-swarming-bot-host-1--1'],
        env={'PYTHONPATH': ':'.join(sys.path)[1:]},
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    output, err = p.communicate()
    self.assertFalse(p.returncode, output or err)
    botconfig = ast.literal_eval(output.decode())
    self.assertTrue(botconfig)
    self.assertIsInstance(botconfig, dict)

  def testHost_maximumBots(self):
    for host in self.configs.GetHosts():
      current_bots = self.configs.GetNumBotConfigs(host)
      message = '%s has more than %s bots [%s]' % (
          host, _MAX_BOTS, current_bots)
      self.assertLessEqual(current_bots, _MAX_BOTS, message)

  def testNoDuplicateHostIp(self):
    self.assertEqual(len(self.configs.GetHostIps()),
                     len(set(self.configs.GetHostIps())))

  def testNoDuplicateGaiaAccounts(self):
    # We can use the same set for both the IDs and the Addresses
    unique_gaia_accounts = set()
    non_unique_gaia_accounts = {}
    for host in self.configs.hosts:
      for bot in self.configs.hosts[host]['hosted_devices']:
        for surface in bot:
          device = bot[surface]
          if 'factory_reset' not in device:
            continue
          try:
            extra_state = device['extra_state']
            email = extra_state.get('email')
            gaia_id = extra_state.get('gaia_id')
            # Need to have a pair of both or none of them
            self.assertEqual(
                bool(email), bool(gaia_id),
                msg='email {} and gaia_id {}'.format(email, gaia_id))
            if not email and not gaia_id:
              continue
            if extra_state.get('allow_duplicate', False):
              if email not in non_unique_gaia_accounts.keys():
                existing_gaia_ids = non_unique_gaia_accounts.values()
                dup_gaia_id_msg = 'gaia_id {} is found in use already'
                self.assertNotIn(gaia_id, existing_gaia_ids,
                                 msg=dup_gaia_id_msg.format(gaia_id))
                non_unique_gaia_accounts[email] = gaia_id
              else:
                err_msg = 'gaia_id {} is inconsistent with this email {}'
                self.assertEqual(gaia_id, non_unique_gaia_accounts[email],
                                 msg=err_msg.format(gaia_id, email))
            else:
              # Email/ID cannot be re-used when allow_duplicate is set to False
              error_msg = 'Duplicate {} is found: {}'
              self.assertNotIn(email, unique_gaia_accounts,
                               msg=error_msg.format('email', email))
              self.assertNotIn(gaia_id, unique_gaia_accounts,
                               msg=error_msg.format('gaia_id', gaia_id))
              unique_gaia_accounts.add(email)
              unique_gaia_accounts.add(gaia_id)
          except KeyError:
            logging.warning('Need account info for %s', str(device))


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