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_wait_for_cwlogs(self):
c = crhelper.resource_helper.CfnResource()
c._context = MockContext
s = Mock()
c._wait_for_cwlogs(sleep=s)
s.assert_not_called()
MockContext.ms_remaining = 140000
c._wait_for_cwlogs(sleep=s)
s.assert_called_once()
def test_set_timeout(self):
c = crhelper.resource_helper.CfnResource()
c._context = MockContext()
def func():
return None
c._set_timeout()
t = threading.Timer(1000, func)
self.assertEqual(type(t), type(c._timer))
t.cancel()
c._timer.cancel()
def test_wrappers(self):
c = crhelper.resource_helper.CfnResource()
def func():
pass
for f in ["create", "update", "delete", "poll_create", "poll_update", "poll_delete"]:
self.assertEqual(None, getattr(c, "_%s_func" % f))
getattr(c, f)(func)
self.assertEqual(func, getattr(c, "_%s_func" % f))
def test_timeout(self, s):
c = crhelper.resource_helper.CfnResource()
c._timeout()
s.assert_called_with('FAILED', "Execution timed out")
def test_cfn_response(self):
c = crhelper.resource_helper.CfnResource()
event = test_events['Create']
c._send = Mock()
orig_pid = c.PhysicalResourceId
self.assertEqual(orig_pid, '')
c._cfn_response(event)
c._send.assert_called_once()
print("RID: [%s]" % [c.PhysicalResourceId])
self.assertEqual(True, c.PhysicalResourceId.startswith('test-stack-id_TestResourceId_'))
c._send = Mock()
c.PhysicalResourceId = 'testpid'
c._cfn_response(event)
c._send.assert_called_once()
self.assertEqual('testpid', c.PhysicalResourceId)
def test_call(self, cfn_response_mock):
c = crhelper.resource_helper.CfnResource()
event = test_events["Create"]
c.__call__(event, MockContext)
self.assertTrue(c._send_response)
cfn_response_mock.assert_called_once_with(event)
c._sam_local = True
c._poll_enabled = Mock(return_value=True)
c._polling_init = Mock()
c.__call__(event, MockContext)
c._polling_init.assert_not_called()
self.assertEqual(1, len(cfn_response_mock.call_args_list))
c._sam_local = False
c._send_response = False
c.__call__(event, MockContext)
c._polling_init.assert_called()
def test_cleanup_response(self):
c = crhelper.resource_helper.CfnResource()
c.Data = {"CrHelperPoll": 1, "CrHelperPermission": 2, "CrHelperRule": 3}
c._cleanup_response()
self.assertEqual({}, c.Data)
def test_setup_polling(self):
c = crhelper.resource_helper.CfnResource()
c._context = MockContext()
c._event = test_events["Update"]
c._lambda_client.add_permission = Mock()
c._events_client.put_rule = Mock(return_value={"RuleArn": "arn:aws:lambda:blah:blah:function:blah/blah"})
c._events_client.put_targets = Mock()
c._setup_polling()
c._events_client.put_targets.assert_called()
c._events_client.put_rule.assert_called()
c._lambda_client.add_permission.assert_called()