Belle II Software  release-05-01-25
test_runningupdate.py
1 #!/usr/bin/env python3
2 # pylama:ignore=W0212
3 import unittest
4 import re
5 from unittest.mock import create_autospec
6 
7 import conditions_db
8 from conditions_db.runningupdate import RunningTagUpdater, RunningTagUpdaterError, RunningTagUpdateMode as Mode
9 from conditions_db.cli_management import command_tag_runningupdate
10 
11 
12 class TestMergeStaging(unittest.TestCase):
13  """Test the class to calculate updates to running tags"""
14 
15  RUNNING_BASE = {
16  # gapless payload with different revisions
17  "A": [(0, 0, 0, -1), (1, 0, 1, 10), (1, 11, 1, 11), (1, 12, -1, -1)],
18  # simple open payload
19  "B": [(0, 0, -1, -1)],
20  # simple closed payload
21  "C": [(0, 0, 0, -1)],
22  }
23 
24  def make_payloads(self, start_revision=1, **payloads):
25  """Create a list of payloads from all keyword arguments Name of the
26  payloads will be the name of the argument, revision will increase
27  monotonic starting at start_revision for each argument and the iovs are
28  the values of the the argument.
29 
30  To create a list with two payloads, A and B with two validities for B
31  and one for A one could use::
32 
33  make_payloads(A=[(0,0,1,2)], B=[(1,2,3,4), (3,5,6,7)])
34  """
35  result = []
36  for name, iovs in payloads.items():
37  for rev, iov in enumerate(iovs, start_revision):
38  result.append(conditions_db.PayloadInformation(0, name, rev, None, None, None, None, iov))
39  return result
40 
41  def create_payloads_from_text(self, text):
42  """
43  Create payloads from a plain text list of payloads in the form of
44 
45  payloadN, rev i, valid from a,b to c,d
46 
47  Return a list of lists of payloads where each block of payload definitions
48  without empty lines in between is returned as one list.
49  """
50  payloads = []
51  all_payloads = []
52  for match in re.findall(r'^\s*(payload\d+), rev (\d+), valid from (\d+),(\d+) to ([0-9-]+),([0-9-]+)$|^\s*$', text, re.M):
53  if not match[0]:
54  if payloads:
55  all_payloads.append(payloads)
56  payloads = []
57  continue
58  payloads.append(conditions_db.PayloadInformation(
59  0, match[0], int(match[1]), None, None, None, None, tuple(map(int, match[2:]))
60  ))
61 
62  if payloads:
63  all_payloads.append(payloads)
64 
65  return all_payloads
66 
67  def make_mock_db(self, running_state="RUNNING", staging_state="VALIDATED", running_payloads=None, staging_payloads=None):
68  """Create a mock object that behaves close enough to the ConditionsDB class but returns
69  fixed values for the globaltags "running" and "staging" for testing"""
70  db = create_autospec(conditions_db.ConditionsDB)
71 
72  if running_payloads is None:
73  running_payloads = []
74  if staging_payloads is None:
75  staging_payloads = []
76 
77  def taginfo(name):
78  return {
79  "running": {"name": "running", "globalTagStatus": {"name": running_state}},
80  "staging": {"name": "staging", "globalTagStatus": {"name": staging_state}},
81  }.get(name, None)
82 
83  def all_iovs(name):
84  return {
85  "running": running_payloads,
86  "staging": staging_payloads}.get(name, None)
87 
88  db.get_globalTagInfo.side_effect = taginfo
89  db.get_all_iovs.side_effect = all_iovs
90  return db
91 
92  def parse_operations(self, operations):
93  """Parse the operations into a single list of [operation type, name,
94  revision, iov] for each operation"""
95  return [[op, p.name, p.revision, p.iov] for (op, p) in operations]
96 
97  def test_states(self):
98  """Test that we get errors if the tags don't exist or are in the wrong state"""
99  db0 = self.make_mock_db()
100  db1 = self.make_mock_db(running_state="PUBLISHED")
101  db2 = self.make_mock_db(staging_state="OPEN")
102  db3 = self.make_mock_db(running_state="INVALID", staging_state="PUBLISHED")
103 
104  with self.assertRaisesRegex(RunningTagUpdaterError, "not in RUNNING state"):
105  RunningTagUpdater(db1, "running", "staging", (0, 0), Mode.ALLOW_CLOSED)
106  with self.assertRaisesRegex(RunningTagUpdaterError, "not in VALIDATED state"):
107  RunningTagUpdater(db2, "running", "staging", (0, 0), Mode.ALLOW_CLOSED)
108  with self.assertRaisesRegex(RunningTagUpdaterError, "not in RUNNING state"):
109  RunningTagUpdater(db3, "running", "staging", (0, 0), Mode.ALLOW_CLOSED)
110  with self.assertRaisesRegex(RunningTagUpdaterError, "'funning' cannot be found"):
111  RunningTagUpdater(db0, "funning", "staging", (0, 0), Mode.ALLOW_CLOSED)
112  with self.assertRaisesRegex(RunningTagUpdaterError, "'caging' cannot be found"):
113  RunningTagUpdater(db0, "running", "caging", (0, 0), Mode.ALLOW_CLOSED)
114  with self.assertRaisesRegex(RunningTagUpdaterError, "'cunning' cannot be found"):
115  RunningTagUpdater(db0, "cunning", "stacking", (0, 0), Mode.ALLOW_CLOSED)
116 
117  def test_overlaps(self):
118  """Test that we get errors on overlaps in the staging"""
119  db = self.make_mock_db()
120  updater = RunningTagUpdater(db, "running", "staging", (0, 0), Mode.ALLOW_CLOSED)
121  # no overlap between different payload names
122  payloads = self.make_payloads(A=[(0, 0, 1, 0)], B=[(0, 1, 1, 0)])
123  self.assertIsNone(updater._check_staging_tag("testoverlaps", payloads))
124 
125  # but otherwise we raise errors
126  payloads = self.make_payloads(A=[(0, 0, 1, 0), (0, 1, 1, 0)])
127  with self.assertRaises(RunningTagUpdaterError) as ctx:
128  updater._check_staging_tag("testoverlaps", payloads)
129  self.assertEqual(ctx.exception.extra_vars['overlaps'], 1)
130 
131  # yes, also a tiny overlap is an overlap
132  payloads = self.make_payloads(A=[(0, 0, 1, 0), (1, 0, 1, -1)])
133  with self.assertRaises(RunningTagUpdaterError) as ctx:
134  updater._check_staging_tag("testoverlaps", payloads)
135  self.assertEqual(ctx.exception.extra_vars['overlaps'], 1)
136 
137  # and we even raise multiple errors on multiple overlaps
138  payloads = self.make_payloads(A=[(0, 0, 1, 0), (1, 0, 1, -1), (1, 1, 1, -1)])
139  with self.assertRaises(RunningTagUpdaterError) as ctx:
140  updater._check_staging_tag("testoverlaps", payloads)
141  self.assertEqual(ctx.exception.extra_vars['overlaps'], 2)
142 
143  # and we even raise multiple errors on multiple overlaps ...
144  # also if the overlaps are not adjacent
145  payloads = self.make_payloads(A=[(0, 0, 1, 0), (3, 0, -1, -1), (1, 0, 1, -1), (2, 0, 2, -1), (1, 1, 1, -1)])
146  with self.assertRaises(RunningTagUpdaterError) as ctx:
147  updater._check_staging_tag("testoverlaps", payloads)
148  self.assertEqual(ctx.exception.extra_vars['overlaps'], 2)
149 
150  # unless there's no overlap
151  payloads = self.make_payloads(A=[(0, 0, 1, 0), (1, 1, 1, -1)])
152  self.assertIsNone(updater._check_staging_tag("testoverlaps", payloads))
153 
154  def test_gaps(self):
155  """Test that we get errors for gaps in the staging"""
156  db = self.make_mock_db()
157  updater = RunningTagUpdater(db, "running", "staging", (0, 0), Mode.ALLOW_CLOSED)
158 
159  # consecutive payloads should be merged
160  payloads = self.make_payloads(A=[(0, 0, 1, 0), (1, 1, -1, -1)])
161  self.assertIsNone(updater._check_staging_tag("testgaps", payloads))
162 
163  # and otherwise we want errors
164  payloads = self.make_payloads(A=[(0, 0, 1, 0), (1, 2, 1, -1), (2, 2, -1, -1)], B=[(0, 0, 1, 2), (2, 1, 3, 3)])
165  with self.assertRaises(RunningTagUpdaterError) as ctx:
166  updater._check_staging_tag("testgaps", payloads)
167  self.assertEqual(ctx.exception.extra_vars['gaps'], 3)
168 
169  def test_closed(self):
170  """Test that we get errors on iovs not being open in the staging"""
171  db = self.make_mock_db()
172  updater = RunningTagUpdater(db, "running", "staging", (0, 0), Mode.STRICT)
173 
174  # consecutive payloads should be merged but there could be gaps
175  payloads = self.make_payloads(A=[(0, 0, 1, 0), (1, 1, 1, -1)], B=[(0, 0, 1, 2), (1, 3, 3, 3)])
176  with self.assertRaises(RunningTagUpdaterError) as ctx:
177  updater._check_staging_tag("testclosed", payloads)
178  self.assertEqual(ctx.exception.extra_vars['closed payloads'], 2)
179 
180  # we can fix one gap and still get an error
181  payloads = self.make_payloads(A=[(0, 0, 1, -1), (2, 0, -1, -1)], B=[(0, 0, 1, 2), (1, 3, 3, 3)])
182  with self.assertRaises(RunningTagUpdaterError) as ctx:
183  updater._check_staging_tag("testclosed", payloads)
184  self.assertEqual(ctx.exception.extra_vars['closed payloads'], 1)
185 
186  # or have a different order ...
187  payloads = self.make_payloads(A=[(0, 0, 1, -1), (2, 0, 99999, 999999)], B=[(10, 0, -1, -1), (0, 0, 9, -1)])
188  with self.assertRaises(RunningTagUpdaterError) as ctx:
189  updater._check_staging_tag("testclosed", payloads)
190  self.assertEqual(ctx.exception.extra_vars['closed payloads'], 1)
191 
192  def test_start_early(self):
193  """Test that we get errors if the payloads in the staging tag start too early"""
194  db = self.make_mock_db()
195  updater = RunningTagUpdater(db, "running", "staging", (2, 5), Mode.ALLOW_CLOSED)
196 
197  # simple start early
198  payloads = self.make_payloads(A=[(2, 0, -1, -1)])
199  with self.assertRaises(RunningTagUpdaterError) as ctx:
200  updater._check_staging_tag("testearly", payloads)
201  self.assertEqual(ctx.exception.extra_vars['starts too early'], 1)
202 
203  # but starting at 0,0 is an exception and is fine
204  payloads = self.make_payloads(A=[(0, 0, -1, -1)])
205  updater._check_staging_tag("testearly", payloads)
206 
207  # overlap and gap free and starting at 0,0 ... but still wrong because
208  # the next payload starts early
209  payloads = self.make_payloads(A=[(0, 0, 1, 0), (1, 1, 1, -1)])
210  with self.assertRaises(RunningTagUpdaterError) as ctx:
211  updater._check_staging_tag("testearly", payloads)
212  self.assertEqual(ctx.exception.extra_vars['starts too early'], 1)
213 
214  payloads = self.make_payloads(A=[(1, 1, 2, 4)])
215  with self.assertRaises(RunningTagUpdaterError) as ctx:
216  updater._check_staging_tag("testearly", payloads)
217  self.assertEqual(ctx.exception.extra_vars['starts too early'], 1)
218 
219  # but allow multiple payloads after the starting point
220  payloads = self.make_payloads(A=[(0, 0, 2, 5), (2, 6, 2, 6), (2, 7, 2, -1), (3, 0, -1, -1)])
221  updater._check_staging_tag("testearly", payloads)
222 
223  # check order dependence
224  payloads = self.make_payloads(A=[(3, 0, -1, -1), (2, 4, 2, -1)])
225  with self.assertRaises(RunningTagUpdaterError) as ctx:
226  updater._check_staging_tag("testearly", payloads)
227  self.assertEqual(ctx.exception.extra_vars['starts too early'], 1)
228 
229  def test_simple_mode(self):
230  """Test that we only allow trivial iovs in simple mode"""
231  db = self.make_mock_db()
232  updater = RunningTagUpdater(db, "running", "staging", (2, 5), Mode.SIMPLE)
233 
234  # One payload only but not the correct one
235  payloads = self.make_payloads(A=[(1, 0, -1, -1)])
236  with self.assertRaises(RunningTagUpdaterError) as ctx:
237  updater._check_staging_tag_simple("testsimple", payloads)
238  self.assertEqual(ctx.exception.extra_vars['wrong validity'], 1)
239 
240  payloads = self.make_payloads(A=[(0, 1, -1, -1)])
241  with self.assertRaises(RunningTagUpdaterError) as ctx:
242  updater._check_staging_tag_simple("testsimple", payloads)
243  self.assertEqual(ctx.exception.extra_vars['wrong validity'], 1)
244 
245  payloads = self.make_payloads(A=[(0, 0, 1, -1)])
246  with self.assertRaises(RunningTagUpdaterError) as ctx:
247  updater._check_staging_tag_simple("testsimple", payloads)
248  self.assertEqual(ctx.exception.extra_vars['wrong validity'], 1)
249 
250  payloads = self.make_payloads(A=[(0, 0, 1, 1)])
251  with self.assertRaises(RunningTagUpdaterError) as ctx:
252  updater._check_staging_tag_simple("testsimple", payloads)
253  self.assertEqual(ctx.exception.extra_vars['wrong validity'], 1)
254 
255  # but 0,0,-1,-1 is fine
256  payloads = self.make_payloads(A=[(0, 0, -1, -1)], B=[(0, 0, -1, -1)])
257  updater._check_staging_tag_simple("testsimple", payloads)
258 
259  # but only one per payload
260  payloads = self.make_payloads(A=[(0, 0, -1, -1), (0, 0, -1, -1), (1, 2, 3, 4)],
261  B=[(0, 0, -1, -1)], C=[(0, 1, 2, 3)])
262  with self.assertRaises(RunningTagUpdaterError) as ctx:
263  updater._check_staging_tag_simple("testsimple", payloads)
264  self.assertEqual(ctx.exception.extra_vars['wrong validity'], 2)
265  self.assertEqual(ctx.exception.extra_vars['duplicate payloads'], 2)
266 
268  """Test that we get errors on iovs not being open in the staging"""
269  db = self.make_mock_db()
270  updater = RunningTagUpdater(db, "running", "staging", (2, 5), Mode.STRICT)
271 
272  # only check we do is that payloads end before the first valid run or are open
273  payloads = self.make_payloads(A=[(0, 0, 1, -1), (2, 0, 2, 4)], B=[(0, 0, -1, -1)])
274  updater._check_running_tag("testrunning", payloads)
275 
276  # so we want an error even if anything closes on the same run
277  payloads = self.make_payloads(A=[(0, 0, 1, -1), (2, 0, 2, 5)], B=[(0, 0, -1, -1)], C=[(0, 0, 10, -1)])
278  with self.assertRaisesRegex(RunningTagUpdaterError, "Given first valid run conflicts with running tag") as ctx:
279  updater._check_running_tag("testrunning", payloads)
280  self.assertEqual(ctx.exception.extra_vars['payloads end after first valid run'], 2)
281 
282  # also complain if it starts after the first valid run
283  payloads = self.make_payloads(A=[(2, 5, -1, -1)], B=[(0, 0, 2, 1), (2, 2, -1, -1)], C=[(10, 0, -1, -1)])
284  with self.assertRaisesRegex(RunningTagUpdaterError, "Given first valid run conflicts with running tag") as ctx:
285  updater._check_running_tag("testrunning", payloads)
286  self.assertEqual(ctx.exception.extra_vars['payloads start after first valid run'], 2)
287 
288  def test_operations_simple(self):
289  running = self.make_payloads(**self.RUNNING_BASE)
290  staging = self.make_payloads(
291  A=[(0, 0, -1, -1)],
292  B=[(0, 0, -1, -1)],
293  C=[(0, 0, -1, -1)],
294  start_revision=10)
295  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
296  updater = RunningTagUpdater(db, "running", "staging", (2, 0), Mode.STRICT)
297  result = updater.calculate_update()
298  self.assertEqual(self.parse_operations(result),
299  [
300  ['CLOSE', 'A', 4, (1, 12, 1, -1)],
301  ['CLOSE', 'B', 1, (0, 0, 1, -1)],
302  ['CREATE', 'A', 10, (2, 0, -1, -1)],
303  ['CREATE', 'B', 10, (2, 0, -1, -1)],
304  ['CREATE', 'C', 10, (2, 0, -1, -1)],
305  ])
306 
308  """
309  In full mode we close B even if there's nothing in the staging tag
310  to allow "removing" payloads from the set of valid ones
311  """
312  running = self.make_payloads(**self.RUNNING_BASE)
313  staging = self.make_payloads(
314  A=[(0, 0, -1, -1)],
315  C=[(0, 0, -1, -1)],
316  start_revision=10)
317  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
318  updater = RunningTagUpdater(db, "running", "staging", (2, 0), Mode.FULL_REPLACEMENT)
319  result = updater.calculate_update()
320  self.assertEqual(self.parse_operations(result),
321  [
322  ['CLOSE', 'A', 4, (1, 12, 1, -1)],
323  ['CLOSE', 'B', 1, (0, 0, 1, -1)],
324  ['CREATE', 'A', 10, (2, 0, -1, -1)],
325  ['CREATE', 'C', 10, (2, 0, -1, -1)],
326  ])
327 
329  """This time B doesn't start on the spot but a bit later. And A is a list of iovs to be preserved"""
330  running = self.make_payloads(**self.RUNNING_BASE)
331  staging = self.make_payloads(
332  A=[(0, 0, 2, 5), (2, 6, 2, 10), (2, 11, -1, -1)],
333  B=[(2, 8, -1, -1)],
334  C=[(2, 0, -1, -1)],
335  start_revision=10)
336  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
337  updater = RunningTagUpdater(db, "running", "staging", (2, 0), Mode.STRICT)
338  result = updater.calculate_update()
339  self.assertEqual(self.parse_operations(result),
340  [
341  ['CLOSE', 'A', 4, (1, 12, 1, -1)],
342  ['CLOSE', 'B', 1, (0, 0, 2, 7)],
343  ['CREATE', 'A', 10, (2, 0, 2, 5)],
344  ['CREATE', 'A', 11, (2, 6, 2, 10)],
345  ['CREATE', 'A', 12, (2, 11, -1, -1)],
346  ['CREATE', 'B', 10, (2, 8, -1, -1)],
347  ['CREATE', 'C', 10, (2, 0, -1, -1)],
348  ])
349 
351  """If the staging payloads have same revision as the last one in running merge them"""
352  running = self.make_payloads(**self.RUNNING_BASE)
353  staging = self.make_payloads(
354  A=[(0, 0, 2, 5), (2, 6, 2, 10), (2, 11, -1, -1)],
355  B=[(2, 8, -1, -1)],
356  C=[(2, 0, -1, -1)],
357  start_revision=10)
358  # we nee to have the same revision for merging though so fudge a bit
359  staging[0].revision = 4
360  staging[-2].revision = 1
361  print([(e.name, e.revision, e.iov) for e in running])
362  print([(e.name, e.revision, e.iov) for e in staging])
363  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
364  updater = RunningTagUpdater(db, "running", "staging", (2, 0), Mode.STRICT)
365  result = updater.calculate_update()
366  self.assertEqual(self.parse_operations(result),
367  [
368  ['CLOSE', 'A', 4, (1, 12, 2, 5)],
369  ['CREATE', 'A', 11, (2, 6, 2, 10)],
370  ['CREATE', 'A', 12, (2, 11, -1, -1)],
371  ['CREATE', 'C', 10, (2, 0, -1, -1)],
372  ])
373 
375  """Test automatic opening of the last iov if necessary"""
376  running = self.make_payloads(**self.RUNNING_BASE)
377  staging = self.make_payloads(
378  A=[(0, 0, 2, 5), (2, 6, 2, 10), (2, 11, 2, -1)],
379  B=[(2, 8, 2, 10), (2, 11, 2, 28)],
380  C=[(2, 0, 2, 100)],
381  start_revision=10)
382  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
383  updater = RunningTagUpdater(db, "running", "staging", (2, 0), Mode.FIX_CLOSED)
384  result = updater.calculate_update()
385  self.assertEqual(self.parse_operations(result),
386  [
387  ['CLOSE', 'A', 4, (1, 12, 1, -1)],
388  ['CLOSE', 'B', 1, (0, 0, 2, 7)],
389  ['CREATE', 'A', 10, (2, 0, 2, 5)],
390  ['CREATE', 'A', 11, (2, 6, 2, 10)],
391  ['CREATE', 'A', 12, (2, 11, -1, -1)],
392  ['CREATE', 'B', 10, (2, 8, 2, 10)],
393  ['CREATE', 'B', 11, (2, 11, -1, -1)],
394  ['CREATE', 'C', 10, (2, 0, -1, -1)],
395  ])
396 
398  """Test that merging fails if we have gaps or overlaps"""
399  running = self.make_payloads(**self.RUNNING_BASE)
400  staging = self.make_payloads(
401  A=[(0, 0, 2, 5), (2, 5, 2, 10), (2, 11, -1, -1)],
402  B=[(2, 8, 2, 10), (2, 12, -1, -1)],
403  C=[(2, 0, -1, -1)],
404  start_revision=10)
405  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
406  updater = RunningTagUpdater(db, "running", "staging", (2, 0), Mode.STRICT)
407  with self.assertRaisesRegex(RunningTagUpdaterError, "tag 'staging' not fit for update") as ctx:
408  updater.calculate_update()
409  # check that we have one overlap and one gap using subset comparison logic
410  self.assertLessEqual({'overlaps': 1, 'gaps': 1}.items(), ctx.exception.extra_vars.items())
411 
412  def test_doc_example(self):
413  """Extract the example from the `b2conditionsdb tag runningupdate` docstring and run it"""
414  running, staging, expected = self.create_payloads_from_text(command_tag_runningupdate.__doc__)
415  # make a copy of the running payloads just to be able to compare later
416  result = running[:]
417  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
418  updater = RunningTagUpdater(db, "running", "staging", (1, 2), Mode.ALLOW_CLOSED)
419  operations = updater.calculate_update()
420  # check that the update does what we actually wrote in the doc. This will
421  # fail if the docs are updated.
422  self.assertEqual(self.parse_operations(operations),
423  [
424  ['CLOSE', 'payload1', 2, (1, 1, 1, 1)],
425  ['CLOSE', 'payload2', 1, (0, 1, 1, 4)],
426  ['CLOSE', 'payload4', 1, (0, 1, 1, 20)],
427  ['CREATE', 'payload1', 3, (1, 2, 1, 8)],
428  ['CREATE', 'payload1', 4, (1, 9, 1, 20)],
429  ['CREATE', 'payload2', 2, (1, 5, 1, 20)],
430  ['CREATE', 'payload3', 2, (1, 2, -1, -1)],
431  ])
432  # and then apply the operations to the input list and make sure that is
433  # what's written in the doc
434  for op, payload in operations:
435  if op == "CLOSE":
436  # adjust the end of the iov
437  i = result.index(payload)
438  result[i].iov = result[i].iov[:2] + payload.iov[2:]
439  else:
440  # or just add the full payload
441  result.append(payload)
442  result.sort()
443  self.assertEqual(expected, result)
444 
446  """Extract the example from the `b2conditionsdb tag runningupdate` docstring and run it"""
447  running, staging, expected = self.create_payloads_from_text(command_tag_runningupdate.__doc__)
448  # make a copy of the running payloads just to be able to compare later
449  result = running[:]
450  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
451  updater = RunningTagUpdater(db, "running", "staging", (1, 2), Mode.FULL_REPLACEMENT)
452  operations = updater.calculate_update()
453  # check that the update does what we actually wrote in the doc. This will
454  # fail if the docs are updated.
455  self.assertEqual(self.parse_operations(operations),
456  [
457  ['CLOSE', 'payload1', 2, (1, 1, 1, 1)],
458  ['CLOSE', 'payload2', 1, (0, 1, 1, 4)],
459  ['CLOSE', 'payload4', 1, (0, 1, 1, 20)],
460  ['CLOSE', 'payload5', 1, (0, 1, 1, 1)],
461  ['CREATE', 'payload1', 3, (1, 2, 1, 8)],
462  ['CREATE', 'payload1', 4, (1, 9, 1, 20)],
463  ['CREATE', 'payload2', 2, (1, 5, 1, 20)],
464  ['CREATE', 'payload3', 2, (1, 2, -1, -1)],
465  ])
466 
468  """Extract the example from the `b2conditionsdb tag runningupdate` docstring and run it"""
469  running, staging, expected = self.create_payloads_from_text(command_tag_runningupdate.__doc__)
470  # make a copy of the running payloads just to be able to compare later
471  result = running[:]
472  db = self.make_mock_db(running_payloads=running, staging_payloads=staging)
473  updater = RunningTagUpdater(db, "running", "staging", (1, 2), Mode.FIX_CLOSED)
474  operations = updater.calculate_update()
475  # check that the update does what we actually wrote in the doc. This will
476  # fail if the docs are updated.
477  self.assertEqual(self.parse_operations(operations),
478  [
479  ['CLOSE', 'payload1', 2, (1, 1, 1, 1)],
480  ['CLOSE', 'payload2', 1, (0, 1, 1, 4)],
481  ['CREATE', 'payload1', 3, (1, 2, 1, 8)],
482  ['CREATE', 'payload1', 4, (1, 9, -1, -1)],
483  ['CREATE', 'payload2', 2, (1, 5, -1, -1)],
484  ['CREATE', 'payload3', 2, (1, 2, -1, -1)],
485  ])
486 
487 
488 if __name__ == "__main__":
489  # test everything
490  unittest.main()
conditions_db.cli_management
Definition: cli_management.py:1
test_runningupdate.TestMergeStaging.test_merge_overlapandgaps
def test_merge_overlapandgaps(self)
Definition: test_runningupdate.py:397
test_runningupdate.TestMergeStaging.test_doc_example_full
def test_doc_example_full(self)
Definition: test_runningupdate.py:445
conditions_db.runningupdate.RunningTagUpdater
Definition: runningupdate.py:47
test_runningupdate.TestMergeStaging.test_simple_mode
def test_simple_mode(self)
Definition: test_runningupdate.py:229
test_runningupdate.TestMergeStaging.test_doc_example_fixclosed
def test_doc_example_fixclosed(self)
Definition: test_runningupdate.py:467
test_runningupdate.TestMergeStaging.make_mock_db
def make_mock_db(self, running_state="RUNNING", staging_state="VALIDATED", running_payloads=None, staging_payloads=None)
Definition: test_runningupdate.py:67
conditions_db.runningupdate
Definition: runningupdate.py:1
test_runningupdate.TestMergeStaging.create_payloads_from_text
def create_payloads_from_text(self, text)
Definition: test_runningupdate.py:41
test_runningupdate.TestMergeStaging.test_doc_example
def test_doc_example(self)
Definition: test_runningupdate.py:412
test_runningupdate.TestMergeStaging.test_operations_full
def test_operations_full(self)
Definition: test_runningupdate.py:307
test_runningupdate.TestMergeStaging.test_running_check
def test_running_check(self)
Definition: test_runningupdate.py:267
test_runningupdate.TestMergeStaging.test_operations_fix_open
def test_operations_fix_open(self)
Definition: test_runningupdate.py:374
test_runningupdate.TestMergeStaging.test_states
def test_states(self)
Definition: test_runningupdate.py:97
test_runningupdate.TestMergeStaging.make_payloads
def make_payloads(self, start_revision=1, **payloads)
Definition: test_runningupdate.py:24
test_runningupdate.TestMergeStaging.parse_operations
def parse_operations(self, operations)
Definition: test_runningupdate.py:92
test_runningupdate.TestMergeStaging.test_start_early
def test_start_early(self)
Definition: test_runningupdate.py:192
test_runningupdate.TestMergeStaging
Definition: test_runningupdate.py:12
test_runningupdate.TestMergeStaging.test_operations_ragged
def test_operations_ragged(self)
Definition: test_runningupdate.py:328
test_runningupdate.TestMergeStaging.test_gaps
def test_gaps(self)
Definition: test_runningupdate.py:154
conditions_db.ConditionsDB
Definition: __init__.py:144
test_runningupdate.TestMergeStaging.test_closed
def test_closed(self)
Definition: test_runningupdate.py:169
test_runningupdate.TestMergeStaging.test_overlaps
def test_overlaps(self)
Definition: test_runningupdate.py:117
test_runningupdate.TestMergeStaging.test_operations_extend
def test_operations_extend(self)
Definition: test_runningupdate.py:350
test_runningupdate.TestMergeStaging.RUNNING_BASE
dictionary RUNNING_BASE
Basic definition of a running tag to be used later.
Definition: test_runningupdate.py:15
conditions_db.PayloadInformation
Definition: __init__.py:49