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)
python类mock()的实例源码
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_find_resources_yield_pages(self):
page_one = json.dumps(load_resource_blob("responses/report-list-twopage-0"))
page_two = json.dumps(load_resource_blob("responses/report-list-twopage-1"))
with requests_mock.mock() as m:
filter_str = "filter%5Bprogram%5D%5B%5D=foo"
m.get(HackerOneClient.BASE_URL + "/reports?" + filter_str, text=page_one)
m.get(HackerOneClient.BASE_URL + "/reports/nextpage", text=page_two)
client = HackerOneClient("a", "b")
listing = client.find_resources(Report, program=["foo"], yield_pages=True)
# The listing should contain two pages each with one report
self.assertEqual(len(listing), 2)
# Flatten the pages into a single page
all_items = list(itertools.chain(*listing))
self.assertEqual(len(all_items), 2)
self.assertTrue(all(isinstance(r, Report) for r in all_items))
def test_get_nonexistent_alert_condition(self, mock):
policy_name = 'IAmNotHerePolicy'
policies_response = {
'policies': []
}
mock.get(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=policies_response,
status_code=200
)
# we don't even need a second endpoint here
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemNotFoundError,
'{}'.format(policy_name)
):
self.client.get_alert_conditions(policy_name)
def test_create_synthetics_alert_condition_no_policy(self, mock):
# Alerts policies list mock
response = {
'policies': []
}
mock.get(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=response,
status_code=200
)
# We don't need other endpoints
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemNotFoundError,
'{}'.format(self.policy_name)
):
self.client.create_synthetics_alert_condition(
policy_name=self.policy_name,
condition_name=self.monitor_name,
monitor_name=self.monitor_name,
)
def test_delete_synthetics_alert_condition_no_policy(self, mock):
response = {
'policies': []
}
# Synthetics monitors list mock
mock.get(
'https://synthetics.newrelic.com/synthetics/api/v3/monitors',
status_code=200,
json=self.monitor_response
)
# Alerts policies list mock
mock.get(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=response,
status_code=200
)
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemNotFoundError,
'{}'.format(self.policy_name)
):
self.client.delete_synthetics_alert_conditions(
policy_name=self.policy_name,
monitor_name=self.monitor_name,
)
def test_delete_synthetics_alert_condition_no_monitor(self, mock):
response = {
'monitors': []
}
# Synthetics monitors list mock
mock.get(
'https://synthetics.newrelic.com/synthetics/api/v3/monitors',
status_code=200,
json=response
)
# Alerts policies list mock
mock.get(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=self.policy_response,
status_code=200
)
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemNotFoundError,
'{}'.format(self.monitor_name)
):
self.client.delete_synthetics_alert_conditions(
policy_name=self.policy_name,
monitor_name=self.monitor_name,
)
def test_create_alert_policy_with_incident_preference(self, mock):
# Now we don't have any policies
response = {
'policies': []
}
mock.get(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=response,
status_code=200
)
# Policy created successfully
policy_created_response = {
'policy': self.first_policy
}
mock.post(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=policy_created_response,
status_code=201
)
created_policy = self.client.create_alert_policy(
self.first_policy['name'],
incident_preference=self.first_policy['incident_preference']
)
self.assertDictEqual(created_policy, self.first_policy)
def test_create_alert_policy_duplicate_success(self, mock):
mock.get(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=self.policy_response,
status_code=200
)
# Policy already exist
policy_created_response = {
'policy': self.first_policy
}
mock.post(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=policy_created_response,
status_code=201
)
created_policy = self.client.create_alert_policy(
self.first_policy['name'],
check_unique=False
)
self.assertDictEqual(created_policy, self.first_policy)
def test_create_alert_policy_duplicate_failure(self, mock):
mock.get(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=self.policy_response,
status_code=200
)
# Policy already exist
policy_created_response = {
'policy': self.first_policy
}
mock.post(
'{}/v2/alerts_policies.json'.format(self.client.base_url),
json=policy_created_response,
status_code=201
)
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemAlreadyExistsError,
'{}'.format(self.first_policy['name'])
):
self.client.create_alert_policy(self.first_policy['name'])
def test_delete_nonexistent_alert_policy(self, mock):
# Now we don't have any policies
response = {
'policies': []
}
mock.get(
'{}/v2/alerts_policies.json?filter[name]={}'
.format(self.client.base_url, self.first_policy['name']),
json=response,
status_code=200
)
mock.delete(
'{}/v2/alerts_policies/{}.json'
.format(self.client.base_url, self.first_policy['id'])
)
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemNotFoundError,
'{}'.format(self.first_policy['name'])
):
self.client.delete_alert_policy(self.first_policy['name'])
def test_get_other_error_with_message(self, mock):
url = '{}/v2/i_am_bogus'.format(self.client.base_url)
mock.get(
url=url,
status_code=400,
json={
'errors': [
{'error': 'I am bogus'},
{'error': 'I am verbose'}
]
}
)
with self.assertRaisesRegexp(
newrelic_cli.exceptions.NewRelicException,
'The following errors were returned by server'
):
self.client._get(url)
def test_create_monitor_with_sla_success(self, mock):
create_monitor_endpoint = '{}/v3/monitors'.format(self.client.base_url)
monitor_data = self.first_monitor
monitor_data['slaThreshold'] = 42
mock.post(
url=create_monitor_endpoint,
status_code=201,
headers={'Location': self.monitor_location}
)
mock.get(
url=self.monitor_location,
status_code=200,
json=monitor_data
)
r = self.client.create_monitor(self.monitor_name, slaThreshold=42)
self.assertDictEqual(r, self.first_monitor)
def test_delete_monitor_success(self, mock):
get_all_monitors_endpoint = '{}/v3/monitors'.format(
self.client.base_url
)
monitor_endpooint = '{}/v3/monitors/{}'.format(
self.client.base_url,
self.monitor_id
)
mock.get(
url=get_all_monitors_endpoint,
status_code=200,
json=self.all_monitors_response
)
mock.delete(
url=monitor_endpooint,
status_code=204
)
# We don't expect any response here.
# Just make sure no exceptions raised
self.client.delete_monitor(self.monitor_name)
def test_delete_nonexistent_monitor(self, mock):
get_all_monitors_endpoint = '{}/v3/monitors'.format(
self.client.base_url
)
monitor_endpooint = '{}/v3/monitors/{}'.format(
self.client.base_url,
self.monitor_id
)
# we don't have any monitors
monitors_list = {'monitors': []}
mock.get(
url=get_all_monitors_endpoint,
status_code=200,
json=monitors_list
)
mock.delete(
url=monitor_endpooint,
status_code=204
)
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemNotFoundError,
'{}'.format(self.monitor_name)
):
self.client.delete_monitor(self.monitor_name)
def test_get_monitor_script_success(self, mock):
get_all_monitors_endpoint = '{}/v3/monitors'.format(
self.client.base_url
)
monitor_script_endpoint = '{}/v3/monitors/{}/script'.format(
self.client.base_url,
self.monitor_id
)
mock.get(
url=get_all_monitors_endpoint,
status_code=200,
json=self.all_monitors_response
)
mock.get(
url=monitor_script_endpoint,
status_code=200,
json={'scriptText': self.monitor_script_base64}
)
script = self.client.get_monitor_script(self.monitor_name)
self.assertEquals(self.monitor_script_plain, script)
def test_get_monitor_without_script(self, mock):
get_all_monitors_endpoint = '{}/v3/monitors'.format(
self.client.base_url
)
monitor_script_endpoint = '{}/v3/monitors/{}/script'.format(
self.client.base_url,
self.monitor_id
)
mock.get(
url=get_all_monitors_endpoint,
status_code=200,
json=self.all_monitors_response
)
mock.get(
url=monitor_script_endpoint,
status_code=404,
)
with self.assertRaisesRegexp(
newrelic_cli.exceptions.ItemNotFoundError,
'{}'.format(self.monitor_name)
):
self.client.get_monitor_script(self.monitor_name)
def test_upload_monitor_script_success(self, mock):
get_all_monitors_endpoint = '{}/v3/monitors'.format(
self.client.base_url
)
monitor_script_endpoint = '{}/v3/monitors/{}/script'.format(
self.client.base_url,
self.monitor_id
)
mock.get(
url=get_all_monitors_endpoint,
status_code=200,
json=self.all_monitors_response
)
mock.put(
url=monitor_script_endpoint,
status_code=204
)
# We don't expect any response here.
# Just make sure no exceptions raised
self.client.upload_monitor_script(
self.monitor_name,
self.monitor_script_plain
)
def test_handle_message_should_add_channel_cmd_buff():
event = {'content': 'hi!',
'channel': 'mychannel',
'sender': {'id': 'abcd1234',
'name': 'myname'}}
context = {'channel': {'endpoint': 'http://localhost'}}
with requests_mock.mock() as m:
m.post('http://localhost/messages')
response = handle_message(event, context, Bot)
assert response['response'] is None
assert m.called
assert m.request_history[0].json() == {
'channel': None,
'receiver': None,
'message': 'hello, you said: hi!',
'event': event,
'extra': None,
'context': {'api_key': '', 'project_id': None}
}
def test_lifecycle_installed_multiple_no_auth(self):
"""Multiple requests should fail unless auth is provided the second time"""
with requests_mock.mock() as m:
m.get('https://gavindev.atlassian.net/plugins/servlet/oauth/consumer-info',
text=consumer_info_response)
client = dict(
baseUrl='https://gavindev.atlassian.net',
clientKey='abc123',
publicKey='public123')
rv = self.client.post('/atlassian_connect/lifecycle/installed',
data=json.dumps(client),
content_type='application/json')
self.assertEqual(204, rv.status_code)
rv = self.client.post('/atlassian_connect/lifecycle/installed',
data=json.dumps(client),
content_type='application/json')
self.assertEqual(401, rv.status_code)
def test_liga_com_nome(self):
# Arrange and Act
with requests_mock.mock() as m:
url = '{api_url}/auth/liga/{slug}'.format(api_url=self.api_url, slug='falydos-fc')
m.get(url, text=self.LIGA)
liga = self.api.liga(nome='Falydos FC')
primeiro_time = liga.times[0]
# Assert
self.assertIsInstance(liga, Liga)
self.assertEqual(liga.id, 6407)
self.assertEqual(liga.nome, 'Virtus Premier League')
self.assertEqual(liga.slug, 'virtus-premier-league')
self.assertEqual(liga.descricao,
u'Prêmios para: \n\n- Melhor de cada Mês (R$50,00)\n- Melhor do 1º e 2º Turno (R$150,00)\n- 2º Lugar Geral (R$50)\n- 1º Lugar Geral (R$250,00)\n\nBoa sorte!')
self.assertIsInstance(liga.times, list)
self.assertIsInstance(primeiro_time, TimeInfo)
self.assertEqual(primeiro_time.id, 453420)
self.assertEqual(primeiro_time.nome, 'Mosqueteiros JPB')
self.assertEqual(primeiro_time.nome_cartola, 'Erick Costa')
self.assertEqual(primeiro_time.slug, 'mosqueteiros-jpb')
self.assertTrue(primeiro_time.assinante)
def test_pontuacao_atleta(self):
# Arrange and Act
with requests_mock.mock() as m:
url = '{api_url}/auth/mercado/atleta/{id}/pontuacao'.format(api_url=self.api_url, id=81682)
m.get(url, text=self.PONTUACAO_ATLETA)
pontuacoes = self.api.pontuacao_atleta(81682)
primeira_rodada = pontuacoes[0]
# Assert
self.assertIsInstance(pontuacoes, list)
self.assertIsInstance(primeira_rodada, PontuacaoInfo)
self.assertEqual(primeira_rodada.atleta_id, 81682)
self.assertEqual(primeira_rodada.rodada_id, 1)
self.assertEqual(primeira_rodada.pontos, 1.1)
self.assertEqual(primeira_rodada.preco, 6.44)
self.assertEqual(primeira_rodada.variacao, -1.56)
self.assertEqual(primeira_rodada.media, 1.1)
def test_ligas(self):
# Arrange and Act
with requests_mock.mock() as m:
url = '{api_url}/ligas'.format(api_url=self.api_url)
m.get(url, text=self.LIGAS)
ligas = self.api.ligas(query='premiere')
primeira_liga = ligas[0]
# Assert
self.assertIsInstance(ligas, list)
self.assertIsInstance(primeira_liga, Liga)
self.assertEqual(primeira_liga.id, 36741)
self.assertEqual(primeira_liga.nome, 'PREMIERE_LIGA_ENTEL')
self.assertEqual(primeira_liga.slug, 'premiere-liga-entel')
self.assertEqual(primeira_liga.descricao, u'“Vale tudo, só não vale...”')
self.assertIsNone(primeira_liga.times)
def test_mercado_atletas(self):
# Arrange and Act
with requests_mock.mock() as m:
url = '{api_url}/atletas/mercado'.format(api_url=self.api_url)
m.get(url, text=self.MERCADO_ATLETAS)
mercado = self.api.mercado_atletas()
primeiro_atleta = mercado[0]
# Assert
self.assertIsInstance(mercado, list)
self.assertIsInstance(primeiro_atleta, Atleta)
self.assertEqual(primeiro_atleta.id, 86935)
self.assertEqual(primeiro_atleta.apelido, 'Rodrigo')
self.assertEqual(primeiro_atleta.pontos, 0)
self.assertEqual(primeiro_atleta.scout, {'CA': 1, 'FC': 3, 'FS': 1, 'PE': 2, 'RB': 2})
self.assertEqual(primeiro_atleta.posicao, _posicoes[4])
self.assertIsInstance(primeiro_atleta.clube, Clube)
self.assertEqual(primeiro_atleta.clube.id, 292)
self.assertEqual(primeiro_atleta.clube.nome, 'Sport')
self.assertEqual(primeiro_atleta.clube.abreviacao, 'SPO')
self.assertEqual(primeiro_atleta.status, _atleta_status[6])
def test_parciais_mercado_fechado(self):
# Arrange
with requests_mock.mock() as m:
url = '{api_url}/mercado/status'.format(api_url=self.api_url)
m.get(url, text=self.MERCADO_STATUS_FECHADO)
url = '{api_url}/atletas/pontuados'.format(api_url=self.api_url)
m.get(url, text=self.PARCIAIS)
# Act
parciais = self.api.parciais()
parcial_juan = parciais[36540]
# Assert
self.assertIsInstance(parciais, dict)
self.assertIsInstance(parcial_juan, Atleta)
self.assertEqual(parcial_juan.id, 36540)
self.assertEqual(parcial_juan.apelido, 'Juan')
self.assertEqual(parcial_juan.pontos, 2.9)
self.assertEqual(parcial_juan.scout, {'CA': 1, 'FC': 1, 'FS': 2, 'PE': 2, 'SG': 1})
self.assertEqual(parcial_juan.posicao, _posicoes[3])
self.assertIsInstance(parcial_juan.clube, Clube)
self.assertEqual(parcial_juan.clube.id, 262)
self.assertEqual(parcial_juan.clube.nome, 'Flamengo')
self.assertEqual(parcial_juan.clube.abreviacao, 'FLA')
def test_pos_rodada_destaques_com_mercado_aberto(self):
# Arrange and Act
with requests_mock.mock() as m:
url = '{api_url}/mercado/status'.format(api_url=self.api_url)
m.get(url, text=self.MERCADO_STATUS_ABERTO)
url = '{api_url}/pos-rodada/destaques'.format(api_url=self.api_url)
m.get(url, text=self.POS_RODADA_DESTAQUES)
destaque_rodada = self.api.pos_rodada_destaques()
# Assert
self.assertIsInstance(destaque_rodada, DestaqueRodada)
self.assertEqual(destaque_rodada.media_cartoletas, 115.8235753058391)
self.assertEqual(destaque_rodada.media_pontos, 46.6480728839843)
self.assertIsInstance(destaque_rodada.mito_rodada, TimeInfo)
self.assertEqual(destaque_rodada.mito_rodada.id, 896224)
self.assertEqual(destaque_rodada.mito_rodada.nome, 'gama campos fc')
self.assertEqual(destaque_rodada.mito_rodada.nome_cartola, 'malmal')
self.assertEqual(destaque_rodada.mito_rodada.slug, 'gama-campos-fc')
self.assertFalse(destaque_rodada.mito_rodada.assinante)
def test_times(self):
# Arrange and Act
with requests_mock.mock() as m:
url = '{api_url}/times'.format(api_url=self.api_url)
m.get(url, text=self.TIMES)
times = self.api.times(query='Faly')
primeiro_time = times[0]
# Assert
self.assertIsInstance(times, list)
self.assertIsInstance(primeiro_time, TimeInfo)
self.assertEqual(primeiro_time.id, 4626963)
self.assertEqual(primeiro_time.nome, 'Falysson29')
self.assertEqual(primeiro_time.nome_cartola, 'Alysson')
self.assertEqual(primeiro_time.slug, 'falysson29')
self.assertFalse(primeiro_time.assinante)
