Unit test for BitSwan

bspump-unittest.py

import unittest

import bspump
import bspump.common
import bspump.trigger

###

class MyProcessor(bspump.Processor):

	'''
	This is the processor, I want to unit test
	'''

	def process(self, context, event):
		return event + 1

#

class MyProcessorUnitTestCase(unittest.TestCase):

	'''
	This is my unit test case
	'''

	def execute(self, processor, input_data):
		'''
		This is universal and become part of bspump.unitest.ProcessorTestCase module
		'''

		app = bspump.BSPumpApplication()
		svc = app.get_service("bspump.PumpService")

		pipeline = UnitTestPipeline(app, MyProcessor)
		svc.add_pipeline(pipeline)

		pipeline.Source.Input = input_data

		app.run()

		return pipeline.Sink.Output


	def test_myprocessor(self):
		'''
		This is the actual unit test code
		'''
		
		Input = [
			(None, 1),
			(None, 2),
			(None, 3),
		]

		Output = self.execute(
			MyProcessor,
			Input
		)

		self.assertEqual(
			[event for context, event in Output],
			[2,3,4]
		)

###

class UnitTestSource(bspump.TriggerSource):

	def __init__(self, app, pipeline, id=None, config=None):
		super().__init__(app, pipeline, id=id, config=config)
		self.Input = []

	async def cycle(self, *args, **kwags):
		for context, event in self.Input:
			await self.process(event, context=context)


class UnitTestSink(bspump.Sink):

	def __init__(self, app, pipeline, id=None, config=None):
		super().__init__(app, pipeline, id=id, config=config)
		self.Output = []

	def process(self, context, event):
		self.Output.append((context, event))


class UnitTestPipeline(bspump.Pipeline):

	def __init__(self, app, processor):
		super().__init__(app, "UnitTestPipeline")
		
		self.Source = UnitTestSource(app, self).on(
			bspump.trigger.RunOnceTrigger(app)
		)
		self.Processor = processor(app, self)
		self.Sink = UnitTestSink(app, self)

		self.build(self.Source, self.Processor, self.Sink)


###

if __name__ == '__main__':
	unittest.main()