xref: /external/autotest/site_utils/server_manager_unittest.py

# 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.BACKUP_DRONE = mox.MockObject(
                server_models.Server,
                attrs={'hostname': 'backup_drone_hostname',
                       'status': server_models.Server.STATUS.BACKUP,
                       '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.BACKUP_SCHEDULER = mox.MockObject(
                server_models.Server,
                attrs={'hostname': 'backup_scheduler_hostname',
                       'status': server_models.Server.STATUS.BACKUP,
                       '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.BACKUP_DRONE.hostname
                                             ).AndReturn(True)
        server_models.Server.objects.get(
                hostname=self.BACKUP_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.BACKUP_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.BACKUP_DRONE.hostname,
                                      role=server_models.ServerRole.ROLE.DRONE)


    def testAddRoleToBackupSuccess(self):
        """Test manager can add a role to a backup 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.BACKUP_DRONE, 'get_role_names')
        self.BACKUP_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.BACKUP_DRONE,
                                 role=server_models.ServerRole.ROLE.DEVSERVER,
                                 action=True)


    def testAddRoleToBackupFail_RoleAlreadyExists(self):
        """Test manager fails to add a role to a backup server if server already
        has the given role.
        """
        server_models.validate(role=server_models.ServerRole.ROLE.DRONE)
        self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names')
        self.BACKUP_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.BACKUP_DRONE,
                          role=server_models.ServerRole.ROLE.DRONE,
                          action=True)


    def testDeleteRoleFromBackupSuccess(self):
        """Test manager can delete a role from a backup 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.BACKUP_DRONE, 'get_role_names')
        self.BACKUP_DRONE.get_role_names().MultipleTimes().AndReturn(
                [server_models.ServerRole.ROLE.DRONE])
        self.mox.StubOutWithMock(self.BACKUP_DRONE.roles, 'get')
        self.BACKUP_DRONE.roles.get(
                role=server_models.ServerRole.ROLE.DRONE
                ).AndReturn(self.DRONE_ROLE)
        self.mox.ReplayAll()
        server_manager._delete_role(server=self.BACKUP_DRONE,
                                    role=server_models.ServerRole.ROLE.DRONE,
                                    action=True)


    def testDeleteRoleFromBackupFail_RoleNotExist(self):
        """Test manager fails to delete a role from a backup server if the
        server does not have the given role.
        """
        server_models.validate(role=server_models.ServerRole.ROLE.DEVSERVER)
        self.mox.StubOutWithMock(self.BACKUP_DRONE, 'get_role_names')
        self.BACKUP_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.BACKUP_DRONE,
                          role=server_models.ServerRole.ROLE.DEVSERVER,
                          action=True)


    def testChangeStatusSuccess_BackupToPrimary(self):
        """Test manager can change the status of a backup 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.BACKUP_DRONE, 'get_role_names')
        self.BACKUP_DRONE.get_role_names().MultipleTimes().AndReturn(
                [server_models.ServerRole.ROLE.DRONE])
        self.mox.StubOutWithMock(self.BACKUP_DRONE.roles, 'filter')
        self.BACKUP_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.BACKUP_DRONE,
                status=server_models.Server.STATUS.PRIMARY,
                action=True)


    def testChangeStatusSuccess_PrimaryToBackup(self):
        """Test manager can change the status of a primary server to backup.
        """
        server_models.validate(status=server_models.Server.STATUS.BACKUP)
        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.BACKUP,
                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.BACKUP)
        self.mox.ReplayAll()
        self.assertRaises(server_manager_utils.ServerActionError,
                          server_manager._change_status,
                          server=self.BACKUP_DRONE,
                          status=server_models.Server.STATUS.BACKUP,
                          action=True)


    def testChangeStatusFail_UniqueInstance(self):
        """Test manager cannot change the status of a server from backup 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.BACKUP_SCHEDULER.roles, 'filter')
        self.BACKUP_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.BACKUP_SCHEDULER,
                          status=server_models.Server.STATUS.PRIMARY,
                          action=True)


    def testAddRoleToBackupFail_CheckServerFail(self):
        """Test manager fails to add a role to a backup 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.BACKUP_DRONE, 'get_role_names')
        self.BACKUP_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.BACKUP_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 '__main__':
    unittest.main()