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
|
# Copyright 2009-2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""
Run the view tests.
"""
import logging
import os
import unittest
from storm.store import Store
from testtools.matchers import LessThan
from lp.services.webapp import canonical_url
from lp.testing import (
login,
logout,
TestCaseWithFactory,
)
from lp.testing._webservice import QueryCollector
from lp.testing.layers import DatabaseFunctionalLayer
from lp.testing.matchers import HasQueryCount
from lp.testing.sampledata import ADMIN_EMAIL
from lp.testing.systemdocs import (
LayeredDocFileSuite,
setUp,
tearDown,
)
class TestAssignments(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def invalidate_and_render(self, browser, dbobj, url):
# Ensure caches have been flushed.
store = Store.of(dbobj)
store.flush()
store.invalidate()
browser.open(url)
def check_query_counts_scaling_with_unique_people(self,
target, targettype):
"""Check that a particular hasSpecifications target scales well.
:param target: A spec target like a product.
:param targettype: The parameter to pass to makeSpecification to
associate the target. e.g. 'product'.
"""
query_baseline = 40
people = []
for _ in range(10):
people.append(self.factory.makePerson())
specs = []
for _ in range(10):
specs.append(self.factory.makeSpecification(
**{targettype: target}))
collector = QueryCollector()
collector.register()
self.addCleanup(collector.unregister)
viewer = self.factory.makePerson(password="test")
browser = self.getUserBrowser(user=viewer)
url = canonical_url(target) + "/+assignments"
# Seed the cookie cache and any other cross-request state we may gain
# in future. See lp.services.webapp.serssion: _get_secret.
browser.open(url)
self.invalidate_and_render(browser, target, url)
# Set a baseline
self.assertThat(collector, HasQueryCount(LessThan(query_baseline)))
no_assignees_count = collector.count
# Assign many unique people, which shouldn't change the page queries.
# Due to storm bug 619017 additional queries can be triggered when
# revalidating people, so we allow -some- fuzz.
login(ADMIN_EMAIL)
for person, spec in zip(people, specs):
spec.assignee = person
logout()
self.invalidate_and_render(browser, target, url)
self.assertThat(
collector,
HasQueryCount(LessThan(no_assignees_count + 5)))
def test_product_query_counts_scale_below_unique_people(self):
self.check_query_counts_scaling_with_unique_people(
self.factory.makeProduct(),
'product')
def test_distro_query_counts_scale_below_unique_people(self):
self.check_query_counts_scaling_with_unique_people(
self.factory.makeDistribution(),
'distribution')
def test_suite():
suite = unittest.TestLoader().loadTestsFromName(__name__)
here = os.path.dirname(os.path.realpath(__file__))
testsdir = os.path.abspath(here)
# Add tests using default setup/teardown
filenames = [filename
for filename in os.listdir(testsdir)
if filename.endswith('.txt')]
# Sort the list to give a predictable order.
filenames.sort()
for filename in filenames:
path = filename
one_test = LayeredDocFileSuite(
path, setUp=setUp, tearDown=tearDown,
layer=DatabaseFunctionalLayer,
stdout_logging_level=logging.WARNING
)
suite.addTest(one_test)
return suite
|