blob: a060ef2059263ae861769e25702d92f51154396e [file] [log] [blame] [edit]
# Copyright 2014 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 mox
import unittest
import common
import django.core.exceptions
from autotest_lib.client.common_lib.cros.network import ping_runner
from autotest_lib.frontend import setup_django_environment
from autotest_lib.frontend.server import models as server_models
from autotest_lib.site_utils import server_manager
from autotest_lib.site_utils import server_manager_utils
from autotest_lib.site_utils.lib import infra
class QueriableList(list):
"""A mock list object supports queries including filter and all.
"""
def filter(self, **kwargs):
"""Mock the filter call in django model.
"""
raise NotImplementedError()
def get(self, **kwargs):
"""Mock the get call in django model.
"""
raise NotImplementedError()
def all(self):
"""Return all items in the list.
@return: All items in the list.
"""
return [item for item in self]
class ServerManagerUnittests(mox.MoxTestBase):
"""Unittest for testing server_manager module.
"""
def setUp(self):
"""Initialize the unittest."""
super(ServerManagerUnittests, self).setUp()
# Initialize test objects.
self.DRONE_ROLE = mox.MockObject(
server_models.ServerRole,
attrs={'role': server_models.ServerRole.ROLE.DRONE})
self.SCHEDULER_ROLE = mox.MockObject(
server_models.ServerRole,
attrs={'role': server_models.ServerRole.ROLE.SCHEDULER})
self.DRONE_ATTRIBUTE = mox.MockObject(
server_models.ServerAttribute,
attrs={'attribute': 'max_processes', 'value':1})
self.PRIMARY_DRONE = mox.MockObject(
server_models.Server,
attrs={'hostname': 'primary_drone_hostname',
'status': server_models.Server.STATUS.PRIMARY,
'roles': QueriableList([self.DRONE_ROLE]),
'attributes': QueriableList([self.DRONE_ATTRIBUTE])})
self.REPAIR_REQUIRED_DRONE = mox.MockObject(
server_models.Server,
attrs={'hostname': 'repair_required_drone_hostname',
'status': server_models.Server.STATUS.REPAIR_REQUIRED,
'roles': QueriableList([self.DRONE_ROLE]),
'attributes': QueriableList([self.DRONE_ATTRIBUTE])})
self.PRIMARY_SCHEDULER = mox.MockObject(
server_models.Server,
attrs={'hostname': 'primary_scheduler_hostname',
'status': server_models.Server.STATUS.PRIMARY,
'roles': QueriableList([self.SCHEDULER_ROLE]),
'attributes': QueriableList([])})
self.REPAIR_REQUIRED_SCHEDULER = mox.MockObject(
server_models.Server,
attrs={'hostname': 'repair_required_scheduler_hostname',
'status': server_models.Server.STATUS.REPAIR_REQUIRED,
'roles': QueriableList([self.SCHEDULER_ROLE]),
'attributes': QueriableList([])})
self.mox.StubOutWithMock(server_manager_utils, 'check_server')
self.mox.StubOutWithMock(server_manager_utils, 'warn_missing_role')
self.mox.StubOutWithMock(server_manager_utils, 'use_server_db')
self.mox.StubOutWithMock(server_models.Server, 'get_role_names')
self.mox.StubOutWithMock(server_models.Server.objects, 'create')
self.mox.StubOutWithMock(server_models.Server.objects, 'filter')
self.mox.StubOutWithMock(server_models.Server.objects, 'get')
self.mox.StubOutWithMock(server_models.ServerRole, 'delete')
self.mox.StubOutWithMock(server_models.ServerRole.objects, 'create')
self.mox.StubOutWithMock(server_models.ServerRole.objects, 'filter')
self.mox.StubOutWithMock(server_models.ServerAttribute.objects,
'create')
self.mox.StubOutWithMock(server_models.ServerAttribute.objects,
'filter')
self.mox.StubOutWithMock(infra, 'execute_command')
self.mox.StubOutWithMock(ping_runner.PingRunner, 'simple_ping')
def testCreateServerSuccess(self):
"""Test create method can create a server successfully.
"""
ping_runner.PingRunner().simple_ping(self.PRIMARY_DRONE.hostname
).AndReturn(True)
server_models.Server.objects.get(
hostname=self.PRIMARY_DRONE.hostname
).AndRaise(django.core.exceptions.ObjectDoesNotExist)
server_models.Server.objects.create(
hostname=mox.IgnoreArg(), status=mox.IgnoreArg(),
date_created=mox.IgnoreArg(), note=mox.IgnoreArg()
).AndReturn(self.PRIMARY_DRONE)
server_models.ServerRole.objects.create(
server=mox.IgnoreArg(), role=server_models.ServerRole.ROLE.DRONE
).AndReturn(self.DRONE_ROLE)
self.mox.ReplayAll()
drone = server_manager.create(hostname=self.PRIMARY_DRONE.hostname,
role=server_models.ServerRole.ROLE.DRONE)
def testAddRoleToRepairRequiredSuccess(self):
"""Test manager can add a role to a repair_failed server successfully.
Confirm that database call is made, and no action is taken, e.g.,
restart scheduler to activate a new devserver.
"""
server_models.validate(role=server_models.ServerRole.ROLE.DEVSERVER)
server_manager_utils.check_server(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(True)
server_manager_utils.use_server_db().MultipleTimes(
).AndReturn(True)
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE, 'get_role_names')
self.REPAIR_REQUIRED_DRONE.get_role_names().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
server_models.ServerRole.objects.create(
server=mox.IgnoreArg(),
role=server_models.ServerRole.ROLE.DEVSERVER
).AndReturn(self.DRONE_ROLE)
self.mox.ReplayAll()
server_manager._add_role(server=self.REPAIR_REQUIRED_DRONE,
role=server_models.ServerRole.ROLE.DEVSERVER,
action=True)
def testAddRoleToRepairRequiredFail_RoleAlreadyExists(self):
"""Test manager fails to add a role to a repair_required server if
server already has the given role.
"""
server_models.validate(role=server_models.ServerRole.ROLE.DRONE)
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE, 'get_role_names')
self.REPAIR_REQUIRED_DRONE.get_role_names().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.mox.ReplayAll()
self.assertRaises(server_manager_utils.ServerActionError,
server_manager._add_role,
server=self.REPAIR_REQUIRED_DRONE,
role=server_models.ServerRole.ROLE.DRONE,
action=True)
def testDeleteRoleFromRepairRequiredSuccess(self):
"""Test manager can delete a role from a repair_required server
successfully.
Confirm that database call is made, and no action is taken, e.g.,
restart scheduler to delete an existing devserver.
"""
server_models.validate(role=server_models.ServerRole.ROLE.DRONE)
server_manager_utils.use_server_db().MultipleTimes(
).AndReturn(True)
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE, 'get_role_names')
self.REPAIR_REQUIRED_DRONE.get_role_names().MultipleTimes().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE.roles, 'get')
self.REPAIR_REQUIRED_DRONE.roles.get(
role=server_models.ServerRole.ROLE.DRONE
).AndReturn(self.DRONE_ROLE)
self.mox.ReplayAll()
server_manager._delete_role(server=self.REPAIR_REQUIRED_DRONE,
role=server_models.ServerRole.ROLE.DRONE,
action=True)
def testDeleteRoleFromRepairRequiredFail_RoleNotExist(self):
"""Test manager fails to delete a role from a repair_required server if
the server does not have the given role.
"""
server_models.validate(role=server_models.ServerRole.ROLE.DEVSERVER)
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE, 'get_role_names')
self.REPAIR_REQUIRED_DRONE.get_role_names().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.mox.ReplayAll()
self.assertRaises(server_manager_utils.ServerActionError,
server_manager._delete_role,
server=self.REPAIR_REQUIRED_DRONE,
role=server_models.ServerRole.ROLE.DEVSERVER,
action=True)
def testChangeStatusSuccess_RepairFailedToPrimary(self):
"""Test manager can change the status of a repair_required server to
primary.
"""
server_models.validate(status=server_models.Server.STATUS.PRIMARY)
server_manager_utils.use_server_db().MultipleTimes(
).AndReturn(True)
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE, 'get_role_names')
self.REPAIR_REQUIRED_DRONE.get_role_names().MultipleTimes().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE.roles, 'filter')
self.REPAIR_REQUIRED_DRONE.roles.filter(
role__in=server_models.ServerRole.ROLES_REQUIRE_UNIQUE_INSTANCE
).AndReturn(None)
server_models.Server.objects.filter(
roles__role=server_models.ServerRole.ROLE.SCHEDULER,
status=server_models.Server.STATUS.PRIMARY
).AndReturn([self.PRIMARY_SCHEDULER])
infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg())
self.mox.ReplayAll()
server_manager._change_status(
server=self.REPAIR_REQUIRED_DRONE,
status=server_models.Server.STATUS.PRIMARY,
action=True)
def testChangeStatusSuccess_PrimaryToRepairFailed(self):
"""Test manager can change the status of a primary server to
repair_required.
"""
server_models.validate(
status=server_models.Server.STATUS.REPAIR_REQUIRED)
self.mox.StubOutWithMock(self.PRIMARY_DRONE.roles, 'filter')
self.mox.StubOutWithMock(self.PRIMARY_DRONE, 'get_role_names')
self.PRIMARY_DRONE.get_role_names().MultipleTimes().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.PRIMARY_DRONE.roles.filter(
role__in=server_models.ServerRole.ROLES_REQUIRE_UNIQUE_INSTANCE
).AndReturn(None)
server_manager_utils.use_server_db().MultipleTimes().AndReturn(True)
server_manager_utils.warn_missing_role(
server_models.ServerRole.ROLE.DRONE, self.PRIMARY_DRONE)
server_models.Server.objects.filter(
roles__role=server_models.ServerRole.ROLE.SCHEDULER,
status=server_models.Server.STATUS.PRIMARY
).AndReturn([self.PRIMARY_SCHEDULER])
infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg())
self.mox.ReplayAll()
server_manager._change_status(
server=self.PRIMARY_DRONE,
status=server_models.Server.STATUS.REPAIR_REQUIRED,
action=True)
def testChangeStatusFail_StatusNoChange(self):
"""Test manager cannot change the status of a server with the same
status.
"""
server_models.validate(
status=server_models.Server.STATUS.REPAIR_REQUIRED)
self.mox.ReplayAll()
self.assertRaises(server_manager_utils.ServerActionError,
server_manager._change_status,
server=self.REPAIR_REQUIRED_DRONE,
status=server_models.Server.STATUS.REPAIR_REQUIRED,
action=True)
def testChangeStatusFail_UniqueInstance(self):
"""Test manager cannot change the status of a server from
repair_required to primary if there is already a primary exists for
role doesn't allow multiple instances.
"""
server_models.validate(status=server_models.Server.STATUS.PRIMARY)
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_SCHEDULER.roles, 'filter')
self.REPAIR_REQUIRED_SCHEDULER.roles.filter(
role__in=server_models.ServerRole.ROLES_REQUIRE_UNIQUE_INSTANCE
).AndReturn(QueriableList([self.SCHEDULER_ROLE]))
server_models.Server.objects.filter(
roles__role=self.SCHEDULER_ROLE.role,
status=server_models.Server.STATUS.PRIMARY
).AndReturn(QueriableList([self.PRIMARY_SCHEDULER]))
self.mox.ReplayAll()
self.assertRaises(server_manager_utils.ServerActionError,
server_manager._change_status,
server=self.REPAIR_REQUIRED_SCHEDULER,
status=server_models.Server.STATUS.PRIMARY,
action=True)
def testAddRoleToRepairFailedFail_CheckServerFail(self):
"""Test manager fails to add a role to a repair_required server if check
server is failed.
"""
server_manager_utils.check_server(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(False)
server_models.validate(role=server_models.ServerRole.ROLE.DRONE)
self.mox.StubOutWithMock(self.REPAIR_REQUIRED_DRONE, 'get_role_names')
self.REPAIR_REQUIRED_DRONE.get_role_names().MultipleTimes().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.mox.ReplayAll()
self.assertRaises(server_manager_utils.ServerActionError,
server_manager._add_role,
server=self.REPAIR_REQUIRED_DRONE,
role=server_models.ServerRole.ROLE.SCHEDULER,
action=True)
def testAddRoleToPrimarySuccess(self):
"""Test manager can add a role to a primary server successfully.
Confirm that actions needs to be taken, e.g., restart scheduler for
new drone to be added.
"""
server_models.validate(role=server_models.ServerRole.ROLE.DRONE)
server_manager_utils.check_server(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(True)
server_manager_utils.use_server_db().MultipleTimes().AndReturn(True)
self.mox.StubOutWithMock(self.PRIMARY_SCHEDULER, 'get_role_names')
self.PRIMARY_SCHEDULER.get_role_names().AndReturn(
[server_models.ServerRole.ROLE.SCHEDULER])
server_models.ServerRole.objects.create(
server=self.PRIMARY_SCHEDULER,
role=server_models.ServerRole.ROLE.DRONE
).AndReturn(self.DRONE_ROLE)
server_models.Server.objects.filter(
roles__role=server_models.ServerRole.ROLE.SCHEDULER,
status=server_models.Server.STATUS.PRIMARY
).AndReturn([self.PRIMARY_SCHEDULER])
infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg())
self.mox.ReplayAll()
server_manager._add_role(self.PRIMARY_SCHEDULER,
server_models.ServerRole.ROLE.DRONE,
action=True)
def testDeleteRoleFromPrimarySuccess(self):
"""Test manager can delete a role from a primary server successfully.
Confirm that database call is made, and actions are taken, e.g.,
restart scheduler to delete an existing drone.
"""
server_manager_utils.use_server_db().MultipleTimes().AndReturn(True)
server_models.validate(role=server_models.ServerRole.ROLE.DRONE)
self.mox.StubOutWithMock(self.PRIMARY_DRONE, 'get_role_names')
self.PRIMARY_DRONE.get_role_names().MultipleTimes().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.mox.StubOutWithMock(self.PRIMARY_DRONE.roles, 'get')
self.PRIMARY_DRONE.roles.get(
role=server_models.ServerRole.ROLE.DRONE
).AndReturn(self.DRONE_ROLE)
server_models.Server.objects.filter(
roles__role=server_models.ServerRole.ROLE.SCHEDULER,
status=server_models.Server.STATUS.PRIMARY
).AndReturn([self.PRIMARY_SCHEDULER])
server_manager.server_manager_utils.warn_missing_role(
server_models.ServerRole.ROLE.DRONE, self.PRIMARY_DRONE)
infra.execute_command(mox.IgnoreArg(), mox.IgnoreArg())
self.mox.ReplayAll()
server_manager._delete_role(self.PRIMARY_DRONE,
server_models.ServerRole.ROLE.DRONE,
action=True)
def testDeleteRoleFromPrimarySuccess_NoAction(self):
"""Test manager can delete a role from a primary server successfully.
Confirm that database call is made, and no action is taken as action
is set to False.
"""
server_manager_utils.use_server_db().MultipleTimes().AndReturn(True)
server_models.validate(role=server_models.ServerRole.ROLE.DRONE)
self.mox.StubOutWithMock(self.PRIMARY_DRONE, 'get_role_names')
self.PRIMARY_DRONE.get_role_names().MultipleTimes().AndReturn(
[server_models.ServerRole.ROLE.DRONE])
self.mox.StubOutWithMock(self.PRIMARY_DRONE.roles, 'get')
self.PRIMARY_DRONE.roles.get(
role=server_models.ServerRole.ROLE.DRONE
).AndReturn(self.DRONE_ROLE)
server_manager.server_manager_utils.warn_missing_role(
server_models.ServerRole.ROLE.DRONE, self.PRIMARY_DRONE)
self.mox.ReplayAll()
server_manager._delete_role(self.PRIMARY_DRONE,
server_models.ServerRole.ROLE.DRONE,
action=False)
if __name__ == "__main__":
unittest.main()