Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
{'images': {'65': {'image_id': '65', 'uri': '65.jpg'}}},
{'images': {'37': {'image_id': '37', 'uri': '37.jpg'}, '79': {'image_id': '79', 'uri': '79.jpg'}}},
)
response = self.client.call_action(
'company_service',
'get_all_companies',
body={},
expansions={'company_type': ['ceo', 'ceo.photo', 'logo']},
)
mock_get_companies.assert_called_once_with({})
mock_get_employees.assert_called_once_with({'employee_ids': AnyOrderEqualsList(['5791', '51'])})
mock_get_images.assert_has_calls(
[
mock.call({'image_ids': ['65']}),
mock.call({'image_ids': AnyOrderEqualsList(['37', '79'])}),
],
)
expected_response = {
'companies': [
{
'_type': 'company_type',
'company_id': '9183',
'name': 'Acme',
'ceo_id': '5791',
'ceo': {
'_type': 'employee_type',
'employee_id': '5791',
'name': 'Julia',
'photo_id': '37',
def test_send_tcp(self):
with mock.patch.object(socket.socket, 'connect') as mock_connect:
handler = SyslogHandler(
address=('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT),
socket_type=socket.SOCK_STREAM,
)
mock_connect.assert_called_once_with(('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT))
with mock.patch.object(socket.socket, 'sendall') as mock_send_all:
handler._send(['this is the first part', 'here is another part', 'one more part'])
mock_send_all.assert_has_calls([
mock.call('this is the first part'),
mock.call('here is another part'),
mock.call('one more part'),
])
def test_send_unix(self):
with mock.patch.object(socket.socket, 'connect') as mock_connect:
handler = SyslogHandler(address='/path/to/unix.socket')
mock_connect.assert_called_once_with('/path/to/unix.socket')
with mock.patch.object(socket.socket, 'send') as mock_send:
handler._send(['this is the first part', 'here is another part', 'one more part'])
mock_send.assert_has_calls([
mock.call('this is the first part'),
mock.call('here is another part'),
mock.call('one more part'),
])
handler._send(['this is the first part', 'here is another part', 'one more part'])
finally:
mock.patch.stopall()
m = mock_sends['first_mock_send']
assert m is not None
m.assert_has_calls([
mock.call('this is the first part'),
mock.call('here is another part'),
])
m = mock_sends['second_mock_send']
assert m is not None
m.assert_has_calls([
mock.call('here is another part'),
mock.call('one more part'),
])
mock_reconnect.assert_called_once_with('/path/to/a/different.socket')
mock_close.assert_called_once_with()
m = first_mock_send_patch.start()
m.side_effect = [True, OSError()]
mock_sends['first_mock_send'] = m
mock_close.side_effect = close_side_effect
mock_reconnect.side_effect = connect_side_effect
handler._send(['this is the first part', 'here is another part', 'one more part'])
finally:
mock.patch.stopall()
m = mock_sends['first_mock_send']
assert m is not None
m.assert_has_calls([
mock.call('this is the first part'),
mock.call('here is another part'),
])
m = mock_sends['second_mock_send']
assert m is not None
m.assert_has_calls([
mock.call('here is another part'),
mock.call('one more part'),
])
mock_reconnect.assert_called_once_with('/path/to/a/different.socket')
mock_close.assert_called_once_with()
def test_send_tcp(self):
with mock.patch.object(socket.socket, 'connect') as mock_connect:
handler = SyslogHandler(
address=('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT),
socket_type=socket.SOCK_STREAM,
)
mock_connect.assert_called_once_with(('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT))
with mock.patch.object(socket.socket, 'sendall') as mock_send_all:
handler._send(['this is the first part', 'here is another part', 'one more part'])
mock_send_all.assert_has_calls([
mock.call('this is the first part'),
mock.call('here is another part'),
mock.call('one more part'),
])
def test_send_unix(self):
with mock.patch.object(socket.socket, 'connect') as mock_connect:
handler = SyslogHandler(address='/path/to/unix.socket')
mock_connect.assert_called_once_with('/path/to/unix.socket')
with mock.patch.object(socket.socket, 'send') as mock_send:
handler._send(['this is the first part', 'here is another part', 'one more part'])
mock_send.assert_has_calls([
mock.call('this is the first part'),
mock.call('here is another part'),
mock.call('one more part'),
])
def test_send_udp(self):
handler = SyslogHandler(address=('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT))
with mock.patch.object(socket.socket, 'sendto') as mock_send_to:
handler._send(['this is the first part', 'here is another part', 'one more part'])
mock_send_to.assert_has_calls([
mock.call('this is the first part', ('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT)),
mock.call('here is another part', ('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT)),
mock.call('one more part', ('127.0.0.1', logging.handlers.SYSLOG_UDP_PORT)),
])
mock.patch.object(socket.socket, 'connect') as mock_reconnect:
m = first_mock_send_patch.start()
m.side_effect = [True, OSError()]
mock_sends['first_mock_send'] = m
mock_close.side_effect = close_side_effect
mock_reconnect.side_effect = connect_side_effect
handler._send(['this is the first part', 'here is another part', 'one more part'])
finally:
mock.patch.stopall()
m = mock_sends['first_mock_send']
assert m is not None
m.assert_has_calls([
mock.call('this is the first part'),
mock.call('here is another part'),
])
m = mock_sends['second_mock_send']
assert m is not None
m.assert_has_calls([
mock.call('here is another part'),
mock.call('one more part'),
])
mock_reconnect.assert_called_once_with('/path/to/a/different.socket')
mock_close.assert_called_once_with()