

def _test_pool_maintenance(self, desired_slots_free=10, slot_capacity=3, host_category=VM_CATEGORY, expected=15, define_credentials=True):
    with self.hosting_device_template(host_category=host_category, slot_capacity=slot_capacity, desired_slots_free=desired_slots_free, plugging_driver=TEST_PLUGGING_DRIVER) as hdt:
        hdt_id = hdt['hosting_device_template']['id']
        creds_id = (DEFAULT_CREDENTIALS_ID if (define_credentials is True) else 'non_existent_id')
        credentials = {
            'user_name': 'bob',
            'password': 'tooEasy',
        }
        with mock.patch.dict(self.plugin._credentials, {
            creds_id: credentials,
        }), self.port(subnet=self._mgmt_subnet, no_delete=True) as mgmt_port1, self.port(subnet=self._mgmt_subnet, no_delete=True) as mgmt_port2:
            with self.hosting_device(template_id=hdt_id, management_port_id=mgmt_port1['port']['id'], auto_delete=True, no_delete=True), self.hosting_device(template_id=hdt_id, management_port_id=mgmt_port2['port']['id'], auto_delete=True, no_delete=True):
                context = self._get_test_context(is_admin=True)
                template = self._devmgr._get_hosting_device_template(context, hdt_id)
                self._devmgr._gt_pool = mock.MagicMock()
                self._devmgr._gt_pool.spawn_n.side_effect = (lambda fcn, ctx, tmplt: fcn(ctx, tmplt))
                self._devmgr._dispatch_pool_maintenance_job(template)
                result_hds = self._list('hosting_devices')['hosting_devices']
                self.assertEqual(expected, (len(result_hds) * slot_capacity))
        self._devmgr.delete_all_hosting_devices(context, True)

dataset/ETHPy150Open openstack/networking-cisco/networking_cisco/tests/unit/cisco/device_manager/test_db_device_manager.py TestDeviceManagerDBPlugin._test_pool_maintenance/original