Python unittest 模块，mock() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用unittest.mock()。
def tests_event_registration(self):
"""Tests that events register correctly."""
# Get the event controller
events = self.abode.events
self.assertIsNotNone(events)
# Create mock callback
callback = Mock()
# Test that a valid event registers
self.assertTrue(
events.add_event_callback(TIMELINE.ALARM_GROUP, callback))
# Test that no event group returns false
self.assertFalse(events.add_event_callback(None, callback))
# Test that an invalid event throws exception
with self.assertRaises(abodepy.AbodeException):
events.add_event_callback("lol", callback)
def tests_timeline_registration(self):
"""Tests that timeline events register correctly."""
# Get the event controller
events = self.abode.events
self.assertIsNotNone(events)
# Create mock callback
callback = Mock()
# Test that a valid timeline event registers
self.assertTrue(
events.add_timeline_callback(
TIMELINE.CAPTURE_IMAGE, callback))
# Test that no timeline event returns false
self.assertFalse(events.add_timeline_callback(None, callback))
# Test that an invalid timeline event string throws exception
with self.assertRaises(abodepy.AbodeException):
events.add_timeline_callback("lol", callback)
# Test that an invalid timeline event dict throws exception
with self.assertRaises(abodepy.AbodeException):
events.add_timeline_callback({"lol": "lol"}, callback)
def tests_multi_events_callback(self):
"""Tests that multiple event updates callback correctly."""
# Get the event controller
events = self.abode.events
self.assertIsNotNone(events)
# Create mock callback
callback = Mock()
# Register our events
self.assertTrue(
events.add_event_callback(
[TIMELINE.ALARM_GROUP, TIMELINE.CAPTURE_GROUP],
callback))
# Call our events callback method and trigger a capture group event
# pylint: disable=protected-access
event_json = json.loads(IRCAMERA.timeline_event())
events._on_timeline_update(event_json)
# Ensure our callback was called
callback.assert_called_with(event_json)
def tests_multi_timeline_callback(self):
"""Tests that multiple timeline updates callback correctly."""
# Get the event controller
events = self.abode.events
self.assertIsNotNone(events)
# Create mock callback
callback = Mock()
# Register our events
self.assertTrue(
events.add_timeline_callback(
[TIMELINE.CAPTURE_IMAGE, TIMELINE.OPENED], callback))
# Call our events callback method and trigger a capture group event
# pylint: disable=protected-access
event_json = json.loads(IRCAMERA.timeline_event())
events._on_timeline_update(event_json)
# Ensure our callback was called
callback.assert_called_with(event_json)
def tests_automations_callback(self):
"""Tests that automation updates callback correctly."""
# Get the event controller
events = self.abode.events
self.assertIsNotNone(events)
# Create mock callbacks
automation_callback = Mock()
# Register our events
self.assertTrue(
events.add_event_callback(
TIMELINE.AUTOMATION_EDIT_GROUP, automation_callback))
# Call our events callback method and trigger a capture group event
# pylint: disable=protected-access
events._on_automation_update('{}')
# Our capture callback should get one, but our alarm should not
automation_callback.assert_called_with('{}')
def test_subscription_recording_processor_for_request(self):
replaced_subscription_id = str(uuid.uuid4())
rp = SubscriptionRecordingProcessor(replaced_subscription_id)
uri_templates = ['https://management.azure.com/subscriptions/{}/providers/Microsoft.ContainerRegistry/'
'checkNameAvailability?api-version=2017-03-01',
'https://graph.windows.net/{}/applications?api-version=1.6']
for template in uri_templates:
mock_sub_id = str(uuid.uuid4())
mock_request = mock.Mock()
mock_request.uri = template.format(mock_sub_id)
mock_request.body = self._mock_subscription_request_body(mock_sub_id)
rp.process_request(mock_request)
self.assertEqual(mock_request.uri, template.format(replaced_subscription_id))
self.assertEqual(mock_request.body,
self._mock_subscription_request_body(replaced_subscription_id))
def setUp(self):
# disable logging while testing
logging.disable(logging.CRITICAL)
self.patched = {}
if hasattr(self, 'patch_these'):
for patch_this in self.patch_these:
namespace = patch_this[0] if isinstance(patch_this, (list, set)) else patch_this
patcher = mock.patch(namespace)
mocked = patcher.start()
mocked.reset_mock()
self.patched[namespace] = mocked
if isinstance(patch_this, (list, set)) and len(patch_this) > 0:
retval = patch_this[1]
if callable(retval):
retval = retval()
mocked.return_value = retval
def test_reset_kill_reconnect(self):
client = AprsClient(aprs_user='testuser', aprs_filter='')
client.connect()
# .run() should be allowed to execute after .connect()
mock_callback = mock.MagicMock(
side_effect=lambda raw_msg: client.disconnect())
self.assertFalse(client._kill)
client.run(callback=mock_callback, autoreconnect=True)
# After .disconnect(), client._kill should be True
self.assertTrue(client._kill)
self.assertEqual(mock_callback.call_count, 1)
# After we reconnect, .run() should be able to run again
mock_callback.reset_mock()
client.connect()
client.run(callback=mock_callback, autoreconnect=True)
self.assertEqual(mock_callback.call_count, 1)
def test_normal_execution(self, mock_isfile, mock_glob):
""" Test the normal behaviour of the function.
"""
# Set the mocked function returned values.
mock_isfile.side_effect = [True]
mock_glob.return_value = ["/my/path/mock_output"]
# Test execution
fslreorient2std(**self.kwargs)
self.assertEqual([
mock.call(["which", "fslreorient2std"],
env={}, stderr=-1, stdout=-1),
mock.call(["fslreorient2std",
self.kwargs["input_image"],
self.kwargs["output_image"]],
cwd=None, env={}, stderr=-1, stdout=-1)],
self.mock_popen.call_args_list)
self.assertEqual(len(self.mock_env.call_args_list), 1)
def mock_ioctl(fd, command, msg):
print("Mocking ioctl")
assert fd == MOCK_FD
assert command is not None
# Reproduce ioctl read operations
if command == I2C_SMBUS and msg.read_write == I2C_SMBUS_READ:
offset = msg.command
if msg.size == I2C_SMBUS_BYTE_DATA:
msg.data.contents.byte = test_buffer[offset]
elif msg.size == I2C_SMBUS_WORD_DATA:
msg.data.contents.word = test_buffer[offset+1]*256 + test_buffer[offset]
elif msg.size == I2C_SMBUS_I2C_BLOCK_DATA:
for k in range(msg.data.contents.byte):
msg.data.contents.block[k+1] = test_buffer[offset+k]
# Override open, close and ioctl with our mock functions
def test_checkOtp(self, mock):
reply = {"requestId": "123", "status": "0000", "statusMessage": "Success", "credentialId": "",
"credentialType": "STANDARD_OTP", "authContext": {"params": {"Key": "authLevel.level", "Value": 1}}}
mock.SymantecUserServices.checkOtp.return_value = Mock()
mock.checkOtp.return_value.hash.return_value = reply
response = symantec_package.lib.userService.SymantecUserServices.checkOut("checkOutyou")
self.assertTrue(response.hash() != reply)
response = symantec_package.lib.userService.SymantecUserServices.checkOtp("PARAMS")
self.assertTrue((response.hash()) == reply)
self.assertTrue(response.hash()["status"] == "0000")
self.assertTrue(response.hash()['requestId'] == "123")
self.assertTrue(response.hash()['statusMessage'] == "Success")
self.assertTrue(response.hash()['credentialId'] == "")
self.assertTrue(response.hash()['credentialType'] == "STANDARD_OTP")
self.assertTrue(response.hash()['authContext']['params']['Key'] == "authLevel.level")
self.assertTrue(response.hash()['authContext']['params']['Value'] == 1)
pass
def test_poll_in_Push(self, mock):
reply = {"requestId": "ac123", "status": "6040", "statusMessage": "Mobile push request sent",
"pushDetail": {"pushCredentialId": "133709001", "pushSent": True},
"transactionId": "RealTransactionId",
"authContext": {"params": {"Key": "authLevel.level", "Value": 10}}}
mock.SymantecServices.authenticateUserWithPushThenPolling.return_value = Mock()
mock.authenticateUserWithPushThenPolling.return_value.hash.return_value = reply
response = symantec_package.lib.allServices.SymantecServices.authenticateUserWithNothing()
self.assertTrue(response.hash() != reply)
response = symantec_package.lib.allServices.SymantecServices.authenticateUserWithPushThenPolling("Parameters Here!")
self.assertTrue((response.hash()) == reply)
self.assertTrue(response.hash()["status"] == "6040")
self.assertTrue(response.hash()['requestId'] == "ac123")
self.assertTrue(response.hash()['statusMessage'] == "Mobile push request sent")
self.assertTrue(response.hash()["pushDetail"]['pushCredentialId'] == "133709001")
self.assertTrue(response.hash()["pushDetail"]['pushSent'] is True)
self.assertTrue(response.hash()['transactionId'] == "RealTransactionId")
self.assertTrue(response.hash()['authContext']['params']['Key'] == "authLevel.level")
self.assertTrue(response.hash()['authContext']['params']['Value'] == 10)
pass
def test_mock_create_user(self, mock_managementservices):
reply = {'requestId': 'create_123', 'status': '0000', 'statusMessage': 'Success'}
# Configure the mock to return a response with an OK status code. Also, the mock should have
# a `json()` method that returns a list of todos.
mock_managementservices.createUser.return_value = Mock()
mock_managementservices.createUser.return_value.json.return_value = reply
# Call the service, which will send a request to the server.
response = symantec_package.lib.managementService.SymantecManagementServices.createUser("create_123",
"new_user3")
print(response.json())
# If the request is sent successfully, then I expect a response to be returned.
self.assertTrue((response.json()) == reply)
self.assertTrue((response.json()["status"] == "0000"))
pass
def test_mock_add_STANDARDOTP_credential(self, mock_managementservices):
reply = {'statusMessage': "Success", 'requestId': 'add_otp_cred', 'status': '0000'}
# Configure the mock to return a response with an OK status code. Also, the mock should have
# a `json()` method that returns a list of todos.
mock_managementservices.addCredentialOtp.return_value = Mock()
mock_managementservices.addCredentialOtp.return_value.json.return_value = reply
response = symantec_package.lib.managementService.SymantecManagementServices.addCredentialOtp("add_otp_cred", "new_user3",
"", "STANDARD_OTP", \
"678066") # change with what's on your device
print(response.json())
# If the request is sent successfully, then I expect a response to be returned.
self.assertTrue((response.json()) == reply)
self.assertTrue((response.json()["status"] == "0000"))
pass
def test_mock_update_STANDARDOTP_credential(self, mock_managementservices):
reply = {'statusMessage': 'Success', 'requestId': 'update_123', 'status': '0000'}
# Configure the mock to return a response with an OK status code. Also, the mock should have
# a `json()` method that returns a list of todos.
mock_managementservices.updateCredential.return_value = Mock()
mock_managementservices.updateCredential.return_value.json.return_value = reply
response = symantec_package.lib.managementService.SymantecManagementServices.updateCredential("update_123", "gabe_phone",
"", "STANDARD_OTP",
"My personal cell phone")
print(response.json())
# If the request is sent successfully, then I expect a response to be returned.
self.assertTrue((response.json()) == reply)
self.assertTrue((response.json()["status"] == "0000"))
pass
def test_mock_setTemporaryPasswordSMSDelivery(self, mock_managementservices):
reply = {'status': '0000', 'requestId': 'setTempPWD', 'statusMessage': 'Success', 'temporaryPassword': '998241'}
# Configure the mock to return a response with an OK status code. Also, the mock should have
# a `json()` method that returns a list of todos.
mock_managementservices.setTemporaryPasswordSMSDelivery.return_value = Mock()
mock_managementservices.setTemporaryPasswordSMSDelivery.return_value.json.return_value = reply
response = symantec_package.lib.managementService.SymantecManagementServices.setTemporaryPasswordSMSDelivery("setTempPWD",
"gabe_phone",
"12313608781",
"17879481605")
print(response.json())
# If the request is sent successfully, then I expect a response to be returned.
self.assertTrue((response.json()) == reply)
self.assertTrue((response.json()["status"] == "0000"))
pass
def request_callback(self, request):
logger.debug('Mock request {} {}'.format(request.method, request.url))
path = self.build_path(request.method, request.url)
if os.path.exists(path):
# Load local file
logger.info('Using mock file {}'.format(path))
with gzip.open(path, 'rb') as f:
response = pickle.load(f)
else:
# Build from actual request
logger.info('Building mock file {}'.format(path))
response = self.real_request(request)
# Save in local file for future use
with gzip.open(path, 'wb') as f:
# Use old pickle ascii protocol (default)
# to be compatible with Python 2
f.write(pickle.dumps(response, protocol=2))
return (
response['status'],
response['headers'],
response['body'],
)
def test_mkdir_p(self, path_mocker_stopall):
# mock
mock_logger_info = path_mocker_stopall.MagicMock(name="mock_logger_info")
mock_dir_exists = path_mocker_stopall.MagicMock(name="mock_dir_exists")
mock_path = path_mocker_stopall.MagicMock(name="mock_path")
# patch
path_mocker_stopall.patch.object(scarlett_os.subprocess.logging.Logger, 'info', mock_logger_info)
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'dir_exists', mock_dir_exists)
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'Path', mock_path)
path = '/home/pi/dev/bossjones-github/scarlett_os/_debug'
mock_dir_exists.return_value = True
# run test
s_path.mkdir_p(path)
# assert
assert mock_logger_info.call_count == 1
mock_path.assert_called_once_with(path)
# from scarlett_os.internal.debugger import dump
mock_path().mkdir.assert_any_call(parents=True, exist_ok=True)
mock_logger_info.assert_any_call("Verify mkdir_p ran: {}".format(mock_dir_exists.return_value))
def test_dir_exists_false(self, path_mocker_stopall):
# mock
mock_logger_error = path_mocker_stopall.MagicMock(name="mock_logger_error")
mock_path = path_mocker_stopall.MagicMock(name="mock_path")
# patch
path_mocker_stopall.patch.object(scarlett_os.subprocess.logging.Logger, 'error', mock_logger_error)
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'Path', mock_path)
path = '/home/pi/dev/bossjones-github/scarlett_os/_debug'
mock_path_instance = mock_path()
mock_path_instance.is_dir.return_value = False
# run test
s_path.dir_exists(path)
# assert
assert mock_logger_error.call_count == 1
assert mock_path_instance.is_dir.call_count == 2
mock_logger_error.assert_any_call("This is not a dir: {}".format(path))
def test_dir_exists_true(self, path_mocker_stopall):
# mock
mock_logger_error = path_mocker_stopall.MagicMock(name="mock_logger_error")
mock_path = path_mocker_stopall.MagicMock(name="mock_path")
# patch
path_mocker_stopall.patch.object(scarlett_os.subprocess.logging.Logger, 'error', mock_logger_error)
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'Path', mock_path)
path = '/home/pi/dev/bossjones-github/scarlett_os/_debug'
mock_path_instance = mock_path()
#
mock_path_instance.is_dir.return_value = True
# run test
s_path.dir_exists(path)
# assert
assert mock_logger_error.call_count == 0
assert mock_path_instance.is_dir.call_count == 2
mock_logger_error.assert_not_called()
def test_mkdir_if_does_not_exist_false(self, path_mocker_stopall):
# mock
mock_mkdir_p = path_mocker_stopall.MagicMock(name="mock_mkdir_p")
mock_dir_exists = path_mocker_stopall.MagicMock(name="mock_dir_exists", return_value=False)
# patch
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'mkdir_p', mock_mkdir_p)
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'dir_exists', mock_dir_exists)
path = '/home/pi/dev/bossjones-github/scarlett_os/_debug'
# run test
result = s_path.mkdir_if_does_not_exist(path)
# assert
assert mock_mkdir_p.call_count == 1
assert mock_dir_exists.call_count == 1
assert result == True
def test_mkdir_if_does_not_exist_true(self, path_mocker_stopall):
# mock
mock_mkdir_p = path_mocker_stopall.MagicMock(name="mock_mkdir_p")
mock_dir_exists = path_mocker_stopall.MagicMock(name="mock_dir_exists", return_value=True)
# patch
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'mkdir_p', mock_mkdir_p)
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'dir_exists', mock_dir_exists)
path = '/home/pi/dev/bossjones-github/scarlett_os/_debug'
# run test
result = s_path.mkdir_if_does_not_exist(path)
# assert
assert mock_mkdir_p.call_count == 0
assert mock_dir_exists.call_count == 1
assert result == False
def test_fname_exists_false(self, path_mocker_stopall):
# mock
mock_path = path_mocker_stopall.MagicMock(name="mock_path")
# patch
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'Path', mock_path)
path = '/home/pi/dev/bossjones-github/scarlett_os/_debug/generator.dot'
mock_path_instance = mock_path()
mock_path_instance.exists.return_value = False
# run test
result = s_path.fname_exists(path)
assert mock_path_instance.exists.call_count == 1
mock_path_instance.exists.assert_called_once_with()
mock_path.assert_any_call(path)
assert result == False
def test_dir_isWritable(self, path_mocker_stopall):
# mock
mock_os_access = path_mocker_stopall.MagicMock(name="mock_os_access")
mock_os_path_isdir = path_mocker_stopall.MagicMock(name="mock_os_path_isdir")
# patch
path_mocker_stopall.patch.object(scarlett_os.internal.path.os, 'access', mock_os_access)
path_mocker_stopall.patch.object(scarlett_os.internal.path.os.path, 'isdir', mock_os_path_isdir)
path = 'file:///tmp'
# patch return values
mock_os_path_isdir.return_value = True
mock_os_access.return_value = True
# run test
result = s_path.isWritable(path)
# tests
mock_os_path_isdir.assert_called_once_with('file:///tmp')
mock_os_access.assert_called_once_with('file:///tmp', os.W_OK)
assert result == True
def test_unicode_decode_error_isWritable(self, path_mocker_stopall):
# mock
mock_os_path_isdir = path_mocker_stopall.MagicMock(name="mock_os_path_isdir")
mock_os_access = path_mocker_stopall.MagicMock(name="mock_os_access")
mock_unicode_error_dialog = path_mocker_stopall.MagicMock(name="mock_unicode_error_dialog")
mock_logger_error = path_mocker_stopall.MagicMock(name="mock_logger_error")
# patch
path_mocker_stopall.patch.object(scarlett_os.internal.path.os.path, 'isdir', mock_os_path_isdir)
path_mocker_stopall.patch.object(scarlett_os.internal.path.os, 'access', mock_os_access)
path_mocker_stopall.patch.object(scarlett_os.internal.path, 'unicode_error_dialog', mock_unicode_error_dialog)
path_mocker_stopall.patch.object(scarlett_os.subprocess.logging.Logger, 'error', mock_logger_error)
path = b'file:///tmp/fake_file'
# patch return values
mock_os_path_isdir.side_effect = UnicodeDecodeError('', b'', 1, 0, '')
s_path.isWritable(path)
assert mock_unicode_error_dialog.call_count == 1
def test_unicode_error_dialog(self, path_mocker_stopall):
# mock
mock_logger_error = path_mocker_stopall.MagicMock(name="mock_logger_error")
# patch
path_mocker_stopall.patch.object(scarlett_os.subprocess.logging.Logger, 'error', mock_logger_error)
s_path.unicode_error_dialog()
assert mock_logger_error.call_count == 1
_message = _("The system's locale that you are using is not UTF-8 capable. "
"Unicode support is required for Python3 software like Pitivi. "
"Please correct your system settings; if you try to use Pitivi "
"with a broken locale, weird bugs will happen.")
mock_logger_error.assert_any_call(_message)
def __get_result_set_from_mock(mock, query):
result_wrapper = Mock()
influx_points = []
result = re.search(r"(\d{19}).*?(\d{19})", query)
start, end = result.groups()
# extract range
points = list(filter(lambda point: point["time"] > int(start) and point["time"] < int(end),
mock.points))
country_result = re.search(r"\"country\"=\'(\w+)\'", query)
if country_result:
country = country_result.groups()[0]
points = list(filter(lambda point: point["tags"]["country"] == country, points))
for point in points:
d = {**point["fields"], **point.get("tags", {})}
d["time"] = datetime.utcfromtimestamp(point["time"] // 1000000000).strftime('%Y-%m-%dT%H:%M:%SZ')
influx_points.append(d)
result_wrapper.get_points.return_value = influx_points
return result_wrapper
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))
self.assertTrue(isinstance(str(ret), str))
self.assertTrue(isinstance(dict(ret), dict))
mock.assert_called_with(
'http://localhost:8000/api/{}/{}/1/'.format(
self.app,
self.name.replace('_', '-')
),
headers=HEADERS
)
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
)
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)