#!/usr/bin/env python2 # Copyright 2015 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. import collections import itertools import logging import os import unittest import common from autotest_lib.frontend.afe.json_rpc import proxy from autotest_lib.server.lib import status_history from autotest_lib.site_utils import lab_inventory # _FAKE_TIME - an arbitrary but plausible time_t value. # You can make your own with `date +%s`. _FAKE_TIME = 1537457599 class _FakeHost(object): """Class to mock `Host` in _FakeHostHistory for testing.""" def __init__(self, hostname): self.hostname = hostname class _FakeHostEvent(object): def __init__(self, time): self.start_time = time self.end_time = time + 1 class _FakeHostHistory(object): """Class to mock `HostJobHistory` for testing.""" def __init__(self, model, pool, status, hostname=''): self.host_model = model self.host_board = model + '_board' self.host_pool = pool self.status = status self.host = _FakeHost(hostname) self.hostname = hostname self.start_time = _FAKE_TIME self.end_time = _FAKE_TIME + 20 self.fake_task = _FakeHostEvent(_FAKE_TIME + 5) self.exception = None def last_diagnosis(self): """Return the recorded diagnosis.""" if self.exception: raise self.exception else: return self.status, self.fake_task class _FakeHostLocation(object): """Class to mock `HostJobHistory` for location sorting.""" _HOSTNAME_FORMAT = 'chromeos%d-row%d-rack%d-host%d' def __init__(self, location): self.hostname = self._HOSTNAME_FORMAT % location @property def host(self): """Return a fake host object with a hostname.""" return self # Status values that may be returned by `HostJobHistory`. # # These merely rename the corresponding values in `status_history` # for convenience. _WORKING = status_history.WORKING _UNUSED = status_history.UNUSED _BROKEN = status_history.BROKEN _UNKNOWN = status_history.UNKNOWN class GetStatusTestCase(unittest.TestCase): """Tests for `_get_diagnosis()`.""" def _get_diagnosis_status(self, history): return lab_inventory._get_diagnosis(history).status def test_working_and_in_range(self): """Test WORKING when task times are in the history range.""" history = _FakeHostHistory('', '', _WORKING) history.fake_task = _FakeHostEvent(history.start_time + 1) self.assertEqual(self._get_diagnosis_status(history), _WORKING) def test_broken_and_in_range(self): """Test BROKEN when task times are in the history range.""" history = _FakeHostHistory('', '', _BROKEN) history.fake_task = _FakeHostEvent(history.start_time + 1) self.assertEqual(self._get_diagnosis_status(history), _BROKEN) def test_broken_and_straddles(self): """Test BROKEN when task time straddles the history start point.""" history = _FakeHostHistory('', '', _BROKEN) history.fake_task = _FakeHostEvent(history.start_time - 1) self.assertEqual(self._get_diagnosis_status(history), _BROKEN) def test_broken_and_out_of_range(self): """Test BROKEN when task times are before the history range.""" history = _FakeHostHistory('', '', _BROKEN) history.fake_task = _FakeHostEvent(history.start_time - 2) self.assertEqual(self._get_diagnosis_status(history), _UNUSED) def test_exception(self): """Test exceptions raised by `last_diagnosis()`.""" history = _FakeHostHistory('', '', _BROKEN) history.exception = proxy.JSONRPCException('exception for testing') self.assertIsNone(self._get_diagnosis_status(history)) class HostSetInventoryTestCase(unittest.TestCase): """Unit tests for class `_HostSetInventory`. Coverage is quite basic: mostly just enough to make sure every function gets called, and to make sure that the counting knows the difference between 0 and 1. The testing also ensures that all known status values that can be returned by `HostJobHistory` are counted as expected. """ def setUp(self): super(HostSetInventoryTestCase, self).setUp() self.histories = lab_inventory._HostSetInventory() def _add_host(self, status): fake = _FakeHostHistory('zebra', lab_inventory.SPARE_POOL, status) self.histories.record_host(fake) def _check_counts(self, working, broken, idle): """Check that pool counts match expectations. Asserts that `get_working()`, `get_broken()`, and `get_idle()` return the given expected values. Also assert that `get_total()` is the sum of all counts. @param working The expected total of working devices. @param broken The expected total of broken devices. @param idle The expected total of idle devices. """ self.assertEqual(self.histories.get_working(), working) self.assertEqual(self.histories.get_broken(), broken) self.assertEqual(self.histories.get_idle(), idle) self.assertEqual(self.histories.get_total(), working + broken + idle) def test_empty(self): """Test counts when there are no DUTs recorded.""" self._check_counts(0, 0, 0) def test_broken(self): """Test counting for broken DUTs.""" self._add_host(_BROKEN) self._check_counts(0, 1, 0) def test_working(self): """Test counting for working DUTs.""" self._add_host(_WORKING) self._check_counts(1, 0, 0) def test_idle(self): """Testing counting for idle status values.""" self._add_host(_UNUSED) self._check_counts(0, 0, 1) self._add_host(_UNKNOWN) self._check_counts(0, 0, 2) def test_working_then_broken(self): """Test counts after adding a working and then a broken DUT.""" self._add_host(_WORKING) self._add_host(_BROKEN) self._check_counts(1, 1, 0) def test_broken_then_working(self): """Test counts after adding a broken and then a working DUT.""" self._add_host(_BROKEN) self._add_host(_WORKING) self._check_counts(1, 1, 0) class PoolSetInventoryTestCase(unittest.TestCase): """Unit tests for class `_PoolSetInventory`. Coverage is quite basic: just enough to make sure every function gets called, and to make sure that the counting knows the difference between 0 and 1. The tests make sure that both individual pool counts and totals are counted correctly. """ _POOL_SET = ['humpty', 'dumpty'] def setUp(self): super(PoolSetInventoryTestCase, self).setUp() self._pool_histories = lab_inventory._PoolSetInventory(self._POOL_SET) def _add_host(self, pool, status): fake = _FakeHostHistory('zebra', pool, status) self._pool_histories.record_host(fake) def _check_all_counts(self, working, broken): """Check that total counts for all pools match expectations. Checks that `get_working()` and `get_broken()` return the given expected values when called without a pool specified. Also check that `get_total()` is the sum of working and broken devices. Additionally, call the various functions for all the pools individually, and confirm that the totals across pools match the given expectations. @param working The expected total of working devices. @param broken The expected total of broken devices. """ self.assertEqual(self._pool_histories.get_working(), working) self.assertEqual(self._pool_histories.get_broken(), broken) self.assertEqual(self._pool_histories.get_total(), working + broken) count_working = 0 count_broken = 0 count_total = 0 for pool in self._POOL_SET: count_working += self._pool_histories.get_working(pool) count_broken += self._pool_histories.get_broken(pool) count_total += self._pool_histories.get_total(pool) self.assertEqual(count_working, working) self.assertEqual(count_broken, broken) self.assertEqual(count_total, working + broken) def _check_pool_counts(self, pool, working, broken): """Check that counts for a given pool match expectations. Checks that `get_working()` and `get_broken()` return the given expected values for the given pool. Also check that `get_total()` is the sum of working and broken devices. @param pool The pool to be checked. @param working The expected total of working devices. @param broken The expected total of broken devices. """ self.assertEqual(self._pool_histories.get_working(pool), working) self.assertEqual(self._pool_histories.get_broken(pool), broken) self.assertEqual(self._pool_histories.get_total(pool), working + broken) def test_empty(self): """Test counts when there are no DUTs recorded.""" self._check_all_counts(0, 0) for pool in self._POOL_SET: self._check_pool_counts(pool, 0, 0) def test_all_working_then_broken(self): """Test counts after adding a working and then a broken DUT. For each pool, add first a working, then a broken DUT. After each DUT is added, check counts to confirm the correct values. """ working = 0 broken = 0 for pool in self._POOL_SET: self._add_host(pool, _WORKING) working += 1 self._check_pool_counts(pool, 1, 0) self._check_all_counts(working, broken) self._add_host(pool, _BROKEN) broken += 1 self._check_pool_counts(pool, 1, 1) self._check_all_counts(working, broken) def test_all_broken_then_working(self): """Test counts after adding a broken and then a working DUT. For each pool, add first a broken, then a working DUT. After each DUT is added, check counts to confirm the correct values. """ working = 0 broken = 0 for pool in self._POOL_SET: self._add_host(pool, _BROKEN) broken += 1 self._check_pool_counts(pool, 0, 1) self._check_all_counts(working, broken) self._add_host(pool, _WORKING) working += 1 self._check_pool_counts(pool, 1, 1) self._check_all_counts(working, broken) class LocationSortTests(unittest.TestCase): """Unit tests for `_sort_by_location()`.""" def setUp(self): super(LocationSortTests, self).setUp() def _check_sorting(self, *locations): """Test sorting a given list of locations. The input is an already ordered list of lists of tuples with row, rack, and host numbers. The test converts the tuples to hostnames, preserving the original ordering. Then it flattens and scrambles the input, runs it through `_sort_by_location()`, and asserts that the result matches the original. """ lab = 0 expected = [] for tuples in locations: lab += 1 expected.append( [_FakeHostLocation((lab,) + t) for t in tuples]) scrambled = [e for e in itertools.chain(*expected)] scrambled = [e for e in reversed(scrambled)] actual = lab_inventory._sort_by_location(scrambled) # The ordering of the labs in the output isn't guaranteed, # so we can't compare `expected` and `actual` directly. # Instead, we create a dictionary keyed on the first host in # each lab, and compare the dictionaries. self.assertEqual({l[0]: l for l in expected}, {l[0]: l for l in actual}) def test_separate_labs(self): """Test that sorting distinguishes labs.""" self._check_sorting([(1, 1, 1)], [(1, 1, 1)], [(1, 1, 1)]) def test_separate_rows(self): """Test for proper sorting when only rows are different.""" self._check_sorting([(1, 1, 1), (9, 1, 1), (10, 1, 1)]) def test_separate_racks(self): """Test for proper sorting when only racks are different.""" self._check_sorting([(1, 1, 1), (1, 9, 1), (1, 10, 1)]) def test_separate_hosts(self): """Test for proper sorting when only hosts are different.""" self._check_sorting([(1, 1, 1), (1, 1, 9), (1, 1, 10)]) def test_diagonal(self): """Test for proper sorting when all parts are different.""" self._check_sorting([(1, 1, 2), (1, 2, 1), (2, 1, 1)]) class InventoryScoringTests(unittest.TestCase): """Unit tests for `_score_repair_set()`.""" def setUp(self): super(InventoryScoringTests, self).setUp() def _make_buffer_counts(self, *counts): """Create a dictionary suitable as `buffer_counts`. @param counts List of tuples with model count data. """ self._buffer_counts = dict(counts) def _make_history_list(self, repair_counts): """Create a list suitable as `repair_list`. @param repair_counts List of (model, count) tuples. """ pool = lab_inventory.SPARE_POOL histories = [] for model, count in repair_counts: for i in range(0, count): histories.append( _FakeHostHistory(model, pool, _BROKEN)) return histories def _check_better(self, repair_a, repair_b): """Test that repair set A scores better than B. Contruct repair sets from `repair_a` and `repair_b`, and score both of them using the pre-existing `self._buffer_counts`. Assert that the score for A is better than the score for B. @param repair_a Input data for repair set A @param repair_b Input data for repair set B """ score_a = lab_inventory._score_repair_set( self._buffer_counts, self._make_history_list(repair_a)) score_b = lab_inventory._score_repair_set( self._buffer_counts, self._make_history_list(repair_b)) self.assertGreater(score_a, score_b) def _check_equal(self, repair_a, repair_b): """Test that repair set A scores the same as B. Contruct repair sets from `repair_a` and `repair_b`, and score both of them using the pre-existing `self._buffer_counts`. Assert that the score for A is equal to the score for B. @param repair_a Input data for repair set A @param repair_b Input data for repair set B """ score_a = lab_inventory._score_repair_set( self._buffer_counts, self._make_history_list(repair_a)) score_b = lab_inventory._score_repair_set( self._buffer_counts, self._make_history_list(repair_b)) self.assertEqual(score_a, score_b) def test_improve_worst_model(self): """Test that improving the worst model improves scoring. Construct a buffer counts dictionary with all models having different counts. Assert that it is both necessary and sufficient to improve the count of the worst model in order to improve the score. """ self._make_buffer_counts(('lion', 0), ('tiger', 1), ('bear', 2)) self._check_better([('lion', 1)], [('tiger', 1)]) self._check_better([('lion', 1)], [('bear', 1)]) self._check_better([('lion', 1)], [('tiger', 2)]) self._check_better([('lion', 1)], [('bear', 2)]) self._check_equal([('tiger', 1)], [('bear', 1)]) def test_improve_worst_case_count(self): """Test that improving the number of worst cases improves the score. Construct a buffer counts dictionary with all models having the same counts. Assert that improving two models is better than improving one. Assert that improving any one model is as good as any other. """ self._make_buffer_counts(('lion', 0), ('tiger', 0), ('bear', 0)) self._check_better([('lion', 1), ('tiger', 1)], [('bear', 2)]) self._check_equal([('lion', 2)], [('tiger', 1)]) self._check_equal([('tiger', 1)], [('bear', 1)]) # Each item is the number of DUTs in that status. STATUS_CHOICES = (_WORKING, _BROKEN, _UNUSED) StatusCounts = collections.namedtuple('StatusCounts', ['good', 'bad', 'idle']) # Each item is a StatusCounts tuple specifying the number of DUTs per status in # the that pool. CRITICAL_POOL = lab_inventory.CRITICAL_POOLS[0] SPARE_POOL = lab_inventory.SPARE_POOL POOL_CHOICES = (CRITICAL_POOL, SPARE_POOL) PoolStatusCounts = collections.namedtuple('PoolStatusCounts', ['critical', 'spare']) def create_inventory(data): """Create a `_LabInventory` instance for testing. This function allows the construction of a complete `_LabInventory` object from a simplified input representation. A single 'critical pool' is arbitrarily chosen for purposes of testing; there's no coverage for testing arbitrary combinations in more than one critical pool. @param data: dict {key: PoolStatusCounts}. @returns: lab_inventory._LabInventory object. """ histories = [] for model, counts in data.iteritems(): for p, pool in enumerate(POOL_CHOICES): for s, status in enumerate(STATUS_CHOICES): fake_host = _FakeHostHistory(model, pool, status) histories.extend([fake_host] * counts[p][s]) inventory = lab_inventory._LabInventory( histories, lab_inventory.MANAGED_POOLS) return inventory class LabInventoryTests(unittest.TestCase): """Tests for the basic functions of `_LabInventory`. Contains basic coverage to show that after an inventory is created and DUTs with known status are added, the inventory counts match the counts of the added DUTs. """ _MODEL_LIST = ['lion', 'tiger', 'bear'] # Oh, my! def _check_inventory_counts(self, inventory, data, msg=None): """Check that all counts in the inventory match `data`. This asserts that the actual counts returned by the various accessor functions for `inventory` match the values expected for the given `data` that created the inventory. @param inventory: _LabInventory object to check. @param data Inventory data to check against. Same type as `create_inventory`. """ self.assertEqual(set(inventory.keys()), set(data.keys())) for model, histories in inventory.iteritems(): expected_counts = data[model] actual_counts = PoolStatusCounts( StatusCounts( histories.get_working(CRITICAL_POOL), histories.get_broken(CRITICAL_POOL), histories.get_idle(CRITICAL_POOL), ), StatusCounts( histories.get_working(SPARE_POOL), histories.get_broken(SPARE_POOL), histories.get_idle(SPARE_POOL), ), ) self.assertEqual(actual_counts, expected_counts, msg) self.assertEqual(len(histories.get_working_list()), sum([p.good for p in expected_counts]), msg) self.assertEqual(len(histories.get_broken_list()), sum([p.bad for p in expected_counts]), msg) self.assertEqual(len(histories.get_idle_list()), sum([p.idle for p in expected_counts]), msg) def test_empty(self): """Test counts when there are no DUTs recorded.""" inventory = create_inventory({}) self.assertEqual(inventory.get_num_duts(), 0) self.assertEqual(inventory.get_boards(), set()) self._check_inventory_counts(inventory, {}) self.assertEqual(inventory.get_num_models(), 0) def _check_model_count(self, model_count): """Parameterized test for testing a specific number of models.""" msg = '[model: %d]' % (model_count,) models = self._MODEL_LIST[:model_count] data = { m: PoolStatusCounts( StatusCounts(1, 1, 1), StatusCounts(1, 1, 1), ) for m in models } inventory = create_inventory(data) self.assertEqual(inventory.get_num_duts(), 6 * model_count, msg) self.assertEqual(inventory.get_num_models(), model_count, msg) for pool in [CRITICAL_POOL, SPARE_POOL]: self.assertEqual(set(inventory.get_pool_models(pool)), set(models)) self._check_inventory_counts(inventory, data, msg=msg) def test_model_counts(self): """Test counts for various numbers of models.""" self.longMessage = True for model_count in range(0, len(self._MODEL_LIST)): self._check_model_count(model_count) def _check_single_dut_counts(self, critical, spare): """Parmeterized test for single dut counts.""" self.longMessage = True counts = PoolStatusCounts(critical, spare) model = self._MODEL_LIST[0] data = {model: counts} msg = '[data: %s]' % (data,) inventory = create_inventory(data) self.assertEqual(inventory.get_num_duts(), 1, msg) self.assertEqual(inventory.get_num_models(), 1, msg) self._check_inventory_counts(inventory, data, msg=msg) def test_single_dut_counts(self): """Test counts when there is a single DUT per board, and it is good.""" status_100 = StatusCounts(1, 0, 0) status_010 = StatusCounts(0, 1, 0) status_001 = StatusCounts(0, 0, 1) status_null = StatusCounts(0, 0, 0) self._check_single_dut_counts(status_100, status_null) self._check_single_dut_counts(status_010, status_null) self._check_single_dut_counts(status_001, status_null) self._check_single_dut_counts(status_null, status_100) self._check_single_dut_counts(status_null, status_010) self._check_single_dut_counts(status_null, status_001) # MODEL_MESSAGE_TEMPLATE - # This is a sample of the output text produced by # _generate_model_inventory_message(). This string is parsed by the # tests below to construct a sample inventory that should produce # the output, and then the output is generated and checked against # this original sample. # # Constructing inventories from parsed sample text serves two # related purposes: # - It provides a way to see what the output should look like # without having to run the script. # - It helps make sure that a human being will actually look at # the output to see that it's basically readable. # This should also help prevent test bugs caused by writing tests # that simply parrot the original output generation code. _MODEL_MESSAGE_TEMPLATE = ''' Model Avail Bad Idle Good Spare Total lion -1 13 2 11 12 26 tiger -1 5 2 9 4 16 bear 0 5 2 10 5 17 platypus 4 2 2 20 6 24 aardvark 7 2 2 6 9 10 ''' class PoolSetInventoryTests(unittest.TestCase): """Tests for `_generate_model_inventory_message()`. The tests create various test inventories designed to match the counts in `_MODEL_MESSAGE_TEMPLATE`, and asserts that the generated message text matches the original message text. Message text is represented as a list of strings, split on the `'\n'` separator. """ def setUp(self): self.maxDiff = None lines = [x.strip() for x in _MODEL_MESSAGE_TEMPLATE.split('\n') if x.strip()] self._header, self._model_lines = lines[0], lines[1:] self._model_data = [] for l in self._model_lines: items = l.split() model = items[0] bad, idle, good, spare = [int(x) for x in items[2:-1]] self._model_data.append((model, (good, bad, idle, spare))) def _make_minimum_spares(self, counts): """Create a counts tuple with as few spare DUTs as possible.""" good, bad, idle, spares = counts if spares > bad + idle: return PoolStatusCounts( StatusCounts(good + bad +idle - spares, 0, 0), StatusCounts(spares - bad - idle, bad, idle), ) elif spares < bad: return PoolStatusCounts( StatusCounts(good, bad - spares, idle), StatusCounts(0, spares, 0), ) else: return PoolStatusCounts( StatusCounts(good, 0, idle + bad - spares), StatusCounts(0, bad, spares - bad), ) def _make_maximum_spares(self, counts): """Create a counts tuple with as many spare DUTs as possible.""" good, bad, idle, spares = counts if good > spares: return PoolStatusCounts( StatusCounts(good - spares, bad, idle), StatusCounts(spares, 0, 0), ) elif good + bad > spares: return PoolStatusCounts( StatusCounts(0, good + bad - spares, idle), StatusCounts(good, spares - good, 0), ) else: return PoolStatusCounts( StatusCounts(0, 0, good + bad + idle - spares), StatusCounts(good, bad, spares - good - bad), ) def _check_message(self, message): """Checks that message approximately matches expected string.""" message = [x.strip() for x in message.split('\n') if x.strip()] self.assertIn(self._header, message) body = message[message.index(self._header) + 1:] self.assertEqual(body, self._model_lines) def test_minimum_spares(self): """Test message generation when the spares pool is low.""" data = { model: self._make_minimum_spares(counts) for model, counts in self._model_data } inventory = create_inventory(data) message = lab_inventory._generate_model_inventory_message(inventory) self._check_message(message) def test_maximum_spares(self): """Test message generation when the critical pool is low.""" data = { model: self._make_maximum_spares(counts) for model, counts in self._model_data } inventory = create_inventory(data) message = lab_inventory._generate_model_inventory_message(inventory) self._check_message(message) def test_ignore_no_spares(self): """Test that messages ignore models with no spare pool.""" data = { model: self._make_maximum_spares(counts) for model, counts in self._model_data } data['elephant'] = ((5, 4, 0), (0, 0, 0)) inventory = create_inventory(data) message = lab_inventory._generate_model_inventory_message(inventory) self._check_message(message) def test_ignore_no_critical(self): """Test that messages ignore models with no critical pools.""" data = { model: self._make_maximum_spares(counts) for model, counts in self._model_data } data['elephant'] = ((0, 0, 0), (1, 5, 1)) inventory = create_inventory(data) message = lab_inventory._generate_model_inventory_message(inventory) self._check_message(message) def test_ignore_no_bad(self): """Test that messages ignore models with no bad DUTs.""" data = { model: self._make_maximum_spares(counts) for model, counts in self._model_data } data['elephant'] = ((5, 0, 1), (5, 0, 1)) inventory = create_inventory(data) message = lab_inventory._generate_model_inventory_message(inventory) self._check_message(message) class _PoolInventoryTestBase(unittest.TestCase): """Parent class for tests relating to generating pool inventory messages. Func `setUp` in the class parses a given |message_template| to obtain header and body. """ def _read_template(self, message_template): """Read message template for PoolInventoryTest and IdleInventoryTest. @param message_template: the input template to be parsed into: header and content (report_lines). """ message_lines = message_template.split('\n') self._header = message_lines[1] self._report_lines = message_lines[2:-1] def _check_report_no_info(self, text): """Test a message body containing no reported info. The input `text` was created from a query to an inventory, which has no objects meet the query and leads to an `empty` return. Assert that the text consists of a single line starting with '(' and ending with ')'. @param text: Message body text to be tested. """ self.assertTrue(len(text) == 1 and text[0][0] == '(' and text[0][-1] == ')') def _check_report(self, text): """Test a message against the passed |expected_content|. @param text: Message body text to be tested. @param expected_content: The ground-truth content to be compared with. """ self.assertEqual(text, self._report_lines) # _POOL_MESSAGE_TEMPLATE - # This is a sample of the output text produced by # _generate_pool_inventory_message(). This string is parsed by the # tests below to construct a sample inventory that should produce # the output, and then the output is generated and checked against # this original sample. # # See the comments on _BOARD_MESSAGE_TEMPLATE above for the # rationale on using sample text in this way. _POOL_MESSAGE_TEMPLATE = ''' Model Bad Idle Good Total lion 5 2 6 13 tiger 4 1 5 10 bear 3 0 7 10 aardvark 2 0 0 2 platypus 1 1 1 3 ''' _POOL_ADMIN_URL = 'http://go/cros-manage-duts' class PoolInventoryTests(_PoolInventoryTestBase): """Tests for `_generate_pool_inventory_message()`. The tests create various test inventories designed to match the counts in `_POOL_MESSAGE_TEMPLATE`, and assert that the generated message text matches the format established in the original message text. The output message text is parsed against the following grammar: -> { "blank line" } -> Instructions to depty mentioning the admin page URL A blank line ->
-> Any number of lines describing one pool
-> The header line from `_POOL_MESSAGE_TEMPLATE` -> Any number of non-blank lines After parsing messages into the parts described above, various assertions are tested against the parsed output, including that the message body matches the body from `_POOL_MESSAGE_TEMPLATE`. Parse message text is represented as a list of strings, split on the `'\n'` separator. """ def setUp(self): super(PoolInventoryTests, self)._read_template(_POOL_MESSAGE_TEMPLATE) self._model_data = [] for l in self._report_lines: items = l.split() model = items[0] bad = int(items[1]) idle = int(items[2]) good = int(items[3]) self._model_data.append((model, (good, bad, idle))) def _create_histories(self, pools, model_data): """Return a list suitable to create a `_LabInventory` object. Creates a list of `_FakeHostHistory` objects that can be used to create a lab inventory. `pools` is a list of strings naming pools, and `model_data` is a list of tuples of the form `(model, (goodcount, badcount))` where `model` is a model name. `goodcount` is the number of working DUTs in the pool. `badcount` is the number of broken DUTs in the pool. @param pools List of pools for which to create histories. @param model_data List of tuples containing models and DUT counts. @return A list of `_FakeHostHistory` objects that can be used to create a `_LabInventory` object. """ histories = [] status_choices = (_WORKING, _BROKEN, _UNUSED) for pool in pools: for model, counts in model_data: for status, count in zip(status_choices, counts): for x in range(0, count): histories.append( _FakeHostHistory(model, pool, status)) return histories def _parse_pool_summaries(self, histories): """Parse message output according to the grammar above. Create a lab inventory from the given `histories`, and generate the pool inventory message. Then parse the message and return a dictionary mapping each pool to the message body parsed after that pool. Tests the following assertions: * Each contains a mention of exactly one pool in the `CRITICAL_POOLS` list. * Each pool is mentioned in exactly one . Note that the grammar requires the header to appear once for each pool, so the parsing implicitly asserts that the output contains the header. @param histories Input used to create the test `_LabInventory` object. @return A dictionary mapping model names to the output (a list of lines) for the model. """ inventory = lab_inventory._LabInventory( histories, lab_inventory.MANAGED_POOLS) message = lab_inventory._generate_pool_inventory_message( inventory).split('\n') poolset = set(lab_inventory.CRITICAL_POOLS) seen_url = False seen_intro = False description = '' model_text = {} current_pool = None for line in message: if not seen_url: if _POOL_ADMIN_URL in line: seen_url = True elif not seen_intro: if not line: seen_intro = True elif current_pool is None: if line == self._header: pools_mentioned = [p for p in poolset if p in description] self.assertEqual(len(pools_mentioned), 1) current_pool = pools_mentioned[0] description = '' model_text[current_pool] = [] poolset.remove(current_pool) else: description += line else: if line: model_text[current_pool].append(line) else: current_pool = None self.assertEqual(len(poolset), 0) return model_text def test_no_shortages(self): """Test correct output when no pools have shortages.""" model_text = self._parse_pool_summaries([]) for text in model_text.values(): self._check_report_no_info(text) def test_one_pool_shortage(self): """Test correct output when exactly one pool has a shortage.""" for pool in lab_inventory.CRITICAL_POOLS: histories = self._create_histories((pool,), self._model_data) model_text = self._parse_pool_summaries(histories) for checkpool in lab_inventory.CRITICAL_POOLS: text = model_text[checkpool] if checkpool == pool: self._check_report(text) else: self._check_report_no_info(text) def test_all_pool_shortages(self): """Test correct output when all pools have a shortage.""" histories = [] for pool in lab_inventory.CRITICAL_POOLS: histories.extend( self._create_histories((pool,), self._model_data)) model_text = self._parse_pool_summaries(histories) for pool in lab_inventory.CRITICAL_POOLS: self._check_report(model_text[pool]) def test_full_model_ignored(self): """Test that models at full strength are not reported.""" pool = lab_inventory.CRITICAL_POOLS[0] full_model = [('echidna', (5, 0, 0))] histories = self._create_histories((pool,), full_model) text = self._parse_pool_summaries(histories)[pool] self._check_report_no_info(text) model_data = self._model_data + full_model histories = self._create_histories((pool,), model_data) text = self._parse_pool_summaries(histories)[pool] self._check_report(text) def test_spare_pool_ignored(self): """Test that reporting ignores the spare pool inventory.""" spare_pool = lab_inventory.SPARE_POOL spare_data = self._model_data + [('echidna', (0, 5, 0))] histories = self._create_histories((spare_pool,), spare_data) model_text = self._parse_pool_summaries(histories) for pool in lab_inventory.CRITICAL_POOLS: self._check_report_no_info(model_text[pool]) _IDLE_MESSAGE_TEMPLATE = ''' Hostname Model Pool chromeos4-row12-rack4-host7 tiger bvt chromeos1-row3-rack1-host2 lion bvt chromeos3-row2-rack2-host5 lion cq chromeos2-row7-rack3-host11 platypus suites ''' class IdleInventoryTests(_PoolInventoryTestBase): """Tests for `_generate_idle_inventory_message()`. The tests create idle duts that match the counts and pool in `_IDLE_MESSAGE_TEMPLATE`. In test, it asserts that the generated idle message text matches the format established in `_IDLE_MESSAGE_TEMPLATE`. Parse message text is represented as a list of strings, split on the `'\n'` separator. """ def setUp(self): super(IdleInventoryTests, self)._read_template(_IDLE_MESSAGE_TEMPLATE) self._host_data = [] for h in self._report_lines: items = h.split() hostname = items[0] model = items[1] pool = items[2] self._host_data.append((hostname, model, pool)) self._histories = [] self._histories.append(_FakeHostHistory('echidna', 'bvt', _BROKEN)) self._histories.append(_FakeHostHistory('lion', 'bvt', _WORKING)) def _add_idles(self): """Add idle duts from `_IDLE_MESSAGE_TEMPLATE`.""" idle_histories = [_FakeHostHistory( model, pool, _UNUSED, hostname) for hostname, model, pool in self._host_data] self._histories.extend(idle_histories) def _check_header(self, text): """Check whether header in the template `_IDLE_MESSAGE_TEMPLATE` is in passed text.""" self.assertIn(self._header, text) def _get_idle_message(self, histories): """Generate idle inventory and obtain its message. @param histories: Used to create lab inventory. @return the generated idle message. """ inventory = lab_inventory._LabInventory( histories, lab_inventory.MANAGED_POOLS) message = lab_inventory._generate_idle_inventory_message( inventory).split('\n') return message def test_check_idle_inventory(self): """Test that reporting all the idle DUTs for every pool, sorted by lab_inventory.MANAGED_POOLS. """ self._add_idles() message = self._get_idle_message(self._histories) self._check_header(message) self._check_report(message[message.index(self._header) + 1 :]) def test_no_idle_inventory(self): """Test that reporting no idle DUTs.""" message = self._get_idle_message(self._histories) self._check_header(message) self._check_report_no_info( message[message.index(self._header) + 1 :]) class CommandParsingTests(unittest.TestCase): """Tests for command line argument parsing in `_parse_command()`.""" # At least one of these options must be specified on every command # line; otherwise, the command line parsing will fail. _REPORT_OPTIONS = [ '--model-notify=', '--pool-notify=', '--report-untestable' ] def setUp(self): dirpath = '/usr/local/fubar' self._command_path = os.path.join(dirpath, 'site_utils', 'arglebargle') self._logdir = os.path.join(dirpath, lab_inventory._LOGDIR) def _parse_arguments(self, argv): """Test parsing with explictly passed report options.""" full_argv = [self._command_path] + argv return lab_inventory._parse_command(full_argv) def _parse_non_report_arguments(self, argv): """Test parsing for non-report command-line options.""" return self._parse_arguments(argv + self._REPORT_OPTIONS) def _check_non_report_defaults(self, report_option): arguments = self._parse_arguments([report_option]) self.assertEqual(arguments.duration, lab_inventory._DEFAULT_DURATION) self.assertIsNone(arguments.recommend) self.assertFalse(arguments.debug) self.assertEqual(arguments.logdir, self._logdir) self.assertEqual(arguments.modelnames, []) return arguments def test_empty_arguments(self): """Test that no reports requested is an error.""" arguments = self._parse_arguments([]) self.assertIsNone(arguments) def test_argument_defaults(self): """Test that option defaults match expectations.""" for report in self._REPORT_OPTIONS: arguments = self._check_non_report_defaults(report) def test_model_notify_defaults(self): """Test defaults when `--model-notify` is specified alone.""" arguments = self._parse_arguments(['--model-notify=']) self.assertEqual(arguments.model_notify, ['']) self.assertEqual(arguments.pool_notify, []) self.assertFalse(arguments.report_untestable) def test_pool_notify_defaults(self): """Test defaults when `--pool-notify` is specified alone.""" arguments = self._parse_arguments(['--pool-notify=']) self.assertEqual(arguments.model_notify, []) self.assertEqual(arguments.pool_notify, ['']) self.assertFalse(arguments.report_untestable) def test_report_untestable_defaults(self): """Test defaults when `--report-untestable` is specified alone.""" arguments = self._parse_arguments(['--report-untestable']) self.assertEqual(arguments.model_notify, []) self.assertEqual(arguments.pool_notify, []) self.assertTrue(arguments.report_untestable) def test_model_arguments(self): """Test that non-option arguments are returned in `modelnames`.""" modellist = ['aardvark', 'echidna'] arguments = self._parse_non_report_arguments(modellist) self.assertEqual(arguments.modelnames, modellist) def test_recommend_option(self): """Test parsing of the `--recommend` option.""" for opt in ['-r', '--recommend']: for recommend in ['5', '55']: arguments = self._parse_non_report_arguments([opt, recommend]) self.assertEqual(arguments.recommend, int(recommend)) def test_debug_option(self): """Test parsing of the `--debug` option.""" arguments = self._parse_non_report_arguments(['--debug']) self.assertTrue(arguments.debug) def test_duration(self): """Test parsing of the `--duration` option.""" for opt in ['-d', '--duration']: for duration in ['1', '11']: arguments = self._parse_non_report_arguments([opt, duration]) self.assertEqual(arguments.duration, int(duration)) def _check_email_option(self, option, getlist): """Test parsing of e-mail address options. This is a helper function to test the `--model-notify` and `--pool-notify` options. It tests the following cases: * `--option a1` gives the list [a1] * `--option ' a1 '` gives the list [a1] * `--option a1 --option a2` gives the list [a1, a2] * `--option a1,a2` gives the list [a1, a2] * `--option 'a1, a2'` gives the list [a1, a2] @param option The option to be tested. @param getlist A function to return the option's value from parsed command line arguments. """ a1 = 'mumble@mumbler.com' a2 = 'bumble@bumbler.org' arguments = self._parse_arguments([option, a1]) self.assertEqual(getlist(arguments), [a1]) arguments = self._parse_arguments([option, ' ' + a1 + ' ']) self.assertEqual(getlist(arguments), [a1]) arguments = self._parse_arguments([option, a1, option, a2]) self.assertEqual(getlist(arguments), [a1, a2]) arguments = self._parse_arguments( [option, ','.join([a1, a2])]) self.assertEqual(getlist(arguments), [a1, a2]) arguments = self._parse_arguments( [option, ', '.join([a1, a2])]) self.assertEqual(getlist(arguments), [a1, a2]) def test_model_notify(self): """Test parsing of the `--model-notify` option.""" self._check_email_option('--model-notify', lambda a: a.model_notify) def test_pool_notify(self): """Test parsing of the `--pool-notify` option.""" self._check_email_option('--pool-notify', lambda a: a.pool_notify) def test_logdir_option(self): """Test parsing of the `--logdir` option.""" logdir = '/usr/local/whatsis/logs' arguments = self._parse_non_report_arguments(['--logdir', logdir]) self.assertEqual(arguments.logdir, logdir) if __name__ == '__main__': # Some of the functions we test log messages. Prevent those # messages from showing up in test output. logging.getLogger().setLevel(logging.CRITICAL) unittest.main()