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

1#!/usr/bin/env python3 

2from os.path import join, dirname 

3 

4import km3pipe as kp 

5from km3pipe.testing import TestCase, data_path 

6from km3pipe.io.offline import OfflinePump 

7 

8 

9# class TestEventPump(TestCase): 

10# def setUp(self): 

11# self.pump = EventPump( 

12# filename=data_path( 

13# "offline/mcv5.0.DAT004340.propa.sirene.jte.jchain.aanet.4340.root" 

14# ) 

15# ) 

16 

17# def test_iteration(self): 

18# i = 0 

19# for blob in self.pump: 

20# i += 1 

21 

22# assert 3 == i 

23 

24# def test_getitem(self): 

25# blob = self.pump[0] 

26# assert "Header" in blob 

27# assert "Hits" in blob 

28# assert "McHits" in blob 

29# assert "McTracks" in blob 

30 

31# assert 86 == len(blob["Hits"]) 

32 

33# with self.assertRaises(IndexError): 

34# self.pump[4] 

35 

36# def test_hits(self): 

37# n_hits = [86, 111, 83] 

38# i = 0 

39# for blob in self.pump: 

40# assert n_hits[i] == len(blob["Hits"]) 

41# i += 1 

42 

43# def test_mc_hits(self): 

44# n_mc_hits = [147, 291, 160] 

45# i = 0 

46# for blob in self.pump: 

47# assert n_mc_hits[i] == len(blob["McHits"]) 

48# i += 1 

49 

50# def test_mc_tracks(self): 

51# n_mc_tracks = [4, 8, 36] 

52# i = 0 

53# for blob in self.pump: 

54# assert n_mc_tracks[i] == len(blob["McTracks"]) 

55# i += 1 

56 

57# def test_header(self): 

58# for blob in self.pump: 

59# pass 

60 

61# self.assertListEqual([0, 1000000000.0, -1, -0.052], list(blob["Header"].cut_in)) 

62 

63 

64class TestOfflinePump(TestCase): 

65 def setUp(self): 

66 self.pump = OfflinePump( 

67 filename=data_path( 

68 "offline/mcv5.0.DAT004340.propa.sirene.jte.jchain.aanet.4340.root" 

69 ) 

70 ) 

71 

72 def test_offline_pump_iteration(self): 

73 i = 0 

74 for blob in self.pump: 

75 i += 1 

76 assert 3 == i 

77 

78 def test_offline_pump_has_header(self): 

79 for blob in self.pump: 

80 assert "header" in blob 

81 

82 def test_offline_pump_in_pipeline(self): 

83 class Observer(kp.Module): 

84 def configure(self): 

85 self.i = 0 

86 

87 def process(self, blob): 

88 assert "header" in blob 

89 assert "event" in blob 

90 self.i += 1 

91 return blob 

92 

93 def finish(self): 

94 return {"n_events": self.i} 

95 

96 pipe = kp.Pipeline() 

97 pipe.attach( 

98 OfflinePump, 

99 filename=data_path( 

100 "offline/mcv5.0.DAT004340.propa.sirene.jte.jchain.aanet.4340.root" 

101 ), 

102 ) 

103 pipe.attach(Observer) 

104 results = pipe.drain() 

105 

106 assert 3 == results["Observer"]["n_events"]