-
Notifications
You must be signed in to change notification settings - Fork 149
Expand file tree
/
Copy pathbin_test.py
More file actions
executable file
·113 lines (87 loc) · 3 KB
/
bin_test.py
File metadata and controls
executable file
·113 lines (87 loc) · 3 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
#!/usr/bin/env python3
''' Integration test for open lambda's native and WebAssembly runtimes '''
# pylint: disable=missing-function-docstring,consider-using-with
# pylint: disable=wrong-import-position
import argparse
import os
import sys
from time import time
# You can either install the OpenLambda Python bindings
# or run the test from the project's root folder
sys.path.append('python/src')
from open_lambda import OpenLambda
from helper import DockerWorker, WasmWorker, SockWorker, TestConfContext
from helper import prepare_open_lambda, setup_config, assert_eq
from helper.test import set_test_filter, start_tests, check_test_results, set_worker_type, test
def get_mem_stat_mb(stat):
with open('/proc/meminfo', 'r', encoding='utf-8') as file:
for line in file:
if line.startswith(stat+":"):
parts = line.strip().split()
assert parts[-1] == 'kB'
return int(parts[1]) / 1024
raise LookupError('could not get stat')
@test
def ping():
open_lambda = OpenLambda()
pings = 1000
t_start = time()
for _ in range(pings):
open_lambda.check_status()
seconds = time() - t_start
return {"pings_per_sec": pings/seconds}
@test
def noop():
open_lambda = OpenLambda()
open_lambda.run("noop", args=[], json=False)
@test
def internal_call():
open_lambda = OpenLambda()
open_lambda.run("internal-call", args=[], json=False)
@test
def hashing():
open_lambda = OpenLambda()
open_lambda.run("hashing", args={"num_hashes": 100, "input_len": 1024}, json=False)
@test
def multiply():
open_lambda = OpenLambda()
result = open_lambda.run("multiply", args={"left": 25, "right": 8}, json=True)
assert_eq(result["result"], 200)
def run_tests():
''' Runs all tests '''
ping()
noop()
hashing()
internal_call()
multiply()
def _main():
parser = argparse.ArgumentParser(description='Run tests for OpenLambda')
parser.add_argument('--test_filter', type=str, default="")
parser.add_argument('--worker_type', type=str, default="sock")
parser.add_argument('--ol_dir', type=str, default="test-dir")
parser.add_argument('--registry', type=str, default="registry")
args = parser.parse_args()
set_test_filter([name for name in args.test_filter.split(",") if name != ''])
wasm = False
if args.worker_type == 'docker':
set_worker_type(DockerWorker)
elif args.worker_type == 'sock':
set_worker_type(SockWorker)
elif args.worker_type in ["webassembly", "wasm"]:
set_worker_type(WasmWorker)
wasm = True
else:
raise RuntimeError(f"Invalid worker type {args.worker_type}")
if wasm:
start_tests()
run_tests()
else:
setup_config(args.ol_dir)
prepare_open_lambda(args.ol_dir)
start_tests()
registry = "file://" + os.path.abspath(args.registry)
with TestConfContext(registry=registry):
run_tests()
check_test_results()
if __name__ == '__main__':
_main()