本文整理汇总了Python中sentry.tasks.post_process.post_process_group函数的典型用法代码示例。如果您正苦于以下问题:Python post_process_group函数的具体用法?Python post_process_group怎么用?Python post_process_group使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了post_process_group函数的17个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_owner_assignment_existing_assignment
def test_owner_assignment_existing_assignment(self):
self.make_ownership()
event = self.store_event(
data={
'message': 'oh no',
'platform': 'python',
'stacktrace': {
'frames': [
{'filename': 'src/app/example.py'}
]
}
},
project_id=self.project.id
)
event.group.assignee_set.create(
team=self.team,
project=self.project)
post_process_group(
event=event,
is_new=False,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
assignee = event.group.assignee_set.first()
assert assignee.user is None
assert assignee.team == self.team
开发者ID:yaoqi,项目名称:sentry,代码行数:27,代码来源:tests.py
示例2: test_invalidates_snooze
def test_invalidates_snooze(self, mock_processor):
group = self.create_group(
project=self.project, status=GroupStatus.IGNORED)
event = self.create_event(group=group)
snooze = GroupSnooze.objects.create(
group=group,
until=timezone.now() - timedelta(hours=1),
)
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
is_new_group_environment=True,
)
mock_processor.assert_called_with(event, True, False, True, True)
assert not GroupSnooze.objects.filter(
id=snooze.id,
).exists()
group = Group.objects.get(id=group.id)
assert group.status == GroupStatus.UNRESOLVED
开发者ID:Kayle009,项目名称:sentry,代码行数:25,代码来源:tests.py
示例3: test_group_refresh
def test_group_refresh(self, mock_processor):
group1 = self.create_group(project=self.project)
group2 = self.create_group(project=self.project)
event = self.create_event(group=group1)
assert event.group_id == group1.id
assert event.group == group1
with self.tasks():
merge_group(group1.id, group2.id)
mock_callback = Mock()
mock_futures = [Mock()]
mock_processor.return_value.apply.return_value = [
(mock_callback, mock_futures),
]
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
is_new_group_environment=True,
)
assert event.group == group2
assert event.group_id == group2.id
开发者ID:hosmelq,项目名称:sentry,代码行数:28,代码来源:tests.py
示例4: test_record_affected_code
def test_record_affected_code(self, mock_record_affected_code):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
with self.settings(SENTRY_ENABLE_EXPLORE_CODE=False):
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert not mock_record_affected_code.delay.called
with self.settings(SENTRY_ENABLE_EXPLORE_CODE=True):
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
mock_record_affected_code.delay.assert_called_once_with(
event=event,
)
开发者ID:carriercomm,项目名称:sentry-1,代码行数:25,代码来源:tests.py
示例5: test_service_hook_fires_on_alert
def test_service_hook_fires_on_alert(self, mock_processor, mock_process_service_hook):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
mock_callback = Mock()
mock_futures = [Mock()]
mock_processor.return_value.apply.return_value = [
(mock_callback, mock_futures),
]
hook = ServiceHook.objects.create(
project_id=self.project.id,
actor_id=self.user.id,
events=['event.alert'],
)
with self.feature('projects:servicehooks'):
post_process_group(
event=event,
is_new=False,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
mock_process_service_hook.delay.assert_called_once_with(
servicehook_id=hook.id,
event=event,
)
开发者ID:hosmelq,项目名称:sentry,代码行数:30,代码来源:tests.py
示例6: test_record_affected_user
def test_record_affected_user(self, mock_record_affected_user):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
mock_record_affected_user.delay.assert_called_once_with(
event=event,
)
开发者ID:carriercomm,项目名称:sentry-1,代码行数:14,代码来源:tests.py
示例7: test_processes_resource_change_task_on_new_group
def test_processes_resource_change_task_on_new_group(self, delay):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
delay.assert_called_once_with(
action='created',
sender='Group',
instance_id=group.id,
)
开发者ID:yaoqi,项目名称:sentry,代码行数:17,代码来源:tests.py
示例8: test_maintains_valid_snooze
def test_maintains_valid_snooze(self):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
snooze = GroupSnooze.objects.create(
group=group,
until=timezone.now() + timedelta(hours=1),
)
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert GroupSnooze.objects.filter(
id=snooze.id,
).exists()
开发者ID:alshopov,项目名称:sentry,代码行数:18,代码来源:tests.py
示例9: test_service_hook_does_not_fire_without_event
def test_service_hook_does_not_fire_without_event(self, mock_process_service_hook):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
ServiceHook.objects.create(
project_id=self.project.id,
actor_id=self.user.id,
events=[],
)
with self.feature('projects:servicehooks'):
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
assert not mock_process_service_hook.delay.mock_calls
开发者ID:hosmelq,项目名称:sentry,代码行数:20,代码来源:tests.py
示例10: test_owner_assignment_ownership_does_not_exist
def test_owner_assignment_ownership_does_not_exist(self):
event = self.store_event(
data={
'message': 'oh no',
'platform': 'python',
'stacktrace': {
'frames': [
{'filename': 'src/app/example.py'}
]
}
},
project_id=self.project.id
)
post_process_group(
event=event,
is_new=False,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
assert not event.group.assignee_set.exists()
开发者ID:yaoqi,项目名称:sentry,代码行数:21,代码来源:tests.py
示例11: test_rule_processor
def test_rule_processor(self, mock_processor):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
mock_callback = Mock()
mock_futures = [Mock()]
mock_processor.return_value.apply.return_value = [
(mock_callback, mock_futures),
]
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
mock_processor.assert_called_once_with(event, True, False, False)
mock_processor.return_value.apply.assert_called_once_with()
mock_callback.assert_called_once_with(event, mock_futures)
开发者ID:carriercomm,项目名称:sentry-1,代码行数:22,代码来源:tests.py
示例12: test_service_hook_fires_on_new_event
def test_service_hook_fires_on_new_event(self, mock_process_service_hook):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
hook = ServiceHook.objects.create(
project_id=self.project.id,
actor_id=self.user.id,
events=['event.created'],
)
with self.feature('projects:servicehooks'):
post_process_group(
event=event,
is_new=False,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
mock_process_service_hook.delay.assert_called_once_with(
servicehook_id=hook.id,
event=event,
)
开发者ID:hosmelq,项目名称:sentry,代码行数:23,代码来源:tests.py
示例13: test_group_created_sends_webhook
def test_group_created_sends_webhook(self, safe_urlopen):
issue = self.create_group(project=self.project)
event = self.create_event(group=issue)
with self.tasks():
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
data = json.loads(faux(safe_urlopen).kwargs['data'])
assert data['action'] == 'created'
assert data['installation']['uuid'] == self.install.uuid
assert data['data']['issue']['id'] == six.text_type(issue.id)
assert faux(safe_urlopen).kwargs_contain('headers.Content-Type')
assert faux(safe_urlopen).kwargs_contain('headers.Request-ID')
assert faux(safe_urlopen).kwargs_contain('headers.Sentry-Hook-Resource')
assert faux(safe_urlopen).kwargs_contain('headers.Sentry-Hook-Timestamp')
assert faux(safe_urlopen).kwargs_contain('headers.Sentry-Hook-Signature')
开发者ID:yaoqi,项目名称:sentry,代码行数:23,代码来源:test_sentry_apps.py
示例14: test_service_hook_does_not_fire_without_alert
def test_service_hook_does_not_fire_without_alert(
self, mock_processor, mock_process_service_hook):
group = self.create_group(project=self.project)
event = self.create_event(group=group)
mock_processor.return_value.apply.return_value = []
self.create_service_hook(
project=self.project,
organization=self.project.organization,
actor=self.user,
events=['event.alert'],
)
with self.feature('projects:servicehooks'):
post_process_group(
event=event,
is_new=False,
is_regression=False,
is_sample=False,
is_new_group_environment=False,
)
assert not mock_process_service_hook.delay.mock_calls
开发者ID:yaoqi,项目名称:sentry,代码行数:24,代码来源:tests.py
示例15: test_execute_rule
def test_execute_rule(self, mock_get_rules, mock_execute_rule):
action_id = 'sentry.rules.actions.notify_event.NotifyEventAction'
condition_id = 'sentry.rules.conditions.first_seen_event.FirstSeenEventCondition'
group = self.create_group(project=self.project)
event = self.create_event(group=group)
mock_get_rules.return_value = [
Rule(
id=1,
data={
'actions': [{'id': action_id}],
'conditions': [{'id': condition_id}],
}
)
]
post_process_group(
group=group,
event=event,
is_new=False,
is_regression=False,
is_sample=False,
)
mock_get_rules.assert_called_once_with(self.project)
assert not mock_execute_rule.delay.called
post_process_group(
group=group,
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert len(mock_execute_rule.delay.mock_calls) == 1
post_process_group(
group=group,
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert len(mock_execute_rule.delay.mock_calls) == 2
开发者ID:animeshinvinci,项目名称:sentry,代码行数:48,代码来源:tests.py
示例16: test_execute_rule
def test_execute_rule(self, mock_get_rules, mock_execute_rule):
action_id = 'sentry.rules.actions.notify_event.NotifyEventAction'
condition_id = 'sentry.rules.conditions.first_seen_event.FirstSeenEventCondition'
group = self.create_group(project=self.project)
event = self.create_event(group=group)
mock_get_rules.return_value = [
Rule(
id=1,
data={
'actions': [{
'id': 'sentry.rules.actions.notify_event.NotifyEventAction',
}],
'conditions': [{
'id': 'sentry.rules.conditions.first_seen_event.FirstSeenEventCondition',
}],
}
),
Rule(
id=2,
data={
'actions': [{
'id': 'sentry.rules.actions.notify_event_service.NotifyEventAction',
'service': 'mail',
}],
'conditions': [{
'id': 'sentry.rules.conditions.every_event.EveryEventCondition',
}],
}
),
]
post_process_group(
event=event,
is_new=False,
is_regression=False,
is_sample=False,
)
mock_get_rules.assert_called_once_with(self.project)
assert len(mock_execute_rule.apply_async.mock_calls) == 1
mock_execute_rule.apply_async.reset_mock()
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert len(mock_execute_rule.apply_async.mock_calls) == 2
mock_execute_rule.apply_async.reset_mock()
post_process_group(
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert len(mock_execute_rule.apply_async.mock_calls) == 2
开发者ID:BlueMoebius,项目名称:sentry,代码行数:65,代码来源:tests.py
示例17: test_execute_rule
def test_execute_rule(self, mock_get_rules, mock_execute_rule):
action_id = 'sentry.rules.actions.notify_event.NotifyEventAction'
condition_id = 'sentry.rules.conditions.first_seen_event.FirstSeenEventCondition'
group = self.create_group(project=self.project)
event = self.create_event(group=group)
mock_get_rules.return_value = [
Rule(
id=1,
data={
'actions': [{'id': action_id}],
'conditions': [{'id': condition_id}],
}
)
]
post_process_group(
group=group,
event=event,
is_new=False,
is_regression=False,
is_sample=False,
)
assert not mock_execute_rule.delay.called
post_process_group(
group=group,
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
mock_execute_rule.delay.assert_called_once_with(
rule_id=1,
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
# ensure we dont execute again since the object hasnt changed state
post_process_group(
group=group,
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert len(mock_execute_rule.mock_calls) == 1
# and finally test the behavior of cycling back to new
post_process_group(
group=group,
event=event,
is_new=False,
is_regression=False,
is_sample=False,
)
post_process_group(
group=group,
event=event,
is_new=True,
is_regression=False,
is_sample=False,
)
assert len(mock_execute_rule.mock_calls) == 2
开发者ID:avdata99,项目名称:sentry,代码行数:70,代码来源:tests.py
注:本文中的sentry.tasks.post_process.post_process_group函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论