1# Lint as: python2, python3 2# Copyright (c) 2019 The Chromium OS Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6import logging 7 8from autotest_lib.client.common_lib import error 9from autotest_lib.server import test 10from autotest_lib.server import frontend 11 12class hardware_StorageQualCheckSetup(test.test): 13 """ 14 Verifies the moblab and DUT setup for storage qual 15 A correct setup consists of 16 At least one pool (the default "no pool" counts as a pool) 17 Each of the labels [retention, trim, suspend] is applied to exactly 18 one DUT in the pool. 19 No duplication of the labels, all labels are applied exactly once 20 per pool 21 22 The test will verify this set up, for the pool that is selected on the 23 RunSuite page (by getting the pool of the DUT running the test). If this 24 test passes, we have confidence that this individual pool is a valid setup 25 for storage qual. 26 """ 27 28 version = 1 29 30 REQUIRED_LABELS = ['retention', 'trim', 'suspend'] 31 32 def _group_hosts_into_pools(self, hosts): 33 pools = {} 34 for host in hosts: 35 labels = [label.name for label in host.get_labels()] 36 37 pool_name = 'none' 38 for label in labels: 39 if 'pool:' in label: 40 pool_name = label.replace('pool:', '') 41 42 if pool_name not in pools: 43 pools[pool_name] = [] 44 45 pools[pool_name].append({ 46 'host': host.hostname, 47 'labels': labels 48 }) 49 50 return pools 51 52 def _get_running_host_pool(self): 53 host = list(self.job.hosts)[0] 54 pools = host.host_info_store.get().pools 55 return list(pools)[0] if len(pools) > 0 else 'none' 56 57 def run_once(self): 58 """ Tests the moblab's connected DUTs to see if the current 59 configuration is valid for storage qual 60 """ 61 62 # get the pool of the host this test is running on 63 pool_name = self._get_running_host_pool() 64 logging.info('Test is running on pool %s', pool_name) 65 66 afe = frontend.AFE(server='localhost', user='moblab') 67 68 # get autotest statuses that indicate a live host 69 live_statuses = afe.host_statuses(live=True) 70 71 # get the hosts connected to autotest, find the live ones 72 hosts = [] 73 for host in afe.get_hosts(): 74 if host.status in live_statuses: 75 logging.info('Host %s is live, status %s', 76 host.hostname, host.status) 77 hosts.append(host) 78 else: 79 logging.info('Host %s is not live, status %s', 80 host.hostname, host.status) 81 82 pools = self._group_hosts_into_pools(hosts) 83 84 # verify that the pool is set up to run storage qual, with correct 85 # number of DUTs and correct labels 86 required_set = set(self.REQUIRED_LABELS) 87 provided_set = set() 88 for host in pools[pool_name]: 89 host_provided_labels = set(host['labels']) & required_set 90 # check that each DUT has at most 1 storage qual label 91 if len(host_provided_labels) > 1: 92 raise error.TestFail( 93 ('Host %s is assigned more than ' 94 'one storage qual label %s') % 95 (host['host'], str(host_provided_labels))) 96 if len(host_provided_labels) == 0: 97 continue 98 99 # check that each label is only on one DUT in the pool 100 provided_label = host_provided_labels.pop() 101 if provided_label in provided_set: 102 raise error.TestFail( 103 ('Host %s is assigned label %s, which is already ' 104 'assigned to another DUT in pool %s') % 105 (host['host'], provided_label, pool_name) 106 ) 107 108 provided_set.add(provided_label) 109 logging.info(' - %s %s', host['host'], provided_label) 110 111 # check that all storage qual labels are accounted for in the pool 112 missing_labels = required_set - provided_set 113 if len(missing_labels) > 0: 114 raise error.TestFail( 115 'Pool %s is missing required labels %s' % 116 (pool_name, str(missing_labels)) 117 ) 118 119 return 120