-
Notifications
You must be signed in to change notification settings - Fork 0
/
test.py
98 lines (73 loc) · 3.2 KB
/
test.py
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
import os.path
import subprocess
import re
import json
import pytest
from AnalysisModule.helper import get_call_with_parmeters
from AnalysisModule.MPIAnalysisModule.CallAnalysis import SingleMPICallAnalysis
from AnalysisModule.MPIAnalysisModule.MPIAPIParameters import mpi_api_dict
from AnalysisModule.repository import Repository
@pytest.mark.parametrize("input,num_params",
[("MPI_Barrier(MPI_COMM_WORLD);", 2),
(
"MPI_Irecv( u + INDEX(i_min[MACSIO_MAIN_Rank] - 1, 1), N, MPI_DOUBLE, left_proc[MACSIO_MAIN_Rank], 0, MACSIO_MAIN_Comm, request + requests++ );",
8),
(
"if (MPI_SUCCESS != MPI_Allgather(&netNum, 1, MPI_UNSIGNED_LONG, netNums, 1, MPI_UNSIGNED_LONG, MPI_COMM_WORLD)) {",
8)])
def test_get_call_with_params(input, num_params):
# print(input)
res = get_call_with_parmeters(input)
# print(res)
assert len(res) == num_params
def test_openmp_normalization():
file = "repositories/Enzo/src/enzo/ffte4X.F90"
if not os.path.isfile(file):
pytest.skip(f" Test code file is not present: {file}")
repo = Repository(repoName="Enzo", repoPath='repositories/Enzo')
normalized = repo.get_normalized_file_content(file)
assert 'OMP' in normalized
def get_analysis_for(call):
info = mpi_api_dict[call]
return SingleMPICallAnalysis(call, info[0], info[1])
def test_multiple_param_count():
analysis = get_analysis_for('MPI_Init_thread')
input = "MPI_Init_thread(NULL,NULL,MPI_THREAD_MULTIPLE,&provided);"
# print(input)
call = get_call_with_parmeters(input)
res = analysis.analyze_call("test", 0, call)
# print(res)
assert res['analysis_successful'].all()
assert res['THREAD_LEVEL'][0] == "MPI_THREAD_MULTIPLE"
assert res['ARGUMENT_COUNT'][0] == "NULL"
input = "MPI_INIT_THREAD(MPI_THREAD_MULTIPLE,provided,ierr);"
# print(input)
call = get_call_with_parmeters(input)
res = analysis.analyze_call("test", 0, call)
# print(res)
assert res['analysis_successful'].all()
input = "MPI_Init_thread(MPI_THREAD_MULTIPLE,provided);"
# print(input)
call = get_call_with_parmeters(input)
res = analysis.analyze_call("test", 0, call)
# print(res)
assert res['analysis_successful'].all()
def test_c_declaration():
analysis = get_analysis_for('MPI_Send')
input = 'int MPI_Send(const void *buf, int count, MPI_Datatype datatype, int dest, int tag, MPI_Comm comm)'
call = get_call_with_parmeters(input)
assert call == []
@pytest.mark.skip(reason="no testcase yet, need to create a file with only fortran decls")
def test_fortran_declaration():
# TODO implement a test for it!
assert False
@pytest.mark.parametrize("input", [
'MPI_Pcontrol(LEVEL)', 'MPI_Pcontrol(LEVEL, vararg1)',
'MPI_Pcontrol(LEVEL, vararg1, vararg2)'])
def test_vararg(input):
analysis = get_analysis_for('MPI_Pcontrol')
call = get_call_with_parmeters(input)
if analysis.is_call_matching(call[0]):
res = analysis.analyze_call("test", 0, call)
print(res)
assert res['analysis_successful'].all()