Coverage for tests / test_tmaUtils.py: 17%
357 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 00:33 +0000
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 00:33 +0000
1# This file is part of summit_utils.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <https://www.gnu.org/licenses/>.
22"""Test cases for utils."""
24import asyncio
25import os
26import unittest
28import matplotlib.pyplot as plt
29import numpy as np
30import pandas as pd
31from astropy.time import TimeDelta
32from utils import getVcr
34import lsst.utils.tests
35from lsst.summit.utils.dateTime import calcNextDay, getDayObsStartTime
36from lsst.summit.utils.efdUtils import makeEfdClient
37from lsst.summit.utils.enums import PowerState
38from lsst.summit.utils.tmaUtils import (
39 AxisMotionState,
40 TMAEvent,
41 TMAEventMaker,
42 TMAState,
43 TMAStateMachine,
44 _initializeTma,
45 filterBadValues,
46 getAxisAndType,
47 getAzimuthElevationDataForEvent,
48 getCommandsDuringEvent,
49 getSlewsFromEventList,
50 getTracksFromEventList,
51 plotEvent,
52)
54__all__ = [
55 "writeNewTmaEventTestTruthValues",
56]
58TESTDIR = os.path.abspath(os.path.dirname(__file__))
59vcr = getVcr()
62def getTmaEventTestTruthValues():
63 """Get the current truth values for the TMA event test cases.
65 Returns
66 -------
67 seqNums : `np.array` of `int`
68 The sequence numbers of the events.
69 startRows : `np.array` of `int`
70 The _startRow numbers of the events.
71 endRows : `np.array` of `int`
72 The _endRow numbers of the events.
73 types : `np.array` of `str`
74 The event types, as a string, i.e. the ``TMAEvent.name`` of the event's
75 ``event.type``.
76 endReasons : `np.array` of `str`
77 The event end reasons, as a string, i.e. the ``TMAEvent.name`` of the
78 event's ``event.endReason``.
79 """
80 dataFilename = os.path.join(TESTDIR, "data", "tmaEventData.txt")
82 seqNums, startRows, endRows, types, endReasons = np.genfromtxt(
83 dataFilename, delimiter=",", dtype=None, names=True, encoding="utf-8", unpack=True
84 )
85 return seqNums, startRows, endRows, types, endReasons
88def writeNewTmaEventTestTruthValues():
89 """This function is used to write out the truth values for the test cases.
91 If the internal event creation logic changes, these values can change, and
92 will need to be updated. Run this function, and check the new values into
93 git.
95 Note: if you have cause to update values with this function, make sure to
96 update the version number on the TMAEvent class.
97 """
98 dayObs = 20241210 # obviously must match the day in the test class
100 eventMaker = TMAEventMaker()
101 events = eventMaker.getEvents(dayObs)
103 dataFilename = os.path.join(TESTDIR, "data", "tmaEventData.txt")
105 columnHeader = "seqNum,startRow,endRow,type,endReason"
106 with open(dataFilename, "w") as f:
107 f.write(columnHeader + "\n")
108 for event in events:
109 line = (
110 f"{event.seqNum},{event._startRow},{event._endRow},{event.type.name},"
111 f"{event.endReason.name}"
112 )
113 f.write(line + "\n")
116def makeValid(tma):
117 """Helper function to turn a TMA into a valid state."""
118 for name, value in tma._parts.items():
119 if value == tma._UNINITIALIZED_VALUE:
120 tma._parts[name] = 1
123def _turnOn(tma):
124 """Helper function to turn TMA axes on for testing.
126 Do not call directly in normal usage or code, as this just arbitrarily
127 sets values to turn the axes on.
129 Parameters
130 ----------
131 tma : `lsst.summit.utils.tmaUtils.TMAStateMachine`
132 The TMA state machine model to initialize.
133 """
134 tma._parts["azimuthSystemState"] = PowerState.ON
135 tma._parts["elevationSystemState"] = PowerState.ON
138class TmaUtilsTestCase(lsst.utils.tests.TestCase):
139 def test_tmaInit(self):
140 tma = TMAStateMachine()
141 self.assertFalse(tma._isValid)
143 # setting one axis should not make things valid
144 tma._parts["azimuthMotionState"] = 1
145 self.assertFalse(tma._isValid)
147 # setting all the other components should make things valid
148 tma._parts["azimuthInPosition"] = 1
149 tma._parts["azimuthSystemState"] = 1
150 tma._parts["elevationInPosition"] = 1
151 tma._parts["elevationMotionState"] = 1
152 tma._parts["elevationSystemState"] = 1
153 self.assertTrue(tma._isValid)
155 def test_tmaReferences(self):
156 """Check the linkage between the component lists and the _parts
157 dict.
158 """
159 tma = TMAStateMachine()
161 # setting one axis should not make things valid
162 self.assertEqual(tma._parts["azimuthMotionState"], tma._UNINITIALIZED_VALUE)
163 self.assertEqual(tma._parts["elevationMotionState"], tma._UNINITIALIZED_VALUE)
164 tma.motion[0] = AxisMotionState.TRACKING # set azimuth to 0
165 tma.motion[1] = AxisMotionState.TRACKING # set azimuth to 0
166 self.assertEqual(tma._parts["azimuthMotionState"], AxisMotionState.TRACKING)
167 self.assertEqual(tma._parts["elevationMotionState"], AxisMotionState.TRACKING)
169 def test_getAxisAndType(self):
170 # check both the long and short form names work
171 for s in ["azimuthMotionState", "lsst.sal.MTMount.logevent_azimuthMotionState"]:
172 self.assertEqual(getAxisAndType(s), ("azimuth", "MotionState"))
174 # check in position, and use elevation instead of azimuth to test that
175 for s in ["elevationInPosition", "lsst.sal.MTMount.logevent_elevationInPosition"]:
176 self.assertEqual(getAxisAndType(s), ("elevation", "InPosition"))
178 for s in ["azimuthSystemState", "lsst.sal.MTMount.logevent_azimuthSystemState"]:
179 self.assertEqual(getAxisAndType(s), ("azimuth", "SystemState"))
181 def test_initStateLogic(self):
182 tma = TMAStateMachine()
183 self.assertFalse(tma._isValid)
184 self.assertFalse(tma.isMoving)
185 self.assertFalse(tma.canMove)
186 self.assertFalse(tma.isTracking)
187 self.assertFalse(tma.isSlewing)
188 self.assertEqual(tma.state, TMAState.UNINITIALIZED)
190 _initializeTma(tma) # we're valid, but still aren't moving and can't
191 self.assertTrue(tma._isValid)
192 self.assertNotEqual(tma.state, TMAState.UNINITIALIZED)
193 self.assertTrue(tma.canMove)
194 self.assertTrue(tma.isNotMoving)
195 self.assertFalse(tma.isMoving)
196 self.assertFalse(tma.isTracking)
197 self.assertFalse(tma.isSlewing)
199 _turnOn(tma) # can now move, still valid, but not in motion
200 self.assertTrue(tma._isValid)
201 self.assertTrue(tma.canMove)
202 self.assertTrue(tma.isNotMoving)
203 self.assertFalse(tma.isMoving)
204 self.assertFalse(tma.isTracking)
205 self.assertFalse(tma.isSlewing)
207 # consider manipulating the axes by hand here and testing these?
208 # it's likely not worth it, given how much this exercised elsewhere,
209 # but these are the only functions not yet being directly tested
210 # tma._axesInFault()
211 # tma._axesOff()
212 # tma._axesOn()
213 # tma._axesInMotion()
214 # tma._axesTRACKING()
215 # tma._axesInPosition()
218@vcr.use_cassette()
219class TMAEventMakerTestCase(lsst.utils.tests.TestCase):
220 @classmethod
221 @vcr.use_cassette()
222 def setUpClass(cls):
223 try:
224 cls.client = makeEfdClient(testing=True)
225 except RuntimeError:
226 raise unittest.SkipTest("Could not instantiate an EFD client")
228 cls.dayObs = 20241210
229 cls.dayObsWithBlockInfo = 20230615
230 # get a sample expRecord here to test expRecordToTimespan
231 cls.tmaEventMaker = TMAEventMaker(cls.client)
232 cls.events = cls.tmaEventMaker.getEvents(cls.dayObs) # does the fetch
233 cls.sampleData = cls.tmaEventMaker._data[cls.dayObs] # pull the data from the object and test length
235 @vcr.use_cassette()
236 def tearDown(self):
237 loop = asyncio.get_event_loop()
238 if self.client.influx_client is not None:
239 loop.run_until_complete(self.client.influx_client.close())
241 @vcr.use_cassette()
242 def test_events(self):
243 data = self.sampleData
244 self.assertIsInstance(data, pd.DataFrame)
245 self.assertEqual(len(data), 800)
247 @vcr.use_cassette()
248 def test_rowDataForValues(self):
249 rowsFor = set(self.sampleData["rowFor"])
250 self.assertEqual(len(rowsFor), 6)
252 # hard coding these ensures that you can't extend the axes/model
253 # without being explicit about it here.
254 correct = {
255 "azimuthInPosition",
256 "azimuthMotionState",
257 "azimuthSystemState",
258 "elevationInPosition",
259 "elevationMotionState",
260 "elevationSystemState",
261 }
262 self.assertSetEqual(rowsFor, correct)
264 @vcr.use_cassette()
265 def test_monotonicTimeInDataframe(self):
266 # ensure that each row is later than the previous
267 times = self.sampleData["private_efdStamp"]
268 self.assertTrue(np.all(np.diff(times) > 0))
270 @vcr.use_cassette()
271 def test_monotonicTimeApplicationOfRows(self):
272 # ensure you can apply rows in the correct order
273 tma = TMAStateMachine()
274 row1 = self.sampleData.iloc[0]
275 row2 = self.sampleData.iloc[1]
277 # just running this check it is OK
278 tma.apply(row1)
279 tma.apply(row2)
281 # and that if you apply them in reverse order then things will raise
282 tma = TMAStateMachine()
283 with self.assertRaises(ValueError):
284 tma.apply(row2)
285 tma.apply(row1)
287 @vcr.use_cassette()
288 def test_fullDaySequence(self):
289 # make sure we can apply all the data from the day without falling
290 # through the logic sieve
291 for engineering in (True, False):
292 tma = TMAStateMachine(engineeringMode=engineering)
294 _initializeTma(tma)
296 for rowNum, row in self.sampleData.iterrows():
297 tma.apply(row)
299 @vcr.use_cassette()
300 def test_endToEnd(self):
301 eventMaker = self.tmaEventMaker
302 events = eventMaker.getEvents(self.dayObs)
303 self.assertIsInstance(events, list)
304 self.assertEqual(len(events), 320)
305 self.assertIsInstance(events[0], TMAEvent)
307 slews = [e for e in events if e.type == TMAState.SLEWING]
308 tracks = [e for e in events if e.type == TMAState.TRACKING]
309 self.assertEqual(len(slews), 172)
310 self.assertEqual(len(tracks), 148)
312 seqNums, startRows, endRows, types, endReasons = getTmaEventTestTruthValues()
313 for eventNum, event in enumerate(events):
314 self.assertEqual(event.seqNum, seqNums[eventNum])
315 self.assertEqual(event._startRow, startRows[eventNum])
316 self.assertEqual(event._endRow, endRows[eventNum])
317 self.assertEqual(event.type.name, types[eventNum])
318 self.assertEqual(event.endReason.name, endReasons[eventNum])
320 eventSet = set(slews) # check we can hash
321 eventSet.update(slews) # check it ignores duplicates
322 self.assertEqual(len(eventSet), len(slews))
324 @vcr.use_cassette()
325 def test_noDataBehaviour(self):
326 eventMaker = self.tmaEventMaker
327 noDataDayObs = 19600101 # do not use 19700101 - there is data for that day!
328 with self.assertLogs(level="WARNING") as cm:
329 correctMsg = f"No EFD data found for dayObs={noDataDayObs}"
330 events = eventMaker.getEvents(noDataDayObs)
331 self.assertIsInstance(events, list)
332 self.assertEqual(len(events), 0)
333 msg = cm.output[0]
334 self.assertIn(correctMsg, msg)
336 @vcr.use_cassette()
337 def test_helperFunctions(self):
338 eventMaker = self.tmaEventMaker
339 events = eventMaker.getEvents(self.dayObs)
341 slews = [e for e in events if e.type == TMAState.SLEWING]
342 tracks = [e for e in events if e.type == TMAState.TRACKING]
343 foundSlews = getSlewsFromEventList(events)
344 foundTracks = getTracksFromEventList(events)
345 self.assertEqual(slews, foundSlews)
346 self.assertEqual(tracks, foundTracks)
348 def test_filterBadValues(self):
349 # NB: if you add enough spurious values that the median is no longer
350 # the value around which your "good" values are oscillating the first
351 # two points will get replaced and this can be very confusing!
353 # test no bad values
354 # mean = median = 1.0
355 values = np.array([1.0, 0.96, 1.0, 1.04, 0.95, 1.0, 1.05, 1.0, 1.05, 1.0, 0.95])
356 mean = np.mean(values)
357 nReplaced = filterBadValues(values)
358 self.assertEqual(nReplaced, 0)
359 self.assertEqual(np.mean(values), mean)
361 # test with one bad values
362 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 1.0, 1.05, 1.0, 1.05, 1.0, 0.95])
363 nReplaced = filterBadValues(values)
364 self.assertEqual(nReplaced, 1)
366 # test with two consecutive bad values
367 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 1.05, 1.0, 1.05, 1.0, 0.95])
368 nReplaced = filterBadValues(values)
369 self.assertEqual(nReplaced, 2)
371 # test with three consecutive bad values
372 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 1.0, 1.05, 1.0, 0.95])
373 nReplaced = filterBadValues(values)
374 self.assertEqual(nReplaced, 3)
376 # test with three consecutive bad values and another at the end
377 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 1.0, 1.05, 1.0, 3.95])
378 nReplaced = filterBadValues(values)
379 self.assertEqual(nReplaced, 4)
381 # test with more than three consecutive bad values
382 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 0.95])
383 nReplaced = filterBadValues(values)
384 self.assertEqual(nReplaced, 3)
385 self.assertIn(5.0, values) # check the last bad value is still there specifically
387 # test with more than three consecutive bad values and another bad
388 # value at the end
389 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 2.95])
390 nReplaced = filterBadValues(values)
391 self.assertEqual(nReplaced, 4)
393 # test with bad values in first two positions
394 values = np.array([2.0, 1.96, 1.0, 1.04, 0.95, 1.0, 1.05, 1.0, 1.05, 1.0, 0.95]) # median = 1.0
395 nReplaced = filterBadValues(values)
396 self.assertEqual(nReplaced, 2)
398 # test with bad values in first two positions and one in the middle
399 values = np.array([2.0, 1.96, 1.0, 1.04, 0.95, 5.0, 1.04, 1.0, 1.05, 1.0, 0.95])
400 nReplaced = filterBadValues(values)
401 self.assertEqual(nReplaced, 3)
403 # check that the last two good values are always used for correction,
404 # including when there are more than three consecutive bad values.
405 values = np.array([1.0, 0.96, 1.0, 1.02, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 2.95])
406 expected = np.array([1.0, 0.96, 1.0, 1.02, 1.01, 1.01, 1.01, 5.0, 1.05, 1.0, 1.025])
407 nReplaced = filterBadValues(values)
408 residuals = np.abs(values - expected)
409 self.assertEqual(nReplaced, 4)
410 self.assertTrue(np.all(residuals < 1e-6))
412 # check with one good point after an overflowing run of bad to make
413 # sure the correction is always applied with good values, not the naive
414 # average of the last two even if they might be bad
415 values = np.array([1.0, 0.96, 1.0, 1.02, 2.95, 3.0, 4.05, 5.0, 1.05, 2.95, 1.0])
416 expected = np.array([1.0, 0.96, 1.0, 1.02, 1.01, 1.01, 1.01, 5.0, 1.05, 1.035, 1.0])
417 nReplaced = filterBadValues(values)
418 residuals = np.abs(values - expected)
419 self.assertEqual(nReplaced, 4)
420 self.assertTrue(np.all(residuals < 1e-6))
422 # check with non-default maxDelta
423 values = np.array([1.0, 0.96, 1.0, 1.02, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 2.95])
424 nReplaced = filterBadValues(values, maxDelta=10)
425 self.assertEqual(nReplaced, 0)
427 values = np.array(
428 [
429 1.0,
430 1.0,
431 1.0,
432 1.1,
433 1.0,
434 1.0,
435 1.0,
436 1.0,
437 1.0,
438 1.0,
439 ]
440 )
441 nReplaced = filterBadValues(values, maxDelta=0.01)
442 self.assertEqual(nReplaced, 1)
444 @vcr.use_cassette()
445 def test_getEvent(self):
446 # test the singular event getter, and what happens if the event doesn't
447 # exist for the day
448 eventMaker = self.tmaEventMaker
449 events = eventMaker.getEvents(self.dayObs)
450 nEvents = len(events)
452 event = eventMaker.getEvent(self.dayObs, 0)
453 self.assertIsInstance(event, TMAEvent)
454 self.assertEqual(event, events[0])
455 event = eventMaker.getEvent(self.dayObs, 100)
456 self.assertIsInstance(event, TMAEvent)
457 self.assertEqual(event, events[100])
459 with self.assertLogs(level="WARNING") as cm:
460 correctMsg = f"Event {nEvents + 1} not found for {self.dayObs}"
461 event = eventMaker.getEvent(self.dayObs, nEvents + 1)
462 msg = cm.output[0]
463 self.assertIn(correctMsg, msg)
465 @vcr.use_cassette()
466 def test_printing(self):
467 eventMaker = self.tmaEventMaker
468 events = eventMaker.getEvents(self.dayObs)
470 # test str(), repr(), and _ipython_display_() for an event
471 print(str(events[0]))
472 print(repr(events[0]))
473 print(events[0]._ipython_display_())
475 # spot-check both a slow and a track to print
476 slews = [e for e in events if e.type == TMAState.SLEWING]
477 tracks = [e for e in events if e.type == TMAState.TRACKING]
478 eventMaker.printEventDetails(slews[0])
479 eventMaker.printEventDetails(tracks[0])
480 eventMaker.printEventDetails(events[-1])
482 # check the full day trick works
483 eventMaker.printFullDayStateEvolution(self.dayObs)
485 tma = TMAStateMachine()
486 _initializeTma(tma) # the uninitialized state contains wrong types for printing
487 eventMaker.printTmaDetailedState(tma)
489 @vcr.use_cassette()
490 def test_getAxisData(self):
491 eventMaker = self.tmaEventMaker
492 events = eventMaker.getEvents(self.dayObs)
494 azData, elData = getAzimuthElevationDataForEvent(self.client, events[0])
495 self.assertIsInstance(azData, pd.DataFrame)
496 self.assertIsInstance(elData, pd.DataFrame)
498 paddedAzData, paddedElData = getAzimuthElevationDataForEvent(
499 self.client, events[0], prePadding=2, postPadding=1
500 )
501 self.assertGreater(len(paddedAzData), len(azData))
502 self.assertGreater(len(paddedElData), len(elData))
504 # just check this doesn't raise when called, and check we can pass the
505 # data in
506 plotEvent(self.client, events[0], azimuthData=azData, elevationData=elData)
508 @vcr.use_cassette()
509 def test_plottingAndCommands(self):
510 eventMaker = self.tmaEventMaker
511 events = eventMaker.getEvents(self.dayObs)
512 event = events[10] # this one has commands, and we'll check that later
514 # check we _can_ plot without a figure, and then stop doing that
515 plotEvent(self.client, event)
517 fig = plt.figure(figsize=(10, 8))
518 # just check this doesn't raise when called
519 plotEvent(self.client, event, fig=fig)
520 plt.close(fig)
522 commandsToPlot = ["raDecTarget", "moveToTarget", "startTracking", "stopTracking"]
523 commands = getCommandsDuringEvent(self.client, event, commandsToPlot, doLog=False)
524 self.assertTrue(not all([time is None for time in commands.values()])) # at least one command
526 plotEvent(self.client, event, fig=fig, commands=commands)
528 del fig
530 @vcr.use_cassette()
531 def test_findEvent(self):
532 eventMaker = self.tmaEventMaker
533 # addBlockInfo=True because it shouldn't affect the comparison, and
534 # this also then ensures that the code is exercised too
535 events = eventMaker.getEvents(self.dayObs, addBlockInfo=True)
536 event = events[28] # this one has a contiguous event before it
538 time = event.begin
539 found = eventMaker.findEvent(time)
540 self.assertEqual(found, event)
542 dt = TimeDelta(0.01, format="sec")
543 # must be just inside to get the same event back, because if a moment
544 # is shared it gives the one which starts with the moment (whilst
545 # logging info messages about it)
546 time = event.end - dt
547 found = eventMaker.findEvent(time)
548 self.assertEqual(found, event)
550 # now check that if we're a hair after, we don't get the same event
551 time = event.end + dt
552 found = eventMaker.findEvent(time)
553 self.assertNotEqual(found, event)
555 # Now check the cases which don't find an event at all. It would be
556 # nice to check the log messages here, but it seems too fragile to be
557 # worth it
558 dt = TimeDelta(1, format="sec")
559 tooEarlyOnDay = getDayObsStartTime(self.dayObs) + dt # 1 second after start of day
560 found = eventMaker.findEvent(tooEarlyOnDay)
561 self.assertIsNone(found)
563 # 1 second before end of day and this day does not end with an open
564 # event
565 tooLateOnDay = getDayObsStartTime(calcNextDay(self.dayObs)) - dt
566 found = eventMaker.findEvent(tooLateOnDay)
567 self.assertIsNone(found)
569 # going just inside the last event of the day should be fine
570 lastEvent = events[-1]
571 found = eventMaker.findEvent(lastEvent.end - dt)
572 self.assertEqual(found, lastEvent)
574 # going at the very end of the last event of the day should actually
575 # find nothing, because the last moment of an event isn't actually in
576 # the event itself, because of how contiguous events are defined to
577 # behave (being half-open intervals)
578 found = eventMaker.findEvent(lastEvent.end)
579 self.assertIsNone(found, lastEvent)
581 @vcr.use_cassette()
582 def test_eventAssociatedWith(self):
583 eventMaker = self.tmaEventMaker
584 events = eventMaker.getEvents(self.dayObsWithBlockInfo)
585 eventsWithBlockInfo = [e for e in events if e.blockInfos]
586 eventsWithoutBlockInfo = [e for e in events if not e.blockInfos]
587 self.assertEqual(len(events), 69)
588 self.assertEqual(len(eventsWithBlockInfo), 65)
589 self.assertEqual(len(eventsWithoutBlockInfo), 4)
591 self.assertIsNotNone(eventsWithoutBlockInfo[0].blockInfos)
592 self.assertIsInstance(eventsWithoutBlockInfo[0].blockInfos, list)
593 self.assertEqual(len(eventsWithoutBlockInfo[0].blockInfos), 0)
595 event = eventsWithBlockInfo[0]
596 self.assertIsInstance(event, TMAEvent)
597 self.assertTrue(event.associatedWith(ticket="SITCOM-906"))
598 self.assertFalse(event.associatedWith(ticket="SITCOM-905"))
600 self.assertTrue(event.associatedWith(salIndex=100017))
601 self.assertFalse(event.associatedWith(salIndex=100018))
603 self.assertTrue(event.associatedWith(block=6))
604 self.assertFalse(event.associatedWith(block=5))
606 # check it works with any and all of the arguments
607 self.assertTrue(event.associatedWith(block=6, salIndex=100017))
608 self.assertTrue(event.associatedWith(block=6, salIndex=100017, ticket="SITCOM-906"))
610 # check it's false if any are false
611 self.assertFalse(event.associatedWith(block=7, salIndex=100017, ticket="SITCOM-906")) # 1 wrong
612 self.assertFalse(event.associatedWith(block=6, salIndex=100018, ticket="SITCOM-906")) # 1 wrong
613 self.assertFalse(event.associatedWith(block=6, salIndex=100017, ticket="SITCOM-907")) # 1 wrong
614 self.assertFalse(event.associatedWith(block=1, salIndex=1, ticket="SITCOM-1")) # all wrong
616 # check with the blockSeqNum, with and without the other items
617 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1))
618 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2))
619 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1, salIndex=100017))
620 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2, salIndex=100017))
621 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1, salIndex=100017, ticket="SITCOM-906"))
622 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2, salIndex=100017, ticket="SITCOM-906"))
624 with self.assertRaises(ValueError):
625 event.associatedWith()
626 event.associatedWith(blockSeqNum=1) # nonsense to ask for a seqNum without a block number
629class TestMemory(lsst.utils.tests.MemoryTestCase):
630 pass
633def setup_module(module):
634 lsst.utils.tests.init()
637if __name__ == "__main__": 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true
638 lsst.utils.tests.init()
639 unittest.main()