Coverage for tests/test_tmaUtils.py: 93%
366 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 11:54 +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
27from typing import Any
29import matplotlib.pyplot as plt
30import numpy as np
31import pandas as pd
32from astropy.time import TimeDelta
33from utils import getVcr
35import lsst.utils.tests
36from lsst.summit.utils.dateTime import calcNextDay, getDayObsStartTime
37from lsst.summit.utils.efdUtils import makeEfdClient
38from lsst.summit.utils.enums import PowerState
39from lsst.summit.utils.tmaUtils import (
40 AxisMotionState,
41 TMAEvent,
42 TMAEventMaker,
43 TMAState,
44 TMAStateMachine,
45 _initializeTma,
46 filterBadValues,
47 getAxisAndType,
48 getAzimuthElevationDataForEvent,
49 getCommandsDuringEvent,
50 getSlewsFromEventList,
51 getTracksFromEventList,
52 plotEvent,
53)
55__all__ = [
56 "writeNewTmaEventTestTruthValues",
57]
59TESTDIR = os.path.abspath(os.path.dirname(__file__))
60vcr = getVcr()
63def getTmaEventTestTruthValues() -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
64 """Get the current truth values for the TMA event test cases.
66 Returns
67 -------
68 seqNums : `np.array` of `int`
69 The sequence numbers of the events.
70 startRows : `np.array` of `int`
71 The _startRow numbers of the events.
72 endRows : `np.array` of `int`
73 The _endRow numbers of the events.
74 types : `np.array` of `str`
75 The event types, as a string, i.e. the ``TMAEvent.name`` of the event's
76 ``event.type``.
77 endReasons : `np.array` of `str`
78 The event end reasons, as a string, i.e. the ``TMAEvent.name`` of the
79 event's ``event.endReason``.
80 """
81 dataFilename = os.path.join(TESTDIR, "data", "tmaEventData.txt")
83 seqNums, startRows, endRows, types, endReasons = np.genfromtxt(
84 dataFilename, delimiter=",", dtype=None, names=True, encoding="utf-8", unpack=True
85 )
86 return seqNums, startRows, endRows, types, endReasons
89def writeNewTmaEventTestTruthValues() -> None:
90 """This function is used to write out the truth values for the test cases.
92 If the internal event creation logic changes, these values can change, and
93 will need to be updated. Run this function, and check the new values into
94 git.
96 Note: if you have cause to update values with this function, make sure to
97 update the version number on the TMAEvent class.
98 """
99 dayObs = 20241210 # obviously must match the day in the test class
101 eventMaker = TMAEventMaker()
102 events = eventMaker.getEvents(dayObs)
104 dataFilename = os.path.join(TESTDIR, "data", "tmaEventData.txt")
106 columnHeader = "seqNum,startRow,endRow,type,endReason"
107 with open(dataFilename, "w") as f:
108 f.write(columnHeader + "\n")
109 for event in events:
110 line = (
111 f"{event.seqNum},{event._startRow},{event._endRow},{event.type.name},"
112 f"{event.endReason.name}"
113 )
114 f.write(line + "\n")
117def makeValid(tma: TMAStateMachine) -> None:
118 """Helper function to turn a TMA into a valid state."""
119 for name, value in tma._parts.items():
120 if value == tma._UNINITIALIZED_VALUE:
121 tma._parts[name] = 1
124def _turnOn(tma: TMAStateMachine) -> None:
125 """Helper function to turn TMA axes on for testing.
127 Do not call directly in normal usage or code, as this just arbitrarily
128 sets values to turn the axes on.
130 Parameters
131 ----------
132 tma : `lsst.summit.utils.tmaUtils.TMAStateMachine`
133 The TMA state machine model to initialize.
134 """
135 tma._parts["azimuthSystemState"] = PowerState.ON
136 tma._parts["elevationSystemState"] = PowerState.ON
139class TmaUtilsTestCase(lsst.utils.tests.TestCase):
140 def test_tmaInit(self) -> None:
141 tma = TMAStateMachine()
142 self.assertFalse(tma._isValid)
144 # setting one axis should not make things valid
145 tma._parts["azimuthMotionState"] = 1
146 self.assertFalse(tma._isValid)
148 # setting all the other components should make things valid
149 tma._parts["azimuthInPosition"] = 1
150 tma._parts["azimuthSystemState"] = 1
151 tma._parts["elevationInPosition"] = 1
152 tma._parts["elevationMotionState"] = 1
153 tma._parts["elevationSystemState"] = 1
154 self.assertTrue(tma._isValid)
156 def test_tmaReferences(self) -> None:
157 """Check the linkage between the component lists and the _parts
158 dict.
159 """
160 tma = TMAStateMachine()
162 # setting one axis should not make things valid
163 self.assertEqual(tma._parts["azimuthMotionState"], tma._UNINITIALIZED_VALUE)
164 self.assertEqual(tma._parts["elevationMotionState"], tma._UNINITIALIZED_VALUE)
165 tma.motion[0] = AxisMotionState.TRACKING # set azimuth to 0
166 tma.motion[1] = AxisMotionState.TRACKING # set azimuth to 0
167 self.assertEqual(tma._parts["azimuthMotionState"], AxisMotionState.TRACKING)
168 self.assertEqual(tma._parts["elevationMotionState"], AxisMotionState.TRACKING)
170 def test_getAxisAndType(self) -> None:
171 # check both the long and short form names work
172 for s in ["azimuthMotionState", "lsst.sal.MTMount.logevent_azimuthMotionState"]:
173 self.assertEqual(getAxisAndType(s), ("azimuth", "MotionState"))
175 # check in position, and use elevation instead of azimuth to test that
176 for s in ["elevationInPosition", "lsst.sal.MTMount.logevent_elevationInPosition"]:
177 self.assertEqual(getAxisAndType(s), ("elevation", "InPosition"))
179 for s in ["azimuthSystemState", "lsst.sal.MTMount.logevent_azimuthSystemState"]:
180 self.assertEqual(getAxisAndType(s), ("azimuth", "SystemState"))
182 def test_initStateLogic(self) -> None:
183 tma = TMAStateMachine()
184 self.assertFalse(tma._isValid)
185 self.assertFalse(tma.isMoving)
186 self.assertFalse(tma.canMove)
187 self.assertFalse(tma.isTracking)
188 self.assertFalse(tma.isSlewing)
189 self.assertEqual(tma.state, TMAState.UNINITIALIZED)
191 _initializeTma(tma) # we're valid, but still aren't moving and can't
192 self.assertTrue(tma._isValid)
193 self.assertNotEqual(tma.state, TMAState.UNINITIALIZED)
194 self.assertTrue(tma.canMove)
195 self.assertTrue(tma.isNotMoving)
196 self.assertFalse(tma.isMoving)
197 self.assertFalse(tma.isTracking)
198 self.assertFalse(tma.isSlewing)
200 _turnOn(tma) # can now move, still valid, but not in motion
201 self.assertTrue(tma._isValid)
202 self.assertTrue(tma.canMove)
203 self.assertTrue(tma.isNotMoving)
204 self.assertFalse(tma.isMoving)
205 self.assertFalse(tma.isTracking)
206 self.assertFalse(tma.isSlewing)
208 # consider manipulating the axes by hand here and testing these?
209 # it's likely not worth it, given how much this exercised elsewhere,
210 # but these are the only functions not yet being directly tested
211 # tma._axesInFault()
212 # tma._axesOff()
213 # tma._axesOn()
214 # tma._axesInMotion()
215 # tma._axesTRACKING()
216 # tma._axesInPosition()
219@vcr.use_cassette()
220class TMAEventMakerTestCase(lsst.utils.tests.TestCase):
221 # class attributes populated in setUpClass
222 client: Any
223 dayObs: int
224 dayObsWithBlockInfo: int
225 tmaEventMaker: TMAEventMaker
226 events: list[TMAEvent]
227 sampleData: Any
229 @classmethod
230 @vcr.use_cassette()
231 def setUpClass(cls) -> None:
232 try:
233 cls.client = makeEfdClient(testing=True)
234 except RuntimeError:
235 raise unittest.SkipTest("Could not instantiate an EFD client")
237 cls.dayObs = 20241210
238 cls.dayObsWithBlockInfo = 20230615
239 # get a sample expRecord here to test expRecordToTimespan
240 cls.tmaEventMaker = TMAEventMaker(cls.client)
241 cls.events = cls.tmaEventMaker.getEvents(cls.dayObs) # does the fetch
242 cls.sampleData = cls.tmaEventMaker._data[cls.dayObs] # pull the data from the object and test length
244 @vcr.use_cassette()
245 def tearDown(self) -> None:
246 loop = asyncio.get_event_loop()
247 if self.client.influx_client is not None: 247 ↛ exitline 247 didn't return from function 'tearDown' because the condition on line 247 was always true
248 loop.run_until_complete(self.client.influx_client.close())
250 @vcr.use_cassette()
251 def test_events(self) -> None:
252 data = self.sampleData
253 self.assertIsInstance(data, pd.DataFrame)
254 self.assertEqual(len(data), 800)
256 @vcr.use_cassette()
257 def test_rowDataForValues(self) -> None:
258 rowsFor = set(self.sampleData["rowFor"])
259 self.assertEqual(len(rowsFor), 6)
261 # hard coding these ensures that you can't extend the axes/model
262 # without being explicit about it here.
263 correct = {
264 "azimuthInPosition",
265 "azimuthMotionState",
266 "azimuthSystemState",
267 "elevationInPosition",
268 "elevationMotionState",
269 "elevationSystemState",
270 }
271 self.assertSetEqual(rowsFor, correct)
273 @vcr.use_cassette()
274 def test_monotonicTimeInDataframe(self) -> None:
275 # ensure that each row is later than the previous
276 times = self.sampleData["private_efdStamp"]
277 self.assertTrue(np.all(np.diff(times) > 0))
279 @vcr.use_cassette()
280 def test_monotonicTimeApplicationOfRows(self) -> None:
281 # ensure you can apply rows in the correct order
282 tma = TMAStateMachine()
283 row1 = self.sampleData.iloc[0]
284 row2 = self.sampleData.iloc[1]
286 # just running this check it is OK
287 tma.apply(row1)
288 tma.apply(row2)
290 # and that if you apply them in reverse order then things will raise
291 tma = TMAStateMachine()
292 with self.assertRaises(ValueError):
293 tma.apply(row2)
294 tma.apply(row1)
296 @vcr.use_cassette()
297 def test_fullDaySequence(self) -> None:
298 # make sure we can apply all the data from the day without falling
299 # through the logic sieve
300 for engineering in (True, False):
301 tma = TMAStateMachine(engineeringMode=engineering)
303 _initializeTma(tma)
305 for rowNum, row in self.sampleData.iterrows():
306 tma.apply(row)
308 @vcr.use_cassette()
309 def test_endToEnd(self) -> None:
310 eventMaker = self.tmaEventMaker
311 events = eventMaker.getEvents(self.dayObs)
312 self.assertIsInstance(events, list)
313 self.assertEqual(len(events), 320)
314 self.assertIsInstance(events[0], TMAEvent)
316 slews = [e for e in events if e.type == TMAState.SLEWING]
317 tracks = [e for e in events if e.type == TMAState.TRACKING]
318 self.assertEqual(len(slews), 172)
319 self.assertEqual(len(tracks), 148)
321 seqNums, startRows, endRows, types, endReasons = getTmaEventTestTruthValues()
322 for eventNum, event in enumerate(events):
323 self.assertEqual(event.seqNum, seqNums[eventNum])
324 self.assertEqual(event._startRow, startRows[eventNum])
325 self.assertEqual(event._endRow, endRows[eventNum])
326 self.assertEqual(event.type.name, types[eventNum])
327 self.assertEqual(event.endReason.name, endReasons[eventNum])
329 eventSet = set(slews) # check we can hash
330 eventSet.update(slews) # check it ignores duplicates
331 self.assertEqual(len(eventSet), len(slews))
333 @vcr.use_cassette()
334 def test_noDataBehaviour(self) -> None:
335 eventMaker = self.tmaEventMaker
336 noDataDayObs = 19600101 # do not use 19700101 - there is data for that day!
337 with self.assertLogs(level="WARNING") as cm:
338 correctMsg = f"No EFD data found for dayObs={noDataDayObs}"
339 events = eventMaker.getEvents(noDataDayObs)
340 self.assertIsInstance(events, list)
341 self.assertEqual(len(events), 0)
342 msg = cm.output[0]
343 self.assertIn(correctMsg, msg)
345 @vcr.use_cassette()
346 def test_helperFunctions(self) -> None:
347 eventMaker = self.tmaEventMaker
348 events = eventMaker.getEvents(self.dayObs)
350 slews = [e for e in events if e.type == TMAState.SLEWING]
351 tracks = [e for e in events if e.type == TMAState.TRACKING]
352 foundSlews = getSlewsFromEventList(events)
353 foundTracks = getTracksFromEventList(events)
354 self.assertEqual(slews, foundSlews)
355 self.assertEqual(tracks, foundTracks)
357 def test_filterBadValues(self) -> None:
358 # NB: if you add enough spurious values that the median is no longer
359 # the value around which your "good" values are oscillating the first
360 # two points will get replaced and this can be very confusing!
362 # test no bad values
363 # mean = median = 1.0
364 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])
365 mean = np.mean(values)
366 nReplaced = filterBadValues(values)
367 self.assertEqual(nReplaced, 0)
368 self.assertEqual(np.mean(values), mean)
370 # test with one bad values
371 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])
372 nReplaced = filterBadValues(values)
373 self.assertEqual(nReplaced, 1)
375 # test with two consecutive bad values
376 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])
377 nReplaced = filterBadValues(values)
378 self.assertEqual(nReplaced, 2)
380 # test with three consecutive bad values
381 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])
382 nReplaced = filterBadValues(values)
383 self.assertEqual(nReplaced, 3)
385 # test with three consecutive bad values and another at the end
386 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])
387 nReplaced = filterBadValues(values)
388 self.assertEqual(nReplaced, 4)
390 # test with more than three consecutive bad values
391 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])
392 nReplaced = filterBadValues(values)
393 self.assertEqual(nReplaced, 3)
394 self.assertIn(5.0, values) # check the last bad value is still there specifically
396 # test with more than three consecutive bad values and another bad
397 # value at the end
398 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])
399 nReplaced = filterBadValues(values)
400 self.assertEqual(nReplaced, 4)
402 # test with bad values in first two positions
403 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
404 nReplaced = filterBadValues(values)
405 self.assertEqual(nReplaced, 2)
407 # test with bad values in first two positions and one in the middle
408 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])
409 nReplaced = filterBadValues(values)
410 self.assertEqual(nReplaced, 3)
412 # check that the last two good values are always used for correction,
413 # including when there are more than three consecutive bad values.
414 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])
415 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])
416 nReplaced = filterBadValues(values)
417 residuals = np.abs(values - expected)
418 self.assertEqual(nReplaced, 4)
419 self.assertTrue(np.all(residuals < 1e-6))
421 # check with one good point after an overflowing run of bad to make
422 # sure the correction is always applied with good values, not the naive
423 # average of the last two even if they might be bad
424 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])
425 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])
426 nReplaced = filterBadValues(values)
427 residuals = np.abs(values - expected)
428 self.assertEqual(nReplaced, 4)
429 self.assertTrue(np.all(residuals < 1e-6))
431 # check with non-default maxDelta
432 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])
433 nReplaced = filterBadValues(values, maxDelta=10)
434 self.assertEqual(nReplaced, 0)
436 values = np.array(
437 [
438 1.0,
439 1.0,
440 1.0,
441 1.1,
442 1.0,
443 1.0,
444 1.0,
445 1.0,
446 1.0,
447 1.0,
448 ]
449 )
450 nReplaced = filterBadValues(values, maxDelta=0.01)
451 self.assertEqual(nReplaced, 1)
453 @vcr.use_cassette()
454 def test_getEvent(self) -> None:
455 # test the singular event getter, and what happens if the event doesn't
456 # exist for the day
457 eventMaker = self.tmaEventMaker
458 events = eventMaker.getEvents(self.dayObs)
459 nEvents = len(events)
461 event = eventMaker.getEvent(self.dayObs, 0)
462 self.assertIsInstance(event, TMAEvent)
463 self.assertEqual(event, events[0])
464 event = eventMaker.getEvent(self.dayObs, 100)
465 self.assertIsInstance(event, TMAEvent)
466 self.assertEqual(event, events[100])
468 with self.assertLogs(level="WARNING") as cm:
469 correctMsg = f"Event {nEvents + 1} not found for {self.dayObs}"
470 event = eventMaker.getEvent(self.dayObs, nEvents + 1)
471 msg = cm.output[0]
472 self.assertIn(correctMsg, msg)
474 @vcr.use_cassette()
475 def test_printing(self) -> None:
476 eventMaker = self.tmaEventMaker
477 events = eventMaker.getEvents(self.dayObs)
479 # test str(), repr(), and _ipython_display_() for an event
480 print(str(events[0]))
481 print(repr(events[0]))
482 events[0]._ipython_display_()
484 # spot-check both a slow and a track to print
485 slews = [e for e in events if e.type == TMAState.SLEWING]
486 tracks = [e for e in events if e.type == TMAState.TRACKING]
487 eventMaker.printEventDetails(slews[0])
488 eventMaker.printEventDetails(tracks[0])
489 eventMaker.printEventDetails(events[-1])
491 # check the full day trick works
492 eventMaker.printFullDayStateEvolution(self.dayObs)
494 tma = TMAStateMachine()
495 _initializeTma(tma) # the uninitialized state contains wrong types for printing
496 eventMaker.printTmaDetailedState(tma)
498 @vcr.use_cassette()
499 def test_getAxisData(self) -> None:
500 eventMaker = self.tmaEventMaker
501 events = eventMaker.getEvents(self.dayObs)
503 azData, elData = getAzimuthElevationDataForEvent(self.client, events[0])
504 self.assertIsInstance(azData, pd.DataFrame)
505 self.assertIsInstance(elData, pd.DataFrame)
507 paddedAzData, paddedElData = getAzimuthElevationDataForEvent(
508 self.client, events[0], prePadding=2, postPadding=1
509 )
510 self.assertGreater(len(paddedAzData), len(azData))
511 self.assertGreater(len(paddedElData), len(elData))
513 # just check this doesn't raise when called, and check we can pass the
514 # data in
515 plotEvent(self.client, events[0], azimuthData=azData, elevationData=elData)
517 @vcr.use_cassette()
518 def test_plottingAndCommands(self) -> None:
519 eventMaker = self.tmaEventMaker
520 events = eventMaker.getEvents(self.dayObs)
521 event = events[10] # this one has commands, and we'll check that later
523 # check we _can_ plot without a figure, and then stop doing that
524 plotEvent(self.client, event)
526 fig = plt.figure(figsize=(10, 8))
527 # just check this doesn't raise when called
528 plotEvent(self.client, event, fig=fig)
529 plt.close(fig)
531 commandsToPlot = ["raDecTarget", "moveToTarget", "startTracking", "stopTracking"]
532 commands = getCommandsDuringEvent(self.client, event, commandsToPlot, doLog=False)
533 self.assertTrue(not all([time is None for time in commands.values()])) # at least one command
535 plotEvent(self.client, event, fig=fig, commands=commands)
537 del fig
539 @vcr.use_cassette()
540 def test_findEvent(self) -> None:
541 eventMaker = self.tmaEventMaker
542 # addBlockInfo=True because it shouldn't affect the comparison, and
543 # this also then ensures that the code is exercised too
544 events = eventMaker.getEvents(self.dayObs, addBlockInfo=True)
545 event = events[28] # this one has a contiguous event before it
547 time = event.begin
548 found = eventMaker.findEvent(time)
549 self.assertEqual(found, event)
551 dt = TimeDelta(0.01, format="sec")
552 # must be just inside to get the same event back, because if a moment
553 # is shared it gives the one which starts with the moment (whilst
554 # logging info messages about it)
555 time = event.end - dt
556 found = eventMaker.findEvent(time)
557 self.assertEqual(found, event)
559 # now check that if we're a hair after, we don't get the same event
560 time = event.end + dt
561 found = eventMaker.findEvent(time)
562 self.assertNotEqual(found, event)
564 # Now check the cases which don't find an event at all. It would be
565 # nice to check the log messages here, but it seems too fragile to be
566 # worth it
567 dt = TimeDelta(1, format="sec")
568 tooEarlyOnDay = getDayObsStartTime(self.dayObs) + dt # 1 second after start of day
569 found = eventMaker.findEvent(tooEarlyOnDay)
570 self.assertIsNone(found)
572 # 1 second before end of day and this day does not end with an open
573 # event
574 tooLateOnDay = getDayObsStartTime(calcNextDay(self.dayObs)) - dt
575 found = eventMaker.findEvent(tooLateOnDay)
576 self.assertIsNone(found)
578 # going just inside the last event of the day should be fine
579 lastEvent = events[-1]
580 found = eventMaker.findEvent(lastEvent.end - dt)
581 self.assertEqual(found, lastEvent)
583 # going at the very end of the last event of the day should actually
584 # find nothing, because the last moment of an event isn't actually in
585 # the event itself, because of how contiguous events are defined to
586 # behave (being half-open intervals)
587 found = eventMaker.findEvent(lastEvent.end)
588 self.assertIsNone(found, lastEvent)
590 @vcr.use_cassette()
591 def test_eventAssociatedWith(self) -> None:
592 eventMaker = self.tmaEventMaker
593 events = eventMaker.getEvents(self.dayObsWithBlockInfo)
594 eventsWithBlockInfo = [e for e in events if e.blockInfos]
595 eventsWithoutBlockInfo = [e for e in events if not e.blockInfos]
596 self.assertEqual(len(events), 69)
597 self.assertEqual(len(eventsWithBlockInfo), 65)
598 self.assertEqual(len(eventsWithoutBlockInfo), 4)
600 firstBlockInfos = eventsWithoutBlockInfo[0].blockInfos
601 self.assertIsNotNone(firstBlockInfos)
602 self.assertIsInstance(firstBlockInfos, list)
603 assert firstBlockInfos is not None # narrow for mypy; asserted non-None just above
604 self.assertEqual(len(firstBlockInfos), 0)
606 event = eventsWithBlockInfo[0]
607 self.assertIsInstance(event, TMAEvent)
608 self.assertTrue(event.associatedWith(ticket="SITCOM-906"))
609 self.assertFalse(event.associatedWith(ticket="SITCOM-905"))
611 self.assertTrue(event.associatedWith(salIndex=100017))
612 self.assertFalse(event.associatedWith(salIndex=100018))
614 self.assertTrue(event.associatedWith(block=6))
615 self.assertFalse(event.associatedWith(block=5))
617 # check it works with any and all of the arguments
618 self.assertTrue(event.associatedWith(block=6, salIndex=100017))
619 self.assertTrue(event.associatedWith(block=6, salIndex=100017, ticket="SITCOM-906"))
621 # check it's false if any are false
622 self.assertFalse(event.associatedWith(block=7, salIndex=100017, ticket="SITCOM-906")) # 1 wrong
623 self.assertFalse(event.associatedWith(block=6, salIndex=100018, ticket="SITCOM-906")) # 1 wrong
624 self.assertFalse(event.associatedWith(block=6, salIndex=100017, ticket="SITCOM-907")) # 1 wrong
625 self.assertFalse(event.associatedWith(block=1, salIndex=1, ticket="SITCOM-1")) # all wrong
627 # check with the blockSeqNum, with and without the other items
628 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1))
629 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2))
630 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1, salIndex=100017))
631 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2, salIndex=100017))
632 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1, salIndex=100017, ticket="SITCOM-906"))
633 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2, salIndex=100017, ticket="SITCOM-906"))
635 with self.assertRaises(ValueError):
636 event.associatedWith()
637 event.associatedWith(blockSeqNum=1) # nonsense to ask for a seqNum without a block number
640class TestMemory(lsst.utils.tests.MemoryTestCase):
641 pass
644def setup_module(module: object) -> None:
645 lsst.utils.tests.init()
648if __name__ == "__main__": 648 ↛ 649line 648 didn't jump to line 649 because the condition on line 648 was never true
649 lsst.utils.tests.init()
650 unittest.main()