add basic stream API tests

gstlal/python/stream.py

@@ -90,7 +90,7 @@ class Stream:
 		self.mainloop = mainloop if mainloop else GLib.MainLoop()
 		self.pipeline = pipeline if pipeline else Gst.Pipeline(self.name)
 		self.handler = handler if handler else StreamHandler(self.mainloop, self.pipeline)
-		self.head = head if head is not None else {}
+		self.head = head
 
 		# set up source elem properties
 		self.source = source if source else None
@@ -153,7 +153,7 @@ class Stream:
 		else:
 			keyed = True
 
-		stream = cls()
+		stream = cls(head={})
 		state_vectors = {}
 		dq_vectors = {}
 		for ifo in ifos:
@@ -171,12 +171,13 @@ class Stream:
 				head=src,
 			)
 
+
 		stream.source = SourceElem(
 			datasource=data_source_info.data_source,
 			is_live=is_live,
 			gps_range=data_source_info.seg,
-			state_vector=state_vectors,
-			dq_vector=dq_vectors
+			state_vector=state_vectors if state_vector else None,
+			dq_vector=dq_vectors if dq_vector else None,
 		)
 
 		if keyed:

gstlal/tests/tests_pytest/test_stream.py

+"""Unittests for Streams
+
+"""
+from gstlal import datasource
+from gstlal.datasource import DataSource, Detector, KNOWN_LIVE_DATASOURCES
+from gstlal.stream import Stream
+
+
+def assert_datasourceinfo_stream_set(stream, info, detectors):
+	"""Utility for testing that a Stream instance was correctly set with DataSourceInfo"""
+	if isinstance(detectors, Detector):
+		assert type(stream.head).__name__ == 'GstAudioConvert'
+	else:
+		for detector in stream.keys():
+			assert type(stream[detector].head).__name__ == 'GstAudioConvert'
+
+	assert stream.source.datasource == info.data_source 
+	assert stream.source.is_live == (stream.source.datasource in KNOWN_LIVE_DATASOURCES)
+	assert stream.source.gps_range == info.seg
+	assert not stream.source.state_vector
+	assert not stream.source.dq_vector
+
+
+class TestStreamFromDataSource:
+	"""Test class for new DataSourceInfo api"""
+
+	def test_datasource_white_one_ifo(self):
+		"""Test Stream.from_datasource with one detector"""
+		info = datasource.DataSourceInfo(
+			data_source=DataSource.White,
+			gps_start_time=1234567,
+			gps_end_time=1234568,
+			channel_name={Detector.H1: 'CHANNEL'}
+		)
+		stream = Stream.from_datasource(info, Detector.H1.value)
+		assert_datasourceinfo_stream_set(stream, info, Detector.H1)
+
+	def test_datasource_white_two_ifo(self):
+		"""Test Stream.from_datasource with two detectors"""
+		info = datasource.DataSourceInfo(
+			data_source=DataSource.White,
+			gps_start_time=1234567,
+			gps_end_time=1234568,
+			channel_name={Detector.H1: 'CHANNEL', Detector.L1: 'CHANNEL'}
+		)
+		stream = Stream.from_datasource(info, [Detector.H1.value, Detector.L1.value])
+		assert_datasourceinfo_stream_set(stream, info, [Detector.H1, Detector.L1])
+
+
+def test_stream_dict_access():
+	"""Test Stream dict access methods (keys, values, items)"""
+	info = datasource.DataSourceInfo(
+		data_source=DataSource.White,
+		gps_start_time=1234567,
+		gps_end_time=1234568,
+		channel_name={Detector.H1: 'CHANNEL', Detector.L1: 'CHANNEL'}
+	)
+	detectors = {Detector.H1.value, Detector.L1.value}
+	stream = Stream.from_datasource(info, detectors)
+
+	assert set(stream.keys()) == detectors
+	for s in stream.values():
+		assert type(s.head).__name__ == 'GstAudioConvert'
+	for detector, s in stream.items():
+		assert detector in detectors
+		assert type(s.head).__name__ == 'GstAudioConvert'
+
+
+def test_stream_remap():
+	"""Test Stream.remap"""
+	info = datasource.DataSourceInfo(
+		data_source=DataSource.White,
+		gps_start_time=1234567,
+		gps_end_time=1234568,
+		channel_name={Detector.H1: 'CHANNEL', Detector.L1: 'CHANNEL'}
+	)
+	detectors = {Detector.H1.value, Detector.L1.value}
+	stream = Stream.from_datasource(info, detectors)
+	assert set(stream.keys()) == detectors
+
+	remapped = stream.remap()
+	assert not remapped.head and isinstance(remapped.head, dict)