← Back to branch summary

~chris-rogers/maus/emr_mc_digitization

« back to all changes in this revision

Viewing changes to tests/py_unit/test_core_go.py

reverse merge from trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/py_unit/test_core_go.py
1
 
import json
 
1
"""Tests for src/common_py/Go.py"""
 
2
 
2
3
import unittest
3
 
import subprocess
4
 
import glob
5
 
import os
6
4
import sys
 
5
import copy
 
6
import tempfile
7
7
from io import StringIO
8
8
 
9
 
from InputPyEmptyDocument import *
10
 
from MapPyDoNothing import *
11
 
from ReducePyDoNothing import *
12
 
from OutputPyJSON import *
13
 
 
14
 
from Go import Go
15
 
 
16
 
class FakeMap():
17
 
    def birth(self, someArg):
 
9
from InputPyEmptyDocument import InputPyEmptyDocument
 
10
from MapPyDoNothing import MapPyDoNothing
 
11
from ReducePyDoNothing import ReducePyDoNothing
 
12
from OutputPyJSON import OutputPyJSON
 
13
 
 
14
from Go import Go, get_possible_dataflows
 
15
 
 
16
class FakeMap(): #pylint: disable = W0232, R0903
 
17
    """Map mock-up that always fails to birth"""
 
18
    def birth(self, some_arg): #pylint: disable = W0613, R0201, R0903
 
19
        """Always return False"""
18
20
        return False
19
21
 
20
 
class GoTestCase(unittest.TestCase):
 
22
class GoTestCase(unittest.TestCase): #pylint: disable = R0904
 
23
    """Tests for src/common_py/Go.py"""
 
24
 
 
25
    def setUp(self): #pylint: disable = C0103
 
26
        """Create temp file"""
 
27
        self.tmp_file = tempfile.mkstemp()[1]
 
28
 
 
29
    def dataflows_test(self):
 
30
        """
 
31
        Make sure get_possible_dataflows() doesn't return nonsense
 
32
        """
 
33
        keys = get_possible_dataflows().keys()
 
34
        self.assertTrue('pipeline_single_thread' in keys)
 
35
 
21
36
    def input_birth_test(self):
 
37
        """Check that Go raises error with bad input"""
22
38
        inputer = FakeMap()
23
 
        mapper = MapPyDoNothing()
24
 
        reducer = ReducePyDoNothing()
25
 
        outputer = OutputPyJSON(open('unit_test', 'w'))  #  this file won't appear since exception
 
39
        transformer = MapPyDoNothing()
 
40
        merger = ReducePyDoNothing()
 
41
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
26
42
        with self.assertRaises(AssertionError):
27
 
            Go(inputer, mapper, reducer, outputer)
 
43
            Go(inputer, transformer, merger, outputer, command_line_args=False)
28
44
 
29
45
 
30
46
    def map_birth_test(self):
 
47
        """Check that Go raises error with bad transformer"""
31
48
        inputer = InputPyEmptyDocument(1)
32
 
        mapper = FakeMap()
33
 
        reducer = ReducePyDoNothing()
34
 
        outputer = OutputPyJSON(open('unit_test', 'w'))  #  this file won't appear since exception
 
49
        transformer = FakeMap()
 
50
        merger = ReducePyDoNothing()
 
51
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
35
52
 
36
53
        with self.assertRaises(AssertionError):
37
 
            Go(inputer, mapper, reducer, outputer)
 
54
            Go(inputer, transformer, merger, outputer, command_line_args=False)
38
55
 
39
56
    def reduce_birth_test(self):
 
57
        """Check that Go raises error with bad merger"""
40
58
        inputer = InputPyEmptyDocument(1)
41
 
        mapper = MapPyDoNothing()
42
 
        reducer = FakeMap()
43
 
        outputer = OutputPyJSON(open('unit_test', 'w'))  #  this file won't appear since exception
 
59
        transformer = MapPyDoNothing()
 
60
        merger = FakeMap()
 
61
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
44
62
 
45
63
        with self.assertRaises(AssertionError):
46
 
            Go(inputer, mapper, reducer, outputer)
 
64
            Go(inputer, transformer, merger, outputer, command_line_args=False)
47
65
 
48
66
    def output_birth_test(self):
 
67
        """Check that Go raises error with bad outputter"""
49
68
        inputer = InputPyEmptyDocument(1)
50
 
        mapper = MapPyDoNothing()
51
 
        reducer = ReducePyDoNothing()
 
69
        transformer = MapPyDoNothing()
 
70
        merger = ReducePyDoNothing()
52
71
        outputer = FakeMap()
53
72
 
54
73
        with self.assertRaises(AssertionError):
55
 
            Go(inputer, mapper, reducer, outputer)
56
 
 
57
 
 
58
 
    def test_map_reduce_type(self):
59
 
        inputer = InputPyEmptyDocument(1)
60
 
        mapper = MapPyDoNothing()
61
 
        reducer = ReducePyDoNothing()
62
 
        outputer = OutputPyJSON(open('unit_test', 'w'))  #  this file won't appear since exception
63
 
 
64
 
        configFile = StringIO(u"""map_reduce_type="bad_type" """)
65
 
 
66
 
        with self.assertRaises(AssertionError):
67
 
            Go(inputer, mapper, reducer, outputer, configFile)
68
 
 
69
 
 
70
 
    def test_map_reduce_type(self):
71
 
        inputer = InputPyEmptyDocument(1)
72
 
        mapper = MapPyDoNothing()
73
 
        reducer = ReducePyDoNothing()
74
 
        outputer = OutputPyJSON(open('unit_test', 'w'))  #  this file won't appear since exception
75
 
        for MapRedType in ["native_python", "native_python_profile"]:
76
 
            configFile = StringIO(u"""map_reduce_type="%s" """ % MapRedType)
77
 
 
78
 
        Go(inputer, mapper, reducer, outputer, configFile)
79
 
 
80
 
 
81
 
 
82
 
 
 
74
            Go(inputer, transformer, merger, outputer, command_line_args=False)
 
75
 
 
76
    def command_line_args_test(self):
 
77
        """Check that Go handles command line args switch correctly"""
 
78
        inputer = InputPyEmptyDocument(1)
 
79
        transformer = MapPyDoNothing()
 
80
        merger = ReducePyDoNothing()
 
81
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
 
82
        arg_temp = copy.deepcopy(sys.argv)
 
83
        sys.argv = [arg_temp[0]]
 
84
    
 
85
        Go(inputer, transformer, merger, outputer, command_line_args = True)
 
86
 
 
87
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
 
88
        arg_temp = copy.deepcopy(sys.argv)
 
89
        sys.argv = [arg_temp[0], "bob"]
 
90
        with self.assertRaises(SystemExit):
 
91
            Go(inputer, transformer, merger, outputer, command_line_args = True)
 
92
 
 
93
        sys.argv = [arg_temp[0], "-verbose_level", "1"]
 
94
        Go(inputer, transformer, merger, outputer, command_line_args = True)
 
95
 
 
96
        sys.argv = arg_temp
 
97
 
 
98
    def test_type_of_dataflow_bad(self):
 
99
        """Check that Go raises error with bad dataflow type"""
 
100
        inputer = InputPyEmptyDocument(1)
 
101
        transformer = MapPyDoNothing()
 
102
        merger = ReducePyDoNothing()
 
103
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
 
104
 
 
105
        config = StringIO(u"""type_of_dataflow="bad_type" """)
 
106
 
 
107
        with self.assertRaises(LookupError):
 
108
            Go(inputer, transformer, merger, outputer, config, \
 
109
               command_line_args = False)
 
110
 
 
111
 
 
112
    def test_type_of_dataflow_good(self):
 
113
        """Check that Go executes okay with good dataflow"""
 
114
        inputer = InputPyEmptyDocument(1)
 
115
        transformer = MapPyDoNothing()
 
116
        merger = ReducePyDoNothing()
 
117
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
 
118
        for map_red_type in ['pipeline_single_thread']:
 
119
            config = StringIO(u"""type_of_dataflow="%s" """ % map_red_type)
 
120
 
 
121
            Go(inputer, transformer, merger, outputer, config, \
 
122
                   command_line_args = False)
 
123
 
 
124
    def test_dataflow_not_implemented(self):
 
125
        """Check that Go notifies user of unimplemented dataflow"""
 
126
        inputer = InputPyEmptyDocument(1)
 
127
        transformer = MapPyDoNothing()
 
128
        merger = ReducePyDoNothing()
 
129
        outputer = OutputPyJSON(open(self.tmp_file, 'w'))
 
130
        for map_red_type in ['many_local_threads','control_room_style']:
 
131
            config = StringIO(u"""type_of_dataflow="%s" """ % map_red_type)
 
132
            
 
133
            with self.assertRaises(NotImplementedError):
 
134
                Go(inputer, transformer, merger, outputer, config, \
 
135
                       command_line_args = False)
83
136
 
84
137
if __name__ == '__main__':
85
138
    unittest.main()