Coverage for src/km3pipe/io/tests/test_clb.py: 100%

1# Filename: test_clb.py 

2# pylint: disable=C0111,R0904,C0103 

3""" 

4... 

5 

6""" 

7from io import StringIO 

8from os.path import join, dirname 

9 

10from km3pipe.testing import TestCase, data_path 

11from km3pipe.io.clb import CLBPump 

12 

13__author__ = "Tamas Gal" 

14__copyright__ = "Copyright 2016, Tamas Gal and the KM3NeT collaboration." 

15__credits__ = [] 

16__license__ = "MIT" 

17__maintainer__ = "Tamas Gal" 

18__email__ = "tgal@km3net.de" 

19__status__ = "Development" 

20 

21 

22class TestCLBPump(TestCase): 

23 def setUp(self): 

24 self.pump = CLBPump(file=data_path("daq/clb.dqd")) 

25 

26 def test_determine_packet_positions_finds_packets(self): 

27 self.assertListEqual([0, 1406, 2284], self.pump._packet_positions) 

28 

29 def test_length(self): 

30 assert 3 == len(self.pump) 

31 

32 def test_getindex(self): 

33 blob = self.pump[0] 

34 self.assertEqual(0, blob["PacketInfo"].run[0]) 

35 self.assertEqual("TTDC", blob["PacketInfo"].data_type[0]) 

36 assert 229 == len(blob["Hits"]) 

37 a_pmt_data = blob["Hits"][2] 

38 self.assertEqual(15, a_pmt_data.channel_id) 

39 self.assertEqual(66541067, a_pmt_data.time) 

40 self.assertEqual(28, a_pmt_data.tot) 

41 

42 assert 229 == len(self.pump[0]["Hits"]) 

43 assert 141 == len(self.pump[1]["Hits"]) 

44 assert 229 == len(self.pump[2]["Hits"]) 

45 

46 def test_iterator(self): 

47 i = 0 

48 for b in self.pump: 

49 i += 1 

50 assert 3 == i 

51 

52 i = 0 

53 for b in self.pump: 

54 i += 1 

55 assert 3 == i