本文整理汇总了Python中sentry.digests.notifications.build_digest函数的典型用法代码示例。如果您正苦于以下问题：Python build_digest函数的具体用法？Python build_digest怎么用？Python build_digest使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了build_digest函数的17个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_notify_digest_single_record
def test_notify_digest_single_record(self, send, notify):
project = self.event.project
rule = project.rule_set.all()[0]
digest = build_digest(project, (event_to_record(self.event, (rule,)),))
self.plugin.notify_digest(project, digest)
assert send.call_count is 1
assert notify.call_count is 1
开发者ID:AyrtonRicardo，项目名称:sentry，代码行数:7，代码来源:tests.py
示例2: deliver_digest
def deliver_digest(key, schedule_timestamp=None):
from sentry.app import digests
plugin, project = split_key(key)
with digests.digest(key) as records:
digest = build_digest(project, records)
if digest:
plugin.notify_digest(project, digest)
开发者ID:haojiang1，项目名称:sentry，代码行数:9，代码来源:digests.py
示例3: deliver_digest
def deliver_digest(key, schedule_timestamp=None):
from sentry.app import digests
plugin, project = split_key(key)
minimum_delay = ProjectOption.objects.get_value(
project,
'{0}:digests:{1}'.format(plugin.get_conf_key(), 'minimum_delay'),
)
with digests.digest(key, minimum_delay=minimum_delay) as records:
digest = build_digest(project, records)
if digest:
plugin.notify_digest(project, digest)
开发者ID:jonashaag，项目名称:sentry，代码行数:13，代码来源:digests.py
示例4: test_simple
def test_simple(self):
rule = self.project.rule_set.all()[0]
records = [event_to_record(event, (rule, ))
for event in self.team1_events + self.team2_events + self.user4_events]
digest = build_digest(self.project, sort_records(records))
expected_result = {
self.user1.id: set(self.team1_events),
self.user2.id: set(self.team2_events),
self.user3.id: set(self.team1_events + self.team2_events),
self.user4.id: set(self.user4_events),
}
self.assert_get_personalized_digests(self.project, digest, self.user_ids, expected_result)
开发者ID:Kayle009，项目名称:sentry，代码行数:13，代码来源:test_utilities.py
示例5: test_notify_digest
def test_notify_digest(self, send_async, notify):
project = self.event.project
rule = project.rule_set.all()[0]
digest = build_digest(
project,
(
event_to_record(self.create_event(group=self.create_group()), (rule,)),
event_to_record(self.event, (rule,)),
),
)
self.plugin.notify_digest(project, digest)
assert send_async.call_count is 1
assert notify.call_count is 0
开发者ID:Cobbyzhang，项目名称:sentry，代码行数:13，代码来源:tests.py
示例6: test_everyone_with_owners
def test_everyone_with_owners(self):
rule = self.project.rule_set.all()[0]
events = self.create_events(
timezone.now(), self.project, [
'hello.moz', 'goodbye.moz', 'hola.moz', 'adios.moz'])
records = [event_to_record(event, (rule, )) for event in events + self.team1_events]
digest = build_digest(self.project, sort_records(records))
expected_result = {
self.user1.id: set(events + self.team1_events),
self.user2.id: set(events),
self.user3.id: set(events + self.team1_events),
self.user4.id: set(events),
self.user5.id: set(events),
}
self.assert_get_personalized_digests(self.project, digest, self.user_ids, expected_result)
开发者ID:Kayle009，项目名称:sentry，代码行数:15，代码来源:test_utilities.py
示例7: test_notify_digest_single_record
def test_notify_digest_single_record(self, has, send, notify):
has.side_effect = lambda label, *a, **k: {
'projects:digests:deliver': True,
}.get(label, False)
project = self.event.project
rule = project.rule_set.all()[0]
digest = build_digest(
project,
(
event_to_record(self.event, (rule,)),
),
)
self.plugin.notify_digest(project, digest)
assert send.call_count is 1
assert notify.call_count is 1
开发者ID:WilliamRen，项目名称:sentry，代码行数:16，代码来源:tests.py
示例8: test_notify_digest_subject_prefix
def test_notify_digest_subject_prefix(self):
project = self.event.project
rule = project.rule_set.all()[0]
digest = build_digest(
project,
(
event_to_record(self.create_event(group=self.create_group()), (rule,)),
event_to_record(self.event, (rule,)),
),
)
self.plugin.notify_digest(project, digest)
assert len(mail.outbox) == 1
msg = mail.outbox[0]
assert msg.subject.startswith('[Example prefix] [foo Bar]')
开发者ID:Andy-hpliu，项目名称:sentry，代码行数:17，代码来源:tests.py
示例9: deliver_digest
def deliver_digest(key, schedule_timestamp=None):
from sentry.app import digests
try:
plugin, project = split_key(key)
except Project.DoesNotExist as error:
logger.info("Cannot deliver digest %r due to error: %s", key, error)
digests.delete(key)
return
minimum_delay = ProjectOption.objects.get_value(
project, "{0}:digests:{1}".format(plugin.get_conf_key(), "minimum_delay")
)
with digests.digest(key, minimum_delay=minimum_delay) as records:
digest = build_digest(project, records)
if digest:
plugin.notify_digest(project, digest)
开发者ID:AyrtonRicardo，项目名称:sentry，代码行数:18，代码来源:digests.py
示例10: test_notify_digest
def test_notify_digest(self, notify):
project = self.event.project
rule = project.rule_set.all()[0]
digest = build_digest(
project,
(
event_to_record(self.create_event(group=self.create_group()), (rule, )),
event_to_record(self.event, (rule, )),
),
)
with self.tasks():
self.plugin.notify_digest(project, digest)
assert notify.call_count is 0
assert len(mail.outbox) == 1
message = mail.outbox[0]
assert 'List-ID' in message.message()
开发者ID:Kayle009，项目名称:sentry，代码行数:19，代码来源:tests.py
示例11: test_get_event_from_groups_in_digest
def test_get_event_from_groups_in_digest(self):
project = self.create_project()
rule = project.rule_set.all()[0]
same_group = self.create_group(project=project)
events = [
self.create_event(group=same_group),
self.create_event(group=same_group),
self.create_event(group=self.create_group(project=project)),
self.create_event(group=self.create_group(project=project)),
self.create_event(group=self.create_group(project=project)),
self.create_event(group=self.create_group(project=project)),
]
digest = build_digest(
project,
sort_records([event_to_record(event, (rule, )) for event in events]),
)
events.pop(0) # remove event with same group
assert get_event_from_groups_in_digest(digest) == set(events)
开发者ID:Kayle009，项目名称:sentry，代码行数:20，代码来源:test_utilities.py
示例12: test_team_without_members
def test_team_without_members(self):
team = self.create_team()
project = self.create_project(teams=[team])
ProjectOwnership.objects.create(
project_id=project.id,
schema=dump_schema([
Rule(Matcher('path', '*.cpp'), [
Owner('team', team.slug),
]),
]),
fallthrough=True,
)
rule = project.rule_set.all()[0]
records = [
event_to_record(event, (rule, )) for event in self.create_events(timezone.now(), project, [
'hello.py', 'goodbye.py', 'hola.py', 'adios.py'])
]
digest = build_digest(project, sort_records(records))
user_ids = [member.user_id for member in team.member_set]
assert not user_ids
for user_id, user_digest in get_personalized_digests(project.id, digest, user_ids):
assert False # no users in this team no digests should be processed
开发者ID:Kayle009，项目名称:sentry，代码行数:22，代码来源:test_utilities.py
示例13: deliver_digest
def deliver_digest(key, schedule_timestamp=None):
from sentry.app import digests
try:
plugin, project = split_key(key)
except Project.DoesNotExist as error:
logger.info('Cannot deliver digest %r due to error: %s', key, error)
digests.delete(key)
return
minimum_delay = ProjectOption.objects.get_value(
project,
get_option_key(plugin.get_conf_key(), 'minimum_delay')
)
try:
with digests.digest(key, minimum_delay=minimum_delay) as records:
digest = build_digest(project, records)
except InvalidState as error:
logger.info('Skipped digest delivery: %s', error, exc_info=True)
return
if digest:
plugin.notify_digest(project, digest)
开发者ID:280185386，项目名称:sentry，代码行数:24，代码来源:digests.py
示例14: digest
def digest(request):
seed = request.GET.get('seed', str(time.time()))
logger.debug('Using random seed value: %s')
random = Random(seed)
now = datetime.utcnow().replace(tzinfo=pytz.utc)
# TODO: Refactor all of these into something more manageable.
org = Organization(
id=1,
slug='example',
name='Example Organization',
)
team = Team(
id=1,
slug='example',
name='Example Team',
organization=org,
)
project = Project(
id=1,
slug='example',
name='Example Project',
team=team,
organization=org,
)
rules = {i: Rule(
id=i,
project=project,
label="Rule #%s" % (i,),
) for i in xrange(1, random.randint(2, 4))}
state = {
'project': project,
'groups': {},
'rules': rules,
'event_counts': {},
'user_counts': {},
}
records = []
group_sequence = itertools.count(1)
event_sequence = itertools.count(1)
for i in xrange(random.randint(1, 30)):
group_id = next(group_sequence)
group = state['groups'][group_id] = Group(
id=group_id,
project=project,
message='This is example event #%s' % (group_id,),
culprit='widget.foo in bar',
)
offset = timedelta(seconds=0)
for i in xrange(random.randint(1, 10)):
offset += timedelta(seconds=random.random() * 120)
event = Event(
id=next(event_sequence),
event_id=uuid.uuid4().hex,
project=project,
group=group,
message=group.message,
data=load_data('python'),
datetime=now - offset,
)
records.append(
Record(
event.event_id,
Notification(
event,
random.sample(state['rules'], random.randint(1, len(state['rules']))),
),
to_timestamp(event.datetime),
)
)
state['event_counts'][group_id] = random.randint(10, 1e4)
state['user_counts'][group_id] = random.randint(10, 1e4)
digest = build_digest(project, records, state)
# TODO(tkaemming): This duplication from ``MailPlugin.notify_digest`` is a code smell
counts = Counter()
for rule, groups in digest.iteritems():
counts.update(groups.keys())
return MailPreview(
html_template='sentry/emails/digests/body.html',
text_template='sentry/emails/digests/body.txt',
context={
'project': project,
'counts': counts,
'digest': digest,
},
#.........这里部分代码省略.........
开发者ID:zooming-tan，项目名称:sentry，代码行数:101，代码来源:mail.py
示例15: digest
def digest(request):
seed = request.GET.get('seed', str(time.time()))
logger.debug('Using random seed value: %s')
random = Random(seed)
now = datetime.utcnow().replace(tzinfo=pytz.utc)
# TODO: Refactor all of these into something more manageable.
org = Organization(
id=1,
slug='example',
name='Example Organization',
)
team = Team(
id=1,
slug='example',
name='Example Team',
organization=org,
)
project = Project(
id=1,
slug='example',
name='Example Project',
team=team,
organization=org,
)
rules = {i: Rule(
id=i,
project=project,
label="Rule #%s" % (i,),
) for i in xrange(1, random.randint(2, 4))}
state = {
'project': project,
'groups': {},
'rules': rules,
'event_counts': {},
'user_counts': {},
}
records = []
group_sequence = itertools.count(1)
event_sequence = itertools.count(1)
for i in xrange(random.randint(1, 30)):
group_id = next(group_sequence)
culprit = '{module} in {function}'.format(
module='.'.join(
''.join(random.sample(WORDS, random.randint(1, int(random.paretovariate(2.2))))) for word in xrange(1, 4)
),
function=random.choice(WORDS)
)
group = state['groups'][group_id] = Group(
id=group_id,
project=project,
message=words(int(random.weibullvariate(8, 4)), common=False),
culprit=culprit,
level=random.choice(LOG_LEVELS.keys()),
)
offset = timedelta(seconds=0)
for i in xrange(random.randint(1, 10)):
offset += timedelta(seconds=random.random() * 120)
event = Event(
id=next(event_sequence),
event_id=uuid.uuid4().hex,
project=project,
group=group,
message=group.message,
data=load_data('python'),
datetime=now - offset,
)
records.append(
Record(
event.event_id,
Notification(
event,
random.sample(state['rules'], random.randint(1, len(state['rules']))),
),
to_timestamp(event.datetime),
)
)
state['event_counts'][group_id] = random.randint(10, 1e4)
state['user_counts'][group_id] = random.randint(10, 1e4)
digest = build_digest(project, records, state)
start, end, counts = get_digest_metadata(digest)
return MailPreview(
html_template='sentry/emails/digests/body.html',
text_template='sentry/emails/digests/body.txt',
context={
'project': project,
#.........这里部分代码省略.........
开发者ID:280185386，项目名称:sentry，代码行数:101，代码来源:mail.py
示例16: digest
def digest(request):
seed = request.GET.get('seed', str(time.time()))
logger.debug('Using random seed value: %s')
random = Random(seed)
now = datetime.utcnow().replace(tzinfo=pytz.utc)
# TODO: Refactor all of these into something more manageable.
org = Organization(
id=1,
slug='example',
name='Example Organization',
)
team = Team(
id=1,
slug='example',
name='Example Team',
organization=org,
)
project = Project(
id=1,
slug='example',
name='Example Project',
team=team,
organization=org,
)
state = {
'project': project,
'groups': {},
'rules': {
1: Rule(id=1, project=project, label="First Rule"),
2: Rule(id=2, project=project, label="Second Rule"),
3: Rule(id=3, project=project, label="Third Rule"),
},
'event_counts': {},
'user_counts': {},
}
records = []
group_sequence = itertools.count(1)
event_sequence = itertools.count(1)
for i in xrange(random.randint(1, 4)):
group_id = next(group_sequence)
group = state['groups'][group_id] = Group(
id=group_id,
project=project,
message='This is example event #%s' % (group_id,),
)
offset = timedelta(seconds=0)
for i in xrange(random.randint(1, 10)):
offset += timedelta(seconds=random.random() * 120)
event = Event(
id=next(event_sequence),
event_id=uuid.uuid4().hex,
project=project,
group=group,
message=group.message,
data=load_data('python'),
datetime=now - offset,
)
records.append(
Record(
event.event_id,
Notification(
event,
random.sample(state['rules'], random.randint(1, len(state['rules']))),
),
to_timestamp(event.datetime),
)
)
state['event_counts'][group_id] = random.randint(10, 1e4)
state['user_counts'][group_id] = random.randint(10, 1e4)
digest = build_digest(project, records, state)
return MailPreview(
html_template='sentry/emails/digests/body.html',
text_template='sentry/emails/digests/body.txt',
context={
'project': project,
'digest': digest,
},
).render()
开发者ID:noah-lee，项目名称:sentry，代码行数:92，代码来源:mail.py
示例17: digest
def digest(request):
random = get_random(request)
# TODO: Refactor all of these into something more manageable.
org = Organization(
id=1,
slug='example',
name='Example Organization',
)
team = Team(
id=1,
slug='example',
name='Example Team',
organization=org,
)
project = Project(
id=1,
slug='example',
name='Example Project',
team=team,
organization=org,
)
rules = {i: Rule(
id=i,
project=project,
label="Rule #%s" % (i,),
) for i in range(1, random.randint(2, 4))}
state = {
'project': project,
'groups': {},
'rules': rules,
'event_counts': {},
'user_counts': {},
}
records = []
event_sequence = itertools.count(1)
group_generator = make_group_generator(random, project)
for i in range(random.randint(1, 30)):
group = next(group_generator)
state['groups'][group.id] = group
offset = timedelta(seconds=0)
for i in range(random.randint(1, 10)):
offset += timedelta(seconds=random.random() * 120)
event = Event(
id=next(event_sequence),
event_id=uuid.uuid4().hex,
project=project,
group=group,
message=group.message,
data=load_data('python'),
datetime=to_datetime(
random.randint(
to_timestamp(group.first_seen),
to_timestamp(group.last_seen),
),
)
)
records.append(
Record(
event.event_id,
Notification(
event,
random.sample(state['rules'], random.randint(1, len(state['rules']))),
),
to_timestamp(event.datetime),
)
)
state['event_counts'][group.id] = random.randint(10, 1e4)
state['user_counts'][group.id] = random.randint(10, 1e4)
digest = build_digest(project, records, state)
start, end, counts = get_digest_metadata(digest)
context = {
'project': project,
'counts': counts,
'digest': digest,
'start': start,
'end': end,
}
add_unsubscribe_link(context)
return MailPreview(
html_template='sentry/emails/digests/body.html',
text_template='sentry/emails/digests/body.txt',
context=context,
).render(request)
开发者ID:faulkner，项目名称:sentry，代码行数:97，代码来源:mail.py
注：本文中的sentry.digests.notifications.build_digest函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论