def test_delete(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name[:-1]
))
) as mock, patch('pynetbox.lib.query.requests.delete') as delete:
ret = getattr(nb, self.name).get(1)
self.assertTrue(ret.delete())
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/1/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
delete.assert_called_with(
'http://localhost:8000/api/{}/{}/1/'.format(
self.app,
self.name.replace('_', '-')
),
headers=AUTH_HEADERS
)
python类mock()的实例源码
def test_create_device_bulk(self, mock):
data = [
{
'name': 'test-device',
'site': 1,
'device_type': 1,
'device_role': 1,
},
{
'name': 'test-device1',
'site': 1,
'device_type': 1,
'device_role': 1,
},
]
ret = nb.devices.create(data)
self.assertTrue(ret)
self.assertTrue(len(ret), 2)
def test_get_all(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name
))
) as mock:
ret = getattr(nb, self.name).all()
self.assertTrue(ret)
self.assertTrue(isinstance(ret, list))
self.assertTrue(isinstance(ret[0], self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_get(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name[:-1]
))
) as mock:
ret = getattr(nb, self.name).get(1)
self.assertTrue(ret)
self.assertTrue(isinstance(ret, self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/1/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_get_all(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name
))
) as mock:
ret = getattr(nb, self.name).all()
self.assertTrue(ret)
self.assertTrue(isinstance(ret, list))
self.assertTrue(isinstance(ret[0], self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_filter(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name
))
) as mock:
ret = getattr(nb, self.name).filter(pk=1)
self.assertTrue(ret)
self.assertTrue(isinstance(ret, list))
self.assertTrue(isinstance(ret[0], self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/?pk=1'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_get(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name[:-1]
))
) as mock:
ret = getattr(nb, self.name).get(1)
self.assertTrue(ret)
self.assertTrue(isinstance(ret, self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/1/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_get_all(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name
))
) as mock:
ret = getattr(nb, self.name).all()
self.assertTrue(ret)
self.assertTrue(isinstance(ret, list))
self.assertTrue(isinstance(ret[0], self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_get(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name[:-1]
))
) as mock:
ret = getattr(nb, self.name).get(1)
self.assertTrue(ret)
self.assertTrue(isinstance(ret, self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/1/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_get_all(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name
))
) as mock:
ret = getattr(nb, self.name).all()
self.assertTrue(ret)
self.assertTrue(isinstance(ret, list))
self.assertTrue(isinstance(ret[0], self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_filter(self):
with patch(
'pynetbox.lib.query.requests.get',
return_value=Response(fixture='{}/{}.json'.format(
self.app,
self.name
))
) as mock:
ret = getattr(nb, self.name).filter(pk=1)
self.assertTrue(ret)
self.assertTrue(isinstance(ret, list))
self.assertTrue(isinstance(ret[0], self.ret))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/?pk=1'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
def test_update_no_active_player_makes_next_active(self):
# Setup, add some players to waiting list.
laser_players = players.Players(1)
laser_players.add_player('192.168.0.1')
# Check none are yet active.
self.assertEqual(None, laser_players.active_player())
# Setup, create mock callbacks that will remember how they were called.
start_active = unittest.mock.Mock()
end_active = unittest.mock.Mock()
# Call update with 1 second of time elapsed.
laser_players.update(0.5, start_active, end_active)
# Check start active was called with the first in line IP, and that the
# currently active player is the first one added (with 1 second of playtime
# left).
start_active.assert_called_once_with('192.168.0.1')
end_active.assert_not_called()
self.assertEqual(('192.168.0.1', 1), laser_players.active_player())
def test_update_moves_to_next_after_playetime_elapsed(self):
# Setup, add two players and make the first active.
laser_players = players.Players(1)
laser_players.add_player('192.168.0.1')
laser_players.add_player('192.168.0.2')
start_active = unittest.mock.Mock()
end_active = unittest.mock.Mock()
laser_players.update(1, start_active, end_active)
start_active.reset_mock()
end_active.reset_mock()
# Update with two seconds of time so the first player time is up, then
# check next player is active.
laser_players.update(2, start_active, end_active)
end_active.assert_called_once_with('192.168.0.1')
start_active.assert_called_once_with('192.168.0.2')
self.assertEqual(('192.168.0.2', 1), laser_players.active_player())
def test_uploading_syslog(self):
'''Tests syslog upload command'''
syslog_uploader_cli = ["syslog_uploader", "-c", self._ndr_config_file, TEST_SYSLOG_DATA]
with unittest.mock.patch.object(sys, 'argv', syslog_uploader_cli):
ndr.tools.syslog_uploader.main()
# Make sure there's only one file in the queue
outbound_queue = os.listdir(self._ncc.outgoing_upload_spool)
self.assertEqual(len(outbound_queue), 1)
this_msg = self._ncc.outgoing_upload_spool + "/" + outbound_queue[0]
loaded_msg = ndr.IngestMessage.verify_and_load_message(
self._ncc, this_msg, only_accept_cn="ndr_test_suite")
os.remove(this_msg)
self.assertEqual(loaded_msg.message_type, ndr.IngestMessageTypes.SYSLOG_UPLOAD)
syslog = ndr.SyslogUploadMessage().from_message(loaded_msg)
def test_uploading_status(self):
'''Tests uploading status messages'''
status_uploader_cli = ["status_uploader", "-c", self._ndr_config_file]
with unittest.mock.patch.object(sys, 'argv', status_uploader_cli):
ndr.tools.status.main()
# Make sure there's only one file in the queue
outbound_queue = os.listdir(self._ncc.outgoing_upload_spool)
self.assertEqual(len(outbound_queue), 1)
this_msg = self._ncc.outgoing_upload_spool + "/" + outbound_queue[0]
loaded_msg = ndr.IngestMessage.verify_and_load_message(
self._ncc, this_msg, only_accept_cn="ndr_test_suite")
os.remove(this_msg)
self.assertEqual(loaded_msg.message_type, ndr.IngestMessageTypes.STATUS)
def test_alert_tester(self):
'''Tests alert tester messages'''
alert_tester_cli = ["alert_tester", "-c", self._ndr_config_file]
with unittest.mock.patch.object(sys, 'argv', alert_tester_cli):
ndr.tools.alert_tester.main()
# Make sure there's only one file in the queue
outbound_queue = os.listdir(self._ncc.outgoing_upload_spool)
self.assertEqual(len(outbound_queue), 1)
this_msg = self._ncc.outgoing_upload_spool + "/" + outbound_queue[0]
loaded_msg = ndr.IngestMessage.verify_and_load_message(
self._ncc, this_msg, only_accept_cn="ndr_test_suite")
os.remove(this_msg)
self.assertEqual(loaded_msg.message_type, ndr.IngestMessageTypes.TEST_ALERT)
def test_pcap_processing(self):
'''Tests uploading and processing a pcap message'''
alert_tester_cli = ["pcap_processor", "-k", "-c", self._ndr_config_file, TSHARK_PCAP]
with unittest.mock.patch.object(sys, 'argv', alert_tester_cli):
ndr.tools.pcap_to_traffic_report.main()
# Make sure there's only one file in the queue
outbound_queue = os.listdir(self._ncc.outgoing_upload_spool)
self.assertEqual(len(outbound_queue), 1)
this_msg = self._ncc.outgoing_upload_spool + "/" + outbound_queue[0]
loaded_msg = ndr.IngestMessage.verify_and_load_message(
self._ncc, this_msg, only_accept_cn="ndr_test_suite")
os.remove(this_msg)
self.assertEqual(loaded_msg.message_type, ndr.IngestMessageTypes.TRAFFIC_REPORT)
def test_recursive_query_basic_failure(self):
resolver = dns.resolver.Resolver()
domain = dns.name.from_text('example.com.')
record_type = 'NS'
with unittest.mock.patch.object(fierce, 'query', return_value=None) as mock_method:
result = fierce.recursive_query(resolver, domain, record_type=record_type)
expected = [
unittest.mock.call(resolver, 'example.com.', record_type),
unittest.mock.call(resolver, 'com.', record_type),
unittest.mock.call(resolver, '', record_type),
]
mock_method.assert_has_calls(expected)
self.assertIsNone(result)
def test_recursive_query_long_domain_failure(self):
resolver = dns.resolver.Resolver()
domain = dns.name.from_text('sd1.sd2.example.com.')
record_type = 'NS'
with unittest.mock.patch.object(fierce, 'query', return_value=None) as mock_method:
result = fierce.recursive_query(resolver, domain, record_type=record_type)
expected = [
unittest.mock.call(resolver, 'sd1.sd2.example.com.', record_type),
unittest.mock.call(resolver, 'sd2.example.com.', record_type),
unittest.mock.call(resolver, 'example.com.', record_type),
unittest.mock.call(resolver, 'com.', record_type),
unittest.mock.call(resolver, '', record_type),
]
mock_method.assert_has_calls(expected)
self.assertIsNone(result)
def test_recursive_query_basic_success(self):
resolver = dns.resolver.Resolver()
domain = dns.name.from_text('example.com.')
record_type = 'NS'
good_response = unittest.mock.MagicMock()
side_effect = [
None,
good_response,
None,
]
with unittest.mock.patch.object(fierce, 'query', side_effect=side_effect) as mock_method:
result = fierce.recursive_query(resolver, domain, record_type=record_type)
expected = [
unittest.mock.call(resolver, 'example.com.', record_type),
unittest.mock.call(resolver, 'com.', record_type),
]
mock_method.assert_has_calls(expected)
self.assertEqual(result, good_response)
