Coverage for skema/skema_py/tests/test_server.py: 100%

1import json 

2import shutil 

3from tempfile import TemporaryDirectory, TemporaryFile 

4from pathlib import Path 

5 

6from fastapi.testclient import TestClient 

7 

8from skema.skema_py.server import app 

9from skema.gromet.metadata.debug import Debug 

10 

11client = TestClient(app) 

12 

13 

14def test_healthcheck(): 

15 """Test case for /code2fn/ping endpoint.""" 

16 response = client.get("/code2fn/healthcheck") 

17 assert response.status_code == 200 

18 

19 

20def test_fn_supported_file_extensions(): 

21 """Test case for /code2fn/fn-supported-file-extensions endpoint.""" 

22 response = client.get("/code2fn/fn-supported-file-extensions") 

23 assert response.status_code == 200 

24 assert len(response.json()) > 0 

25 

26 

27def test_fn_given_filepaths(): 

28 """Test case for /code2fn/fn-given-filpaths endpoint with all optional fields included.""" 

29 system = { 

30 "files": ["example1.py", "dir/example2.py"], 

31 "blobs": [ 

32 "greet = lambda: print('howdy!')\ngreet()", 

33 "#Variable declaration\nx=2\n#Function definition\ndef foo(x):\n '''Increment the input variable'''\n return x+1", # Content of dir/example2.py 

34 ], 

35 "system_name": "example-system", 

36 "root_name": "example-system", 

37 "comments": { 

38 "files": { 

39 "example-system/dir/example2.py": { 

40 "single": [ 

41 {"content": "Variable declaration", "line_number": 0}, 

42 {"content": "Function definition", "line_number": 2}, 

43 ], 

44 "multi": [], 

45 "docstring": [ 

46 { 

47 "function_name": "foo", 

48 "content": ["Increment the input variable"], 

49 "start_line_number": 4, 

50 "end_line_number": 4, 

51 } 

52 ], 

53 } 

54 } 

55 }, 

56 } 

57 response = client.post("/code2fn/fn-given-filepaths", json=system) 

58 print(response.json()) 

59 assert response.status_code == 200 

60 assert "modules" in response.json() 

61 

62 

63def test_fn_given_filepaths_optional_fields(): 

64 """Test case for /code2fn/fn-given-filpaths endpoint with all optional fields excluded.""" 

65 system = { 

66 "files": ["example1.py", "dir/example2.py"], 

67 "blobs": [ 

68 "greet = lambda: print('howdy!')\ngreet()", 

69 "#Variable declaration\nx=2\n#Function definition\ndef foo(x):\n '''Increment the input variable'''\n return x+1", # Content of dir/example2.py 

70 ], 

71 } 

72 response = client.post("/code2fn/fn-given-filepaths", json=system) 

73 

74 assert response.status_code == 200 

75 assert "modules" in response.json() 

76 

77 

78def test_fn_given_filepaths_zip(): 

79 """Test case for /code2fn/fn-given-filpaths-zip endpoint.""" 

80 system = { 

81 "files": ["example1.py", "dir/example2.py"], 

82 "blobs": [ 

83 "greet = lambda: print('howdy!')\ngreet()", 

84 "#Variable declaration\nx=2\n#Function definition\ndef foo(x):\n '''Increment the input variable'''\n return x+1", # Content of dir/example2.py 

85 ], 

86 "system_name": "example-system", 

87 "root_name": "example-system", 

88 } 

89 with TemporaryDirectory() as tmp: 

90 system_path = Path(tmp) / system["root_name"] 

91 system_path.mkdir() 

92 system_zip_path = Path(tmp) / f"{system['root_name']}.zip" 

93 

94 for index, file in enumerate(system["files"]): 

95 file_path = Path(tmp, system["root_name"], file) 

96 file_path.parent.mkdir(parents=True, exist_ok=True) 

97 file_path.write_text(system["blobs"][index]) 

98 

99 input_path = Path(tmp, system["root_name"]) 

100 output_path = Path(tmp, f"{system['root_name']}.zip") 

101 shutil.make_archive(input_path, "zip", input_path) 

102 

103 response = client.post( 

104 "/code2fn/fn-given-filepaths-zip", 

105 files={"zip_file": open(output_path, "rb")}, 

106 ) 

107 assert response.status_code == 200 

108 assert "modules" in response.json() 

109 

110 

111def test_no_supported_files(): 

112 system = { 

113 "files": ["unsupported1.git", "unsupported2.lock"], 

114 "blobs": [ 

115 "This is not a source code file.", 

116 "This is not a source code file.", 

117 ], 

118 "system_name": "unsupported-system", 

119 "root_name": "unsupported-system", 

120 } 

121 

122 response = client.post("/code2fn/fn-given-filepaths", json=system) 

123 assert response.status_code == 200 

124 

125 gromet_collection = response.json() 

126 assert "metadata_collection" in gromet_collection 

127 assert ( 

128 len(gromet_collection["metadata_collection"]) == 1 

129 ) # Only one element (GrometFNModuleCollection) should create metadata in this metadata_collection 

130 assert ( 

131 len(gromet_collection["metadata_collection"][0]) == 1 

132 ) # There should only be one ERROR Debug metadata since there are no source files to process. 

133 assert gromet_collection["metadata_collection"][0][0]["gromet_type"] == "Debug" 

134 assert gromet_collection["metadata_collection"][0][0]["severity"] == "ERROR" 

135 

136 

137def test_partial_supported_files(): 

138 system = { 

139 "files": ["supported.py", "unsupported.lock"], 

140 "blobs": [ 

141 "x=2", 

142 "This is not a source code file.", 

143 ], 

144 "system_name": " mixed-system", 

145 "root_name": "mixed-system", 

146 } 

147 

148 response = client.post("/code2fn/fn-given-filepaths", json=system) 

149 assert response.status_code == 200 

150 

151 gromet_collection = response.json() 

152 assert "metadata_collection" in gromet_collection 

153 assert ( 

154 len(gromet_collection["metadata_collection"]) == 1 

155 ) # Only one element (GrometFNModuleCollection) should create metadata in this metadata_collection 

156 assert ( 

157 len(gromet_collection["metadata_collection"][0]) == 1 

158 ) # There should only be one WARNING Debug metadata since is a single unsupported file. 

159 assert gromet_collection["metadata_collection"][0][0]["gromet_type"] == "Debug" 

160 assert gromet_collection["metadata_collection"][0][0]["severity"] == "WARNING" 

161 

162def test_hidden_files(): 

163 """Test case for hidden files and MACOSX artifacts""" 

164 system = { 

165 "files": [".hidden/source.py", "root/_MACOSX/hello.py"], 

166 "blobs": [ 

167 "x=2", 

168 "print('hello world')", 

169 ], 

170 "system_name": "hidden-system", 

171 "root_name": "hidden-system", 

172 } 

173 

174 response = client.post("/code2fn/fn-given-filepaths", json=system) 

175 assert response.status_code == 200 

176 

177 gromet_collection = response.json() 

178 assert "metadata_collection" in gromet_collection 

179 assert ( 

180 len(gromet_collection["metadata_collection"]) == 1 

181 ) # Only one element (GrometFNModuleCollection) should create metadata in this metadata_collection 

182 assert ( 

183 len(gromet_collection["metadata_collection"][0]) == 1 

184 ) # There should only be one ERROR Debug metadata since there are no source files to process. 

185 assert gromet_collection["metadata_collection"][0][0]["gromet_type"] == "Debug" 

186 assert gromet_collection["metadata_collection"][0][0]["severity"] == "ERROR" 

187 

188def test_dependency_depth(): 

189 system = { 

190 "files": ["dependency.py"], 

191 "blobs": [ 

192 "import minimal" 

193 ], 

194 "dependency_depth": 1 

195 } 

196 

197 response = client.post("/code2fn/fn-given-filepaths", json=system) 

198 gromet_collection = response.json() 

199 assert response.status_code == 200 

200 assert len(gromet_collection["module_dependencies"]) > 1 

201 

202 

203def test_dependency_depth_missing(): 

204 system = { 

205 "files": ["dependency.py"], 

206 "blobs": [ 

207 "import minimal" 

208 ], 

209 } 

210 

211 response = client.post("/code2fn/fn-given-filepaths", json=system) 

212 gromet_collection = response.json() 

213 assert response.status_code == 200 

214 assert len(gromet_collection["module_dependencies"]) == 1 

215 

216 

217def test_dependency_depth_invalid(): 

218 system = { 

219 "files": ["dependency.py"], 

220 "blobs": [ 

221 "import minimal" 

222 ], 

223 "dependency_depth": -1 

224 } 

225 

226 response = client.post("/code2fn/fn-given-filepaths", json=system) 

227 gromet_collection = response.json() 

228 assert response.status_code == 422 

229 

230 system["dependency_depth"] = 5 

231 response = client.post("/code2fn/fn-given-filepaths", json=system) 

232 gromet_collection = response.json() 

233 assert response.status_code == 422 

234 

235def test_gromet_object_count(): 

236 """Test case for get-object-count endpoint""" 

237 system = { 

238 "files": ["example1.py", "dir/example2.py"], 

239 "blobs": [ 

240 "greet = lambda: print('howdy!')\ngreet()", 

241 "#Variable declaration\nx=2\n#Function definition\ndef foo(x):\n '''Increment the input variable'''\n return x+1", # Content of dir/example2.py 

242 ], 

243 } 

244 

245 response = client.post("/code2fn/fn-given-filepaths", json=system) 

246 gromet_collection = response.json() 

247 response = client.post("/code2fn/gromet-object-count", json=gromet_collection) 

248 assert response.status_code == 200 

249 gromet_object_count = response.json() 

250 assert sum([value for key, value in gromet_object_count.items()]) > 0 

251 assert gromet_object_count["boxes"] == 16 

252 assert gromet_object_count["wires"] == 6 

253 assert gromet_object_count["ports"] == 14 

254 

255 

256# TODO: Add more complex test case to test_get_pyacset 

257def test_get_pyacset(): 

258 """Test case for /code2fn/get_pyacset endpoint.""" 

259 ports = { 

260 "opis": ["opi1", "opi2"], 

261 "opos": ["opo1", "opo2"], 

262 } 

263 response = client.post("/code2fn/get-pyacset", json=ports) 

264 assert response.status_code == 200