Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
full_name="User 1")
pd_user2 = PagerdutyUser(user_id="PDUSER2",
full_name="User 2")
pd_user3 = PagerdutyUser(user_id="PDUSER3",
full_name="User 3")
pd_sched1 = PagerdutySchedule(name="Schedule1",
oncall_users=[pd_user1])
pd_sched2 = PagerdutySchedule(name="Schedule2",
oncall_users=[pd_user1, pd_user2, pd_user3]) # NOQA
attachment = SlackAttachment(
color="#36a64f",
fallback="*Schedule1* - User 1\n*Schedule2* - User 1, User 2, User 3", # NOQA
text="*Schedule1* - User 1\n*Schedule2* - User 1, User 2, User 3",
mrkdwn_in=["text"]
)
expected = call('chat.postMessage',
channel="chan1",
attachments=attachment,
as_user=False,
username="Currently On-Call",
icon_url="https://avatars.slack-edge.com/2015-07-31/8502215814_6662f69db3bed43d32e6_48.jpg") # NOQA
yield from self.send_response(self.slack_connection,
[pd_sched1, pd_sched2],
"chan1")
self.assertEqual(self.mock_api_call.mock_calls, [expected])
listen.listener_event("event", "test1")
listener.event.assert_has_calls(
[mock.call("test1"), mock.call("test1")], any_order=True
)
context_listener.event.assert_has_calls(
[mock.call(listen, "test1")], any_order=True
)
broken_listener.event.assert_has_calls([mock.call("test1")], any_order=True)
assert listener.event.call_count == 2
assert context_listener.event.call_count == 1
assert broken_listener.event.call_count == 1
listen.listener_event("non_existing_event", "test2")
listener.event.assert_has_calls(
[mock.call("test1"), mock.call("test1")], any_order=True
)
context_listener.event.assert_has_calls(
[mock.call(listen, "test1")], any_order=True
)
broken_listener.event.assert_has_calls([mock.call("test1")], any_order=True)
assert listener.event.call_count == 2
assert context_listener.event.call_count == 1
assert broken_listener.event.call_count == 1
def test_one_msg_one_good(self):
messages = [
{
"type": "message",
"user": "user1",
"channel": "C1",
"text": "!oncall",
}
]
yield from self.pd.process_message(messages)
expected = [call("C1")]
self.pd.handle_single_prefixed_message.assert_has_calls(expected)
def test_callback_multi(ezsp_f):
testcb = mock.MagicMock()
cbid1 = ezsp_f.add_callback(testcb)
ezsp_f.add_callback(testcb)
ezsp_f.handle_callback(1, 2, 3)
assert testcb.call_count == 2
ezsp_f.remove_callback(cbid1)
ezsp_f.handle_callback(4, 5, 6)
testcb.assert_has_calls(
[mock.call(1, 2, 3), mock.call(1, 2, 3), mock.call(4, 5, 6)]
)
def test_multiple_tickets_sentence(self):
self.sm.text = "Tickets ATL-1, ATL-2,,ATL-3\nOh, Also ATL-04 and ATL05"
yield from self.jira.process_message(self.sm)
expected_calls = [
call("https://jira.atlassian.com", "ATL-1"),
call("https://jira.atlassian.com", "ATL-2"),
call("https://jira.atlassian.com", "ATL-3"),
call("https://jira.atlassian.com", "ATL-04"),
]
self.assertEqual(self.mock_get_jira_issue_info.mock_calls,
expected_calls)
def test_two_msgs_two_good(self):
msg = SlackMessage(type="message",
user="U2147483697",
channel="C1",
text="!help fine, now I need help")
yield from self.hp.process_message(msg)
msg = SlackMessage(type="message",
user="U2147483697",
channel="C2",
text="!help Yep, still need help")
yield from self.hp.process_message(msg)
expected = [call('C1'), call('C2')]
self.hp.send_help_message.assert_has_calls(expected, any_order=True)
listener = mock.MagicMock(spec_set=["event"])
listen.add_listener(listener)
listen.add_listener(listener)
context_listener = mock.MagicMock(spec_set=["event"])
listen.add_context_listener(context_listener)
listen.listener_event("event", "test1")
listener.event.assert_has_calls(
[mock.call("test1"), mock.call("test1")], any_order=True
)
context_listener.event.assert_has_calls(
[mock.call(listen, "test1")], any_order=True
)
broken_listener.event.assert_has_calls([mock.call("test1")], any_order=True)
assert listener.event.call_count == 2
assert context_listener.event.call_count == 1
assert broken_listener.event.call_count == 1
listen.listener_event("non_existing_event", "test2")
listener.event.assert_has_calls(
[mock.call("test1"), mock.call("test1")], any_order=True
)
context_listener.event.assert_has_calls(
[mock.call(listen, "test1")], any_order=True
)
broken_listener.event.assert_has_calls([mock.call("test1")], any_order=True)
assert listener.event.call_count == 2
assert context_listener.event.call_count == 1
assert broken_listener.event.call_count == 1
mock_feed.return_value.update.return_value = "OK", [mock_entry_1]
with assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(
hass, geo_location.DOMAIN, CONFIG_WITH_CUSTOM_LOCATION
)
# Artificially trigger update.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
assert mock_feed.call_args == call((15.1, 25.2), URL, filter_radius=200.0)
def test_two_schedules_one_oncall_person(self):
pd_user1 = PagerdutyUser(user_id="PDUSER1",
full_name="User 1")
pd_sched1 = PagerdutySchedule(name="Schedule1",
oncall_users=[pd_user1])
pd_sched2 = PagerdutySchedule(name="Schedule2",
oncall_users=[])
attachment = SlackAttachment(
color="#36a64f",
fallback="*Schedule1* - User 1\n*Schedule2* - could not determine who's on call :disappointed:", # NOQA
text="*Schedule1* - User 1\n*Schedule2* - could not determine who's on call :disappointed:", # NOQA
mrkdwn_in=["text"]
)
expected = call('chat.postMessage',
channel="chan1",
attachments=attachment,
as_user=False,
username="Currently On-Call",
icon_url="https://avatars.slack-edge.com/2015-07-31/8502215814_6662f69db3bed43d32e6_48.jpg") # NOQA
yield from self.send_response(self.slack_connection,
[pd_sched1, pd_sched2],
"chan1")
self.assertEqual(self.mock_api_call.mock_calls, [expected])
async def test_agent_user_id_connect():
"""Test the connection and disconnection of users."""
config = MockConfig()
store = config._store
await config.async_connect_agent_user("agent_2")
assert store.add_agent_user_id.call_args == call("agent_2")
await config.async_connect_agent_user("agent_1")
assert store.add_agent_user_id.call_args == call("agent_1")
await config.async_disconnect_agent_user("agent_2")
assert store.pop_agent_user_id.call_args == call("agent_2")
await config.async_disconnect_agent_user("agent_1")
assert store.pop_agent_user_id.call_args == call("agent_1")