1 # Copyright 2015 The Chromium Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 """Operations for updating test owners. 6 7 We want to allow editing test ownership through incoming chart metadata as well 8 as through dashboard UI, so we keep track of changes in test ownership that come 9 in from incoming chart metadata and apply them to a master copy of the test 10 owners data. The dashboard UI, on the other hand, can modify the master copy 11 directly. 12 13 Test owners data are stored in layered_cache as dictionary of test suite path to 14 set of its owners' email. 15 { 16 'a/a': {'a (at] foo.com'}, 17 'b/b': {'b (at] foo.com'}, 18 } 19 """ 20 21 from dashboard import layered_cache 22 23 # Cache keys for layered cache test owner dictionary. 24 _CHARTJSON_OWNER_CACHE_KEY = 'ChartjsonOwner' 25 _MASTER_OWNER_CACHE_KEY = 'MasterOwner' 26 27 _MAX_TEST_SUITE_PATH_LENGTH = 500 28 _MAX_OWNER_EMAIL_LENGTH = 254 29 30 31 def UpdateOwnerFromChartjson(owner_dict): 32 """Updates test owners with test owner data from chartjson. 33 34 Checks if tests owners have changed by matching |owner_dict| with the stored 35 owner dict for chartjson and update the master owner dict accordingly. 36 37 Args: 38 owner_dict: A dictionary of Master/Test suite to set of owners. 39 """ 40 add_owner_dict = {} 41 remove_owner_dict = {} 42 owner_dict_cache = layered_cache.GetExternal(_CHARTJSON_OWNER_CACHE_KEY) or {} 43 44 for path, owners in owner_dict.iteritems(): 45 owners = owners or set() 46 owners_cache = owner_dict_cache.get(path, set()) 47 if owners_cache: 48 diff = owners_cache - owners 49 if diff: 50 remove_owner_dict[path] = diff 51 diff = owners - owners_cache 52 if diff: 53 add_owner_dict[path] = diff 54 else: 55 add_owner_dict[path] = owners 56 57 if owners: 58 owner_dict_cache[path] = owners 59 elif path in owner_dict_cache: 60 del owner_dict_cache[path] 61 62 if add_owner_dict or remove_owner_dict: 63 layered_cache.SetExternal(_CHARTJSON_OWNER_CACHE_KEY, owner_dict_cache) 64 if add_owner_dict: 65 AddOwnerFromDict(add_owner_dict) 66 if remove_owner_dict: 67 RemoveOwnerFromDict(remove_owner_dict) 68 69 70 def AddOwner(test_suite_path, owner_email): 71 """Adds an owner for a test suite path. 72 73 Args: 74 test_suite_path: A string of "Master/Test suite". 75 owner_email: An email string. 76 """ 77 owner_dict_cache = GetMasterCachedOwner() 78 owners = owner_dict_cache.get(test_suite_path, set()) 79 owners.add(owner_email) 80 owner_dict_cache[test_suite_path] = owners 81 layered_cache.SetExternal(_MASTER_OWNER_CACHE_KEY, owner_dict_cache) 82 owner_dict_cache = GetMasterCachedOwner() 83 84 85 def AddOwnerFromDict(owner_dict): 86 """Adds test owner from |owner_dict| to owner dict in layered_cache. 87 88 For example, if owner cache dict is: 89 { 90 'a/a': {'a (at] foo.com'}, 91 'a/b': {'b (at] foo.com'}, 92 } 93 and parameter owner_dict is: 94 { 95 'a/a': {'c (at] foo.com'}, 96 'c/c': {'c (at] foo.com'}, 97 } 98 then the cached will be updated to: 99 { 100 'a/a': {'a (at] foo.com', 'c (at] foo.com'}, 101 'a/b': {'b (at] foo.com'}, 102 'c/c': {'c (at] foo.com'}, 103 } 104 105 Args: 106 owner_dict: A dictionary of "Master/Test suite" to set of owners' email. 107 """ 108 owner_dict_cache = GetMasterCachedOwner() 109 for path, owners in owner_dict.iteritems(): 110 owners_cache = owner_dict_cache.get(path, set()) 111 owners_cache.update(owners) 112 owner_dict_cache[path] = owners_cache 113 layered_cache.SetExternal(_MASTER_OWNER_CACHE_KEY, owner_dict_cache) 114 115 116 def RemoveOwner(test_suite_path, owner_email=None): 117 """Removes test owners for |test_suite_path|. 118 119 Args: 120 test_suite_path: A string of "Master/Test suite". 121 owner_email: Optional email string. If not specified, dict entry 122 for |test_suite_path| will be deleted. 123 """ 124 owner_dict_cache = GetMasterCachedOwner() 125 if test_suite_path in owner_dict_cache: 126 if owner_email: 127 owners = owner_dict_cache[test_suite_path] 128 owners.remove(owner_email) 129 if not owners: 130 del owner_dict_cache[test_suite_path] 131 else: 132 del owner_dict_cache[test_suite_path] 133 layered_cache.SetExternal(_MASTER_OWNER_CACHE_KEY, owner_dict_cache) 134 135 136 def RemoveOwnerFromDict(owner_dict): 137 """Adds test owner from |owner_dict| to owner dict in layered_cache. 138 139 Args: 140 owner_dict: A dictionary of Master/Test suite to set of owners to be 141 removed. 142 """ 143 owner_dict_cache = GetMasterCachedOwner() 144 for path, owners in owner_dict.iteritems(): 145 owners_cache = owner_dict_cache.get(path, set()) 146 owner_dict_cache[path] = owners_cache - owners 147 if not owner_dict_cache[path]: 148 del owner_dict_cache[path] 149 layered_cache.SetExternal(_MASTER_OWNER_CACHE_KEY, owner_dict_cache) 150 151 152 def GetOwners(test_suite_paths): 153 """Gets a list of owners for a list of test suite paths.""" 154 owners = set() 155 owner_dict_cache = GetMasterCachedOwner() 156 for path in test_suite_paths: 157 if path in owner_dict_cache: 158 owners.update(owner_dict_cache[path]) 159 return sorted(owners) 160 161 162 def GetTestSuitePaths(owner_email): 163 """Gets a list of test suite paths for an owner.""" 164 test_suite_paths = [] 165 owner_dict_cache = GetMasterCachedOwner() 166 for path, owners in owner_dict_cache.iteritems(): 167 if owner_email in owners: 168 test_suite_paths.append(path) 169 return sorted(test_suite_paths) 170 171 172 def GetMasterCachedOwner(): 173 """Gets test owner cached dictionary from layered_cache.""" 174 return layered_cache.GetExternal(_MASTER_OWNER_CACHE_KEY) or {} 175 176 177 def ValidateTestSuitePath(test_suite_path): 178 if test_suite_path and len(test_suite_path) > _MAX_TEST_SUITE_PATH_LENGTH: 179 raise ValueError('Test suite path is too long: %s.' % test_suite_path) 180 181 182 def ValidateOwnerEmail(owner_email): 183 if owner_email and len(owner_email) > _MAX_OWNER_EMAIL_LENGTH: 184 raise ValueError('Owner\'s email is too long: %s.' % owner_email) 185