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
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests for IBranchCloud provider."""
__metaclass__ = type
from datetime import datetime, timedelta
import transaction
import pytz
from storm.locals import Store
from zope.component import getUtility
from canonical.launchpad.testing.databasehelpers import (
remove_all_sample_data_branches,
)
from canonical.testing.layers import DatabaseFunctionalLayer
from lp.code.interfaces.branch import IBranchCloud
from lp.code.model.revision import RevisionCache
from lp.code.tests.helpers import make_project_branch_with_revisions
from lp.testing import TestCaseWithFactory, time_counter
class TestBranchCloud(TestCaseWithFactory):
layer = DatabaseFunctionalLayer
def setUp(self):
TestCaseWithFactory.setUp(self)
remove_all_sample_data_branches()
self._branch_cloud = getUtility(IBranchCloud)
def getProductsWithInfo(self, num_products=None):
"""Get product cloud information."""
# Since we use the slave store to get the information, we need to
# commit the transaction to make the information visible to the slave.
transaction.commit()
cloud_info = self._branch_cloud.getProductsWithInfo(num_products)
# The last commit time is timezone unaware as the storm Max function
# doesn't take into account the type that it is aggregating, so whack
# the UTC tz on it here for easier comparing in the tests.
def add_utc(value):
return value.replace(tzinfo=pytz.UTC)
return [
(name, commits, authors, add_utc(last_commit))
for name, commits, authors, last_commit in cloud_info]
def makeBranch(self, product=None, last_commit_date=None, private=False,
revision_count=None):
"""Make a product branch with a particular last commit date"""
if revision_count is None:
revision_count = 5
delta = timedelta(days=1)
if last_commit_date is None:
# By default we create revisions that are within the last 30 days.
date_generator = time_counter(
datetime.now(pytz.UTC) - timedelta(days=25), delta)
else:
start_date = last_commit_date - delta * (revision_count - 1)
date_generator = time_counter(start_date, delta)
branch = make_project_branch_with_revisions(
self.factory, date_generator, product, private, revision_count)
return branch
def test_empty_with_no_branches(self):
# getProductsWithInfo returns an empty result set if there are no
# branches in the database.
self.assertEqual([], self.getProductsWithInfo())
def test_empty_products_not_counted(self):
# getProductsWithInfo doesn't include products that don't have any
# branches.
#
# Note that this is tested implicitly by test_empty_with_no_branches,
# since there are such products in the sample data.
self.factory.makeProduct()
self.assertEqual([], self.getProductsWithInfo())
def test_empty_branches_not_counted(self):
# getProductsWithInfo doesn't consider branches that lack revision
# data, 'empty branches', to contribute to the count of branches on a
# product.
self.factory.makeProductBranch()
self.assertEqual([], self.getProductsWithInfo())
def test_private_branches_not_counted(self):
# getProductsWithInfo doesn't count private branches.
self.makeBranch(private=True)
self.assertEqual([], self.getProductsWithInfo())
def test_revisions_counted(self):
# getProductsWithInfo includes products that public revisions.
last_commit_date = datetime.now(pytz.UTC) - timedelta(days=5)
product = self.factory.makeProduct()
self.makeBranch(product=product, last_commit_date=last_commit_date)
self.assertEqual(
[(product.name, 5, 1, last_commit_date)],
self.getProductsWithInfo())
def test_only_recent_revisions_counted(self):
# If the revision cache has revisions for the project, but they are
# over 30 days old, we don't count them.
product = self.factory.makeProduct()
date_generator = time_counter(
datetime.now(pytz.UTC) - timedelta(days=33),
delta=timedelta(days=2))
store = Store.of(product)
for i in range(4):
revision = self.factory.makeRevision(
revision_date=date_generator.next())
cache = RevisionCache(revision)
cache.product = product
store.add(cache)
self.assertEqual(
[(product.name, 2, 2, revision.revision_date)],
self.getProductsWithInfo())
def test_sorted_by_commit_count(self):
# getProductsWithInfo returns a result set sorted so that the products
# with the most commits come first.
product1 = self.factory.makeProduct()
for i in range(3):
self.makeBranch(product=product1)
product2 = self.factory.makeProduct()
for i in range(5):
self.makeBranch(product=product2)
self.assertEqual(
[product2.name, product1.name],
[name for name, commits, count, last_commit
in self.getProductsWithInfo()])
def test_limit(self):
# If num_products is passed to getProductsWithInfo, it limits the
# number of products in the result set. The products with the fewest
# branches are discarded first.
product1 = self.factory.makeProduct()
for i in range(3):
self.makeBranch(product=product1)
product2 = self.factory.makeProduct()
for i in range(5):
self.makeBranch(product=product2)
product3 = self.factory.makeProduct()
for i in range(7):
self.makeBranch(product=product3)
self.assertEqual(
[product3.name, product2.name],
[name for name, commits, count, last_commit
in self.getProductsWithInfo(num_products=2)])
|