1 # Copyright 2014 The Chromium OS 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 import mox 6 import unittest 7 8 import common 9 10 import django.core.exceptions 11 from autotest_lib.client.common_lib.cros.network import ping_runner 12 from autotest_lib.frontend import setup_django_environment 13 from autotest_lib.frontend.server import models as server_models 14 from autotest_lib.site_utils import server_manager 15 from autotest_lib.site_utils import server_manager_utils 16 from autotest_lib.site_utils.lib import infra 17 18 19 class QueriableList(list): 20 """A mock list object supports queries including filter and all. 21 """ 22 23 def filter(self, **kwargs): 24 """Mock the filter call in django model. 25 """ 26 raise NotImplementedError() 27 28 29 def get(self, **kwargs): 30 """Mock the get call in django model. 31 """ 32 raise NotImplementedError() 33 34 35 def all(self): 36 """Return all items in the list. 37 38 @return: All items in the list. 39 """ 40 return [item for item in self] 41 42 43 class ServerManagerUnittests(mox.MoxTestBase): 44 """Unittest for testing server_manager module. 45 """ 46 47 def setUp(self): 48 """Initialize the unittest.""" 49 super(ServerManagerUnittests, self).setUp() 50 51 # Initialize test objects. 52 self.DRONE_ROLE = mox.MockObject( 53 server_models.ServerRole, 54 attrs={'role': server_models.ServerRole.ROLE.DRONE}) 55 self.SCHEDULER_ROLE = mox.MockObject( 56 server_models.ServerRole, 57 attrs={'role': server_models.ServerRole.ROLE.SCHEDULER}) 58 self.DRONE_ATTRIBUTE = mox.MockObject( 59 server_models.ServerAttribute, 60 attrs={'attribute': 'max_processes', 'value':1}) 61 self.PRIMARY_DRONE = mox.MockObject( 62 server_models.Server, 63 attrs={'hostname': 'primary_drone_hostname', 64 'status': server_models.Server.STATUS.PRIMARY, 65 'roles': QueriableList([self.DRONE_ROLE]), 66 'attributes': QueriableList([self.DRONE_ATTRIBUTE])}) 67 self.BACKUP_DRONE = mox.MockObject( 68 server_models.Server, 69 attrs={'hostname': 'backup_drone_hostname', 70 'status': server_models.Server.STATUS.BACKUP, 71 'roles': QueriableList([self.DRONE_ROLE]), 72 'attributes': QueriableList([self.DRONE_ATTRIBUTE])}) 73 self.PRIMARY_SCHEDULER = mox.MockObject( 74 server_models.Server, 75 attrs={'hostname': 'primary_scheduler_hostname', 76 'status': server_models.Server.STATUS.PRIMARY, 77 'roles': QueriableList([self.SCHEDULER_ROLE]), 78 'attributes': QueriableList([])}) 79 self.BACKUP_SCHEDULER = mox.MockObject( 80 server_models.Server, 81 attrs={'hostname': 'backup_scheduler_hostname', 82 'status': server_models.Server.STATUS.BACKUP, 83 'roles': QueriableList([self.SCHEDULER_ROLE]), 84 'attributes': QueriableList([])}) 85 86 self.mox.StubOutWithMock(server_manager_utils, 'check_server') 87 self.mox.StubOutWithMock(server_manager_utils, 'warn_missing_role') 88 self.mox.StubOutWithMock(server_manager_utils, 'use_server_db') 89 self.mox.StubOutWithMock(server_models.Server, 'get_role_names') 90 self.mox.StubOutWithMock(server_models.Server.objects, 'create') 91 self.mox.StubOutWithMock(server_models.Server.objects, 'filter') 92 self.mox.StubOutWithMock(server_models.Server.objects, 'get') 93 self.mox.StubOutWithMock(server_models.ServerRole, 'delete') 94 self.mox.StubOutWithMock(server_models.ServerRole.objects, 'create') 95 self.mox.StubOutWithMock(server_models.ServerRole.objects, 'filter') 96 self.mox.StubOutWithMock(server_models.ServerAttribute.objects, 97 'create') 98 self.mox.StubOutWithMock(server_models.ServerAttribute.objects, 99 'filter') 100 self.mox.StubOutWithMock(infra, 'execute_command') 101 self.mox.StubOutWithMock(ping_runner.PingRunner, 'simple_ping') 102 103 104 def testCreateServerSuccess(self): 105 """Test create method can create a server successfully. 106 """ 107 ping_runner.PingRunner().simple_ping(self.BACKUP_DRONE.hostname 108 ).AndReturn(True) 109 server_models.Server.objects.get( 110 hostname=self.BACKUP_DRONE.hostname 111 ).AndRaise(django.core.exceptions.ObjectDoesNotExist) 112 server_models.Server.objects.create( 113 hostname=mox.IgnoreArg(), status=mox.IgnoreArg(), 114 date_created=mox.IgnoreArg(), note=mox.IgnoreArg() 115 ).AndReturn(self.BACKUP_DRONE) 116 server_models.ServerRole.objects.create( 117 server=mox.IgnoreArg(), role=server_models.ServerRole.ROLE.DRONE 118 ).AndReturn(self.DRONE_ROLE) 119 self.mox.ReplayAll() 120 drone = server_manager.create(hostname=self.BACKUP_DRONE.hostname, 121 role=server_models.ServerRole.ROLE.DRONE) 122 123 124 def testAddRoleToBackupSuccess(self): 125 """Test manager can add a role to a backup server successfully. 126 127 Confirm that database call is made, and no action is taken, e.g., 128 restart scheduler to activate a new devserver. 129 """ 130 server_models.validate(role=server_models.ServerRole.ROLE.DEVSERVER) 131 server_manager_utils.check_server(mox.IgnoreArg(), 132 mox.IgnoreArg()).AndReturn(True) 133 server_manager_utils.use_server_db().MultipleTimes( 134 ).AndReturn(True) 135 self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names') 136 self.BACKUP_DRONE.get_role_names().AndReturn( 137 [server_models.ServerRole.ROLE.DRONE]) 138 server_models.ServerRole.objects.create( 139 server=mox.IgnoreArg(), 140 role=server_models.ServerRole.ROLE.DEVSERVER 141 ).AndReturn(self.DRONE_ROLE) 142 self.mox.ReplayAll() 143 server_manager._add_role(server=self.BACKUP_DRONE, 144 role=server_models.ServerRole.ROLE.DEVSERVER, 145 action=True) 146 147 148 def testAddRoleToBackupFail_RoleAlreadyExists(self): 149 """Test manager fails to add a role to a backup server if server already 150 has the given role. 151 """ 152 server_models.validate(role=server_models.ServerRole.ROLE.DRONE) 153 self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names') 154 self.BACKUP_DRONE.get_role_names().AndReturn( 155 [server_models.ServerRole.ROLE.DRONE]) 156 self.mox.ReplayAll() 157 self.assertRaises(server_manager_utils.ServerActionError, 158 server_manager._add_role, 159 server=self.BACKUP_DRONE, 160 role=server_models.ServerRole.ROLE.DRONE, 161 action=True) 162 163 164 def testDeleteRoleFromBackupSuccess(self): 165 """Test manager can delete a role from a backup server successfully. 166 167 Confirm that database call is made, and no action is taken, e.g., 168 restart scheduler to delete an existing devserver. 169 """ 170 server_models.validate(role=server_models.ServerRole.ROLE.DRONE) 171 server_manager_utils.use_server_db().MultipleTimes( 172 ).AndReturn(True) 173 self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names') 174 self.BACKUP_DRONE.get_role_names().MultipleTimes().AndReturn( 175 [server_models.ServerRole.ROLE.DRONE]) 176 self.mox.StubOutWithMock(self.BACKUP_DRONE.roles, 'get') 177 self.BACKUP_DRONE.roles.get( 178 role=server_models.ServerRole.ROLE.DRONE 179 ).AndReturn(self.DRONE_ROLE) 180 self.mox.ReplayAll() 181 server_manager._delete_role(server=self.BACKUP_DRONE, 182 role=server_models.ServerRole.ROLE.DRONE, 183 action=True) 184 185 186 def testDeleteRoleFromBackupFail_RoleNotExist(self): 187 """Test manager fails to delete a role from a backup server if the 188 server does not have the given role. 189 """ 190 server_models.validate(role=server_models.ServerRole.ROLE.DEVSERVER) 191 self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names') 192 self.BACKUP_DRONE.get_role_names().AndReturn( 193 [server_models.ServerRole.ROLE.DRONE]) 194 self.mox.ReplayAll() 195 self.assertRaises(server_manager_utils.ServerActionError, 196 server_manager._delete_role, server=self.BACKUP_DRONE, 197 role=server_models.ServerRole.ROLE.DEVSERVER, 198 action=True) 199 200 201 def testChangeStatusSuccess_BackupToPrimary(self): 202 """Test manager can change the status of a backup server to primary. 203 """ 204 server_models.validate(status=server_models.Server.STATUS.PRIMARY) 205 server_manager_utils.use_server_db().MultipleTimes( 206 ).AndReturn(True) 207 self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names') 208 self.BACKUP_DRONE.get_role_names().MultipleTimes().AndReturn( 209 [server_models.ServerRole.ROLE.DRONE]) 210 self.mox.StubOutWithMock(self.BACKUP_DRONE.roles, 'filter') 211 self.BACKUP_DRONE.roles.filter( 212 role__in=server_models.ServerRole.ROLES_REQUIRE_UNIQUE_INSTANCE 213 ).AndReturn(None) 214 server_models.Server.objects.filter( 215 roles__role=server_models.ServerRole.ROLE.SCHEDULER, 216 status=server_models.Server.STATUS.PRIMARY 217 ).AndReturn([self.PRIMARY_SCHEDULER]) 218 infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg()) 219 self.mox.ReplayAll() 220 server_manager._change_status( 221 server=self.BACKUP_DRONE, 222 status=server_models.Server.STATUS.PRIMARY, 223 action=True) 224 225 226 def testChangeStatusSuccess_PrimaryToBackup(self): 227 """Test manager can change the status of a primary server to backup. 228 """ 229 server_models.validate(status=server_models.Server.STATUS.BACKUP) 230 self.mox.StubOutWithMock(self.PRIMARY_DRONE.roles, 'filter') 231 self.mox.StubOutWithMock(self.PRIMARY_DRONE, 'get_role_names') 232 self.PRIMARY_DRONE.get_role_names().MultipleTimes().AndReturn( 233 [server_models.ServerRole.ROLE.DRONE]) 234 self.PRIMARY_DRONE.roles.filter( 235 role__in=server_models.ServerRole.ROLES_REQUIRE_UNIQUE_INSTANCE 236 ).AndReturn(None) 237 server_manager_utils.use_server_db().MultipleTimes().AndReturn(True) 238 server_manager_utils.warn_missing_role( 239 server_models.ServerRole.ROLE.DRONE, self.PRIMARY_DRONE) 240 server_models.Server.objects.filter( 241 roles__role=server_models.ServerRole.ROLE.SCHEDULER, 242 status=server_models.Server.STATUS.PRIMARY 243 ).AndReturn([self.PRIMARY_SCHEDULER]) 244 infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg()) 245 self.mox.ReplayAll() 246 server_manager._change_status( 247 server=self.PRIMARY_DRONE, 248 status=server_models.Server.STATUS.BACKUP, 249 action=True) 250 251 252 def testChangeStatusFail_StatusNoChange(self): 253 """Test manager cannot change the status of a server with the same 254 status. 255 """ 256 server_models.validate(status=server_models.Server.STATUS.BACKUP) 257 self.mox.ReplayAll() 258 self.assertRaises(server_manager_utils.ServerActionError, 259 server_manager._change_status, 260 server=self.BACKUP_DRONE, 261 status=server_models.Server.STATUS.BACKUP, 262 action=True) 263 264 265 def testChangeStatusFail_UniqueInstance(self): 266 """Test manager cannot change the status of a server from backup to 267 primary if there is already a primary exists for role doesn't allow 268 multiple instances. 269 """ 270 server_models.validate(status=server_models.Server.STATUS.PRIMARY) 271 self.mox.StubOutWithMock(self.BACKUP_SCHEDULER.roles, 'filter') 272 self.BACKUP_SCHEDULER.roles.filter( 273 role__in=server_models.ServerRole.ROLES_REQUIRE_UNIQUE_INSTANCE 274 ).AndReturn(QueriableList([self.SCHEDULER_ROLE])) 275 server_models.Server.objects.filter( 276 roles__role=self.SCHEDULER_ROLE.role, 277 status=server_models.Server.STATUS.PRIMARY 278 ).AndReturn(QueriableList([self.PRIMARY_SCHEDULER])) 279 self.mox.ReplayAll() 280 self.assertRaises(server_manager_utils.ServerActionError, 281 server_manager._change_status, 282 server=self.BACKUP_SCHEDULER, 283 status=server_models.Server.STATUS.PRIMARY, 284 action=True) 285 286 287 def testAddRoleToBackupFail_CheckServerFail(self): 288 """Test manager fails to add a role to a backup server if check_server 289 is failed. 290 """ 291 server_manager_utils.check_server(mox.IgnoreArg(), 292 mox.IgnoreArg()).AndReturn(False) 293 server_models.validate(role=server_models.ServerRole.ROLE.DRONE) 294 self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names') 295 self.BACKUP_DRONE.get_role_names().MultipleTimes().AndReturn( 296 [server_models.ServerRole.ROLE.DRONE]) 297 self.mox.ReplayAll() 298 self.assertRaises(server_manager_utils.ServerActionError, 299 server_manager._add_role, server=self.BACKUP_DRONE, 300 role=server_models.ServerRole.ROLE.SCHEDULER, 301 action=True) 302 303 304 def testAddRoleToPrimarySuccess(self): 305 """Test manager can add a role to a primary server successfully. 306 307 Confirm that actions needs to be taken, e.g., restart scheduler for 308 new drone to be added. 309 """ 310 server_models.validate(role=server_models.ServerRole.ROLE.DRONE) 311 server_manager_utils.check_server(mox.IgnoreArg(), 312 mox.IgnoreArg()).AndReturn(True) 313 server_manager_utils.use_server_db().MultipleTimes().AndReturn(True) 314 self.mox.StubOutWithMock(self.PRIMARY_SCHEDULER, 'get_role_names') 315 self.PRIMARY_SCHEDULER.get_role_names().AndReturn( 316 [server_models.ServerRole.ROLE.SCHEDULER]) 317 server_models.ServerRole.objects.create( 318 server=self.PRIMARY_SCHEDULER, 319 role=server_models.ServerRole.ROLE.DRONE 320 ).AndReturn(self.DRONE_ROLE) 321 server_models.Server.objects.filter( 322 roles__role=server_models.ServerRole.ROLE.SCHEDULER, 323 status=server_models.Server.STATUS.PRIMARY 324 ).AndReturn([self.PRIMARY_SCHEDULER]) 325 infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg()) 326 self.mox.ReplayAll() 327 server_manager._add_role(self.PRIMARY_SCHEDULER, 328 server_models.ServerRole.ROLE.DRONE, 329 action=True) 330 331 332 def testDeleteRoleFromPrimarySuccess(self): 333 """Test manager can delete a role from a primary server successfully. 334 335 Confirm that database call is made, and actions are taken, e.g., 336 restart scheduler to delete an existing drone. 337 """ 338 server_manager_utils.use_server_db().MultipleTimes().AndReturn(True) 339 server_models.validate(role=server_models.ServerRole.ROLE.DRONE) 340 self.mox.StubOutWithMock(self.PRIMARY_DRONE, 'get_role_names') 341 self.PRIMARY_DRONE.get_role_names().MultipleTimes().AndReturn( 342 [server_models.ServerRole.ROLE.DRONE]) 343 344 self.mox.StubOutWithMock(self.PRIMARY_DRONE.roles, 'get') 345 self.PRIMARY_DRONE.roles.get( 346 role=server_models.ServerRole.ROLE.DRONE 347 ).AndReturn(self.DRONE_ROLE) 348 349 server_models.Server.objects.filter( 350 roles__role=server_models.ServerRole.ROLE.SCHEDULER, 351 status=server_models.Server.STATUS.PRIMARY 352 ).AndReturn([self.PRIMARY_SCHEDULER]) 353 server_manager.server_manager_utils.warn_missing_role( 354 server_models.ServerRole.ROLE.DRONE, self.PRIMARY_DRONE) 355 infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg()) 356 self.mox.ReplayAll() 357 server_manager._delete_role(self.PRIMARY_DRONE, 358 server_models.ServerRole.ROLE.DRONE, 359 action=True) 360 361 362 def testDeleteRoleFromPrimarySuccess_NoAction(self): 363 """Test manager can delete a role from a primary server successfully. 364 365 Confirm that database call is made, and no action is taken as action 366 is set to False. 367 """ 368 server_manager_utils.use_server_db().MultipleTimes().AndReturn(True) 369 server_models.validate(role=server_models.ServerRole.ROLE.DRONE) 370 self.mox.StubOutWithMock(self.PRIMARY_DRONE, 'get_role_names') 371 self.PRIMARY_DRONE.get_role_names().MultipleTimes().AndReturn( 372 [server_models.ServerRole.ROLE.DRONE]) 373 374 self.mox.StubOutWithMock(self.PRIMARY_DRONE.roles, 'get') 375 self.PRIMARY_DRONE.roles.get( 376 role=server_models.ServerRole.ROLE.DRONE 377 ).AndReturn(self.DRONE_ROLE) 378 379 server_manager.server_manager_utils.warn_missing_role( 380 server_models.ServerRole.ROLE.DRONE, self.PRIMARY_DRONE) 381 self.mox.ReplayAll() 382 server_manager._delete_role(self.PRIMARY_DRONE, 383 server_models.ServerRole.ROLE.DRONE, 384 action=False) 385 386 387 if '__main__': 388 unittest.main() 389
