Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_attributes(dut, moku, attr, value):
'''
TODO Default test
'''
setattr(dut, attr, value)
dut.commit()
moku._write_regs.assert_called_with(ANY)
yield None
m_get_arcrc_path.return_value = ".arcrc"
git.api_url = "http://test/api/"
mozphab.conduit.set_repo(git)
m_open.side_effect = PermissionError
with pytest.raises(PermissionError):
save_api_token("abc")
m_chmod.reset_mock()
m_open.side_effect = (FileNotFoundError, with_open())
save_api_token("abc")
m_chmod.assert_called_once_with(".arcrc", 0o600)
m_json.dump.assert_called_once_with(
{"hosts": {git.api_url: {"token": "abc"}}}, mock.ANY, sort_keys=True, indent=2
)
m_chmod.reset_mock()
m_json.reset_mock()
m_open.side_effect = None
m_json.load.return_value = {"existing_key": "existing_value"}
save_api_token("abc")
m_json.dump.assert_called_once_with(
{"hosts": {git.api_url: {"token": "abc"}}, "existing_key": "existing_value"},
mock.ANY,
sort_keys=True,
indent=2,
)
m_chmod.assert_not_called()
m_json.reset_mock()
def test_get_assertion(self):
ctap = CTAP2(mock.MagicMock())
ctap.device.call.return_value = b"\0" + _GA_RESP
resp = ctap.get_assertion(1, 2)
ctap.device.call.assert_called_with(
0x10, b"\2" + cbor.encode({1: 1, 2: 2}), mock.ANY, None
)
self.assertIsInstance(resp, AssertionResponse)
self.assertEqual(resp, _GA_RESP)
self.assertEqual(resp.credential, _CRED)
self.assertEqual(resp.auth_data, _AUTH_DATA_GA)
self.assertEqual(resp.signature, _SIGNATURE)
self.assertIsNone(resp.user)
self.assertIsNone(resp.number_of_credentials)
def test_list_with_formatting(self, mock_echo, mock_index, mock_version):
rsp = [self._get_patch()]
mock_index.return_value = rsp
runner = CLIRunner()
result = runner.invoke(patch.list_cmd, [
'--format', 'simple', '--column', 'ID', '--column', 'Name'])
assert result.exit_code == 0, result
mock_echo.assert_called_once_with(mock.ANY, ('ID', 'Name'),
fmt='simple')
# -*- coding: utf-8 -*-
import sys
import mock
from mock import call, ANY
import odooly
from ._common import XmlRpcTestCase
class TestInteract(XmlRpcTestCase):
server_version = '6.1'
startup_calls = (
call(ANY, 'db', ANY, verbose=ANY),
'db.server_version',
call(ANY, 'db', ANY, verbose=ANY),
call(ANY, 'common', ANY, verbose=ANY),
call(ANY, 'object', ANY, verbose=ANY),
call(ANY, 'report', ANY, verbose=ANY),
call(ANY, 'wizard', ANY, verbose=ANY),
'db.list',
)
def setUp(self):
super(TestInteract, self).setUp()
# Hide readline module
mock.patch.dict('sys.modules', {'readline': None}).start()
mock.patch('odooly.Client._globals', None).start()
mock.patch('odooly.Client._set_interactive', wraps=odooly.Client._set_interactive).start()
self.interact = mock.patch('odooly._interact', wraps=odooly._interact).start()
def assert_request_sent(self, headers=None, body=None, url='/', chunked=False):
if headers is None:
headers = {}
self.connection.urlopen.assert_called_once_with(
method=self.request.method,
url=url,
body=body,
headers=headers,
retries=ANY,
assert_same_host=False,
preload_content=False,
decode_content=False,
chunked=chunked,
)
worker_ctx = WorkerContext(container, service, DummyProvider("dispatch"))
custom_retry_policy = {'max_retries': 5}
event_dispatcher = EventDispatcher(retry_policy=custom_retry_policy).bind(
container, attr_name="dispatch")
event_dispatcher.setup()
service.dispatch = event_dispatcher.get_dependency(worker_ctx)
service.dispatch('eventtype', 'msg')
headers = event_dispatcher.get_message_headers(worker_ctx)
expected_args = ('msg',)
expected_kwargs = {
'exchange': ANY,
'routing_key': 'eventtype',
'headers': headers,
'declare': event_dispatcher.declare,
'retry': event_dispatcher.publisher_cls.retry,
'retry_policy': custom_retry_policy,
'compression': event_dispatcher.publisher_cls.compression,
'mandatory': event_dispatcher.publisher_cls.mandatory,
'expiration': event_dispatcher.publisher_cls.expiration,
'delivery_mode': event_dispatcher.publisher_cls.delivery_mode,
'priority': event_dispatcher.publisher_cls.priority,
'serializer': event_dispatcher.serializer,
}
assert mock_producer.publish.call_count == 1
args, kwargs = mock_producer.publish.call_args
assert args == expected_args
with requests_mock.Mocker() as mock:
add_org_options(mock, DF.options())
add_org_members(mock, DF.members())
allow_org_method(mock, "PUT", "memberships/member3")
org.members = dict(
member1=dict(
role="owner"
),
member2=dict(
role="member"
),
member3=dict(
role="member"
),
)
org.spy.put.assert_called_once_with(ANY, dict(
role="member",
))
params_for_present = deepcopy(PARAMS_FOR_PRESENT)
params_for_present['data']['serverHardwareName'] = "ServerHardwareName"
params_for_present['data']['serverProfileTemplateName'] = "TemplateA200"
mock_resource_compare.return_value = False
self.resource.data = profile_data
self.mock_ov_client.server_hardware.get_by.return_value = [FAKE_SERVER_HARDWARE]
self.mock_ov_client.server_profile_templates.get_by_name.return_value = None
self.mock_ov_client.api_version = 1200
self.mock_ansible_module.params = params_for_present
ServerProfileModule().run()
self.mock_ansible_module.fail_json.assert_called_once_with(exception=mock.ANY, msg="Informed Server Profile Template 'TemplateA200' not found")