def mock_ec2():
return mock.Mock(spec=botocore.client)
python类Mock()的实例源码
test_rolling_upgrades.py 文件源码
项目:asg-rolling-upgrade
作者: crunch-accounting
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
test_rolling_upgrades.py 文件源码
项目:asg-rolling-upgrade
作者: crunch-accounting
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def mock_ec2_client():
return mock.Mock()
test_rolling_upgrades.py 文件源码
项目:asg-rolling-upgrade
作者: crunch-accounting
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def mock_ssh_connection():
return mock.Mock(spec=SSHClient)
test_rolling_upgrades.py 文件源码
项目:asg-rolling-upgrade
作者: crunch-accounting
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def mock_aws_manager():
return mock.Mock(spec=AwsManager)
test_rolling_upgrades.py 文件源码
项目:asg-rolling-upgrade
作者: crunch-accounting
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def mock_instance_manager():
return mock.Mock(spec=InstanceSshManager)
test_rolling_upgrades.py 文件源码
项目:asg-rolling-upgrade
作者: crunch-accounting
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def test_rum_gets_list_of_instances_to_upgrade(
rolling_upgrade_manager,
mock_aws_manager
):
Instance = namedtuple('Instance', ['id'])
instances = [Instance('instance1'),
Instance('instance2'),
Instance('instance3')]
rolling_upgrade_manager.compare_instance_to_config = mock.Mock()
mock_aws_manager.get_instances_for_asg.return_value = instances
rolling_upgrade_manager.compare_instance_to_config.side_effect = \
([], [], ['diff'])
result = rolling_upgrade_manager.get_instances_to_upgrade("test-asg", {})
assert len(result) == 1
assert instances[2] in result
rolling_upgrade_manager.compare_instance_to_config.side_effect = \
(['diff1', 'diff2'], [], ['diff'])
result = rolling_upgrade_manager.get_instances_to_upgrade("test-asg", {})
assert len(result) == 2
assert instances[0] in result
assert instances[2] in result
rolling_upgrade_manager.compare_instance_to_config.side_effect = \
([], [], [])
result = rolling_upgrade_manager.get_instances_to_upgrade("test-asg", {})
assert len(result) == 0
mock_aws_manager.get_instances_for_asg.assert_has_calls((
mock.call('test-asg'),
mock.call('test-asg'),
mock.call('test-asg'),
))
def test_poster_with_tags_facebook(self, facebook_mock, phantom_driver):
mock_body = open(os.path.join(os.path.dirname(__file__), 'data', 'crs_page.html')).read()
posters_register._posters = []
posters_register.register_poster(facebook_status, 'facebook_done')
m = mock.Mock()
m.put_photo.side_effect = mock.Mock(id='1')
facebook_mock.return_value = m
phantom_driver.return_value = MockPhantomJS(mock_body)
self.assertEqual(len(Log.objects.all()), 0)
meteo_fake_status = '?? Emergenza: A+ , A- , O+ , O- , B- , AB- , AB+\n'
meteo_upload_fake = os.path.join(os.path.dirname(__file__), 'data', 'meteo_fake_upload.png')
facebook_status(meteo_fake_status, meteo_upload_fake)
self.assertEqual(m.put_wall_post.call_count, 0)
self.assertEqual(m.put_photo.call_count, 1)
self.assertEqual(
m.put_photo.call_args[1]['tags'],
'[{"tag_uid": "andrea.stagi"}, {"tag_uid": "patrick.arminio"}]'
)
def core_mock(self):
return mock.Mock()
def setUp(self):
super(TestSnmpClient, self).setUp()
self.command_generator_mock = mock.Mock()
self.pysnmp_mock = mock.Mock()
self.oneliner_cmdgen = mock.Mock()
self.oneliner_cmdgen.CommandGenerator.return_value = \
self.command_generator_mock
self.oneliner_cmdgen.CommunityData.return_value = \
sentinel.community_data
self.oneliner_cmdgen.UdpTransportTarget.return_value = \
sentinel.udp_transport_target
self.snmp_client = snmp_client.SnmpClient(
oneliner_cmdgen=self.oneliner_cmdgen,
host=sentinel.hostname,
port=sentinel.port,
community=sentinel.community,
timeout=sentinel.timeout,
retries=sentinel.retries,
)
self.all_error_indications = [
(errind.__dict__.get(a).__class__.__name__, errind.__dict__.get(a))
for a in dir(errind)
if isinstance(errind.__dict__.get(a), ErrorIndication)]
def test_power_on(self, mock_open):
domain_mock = mock.Mock()
connection_mock = mock.Mock()
connection_mock.lookupByName.return_value = domain_mock
mock_open.return_value = connection_mock
self.driver.power_on('domainA')
mock_open.assert_called_with('hello')
connection_mock.lookupByName.assert_called_with('domainA')
domain_mock.create.assert_called_with()
connection_mock.close.assert_called_with()
def test_power_off(self, mock_open):
domain_mock = mock.Mock()
connection_mock = mock.Mock()
connection_mock.lookupByName.return_value = domain_mock
mock_open.return_value = connection_mock
self.driver.power_off('domainA')
mock_open.assert_called_with('hello')
connection_mock.lookupByName.assert_called_with('domainA')
domain_mock.destroy.assert_called_with()
connection_mock.close.assert_called_with()
def test_get_power_state_on(self, mock_open):
domain_mock = mock.Mock()
domain_mock.isActive.return_value = 1
connection_mock = mock.Mock()
connection_mock.lookupByName.return_value = domain_mock
mock_open.return_value = connection_mock
self.assertEqual(drivers.POWER_ON,
self.driver.get_power_state('domainA'))
mock_open.assert_called_with('hello')
connection_mock.lookupByName.assert_called_with('domainA')
domain_mock.isActive.assert_called_with()
connection_mock.close.assert_called_with()
def test_get_power_state_off(self, mock_open):
domain_mock = mock.Mock()
domain_mock.isActive.return_value = 0
connection_mock = mock.Mock()
connection_mock.lookupByName.return_value = domain_mock
mock_open.return_value = connection_mock
self.assertEqual(drivers.POWER_OFF,
self.driver.get_power_state('domainA'))
mock_open.assert_called_with('hello')
connection_mock.lookupByName.assert_called_with('domainA')
domain_mock.isActive.assert_called_with()
connection_mock.close.assert_called_with()
def test_get_power_state_domain_not_found(self, mock_open):
connection_mock = mock.Mock()
connection_mock.lookupByName.side_effect = \
libvirt.libvirtError('virDomainLookupByName() failed',
conn=connection_mock)
mock_open.return_value = connection_mock
self.assertRaises(drivers.DeviceNotFound,
self.driver.get_power_state, 'domainA')
connection_mock.close.assert_called_with()
def test_power_on_domain_not_found(self, mock_open):
connection_mock = mock.Mock()
connection_mock.lookupByName.side_effect = \
libvirt.libvirtError('virDomainLookupByName() failed',
conn=connection_mock)
mock_open.return_value = connection_mock
self.assertRaises(drivers.DeviceNotFound,
self.driver.power_on, 'domainA')
connection_mock.close.assert_called_with()
def test_power_off_domain_not_found(self, mock_open):
connection_mock = mock.Mock()
connection_mock.lookupByName.side_effect = \
libvirt.libvirtError('virDomainLookupByName() failed',
conn=connection_mock)
mock_open.return_value = connection_mock
self.assertRaises(drivers.DeviceNotFound,
self.driver.power_off, 'domainA')
connection_mock.close.assert_called_with()
def test_power_on_generic_error_goes_through(self, mock_open):
domain_mock = mock.Mock()
connection_mock = mock.Mock()
connection_mock.lookupByName.return_value = domain_mock
domain_mock.create.side_effect = SpecificException()
mock_open.return_value = connection_mock
self.assertRaises(SpecificException, self.driver.power_on, 'domainA')
connection_mock.close.assert_called_with()
def test_power_on_stays_connected(self, mock_open):
domain_mock = mock.Mock()
connection_mock = mock.Mock()
connection_mock.lookupByName.return_value = domain_mock
mock_open.return_value = connection_mock
self.driver.power_on('domainA')
mock_open.assert_called_with('hello')
connection_mock.lookupByName.assert_called_with('domainA')
domain_mock.create.assert_called_with()
self.assertFalse(connection_mock.close.called)
def setUp(self):
super(PDUTestCase, self).setUp()
self.community = 'test1212'
self.core_mock = mock.Mock()
self.pdu = self.pdu_class(name='test_pdu', core=self.core_mock)
self.pdu_test_harness = pysnmp_handler.SNMPPDUHarness(
self.pdu,
'127.0.0.1',
random.randint(20000, 30000),
self.community)
self.pdu_test_harness.start()