-
Notifications
You must be signed in to change notification settings - Fork 38
Expand file tree
/
Copy pathhandler_test.py
More file actions
118 lines (100 loc) · 3.86 KB
/
Copy pathhandler_test.py
File metadata and controls
118 lines (100 loc) · 3.86 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
import sys
import json
import unittest
import os
from Test.handler_algorithms import *
class HandlerTest(unittest.TestCase):
fifo_pipe_path = "/tmp/algoout"
fifo_pipe = None
def setUp(self):
try:
os.mkfifo(self.fifo_pipe_path)
except Exception:
pass
def tearDown(self):
os.remove(self.fifo_pipe_path)
def read_from_pipe(self):
read_obj = os.read(self.fifo_pipe, 10000)
if isinstance(read_obj, bytes):
read_obj = read_obj.decode("utf-8")
actual_output = json.loads(read_obj)
os.close(self.fifo_pipe)
return actual_output
def open_pipe(self):
self.fifo_pipe = os.open(self.fifo_pipe_path, os.O_RDONLY | os.O_NONBLOCK)
def execute_example(self, input, apply, load=lambda: None):
self.open_pipe()
algo = Algorithmia.handler(apply, load)
sys.stdin = input
algo.serve()
output = self.read_from_pipe()
return output
def execute_without_load(self, input, apply):
self.open_pipe()
algo = Algorithmia.handler(apply)
sys.stdin = input
algo.serve()
output = self.read_from_pipe()
return output
# ----- Tests ----- #
def test_basic(self):
input = {'content_type': 'json', 'data': 'Algorithmia'}
expected_output = {"metadata":
{
"content_type": "text"
},
"result": "hello Algorithmia"
}
input = [str(json.dumps(input))]
actual_output = self.execute_example(input, apply_input_or_context)
self.assertEqual(expected_output, actual_output)
def test_basic_2(self):
input = {'content_type': 'json', 'data': 'Algorithmia'}
expected_output = {"metadata":
{
"content_type": "text"
},
"result": "hello Algorithmia"
}
input = [str(json.dumps(input))]
actual_output = self.execute_without_load(input, apply_basic)
self.assertEqual(expected_output, actual_output)
def test_algorithm_loading_basic(self):
input = {'content_type': 'json', 'data': 'ignore me'}
expected_output = {'metadata':
{
'content_type': 'json'
},
'result': {'message': 'This message was loaded prior to runtime'}
}
input = [str(json.dumps(input))]
actual_output = self.execute_example(input, apply_input_or_context, loading_text)
self.assertEqual(expected_output, actual_output)
def test_algorithm_loading_algorithmia(self):
input = {'content_type': 'json', 'data': 'ignore me'}
expected_output = {'metadata':
{
'content_type': 'json'
},
'result': {
'data_url': 'data://demo/collection/somefile.json',
'data': {'foo': 'bar'}
}
}
input = [str(json.dumps(input))]
actual_output = self.execute_example(input, apply_input_or_context, loading_file_from_algorithmia)
self.assertEqual(expected_output, actual_output)
def test_error_loading(self):
input = {'content_type': 'json', 'data': 'Algorithmia'}
expected_output = {'error':
{'message': 'This exception was thrown in loading',
'error_type': 'AlgorithmError',
'stacktrace': ''
}
}
input = [str(json.dumps(input))]
actual_output = self.execute_example(input, apply_input_or_context, loading_exception)
# beacuse the stacktrace is local path specific,
# we're going to assume it's setup correctly and remove it from our equality check
actual_output["error"]["stacktrace"] = ''
self.assertEqual(expected_output, actual_output)