forked from mozilla-releng/balrog
-
Notifications
You must be signed in to change notification settings - Fork 5
/
test_AUS.py
147 lines (111 loc) · 5.34 KB
/
test_AUS.py
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
import logging
import unittest
import mock
import pytest
from auslib.AUS import AUS, FORCE_FALLBACK_MAPPING, FORCE_MAIN_MAPPING
from auslib.blobs.base import createBlob
from auslib.global_state import dbo
ENTIRE_RANGE = range(0, 100)
def setUpModule():
# Silence SQLAlchemy-Migrate's debugging logger
logging.getLogger("migrate").setLevel(logging.CRITICAL)
@pytest.mark.usefixtures("current_db_schema")
class TestAUSThrottlingWithoutFallback(unittest.TestCase):
def setUp(self):
dbo.setDb("sqlite:///:memory:")
self.metadata.create_all(dbo.engine)
dbo.releases.t.insert().execute(
name="b",
product="b",
data_version=1,
data=createBlob({"name": "b", "extv": "1.0", "schema_version": 1, "platforms": {"a": {"buildID": "1", "locales": {"a": {}}}}}),
)
dbo.releases.t.insert().execute(
name="fallback",
product="c",
data_version=1,
data=createBlob({"name": "fallback", "extv": "1.0", "schema_version": 1, "platforms": {"a": {"buildID": "1", "locales": {"a": {}}}}}),
)
def tearDown(self):
dbo.reset()
def random_aus_test(self, background_rate, force=None, fallback=False):
mapping = "b"
with mock.patch("auslib.db.Rules.getRulesMatchingQuery") as m:
fallback = fallback and "fallback" # convert True to string
m.return_value = [dict(backgroundRate=background_rate, priority=1, mapping=mapping, update_type="minor", fallbackMapping=fallback)]
results = list(ENTIRE_RANGE)
resultsLength = len(results)
def se(*args, **kwargs):
return results.pop()
aus = AUS()
aus.rand = mock.Mock(side_effect=se)
served_mapping = 0
served_fallback = 0
tested = 0
while len(results) > 0:
updateQuery = dict(channel="foo", force=force, buildTarget="a", buildID="0", locale="a", version="1.0", product="bar")
r, _ = aus.evaluateRules(updateQuery)
tested += 1
if r and r["name"] == mapping:
served_mapping += 1
elif fallback and r["name"] == fallback:
served_fallback += 1
# bail out if we're not asking for any randint's
if resultsLength == len(results):
break
return (served_mapping, served_fallback, tested)
def testThrottling100(self):
(served, _, tested) = self.random_aus_test(background_rate=100)
self.assertEqual(served, 1)
self.assertEqual(tested, 1)
def testThrottling50(self):
(served, _, tested) = self.random_aus_test(background_rate=50)
self.assertEqual(served, 50)
self.assertEqual(tested, 100)
def testThrottling25(self):
(served, _, tested) = self.random_aus_test(background_rate=25)
self.assertEqual(served, 25)
self.assertEqual(tested, 100)
def testThrottlingZero(self):
(served, _, tested) = self.random_aus_test(background_rate=0)
self.assertEqual(served, 0)
self.assertEqual(tested, 100)
def testThrottling25WithForcing(self):
(served, _, tested) = self.random_aus_test(background_rate=25, force=FORCE_MAIN_MAPPING)
self.assertEqual(served, 1)
self.assertEqual(tested, 1)
def testThrottling25WithForcingFailure(self):
(served, fallback, tested) = self.random_aus_test(background_rate=25, force=FORCE_FALLBACK_MAPPING)
self.assertEqual(served, 0)
self.assertEqual(fallback, 0)
self.assertEqual(tested, 1)
def testThrottling100WithFallback(self):
(served_mapping, served_fallback, tested) = self.random_aus_test(background_rate=100, fallback=True)
self.assertEqual(served_mapping, 1)
self.assertEqual(served_fallback, 0)
self.assertEqual(tested, 1)
def testThrottling50WithFallback(self):
(served_mapping, served_fallback, tested) = self.random_aus_test(background_rate=50, fallback=True)
self.assertEqual(served_mapping, 50)
self.assertEqual(served_fallback, 50)
self.assertEqual(tested, 100)
def testThrottling25WithFallback(self):
(served_mapping, served_fallback, tested) = self.random_aus_test(background_rate=25, fallback=True)
self.assertEqual(served_mapping, 25)
self.assertEqual(served_fallback, 75)
self.assertEqual(tested, 100)
def testThrottlingZeroWithFallback(self):
(served_mapping, served_fallback, tested) = self.random_aus_test(background_rate=0, fallback=True)
self.assertEqual(served_mapping, 0)
self.assertEqual(served_fallback, 100)
self.assertEqual(tested, 100)
def testThrottling25WithForcingAndFallback(self):
(served_mapping, served_fallback, tested) = self.random_aus_test(background_rate=25, force=FORCE_MAIN_MAPPING, fallback=True)
self.assertEqual(served_mapping, 1)
self.assertEqual(served_fallback, 0)
self.assertEqual(tested, 1)
def testThrottling25WithForcingFailureAndFallback(self):
(served, fallback, tested) = self.random_aus_test(background_rate=25, force=FORCE_FALLBACK_MAPPING, fallback=True)
self.assertEqual(served, 0)
self.assertEqual(fallback, 1)
self.assertEqual(tested, 1)