• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python post_process.post_process_group函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了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;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python post_process_group.delay函数代码示例发布时间:2022-05-27
下一篇:
Python merge_group.delay函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap