#!/usr/bin/env python
|
# 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:
|
<message> -> <intro> <pool> { "blank line" <pool> }
|
<intro> ->
|
Instructions to depty mentioning the admin page URL
|
A blank line
|
<pool> ->
|
<description>
|
<header line>
|
<message body>
|
<description> ->
|
Any number of lines describing one pool
|
<header line> ->
|
The header line from `_POOL_MESSAGE_TEMPLATE`
|
<message body> ->
|
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 <description> contains a mention of exactly one
|
pool in the `CRITICAL_POOLS` list.
|
* Each pool is mentioned in exactly one <description>.
|
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()
|
