python类call()的实例源码

def test_ovs_set_if_external_id(self):
        interface = 'fake_interface'
        extneral_id = 'fake_extneral_id'
        value = 'fake_value'
        side_effects = [interface, extneral_id, value]
        self.mock_patch_object(self.pluginlib,
                               'exists')
        self.pluginlib.exists.side_effect = side_effects
        expected_cmd_args = ['ovs-vsctl', 'set', 'Interface', interface,
                             'external-ids:%s=%s' % (extneral_id, value)]
        expected_pluginlib_arg_list = [call('fake_args', 'interface'),
                                       call('fake_args', 'extneral_id'),
                                       call('fake_args', 'value')]

        self.host._ovs_set_if_external_id('fake_args')
        self.host._run_command.assert_called_with(expected_cmd_args)
        self.assertEqual(self.pluginlib.exists.call_args_list,
                         expected_pluginlib_arg_list)

def test_ip_link_add_veth_pair(self):
        dev1_name = 'fake_brige_name'
        dev2_name = 'fake_port_name'
        side_effects = [dev1_name, dev2_name]
        self.mock_patch_object(self.pluginlib,
                               'exists')
        self.pluginlib.exists.side_effect = side_effects
        expected_cmd_args = ['ip', 'link', 'add', dev1_name, 'type',
                             'veth', 'peer', 'name', dev2_name]
        expected_pluginlib_arg_list = [call('fake_args', 'dev1_name'),
                                       call('fake_args', 'dev2_name')]

        self.host._ip_link_add_veth_pair('fake_args')
        self.host._run_command.assert_called_with(expected_cmd_args)
        self.assertEqual(self.pluginlib.exists.call_args_list,
                         expected_pluginlib_arg_list)

def test_ip_link_set_promisc(self):
        device_name = 'fake_device_name'
        option = 'fake_option'
        side_effects = [device_name, option]
        self.mock_patch_object(self.pluginlib,
                               'exists')
        self.pluginlib.exists.side_effect = side_effects
        expected_cmd_args = ['ip', 'link', 'set', device_name, 'promisc',
                             option]
        expected_pluginlib_arg_list = [call('fake_args', 'device_name'),
                                       call('fake_args', 'option')]

        self.host._ip_link_set_promisc('fake_args')
        self.host._run_command.assert_called_with(expected_cmd_args)
        self.assertEqual(self.pluginlib.exists.call_args_list,
                         expected_pluginlib_arg_list)

def test_no_req_ids(self, *args):
        in_flight = 3

        get_holders = self.make_get_holders(1)
        max_connection = Mock(spec=Connection, host='localhost',
                              lock=Lock(),
                              max_request_id=in_flight - 1, in_flight=in_flight,
                              is_idle=True, is_defunct=False, is_closed=False)
        holder = get_holders.return_value[0]
        holder.get_connections.return_value.append(max_connection)

        self.run_heartbeat(get_holders)

        holder.get_connections.assert_has_calls([call()] * get_holders.call_count)
        self.assertEqual(max_connection.in_flight, in_flight)
        self.assertEqual(max_connection.send_msg.call_count, 0)
        self.assertEqual(max_connection.send_msg.call_count, 0)
        max_connection.defunct.assert_has_calls([call(ANY)] * get_holders.call_count)
        holder.return_connection.assert_has_calls(
            [call(max_connection)] * get_holders.call_count)

def test_scheduler_persistent(self):
        # TODO: Improve this test to avoid the need to check for log messages.
        self.spider.log = mock.Mock(spec=self.spider.log)

        self.scheduler.persist = True
        self.scheduler.open(self.spider)

        self.assertEqual(self.spider.log.call_count, 0)

        self.scheduler.enqueue_request(Request('http://example.com/page1'))
        self.scheduler.enqueue_request(Request('http://example.com/page2'))

        self.assertTrue(self.scheduler.has_pending_requests())
        self.scheduler.close('finish')

        self.scheduler.open(self.spider)
        self.spider.log.assert_has_calls([
            mock.call("Resuming crawl (2 requests scheduled)"),
        ])
        self.assertEqual(len(self.scheduler), 2)

        self.scheduler.persist = False
        self.scheduler.close('finish')

        self.assertEqual(len(self.scheduler), 0)

def test_register_configs_apache(self, resource_map, exists, renderer):
        exists.return_value = False
        self.os_release.return_value = 'havana'
        fake_renderer = MagicMock()
        fake_renderer.register = MagicMock()
        renderer.return_value = fake_renderer

        resource_map.return_value = self.rsc_map
        utils.register_configs()
        renderer.assert_called_with(
            openstack_release='havana', templates_dir='templates/')

        ex_reg = [
            call('/etc/keystone/keystone.conf', [self.ctxt]),
            call(
                '/etc/apache2/sites-available/openstack_https_frontend',
                [self.ctxt]),
            call(
                '/etc/apache2/sites-available/openstack_https_frontend.conf',
                [self.ctxt]),
        ]
        self.assertEqual(fake_renderer.register.call_args_list, ex_reg)

def test_create_user_credentials_no_roles(self, mock_create_user,
                                              mock_create_role,
                                              mock_grant_role,
                                              mock_user_exists,
                                              get_callback, set_callback):
        mock_user_exists.return_value = False
        get_callback.return_value = 'passA'
        utils.create_user_credentials('userA',
                                      get_callback,
                                      set_callback,
                                      tenant='tenantA')
        mock_create_user.assert_has_calls([call('userA', 'passA',
                                                domain=None,
                                                tenant='tenantA')])
        mock_create_role.assert_has_calls([])
        mock_grant_role.assert_has_calls([])

def test_create_user_credentials(self, mock_create_user, mock_create_role,
                                     mock_grant_role, mock_user_exists,
                                     get_callback, set_callback):
        mock_user_exists.return_value = False
        get_callback.return_value = 'passA'
        utils.create_user_credentials('userA',
                                      get_callback,
                                      set_callback,
                                      tenant='tenantA',
                                      grants=['roleA'], new_roles=['roleB'])
        mock_create_user.assert_has_calls([call('userA', 'passA',
                                                tenant='tenantA',
                                                domain=None)])
        mock_create_role.assert_has_calls([call('roleB', user='userA',
                                                tenant='tenantA',
                                                domain=None)])
        mock_grant_role.assert_has_calls([call('userA', 'roleA',
                                               tenant='tenantA',
                                               user_domain=None,
                                               project_domain=None)])

def test_git_pre_install(self, adduser, add_group, add_user_to_group,
                             write_file, mkdir):
        utils.git_pre_install()
        adduser.assert_called_with('keystone', shell='/bin/bash',
                                   system_user=True,
                                   home_dir='/var/lib/keystone')
        add_group.assert_called_with('keystone', system_group=True)
        add_user_to_group.assert_called_with('keystone', 'keystone')
        expected = [
            call('/var/lib/keystone', owner='keystone',
                 group='keystone', perms=0755, force=False),
            call('/var/lib/keystone/cache', owner='keystone',
                 group='keystone', perms=0755, force=False),
            call('/var/log/keystone', owner='keystone',
                 group='keystone', perms=0755, force=False),
        ]
        self.assertEqual(mkdir.call_args_list, expected)
        write_file.assert_called_with('/var/log/keystone/keystone.log',
                                      '', owner='keystone', group='keystone',
                                      perms=0600)

def test_ports_build(self, mock_iface_updown):
        expected_port_build = self.read_config_xml('irf_port_build')
        expected_activate = self.read_action_xml('irf_port_build_activate')

        expected_call_list = [
            [expected_port_build, 'edit_config'],
            ['startup.cfg', 'save'],
            [expected_activate, 'action']

        ]

        down_ifaces = ['FortyGigE1/0/2', 'FortyGigE1/0/1', 'FortyGigE1/0/3', 'FortyGigE1/0/4']
        up_ifaces = ['FortyGigE1/0/1', 'FortyGigE1/0/2', 'FortyGigE1/0/3', 'FortyGigE1/0/4']

        self.irf_port.build('1', [], [], ['FortyGigE1/0/1', 'FortyGigE1/0/2'], ['FortyGigE1/0/3', 'FortyGigE1/0/4'])

        self.assert_stage_requests_multiple(expected_call_list)
        mock_iface_updown.assert_has_calls([mock.call(down_ifaces, 'down'), mock.call(up_ifaces, 'up')])

def test_ports_build_remove(self, mock_iface_updown):
        expected_port_build = self.read_config_xml('irf_port_build_remove')
        expected_activate = self.read_action_xml('irf_port_build_activate')

        expected_call_list = [
            [expected_port_build, 'edit_config'],
            ['startup.cfg', 'save'],
            [expected_activate, 'action']

        ]

        down_ifaces = ['FortyGigE1/0/2', 'FortyGigE1/0/1', 'FortyGigE1/0/3', 'FortyGigE1/0/4']
        up_ifaces = []

        self.irf_port.build('1', ['FortyGigE1/0/1', 'FortyGigE1/0/2'], ['FortyGigE1/0/3', 'FortyGigE1/0/4'], [], [])

        self.assert_stage_requests_multiple(expected_call_list)
        mock_iface_updown.assert_has_calls([mock.call(down_ifaces, 'down'), mock.call(up_ifaces, 'up')])

def test_init(self, mock_ping):
        ping = Ping(self.device, TARGET)

        self.assertEqual(ping.vrf, '')
        self.assertEqual(ping.host, TARGET)
        self.assertEqual(ping.v6, False)
        self.assertEqual(ping.detail, False)

        ping6 = Ping(self.device, TARGET, v6=True)

        self.assertEqual(ping6.vrf, '')
        self.assertEqual(ping6.host, TARGET)
        self.assertEqual(ping6.v6, True)
        self.assertEqual(ping6.detail, False)

        mock_ping.assert_has_calls([mock.call(), mock.call()])

def test_migrate_nova_databases_ocata(self, cellv2_ready, get_cell_uuid,
                                          check_output):
        "Migrate database with nova-manage in a clustered env"
        get_cell_uuid.return_value = 'c83121db-f1c7-464a-b657-38c28fac84c6'
        self.relation_ids.return_value = ['cluster:1']
        self.os_release.return_value = 'ocata'
        utils.migrate_nova_databases()
        check_output.assert_has_calls([
            call(['nova-manage', 'api_db', 'sync']),
            call(['nova-manage', 'cell_v2', 'map_cell0']),
            call(['nova-manage', 'cell_v2', 'create_cell', '--name', 'cell1',
                  '--verbose']),
            call(['nova-manage', 'db', 'sync']),
            call(['nova-manage', 'db', 'online_data_migrations']),
            call(['nova-manage', 'cell_v2', 'discover_hosts', '--cell_uuid',
                  'c83121db-f1c7-464a-b657-38c28fac84c6', '--verbose']),
        ])
        self.peer_store.assert_called_with('dbsync_state', 'complete')
        self.assertTrue(self.enable_services.called)
        self.cmd_all_services.assert_called_with('start')

def test_upgrade_icehouse_juno(self, determine_packages,
                                   migrate_nova_databases,
                                   get_step_upgrade_source):
        "Simulate a call to do_openstack_upgrade() for icehouse->juno"
        self.test_config.set('openstack-origin', 'cloud:trusty-juno')
        get_step_upgrade_source.return_value = None
        self.os_release.return_value = 'icehouse'
        self.get_os_codename_install_source.return_value = 'juno'
        self.is_leader.return_value = True
        self.relation_ids.return_value = []
        utils.do_openstack_upgrade(self.register_configs())
        self.apt_update.assert_called_with(fatal=True)
        self.apt_upgrade.assert_called_with(options=DPKG_OPTS, fatal=True,
                                            dist=True)
        self.apt_install.assert_called_with(determine_packages(), fatal=True)
        self.register_configs.assert_called_with(release='juno')
        self.assertTrue(migrate_nova_databases.call_count, 1)

def test_upgrade_kilo_liberty(self, determine_packages,
                                  migrate_nova_databases,
                                  migrate_nova_flavors,
                                  get_step_upgrade_source):
        "Simulate a call to do_openstack_upgrade() for kilo->liberty"
        self.test_config.set('openstack-origin', 'cloud:trusty-liberty')
        get_step_upgrade_source.return_value = None
        self.os_release.return_value = 'kilo'
        self.get_os_codename_install_source.return_value = 'liberty'
        self.is_leader.return_value = True
        self.relation_ids.return_value = []
        utils.do_openstack_upgrade(self.register_configs())
        self.apt_update.assert_called_with(fatal=True)
        self.apt_upgrade.assert_called_with(options=DPKG_OPTS, fatal=True,
                                            dist=True)
        self.apt_install.assert_called_with(determine_packages(), fatal=True)
        self.register_configs.assert_called_with(release='liberty')
        self.assertTrue(migrate_nova_flavors.call_count, 1)
        self.assertTrue(migrate_nova_databases.call_count, 1)

def test_upgrade_liberty_mitaka(self, determine_packages,
                                    migrate_nova_databases,
                                    get_step_upgrade_source,
                                    database_setup):
        "Simulate a call to do_openstack_upgrade() for liberty->mitaka"
        self.test_config.set('openstack-origin', 'cloud:trusty-kilo')
        get_step_upgrade_source.return_value = None
        self.os_release.return_value = 'liberty'
        self.get_os_codename_install_source.return_value = 'mitaka'
        self.is_leader.return_value = True
        self.relation_ids.return_value = []
        database_setup.return_value = False
        utils.do_openstack_upgrade(self.register_configs())
        self.apt_update.assert_called_with(fatal=True)
        self.apt_upgrade.assert_called_with(options=DPKG_OPTS, fatal=True,
                                            dist=True)
        self.apt_install.assert_called_with(determine_packages(), fatal=True)
        self.register_configs.assert_called_with(release='mitaka')
        self.assertFalse(migrate_nova_databases.called)
        database_setup.assert_called_with(prefix='novaapi')

def test_config_changed_region_change(self, mock_compute_changed,
                                          mock_config_https,
                                          mock_filter_packages,
                                          mock_service_resume,
                                          mock_is_db_initialised,
                                          mock_update_nova_consoleauth_config,
                                          mock_update_aws_compat_services):
        self.git_install_requested.return_value = False
        self.openstack_upgrade_available.return_value = False
        self.config_value_changed.return_value = True
        self.related_units.return_value = ['unit/0']
        self.relation_ids.side_effect = \
            lambda x: ['generic_rid'] if x == 'cloud-compute' else []
        mock_is_db_initialised.return_value = False
        self.os_release.return_value = 'diablo'
        hooks.config_changed()
        mock_compute_changed.assert_has_calls([call('generic_rid', 'unit/0')])
        self.assertTrue(mock_update_nova_consoleauth_config.called)
        self.assertTrue(mock_update_aws_compat_services.called)

def test_db_joined_mitaka(self):
        self.get_relation_ip.return_value = '10.10.10.10'
        self.os_release.return_value = 'mitaka'
        self.is_relation_made.return_value = False
        hooks.db_joined()
        self.relation_set.assert_has_calls([
            call(nova_database='nova',
                 nova_username='nova',
                 nova_hostname='10.10.10.10',
                 relation_id=None),
            call(novaapi_database='nova_api',
                 novaapi_username='nova',
                 novaapi_hostname='10.10.10.10',
                 relation_id=None),
        ])

        self.get_relation_ip.assert_called_with('shared-db',
                                                cidr_network=None)

def test_amqp_changed_api_rel(self, configs, cell_joined, api_joined,
                                  quantum_joined, mock_is_db_initialised,
                                  update_db_allowed, init_db_allowed):
        self.relation_ids.side_effect = [
            ['nova-cell-api/0'],
            ['nova-api/0'],
            ['quantum-service/0'],
        ]
        mock_is_db_initialised.return_value = False
        configs.complete_contexts = MagicMock()
        configs.complete_contexts.return_value = ['amqp']
        configs.write = MagicMock()
        self.os_release.return_value = 'diablo'
        self.is_relation_made.return_value = True
        hooks.amqp_changed()
        self.assertEqual(configs.write.call_args_list,
                         [call('/etc/nova/nova.conf')])
        cell_joined.assert_called_with(rid='nova-cell-api/0')
        api_joined.assert_called_with(rid='nova-api/0')
        quantum_joined.assert_called_with(rid='quantum-service/0',
                                          remote_restart=True)

def test_execute(self, mock_finalize, mock_execute, mock_initialize):
        c1 = Chain('c1')
        l1 = Link('l1')
        l2 = Link('l2')
        l3 = Link('l3')

        # test happy flow
        c1.links = [l1, l2, l3]
        mock_initialize.return_value = StatusCode.Success
        mock_execute.return_value = StatusCode.Success
        mock_finalize.return_value = StatusCode.Success
        mock_parent = mock.MagicMock(autospec=True)
        mock_parent.attach_mock(mock_initialize, 'initialize')
        mock_parent.attach_mock(mock_execute, 'execute')
        mock_parent.attach_mock(mock_finalize, 'finalize')
        status = c1.initialize()
        self.assertEqual(status, StatusCode.Success)
        status = c1.execute()
        self.assertEqual(status, StatusCode.Success)
        status = c1.finalize()
        self.assertEqual(status, StatusCode.Success)
        calls = [mock.call.initialize()]*3 + [mock.call.execute()]*3 + [mock.call.finalize()]*3
        mock_parent.assert_has_calls(calls, any_order=False)