1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
|
# Copyright 2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Test `BugSubscriptionInfo`."""
__metaclass__ = type
from contextlib import contextmanager
from storm.store import Store
from testtools.matchers import Equals
from zope.component import queryAdapter
from zope.security.checker import getChecker
from lp.app.interfaces.security import IAuthorization
from lp.bugs.enum import BugNotificationLevel
from lp.bugs.model.bug import (
BugSubscriberSet,
BugSubscriptionInfo,
BugSubscriptionSet,
load_people,
StructuralSubscriptionSet,
)
from lp.bugs.security import (
PublicToAllOrPrivateToExplicitSubscribersForBugTask,
)
from lp.registry.model.person import Person
from lp.testing import (
person_logged_in,
StormStatementRecorder,
TestCaseWithFactory,
)
from lp.testing.layers import DatabaseFunctionalLayer
from lp.testing.matchers import HasQueryCount
class TestLoadPeople(TestCaseWithFactory):
"""Tests for `load_people`."""
layer = DatabaseFunctionalLayer
def test(self):
expected = [
self.factory.makePerson(),
self.factory.makeTeam(),
]
observed = load_people(
Person.id.is_in(person.id for person in expected))
self.assertContentEqual(expected, observed)
class TestSubscriptionRelatedSets(TestCaseWithFactory):
"""Tests for *Set classes related to subscriptions."""
layer = DatabaseFunctionalLayer
name_pairs = ("A", "xa"), ("C", "xd"), ("B", "xb"), ("C", "xc")
name_pairs_sorted = ("A", "xa"), ("B", "xb"), ("C", "xc"), ("C", "xd")
def setUp(self):
super(TestSubscriptionRelatedSets, self).setUp()
make_person = lambda (displayname, name): (
self.factory.makePerson(displayname=displayname, name=name))
subscribers = dict(
(name_pair, make_person(name_pair))
for name_pair in self.name_pairs)
self.subscribers_set = frozenset(subscribers.itervalues())
self.subscribers_sorted = tuple(
subscribers[name_pair] for name_pair in self.name_pairs_sorted)
def test_BugSubscriberSet(self):
subscriber_set = BugSubscriberSet(self.subscribers_set)
self.assertIsInstance(subscriber_set, frozenset)
self.assertEqual(self.subscribers_set, subscriber_set)
self.assertEqual(self.subscribers_sorted, subscriber_set.sorted)
def test_BugSubscriptionSet(self):
bug = self.factory.makeBug()
with person_logged_in(bug.owner):
subscriptions = frozenset(
bug.subscribe(subscriber, subscriber)
for subscriber in self.subscribers_set)
subscription_set = BugSubscriptionSet(subscriptions)
self.assertIsInstance(subscription_set, frozenset)
self.assertEqual(subscriptions, subscription_set)
# BugSubscriptionSet.sorted returns a tuple of subscriptions ordered
# by subscribers.
self.assertEqual(
self.subscribers_sorted, tuple(
subscription.person
for subscription in subscription_set.sorted))
# BugSubscriptionSet.subscribers returns a BugSubscriberSet of the
# subscription's subscribers.
self.assertIsInstance(subscription_set.subscribers, BugSubscriberSet)
self.assertEqual(self.subscribers_set, subscription_set.subscribers)
def test_StructuralSubscriptionSet(self):
product = self.factory.makeProduct()
with person_logged_in(product.owner):
subscriptions = frozenset(
product.addSubscription(subscriber, subscriber)
for subscriber in self.subscribers_set)
subscription_set = StructuralSubscriptionSet(subscriptions)
self.assertIsInstance(subscription_set, frozenset)
self.assertEqual(subscriptions, subscription_set)
# StructuralSubscriptionSet.sorted returns a tuple of subscriptions
# ordered by subscribers.
self.assertEqual(
self.subscribers_sorted, tuple(
subscription.subscriber
for subscription in subscription_set.sorted))
# StructuralSubscriptionSet.subscribers returns a BugSubscriberSet of
# the subscription's subscribers.
self.assertIsInstance(subscription_set.subscribers, BugSubscriberSet)
self.assertEqual(self.subscribers_set, subscription_set.subscribers)
class TestBugSubscriptionInfo(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def setUp(self):
super(TestBugSubscriptionInfo, self).setUp()
self.target = self.factory.makeProduct(
bug_supervisor=self.factory.makePerson())
self.bug = self.factory.makeBug(product=self.target)
# Unsubscribe the bug filer to make the tests more readable.
with person_logged_in(self.bug.owner):
self.bug.unsubscribe(self.bug.owner, self.bug.owner)
def getInfo(self, level=BugNotificationLevel.LIFECYCLE):
return BugSubscriptionInfo(self.bug, level)
def _create_direct_subscriptions(self):
subscribers = (
self.factory.makePerson(),
self.factory.makePerson())
with person_logged_in(self.bug.owner):
subscriptions = tuple(
self.bug.subscribe(subscriber, subscriber)
for subscriber in subscribers)
return subscribers, subscriptions
def test_forTask(self):
# `forTask` returns a new `BugSubscriptionInfo` narrowed to the given
# bugtask.
info = self.getInfo()
self.assertIs(None, info.bugtask)
# If called with the current bugtask the same `BugSubscriptionInfo`
# instance is returned.
self.assertIs(info, info.forTask(info.bugtask))
# If called with a different bugtask a new `BugSubscriptionInfo` is
# created.
bugtask = self.bug.default_bugtask
info_for_task = info.forTask(bugtask)
self.assertIs(bugtask, info_for_task.bugtask)
self.assertIsNot(info, info_for_task)
# The instances share a cache of `BugSubscriptionInfo` instances.
expected_cache = {
info.cache_key: info,
info_for_task.cache_key: info_for_task,
}
self.assertEqual(expected_cache, info.cache)
self.assertIs(info.cache, info_for_task.cache)
# Calling `forTask` again looks in the cache first.
self.assertIs(info, info_for_task.forTask(info.bugtask))
self.assertIs(info_for_task, info.forTask(info_for_task.bugtask))
# The level is the same.
self.assertEqual(info.level, info_for_task.level)
def test_forLevel(self):
# `forLevel` returns a new `BugSubscriptionInfo` narrowed to the given
# subscription level.
info = self.getInfo(BugNotificationLevel.LIFECYCLE)
# If called with the current level the same `BugSubscriptionInfo`
# instance is returned.
self.assertIs(info, info.forLevel(info.level))
# If called with a different level a new `BugSubscriptionInfo` is
# created.
level = BugNotificationLevel.METADATA
info_for_level = info.forLevel(level)
self.assertEqual(level, info_for_level.level)
self.assertIsNot(info, info_for_level)
# The instances share a cache of `BugSubscriptionInfo` instances.
expected_cache = {
info.cache_key: info,
info_for_level.cache_key: info_for_level,
}
self.assertEqual(expected_cache, info.cache)
self.assertIs(info.cache, info_for_level.cache)
# Calling `forLevel` again looks in the cache first.
self.assertIs(info, info_for_level.forLevel(info.level))
self.assertIs(info_for_level, info.forLevel(info_for_level.level))
# The bugtask is the same.
self.assertIs(info.bugtask, info_for_level.bugtask)
def test_muted(self):
# The set of muted subscribers for the bug.
subscribers, subscriptions = self._create_direct_subscriptions()
sub1, sub2 = subscribers
with person_logged_in(sub1):
self.bug.mute(sub1, sub1)
self.assertContentEqual([sub1], self.getInfo().muted_subscribers)
def test_direct(self):
# The set of direct subscribers.
subscribers, subscriptions = self._create_direct_subscriptions()
found_subscriptions = self.getInfo().direct_subscriptions
self.assertContentEqual(subscriptions, found_subscriptions)
self.assertContentEqual(subscribers, found_subscriptions.subscribers)
def test_direct_muted(self):
# If a direct is muted, it is not listed.
subscribers, subscriptions = self._create_direct_subscriptions()
with person_logged_in(subscribers[0]):
self.bug.mute(subscribers[0], subscribers[0])
found_subscriptions = self.getInfo().direct_subscriptions
self.assertContentEqual([subscriptions[1]], found_subscriptions)
def test_all_direct(self):
# The set of all direct subscribers, regardless of level.
subscribers, subscriptions = self._create_direct_subscriptions()
# Change the first subscription to be for comments only.
sub1, sub2 = subscriptions
with person_logged_in(sub1.person):
sub1.bug_notification_level = BugNotificationLevel.LIFECYCLE
info = self.getInfo(BugNotificationLevel.COMMENTS)
self.assertContentEqual([sub2], info.direct_subscriptions)
self.assertContentEqual(
[sub1, sub2], info.direct_subscriptions_at_all_levels)
def _create_duplicate_subscription(self):
duplicate_bug = self.factory.makeBug(product=self.target)
with person_logged_in(duplicate_bug.owner):
duplicate_bug.markAsDuplicate(self.bug)
duplicate_bug_subscription = (
duplicate_bug.getSubscriptionForPerson(
duplicate_bug.owner))
return duplicate_bug, duplicate_bug_subscription
def test_duplicate(self):
# The set of subscribers from duplicate bugs.
found_subscriptions = self.getInfo().duplicate_subscriptions
self.assertContentEqual([], found_subscriptions)
self.assertContentEqual([], found_subscriptions.subscribers)
duplicate_bug, duplicate_bug_subscription = (
self._create_duplicate_subscription())
found_subscriptions = self.getInfo().duplicate_subscriptions
self.assertContentEqual(
[duplicate_bug_subscription],
found_subscriptions)
self.assertContentEqual(
[duplicate_bug.owner],
found_subscriptions.subscribers)
def test_duplicate_muted(self):
# If a duplicate is muted, it is not listed.
duplicate_bug, duplicate_bug_subscription = (
self._create_duplicate_subscription())
with person_logged_in(duplicate_bug.owner):
self.bug.mute(duplicate_bug.owner, duplicate_bug.owner)
found_subscriptions = self.getInfo().duplicate_subscriptions
self.assertContentEqual([], found_subscriptions)
def test_duplicate_for_private_bug(self):
# The set of subscribers from duplicate bugs is always empty when the
# master bug is private.
duplicate_bug = self.factory.makeBug(product=self.target)
with person_logged_in(duplicate_bug.owner):
duplicate_bug.markAsDuplicate(self.bug)
with person_logged_in(self.bug.owner):
self.bug.setPrivate(True, self.bug.owner)
found_subscriptions = self.getInfo().duplicate_subscriptions
self.assertContentEqual([], found_subscriptions)
self.assertContentEqual([], found_subscriptions.subscribers)
def test_duplicate_only(self):
# The set of duplicate subscriptions where the subscriber has no other
# subscriptions.
duplicate_bug = self.factory.makeBug(product=self.target)
with person_logged_in(duplicate_bug.owner):
duplicate_bug.markAsDuplicate(self.bug)
duplicate_bug_subscription = (
duplicate_bug.getSubscriptionForPerson(
duplicate_bug.owner))
found_subscriptions = self.getInfo().duplicate_only_subscriptions
self.assertContentEqual(
[duplicate_bug_subscription],
found_subscriptions)
# If a user is subscribed to a duplicate bug and is a bugtask
# assignee, for example, their duplicate subscription will not be
# included.
with person_logged_in(self.target.owner):
self.bug.default_bugtask.transitionToAssignee(
duplicate_bug_subscription.person)
found_subscriptions = self.getInfo().duplicate_only_subscriptions
self.assertContentEqual([], found_subscriptions)
def test_structural_subscriptions(self):
# The set of structural subscriptions.
subscribers = (
self.factory.makePerson(),
self.factory.makePerson())
with person_logged_in(self.bug.owner):
subscriptions = tuple(
self.target.addBugSubscription(subscriber, subscriber)
for subscriber in subscribers)
found_subscriptions = self.getInfo().structural_subscriptions
self.assertContentEqual(subscriptions, found_subscriptions)
def test_structural_subscriptions_muted(self):
# The set of structural subscriptions DOES NOT exclude muted
# subscriptions.
subscriber = self.factory.makePerson()
with person_logged_in(subscriber):
self.bug.mute(subscriber, subscriber)
with person_logged_in(self.bug.owner):
subscription = self.target.addBugSubscription(
subscriber, subscriber)
found_subscriptions = self.getInfo().structural_subscriptions
self.assertContentEqual([subscription], found_subscriptions)
def test_structural_subscribers(self):
# The set of structural subscribers.
subscribers = (
self.factory.makePerson(),
self.factory.makePerson())
with person_logged_in(self.bug.owner):
for subscriber in subscribers:
self.target.addBugSubscription(subscriber, subscriber)
found_subscribers = self.getInfo().structural_subscribers
self.assertContentEqual(subscribers, found_subscribers)
def test_structural_subscribers_muted(self):
# The set of structural subscribers DOES NOT exclude muted
# subscribers.
subscriber = self.factory.makePerson()
with person_logged_in(subscriber):
self.bug.mute(subscriber, subscriber)
with person_logged_in(self.bug.owner):
self.target.addBugSubscription(subscriber, subscriber)
found_subscribers = self.getInfo().structural_subscribers
self.assertContentEqual([subscriber], found_subscribers)
def test_all_assignees(self):
# The set of bugtask assignees for bugtasks that have been assigned.
found_assignees = self.getInfo().all_assignees
self.assertContentEqual([], found_assignees)
bugtask = self.bug.default_bugtask
with person_logged_in(bugtask.pillar.bug_supervisor):
bugtask.transitionToAssignee(self.bug.owner)
found_assignees = self.getInfo().all_assignees
self.assertContentEqual([self.bug.owner], found_assignees)
bugtask2 = self.factory.makeBugTask(bug=self.bug)
with person_logged_in(bugtask2.pillar.owner):
bugtask2.transitionToAssignee(bugtask2.owner)
found_assignees = self.getInfo().all_assignees
self.assertContentEqual(
[self.bug.owner, bugtask2.owner],
found_assignees)
# Getting info for a specific bugtask will return the assignee for
# that bugtask only.
self.assertContentEqual(
[bugtask2.owner],
self.getInfo().forTask(bugtask2).all_assignees)
def test_all_pillar_owners_without_bug_supervisors(self):
# The set of owners of pillars for which no bug supervisor is
# configured and which use Launchpad for bug tracking.
[bugtask] = self.bug.bugtasks
found_owners = (
self.getInfo().all_pillar_owners_without_bug_supervisors)
self.assertContentEqual([], found_owners)
# Clear the supervisor for the bugtask's target and ensure that the
# project uses Launchpad Bugs.
with person_logged_in(bugtask.target.owner):
bugtask.target.setBugSupervisor(None, bugtask.owner)
bugtask.pillar.official_malone = True
# The collection includes the pillar's owner.
found_owners = (
self.getInfo().all_pillar_owners_without_bug_supervisors)
self.assertContentEqual([bugtask.pillar.owner], found_owners)
# Add another bugtask for a pillar that uses Launchpad but does not
# have a bug supervisor.
target2 = self.factory.makeProduct(
bug_supervisor=None, official_malone=True)
bugtask2 = self.factory.makeBugTask(bug=self.bug, target=target2)
found_owners = (
self.getInfo().all_pillar_owners_without_bug_supervisors)
self.assertContentEqual(
[bugtask.pillar.owner, bugtask2.pillar.owner],
found_owners)
def test_all_pillar_owners_without_bug_supervisors_not_using_malone(self):
# The set of owners of pillars for which no bug supervisor is
# configured and which do not use Launchpad for bug tracking is empty.
[bugtask] = self.bug.bugtasks
# Clear the supervisor for the first bugtask's target and ensure the
# project does not use Launchpad Bugs.
with person_logged_in(bugtask.target.owner):
bugtask.target.setBugSupervisor(None, bugtask.owner)
bugtask.pillar.official_malone = False
found_owners = (
self.getInfo().all_pillar_owners_without_bug_supervisors)
self.assertContentEqual([], found_owners)
def test_all_pillar_owners_without_bug_supervisors_for_bugtask(self):
# The set of the owner of the chosen bugtask's pillar when no bug
# supervisor is configured and which uses Launchpad for bug tracking.
[bugtask] = self.bug.bugtasks
# Clear the supervisor for the bugtask's target and ensure that the
# project uses Launchpad Bugs.
with person_logged_in(bugtask.target.owner):
bugtask.target.setBugSupervisor(None, bugtask.owner)
bugtask.pillar.official_malone = True
# Add another bugtask for a pillar that uses Launchpad but does not
# have a bug supervisor.
target2 = self.factory.makeProduct(
bug_supervisor=None, official_malone=True)
bugtask2 = self.factory.makeBugTask(bug=self.bug, target=target2)
# Getting subscription info for just a specific bugtask will yield
# owners for only the pillar associated with that bugtask.
info_for_bugtask2 = self.getInfo().forTask(bugtask2)
self.assertContentEqual(
[bugtask2.pillar.owner],
info_for_bugtask2.all_pillar_owners_without_bug_supervisors)
def _create_also_notified_subscribers(self):
# Add an assignee, a bug supervisor and a structural subscriber.
bugtask = self.bug.default_bugtask
assignee = self.factory.makePerson()
with person_logged_in(bugtask.pillar.bug_supervisor):
bugtask.transitionToAssignee(assignee)
supervisor = self.factory.makePerson()
with person_logged_in(bugtask.target.owner):
bugtask.target.setBugSupervisor(supervisor, supervisor)
structural_subscriber = self.factory.makePerson()
with person_logged_in(structural_subscriber):
bugtask.target.addSubscription(
structural_subscriber, structural_subscriber)
return assignee, supervisor, structural_subscriber
def test_also_notified_subscribers(self):
# The set of also notified subscribers.
found_subscribers = self.getInfo().also_notified_subscribers
self.assertContentEqual([], found_subscribers)
assignee, supervisor, structural_subscriber = (
self._create_also_notified_subscribers())
# Add a direct subscription.
direct_subscriber = self.factory.makePerson()
with person_logged_in(self.bug.owner):
self.bug.subscribe(direct_subscriber, direct_subscriber)
# The direct subscriber does not appear in the also notified set, but
# the assignee, supervisor and structural subscriber do.
found_subscribers = self.getInfo().also_notified_subscribers
self.assertContentEqual(
[assignee, supervisor, structural_subscriber],
found_subscribers)
def test_also_notified_subscribers_muted(self):
# If someone is muted, they are not listed in the
# also_notified_subscribers.
assignee, supervisor, structural_subscriber = (
self._create_also_notified_subscribers())
# As a control, we first show that the
# the assignee, supervisor and structural subscriber do show up
# when they are not muted.
found_subscribers = self.getInfo().also_notified_subscribers
self.assertContentEqual(
[assignee, supervisor, structural_subscriber],
found_subscribers)
# Now we mute all of the subscribers.
with person_logged_in(assignee):
self.bug.mute(assignee, assignee)
with person_logged_in(supervisor):
self.bug.mute(supervisor, supervisor)
with person_logged_in(structural_subscriber):
self.bug.mute(structural_subscriber, structural_subscriber)
# Now we don't see them.
found_subscribers = self.getInfo().also_notified_subscribers
self.assertContentEqual([], found_subscribers)
def test_also_notified_subscribers_for_private_bug(self):
# The set of also notified subscribers is always empty when the master
# bug is private.
assignee = self.factory.makePerson()
bugtask = self.bug.default_bugtask
with person_logged_in(bugtask.pillar.bug_supervisor):
bugtask.transitionToAssignee(assignee)
with person_logged_in(self.bug.owner):
self.bug.setPrivate(True, self.bug.owner)
found_subscribers = self.getInfo().also_notified_subscribers
self.assertContentEqual([], found_subscribers)
def test_indirect_subscribers(self):
# The set of indirect subscribers is the union of also notified
# subscribers and subscribers to duplicates.
assignee = self.factory.makePerson()
bugtask = self.bug.default_bugtask
with person_logged_in(bugtask.pillar.bug_supervisor):
bugtask.transitionToAssignee(assignee)
duplicate_bug = self.factory.makeBug(product=self.target)
with person_logged_in(duplicate_bug.owner):
duplicate_bug.markAsDuplicate(self.bug)
found_subscribers = self.getInfo().indirect_subscribers
self.assertContentEqual(
[assignee, duplicate_bug.owner],
found_subscribers)
class TestBugSubscriptionInfoPermissions(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def test(self):
bug = self.factory.makeBug()
info = bug.getSubscriptionInfo()
checker = getChecker(info)
# BugSubscriptionInfo objects are immutable.
self.assertEqual({}, checker.set_permissions)
# All attributes require launchpad.View.
permissions = set(checker.get_permissions.itervalues())
self.assertContentEqual(["launchpad.View"], permissions)
# The security adapter for launchpad.View lets anyone reference the
# attributes unless the bug is private, in which case only explicit
# subscribers are permitted.
adapter = queryAdapter(info, IAuthorization, "launchpad.View")
self.assertIsInstance(
adapter, PublicToAllOrPrivateToExplicitSubscribersForBugTask)
class TestBugSubscriptionInfoQueries(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def setUp(self):
super(TestBugSubscriptionInfoQueries, self).setUp()
self.target = self.factory.makeProduct()
self.bug = self.factory.makeBug(product=self.target)
self.info = BugSubscriptionInfo(
self.bug, BugNotificationLevel.LIFECYCLE)
# Get the Storm cache into a known state.
self.store = Store.of(self.bug)
self.store.invalidate()
self.store.reload(self.bug)
self.bug.bugtasks
self.bug.tags
@contextmanager
def exactly_x_queries(self, count):
# Assert that there are exactly `count` queries sent to the database
# in this context. Flush first to ensure we don't count things that
# happened before entering this context.
self.store.flush()
condition = HasQueryCount(Equals(count))
with StormStatementRecorder() as recorder:
yield recorder
self.assertThat(recorder, condition)
def exercise_subscription_set(self, set_name, counts=(1, 1, 0)):
"""Test the number of queries it takes to inspect a subscription set.
:param set_name: The name of the set, e.g. "direct_subscriptions".
:param counts: A triple of the expected query counts for each of three
operations: get the set, get the set's subscribers, get the set's
subscribers in order.
"""
# Looking up subscriptions takes a single query.
with self.exactly_x_queries(counts[0]):
getattr(self.info, set_name)
# Getting the subscribers results in one additional query.
with self.exactly_x_queries(counts[1]):
getattr(self.info, set_name).subscribers
# Everything is now cached so no more queries are needed.
with self.exactly_x_queries(counts[2]):
getattr(self.info, set_name).subscribers
getattr(self.info, set_name).subscribers.sorted
def exercise_subscription_set_sorted_first(
self, set_name, counts=(1, 1, 0)):
"""Test the number of queries it takes to inspect a subscription set.
This differs from `exercise_subscription_set` in its second step, when
it looks at the sorted subscription list instead of the subscriber
set.
:param set_name: The name of the set, e.g. "direct_subscriptions".
:param counts: A triple of the expected query counts for each of three
operations: get the set, get the set in order, get the set's
subscribers in order.
"""
# Looking up subscriptions takes a single query.
with self.exactly_x_queries(counts[0]):
getattr(self.info, set_name)
# Getting the sorted subscriptions takes one additional query.
with self.exactly_x_queries(counts[1]):
getattr(self.info, set_name).sorted
# Everything is now cached so no more queries are needed.
with self.exactly_x_queries(counts[2]):
getattr(self.info, set_name).subscribers
getattr(self.info, set_name).subscribers.sorted
def test_direct_subscriptions(self):
self.exercise_subscription_set(
"direct_subscriptions")
def test_direct_subscriptions_sorted_first(self):
self.exercise_subscription_set_sorted_first(
"direct_subscriptions")
def test_direct_subscriptions_at_all_levels(self):
self.exercise_subscription_set(
"direct_subscriptions_at_all_levels")
def make_duplicate_bug(self):
duplicate_bug = self.factory.makeBug(product=self.target)
with person_logged_in(duplicate_bug.owner):
duplicate_bug.markAsDuplicate(self.bug)
def test_duplicate_subscriptions(self):
self.make_duplicate_bug()
self.exercise_subscription_set(
"duplicate_subscriptions")
def test_duplicate_subscriptions_sorted_first(self):
self.make_duplicate_bug()
self.exercise_subscription_set_sorted_first(
"duplicate_subscriptions")
def test_duplicate_subscriptions_for_private_bug(self):
self.make_duplicate_bug()
with person_logged_in(self.bug.owner):
self.bug.setPrivate(True, self.bug.owner)
with self.exactly_x_queries(0):
self.info.duplicate_subscriptions
with self.exactly_x_queries(0):
self.info.duplicate_subscriptions.subscribers
def add_structural_subscriber(self):
subscriber = self.factory.makePerson()
with person_logged_in(subscriber):
self.target.addSubscription(subscriber, subscriber)
def test_structural_subscriptions(self):
self.add_structural_subscriber()
self.exercise_subscription_set(
"structural_subscriptions", (2, 1, 0))
def test_structural_subscriptions_sorted_first(self):
self.add_structural_subscriber()
self.exercise_subscription_set_sorted_first(
"structural_subscriptions", (2, 1, 0))
def test_all_assignees(self):
with self.exactly_x_queries(1):
self.info.all_assignees
def test_all_pillar_owners_without_bug_supervisors(self):
# Getting all bug supervisors and pillar owners can take several
# queries. However, there are typically few tasks so the trade for
# simplicity of implementation is acceptable. Only the simplest case
# is tested here (everything is already cached).
with self.exactly_x_queries(0):
self.info.all_pillar_owners_without_bug_supervisors
def test_also_notified_subscribers(self):
with self.exactly_x_queries(5):
self.info.also_notified_subscribers
def test_also_notified_subscribers_later(self):
# When also_notified_subscribers is referenced after some other sets
# in BugSubscriptionInfo are referenced, everything comes from cache.
self.info.all_assignees
self.info.all_pillar_owners_without_bug_supervisors
self.info.direct_subscriptions.subscribers
self.info.structural_subscribers
with self.exactly_x_queries(1):
self.info.also_notified_subscribers
def test_indirect_subscribers(self):
with self.exactly_x_queries(6):
self.info.indirect_subscribers
def test_indirect_subscribers_later(self):
# When indirect_subscribers is referenced after some other sets in
# BugSubscriptionInfo are referenced, everything comes from cache.
self.info.also_notified_subscribers
self.info.duplicate_subscriptions.subscribers
with self.exactly_x_queries(0):
self.info.indirect_subscribers
|