Coverage for tests/test_project_scan.py: 100%

1import pathlib 

2import types 

3import typing 

4 

5import pytest 

6 

7from archdocs.main import ArchitectureParserAndRenderer, SettingsForArchdocs 

8from tests import diagram_rendering 

9 

10 

11_OWN_SOURCE: typing.Final = """import redis 

12 

13 

14cache_client = redis.Redis(host="localhost") 

15""" 

16_VENDORED_SOURCE: typing.Final = """import celery 

17import uvicorn 

18 

19 

20app = celery.Celery(__name__) 

21""" 

22_APPLICATION_SOURCE: typing.Final = "import fastapi\n\napp = fastapi.FastAPI()\n" 

23_CHART_RELATIVE_PATH: typing.Final = "deploy/mychart" 

24_SOURCES_RELATIVE_PATH: typing.Final = "src" 

25_NEIGHBOUR_CHART_VALUES: typing.Final = """replicaCount: 4 

26 

27ingress: 

28 enabled: true 

29 hosts: 

30 - host: neighbour.example.com 

31""" 

32_RAW_INGRESS_MANIFEST: typing.Final = """apiVersion: networking.k8s.io/v1 

33kind: Ingress 

34metadata: 

35 name: mychart 

36spec: 

37 tls: 

38 - secretName: from-templates-tls 

39 rules: 

40 - host: from-templates.example.com 

41""" 

42_DECOY_CHART_VALUES: typing.Final = """replicaCount: 9 

43 

44ingress: 

45 enabled: true 

46 hosts: 

47 - host: decoy.example.com 

48""" 

49_SUBCHART_VALUES: typing.Final = "replicaCount: 9\n" 

50_NEIGHBOUR_HOST_EDGE: typing.Final = "HTTP neighbour.example.com" 

51 

52 

53def _write_legacy_encoded_source(project_path: pathlib.Path, /) -> None: 

54 (project_path / "legacy.py").write_bytes("HEADING = 'café'\n".encode("latin-1")) 

55 

56 

57def _write_dangling_symlink(project_path: pathlib.Path, /) -> None: 

58 (project_path / "removed.py").symlink_to(project_path / "never-existed.py") 

59 

60 

61def _write_legacy_encoded_manifest(chart_path: pathlib.Path, /) -> None: 

62 (chart_path / "legacy.yaml").write_bytes( 

63 "apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: café\n".encode("latin-1"), 

64 ) 

65 

66 

67def _write_dangling_manifest_symlinks(chart_path: pathlib.Path, /) -> None: 

68 (chart_path / "broken.yaml").symlink_to(chart_path / "never-existed.yaml") 

69 (chart_path.parent / "values.yaml").symlink_to(chart_path.parent / "never-existed-values.yaml") 

70 

71 

72_ALL_UNREADABLE_SOURCES: typing.Final = types.MappingProxyType( 

73 { 

74 "legacy encoding": _write_legacy_encoded_source, 

75 "dangling symlink": _write_dangling_symlink, 

76 }, 

77) 

78_ALL_UNREADABLE_MANIFESTS: typing.Final = types.MappingProxyType( 

79 { 

80 "legacy encoding": _write_legacy_encoded_manifest, 

81 "dangling symlink": _write_dangling_manifest_symlinks, 

82 }, 

83) 

84 

85 

86def _build_charted_project( 

87 project_path: pathlib.Path, 

88 /, 

89 *, 

90 chart_path: pathlib.Path | None = None, 

91 chart_values: str = _NEIGHBOUR_CHART_VALUES, 

92) -> pathlib.Path: 

93 source_dir: typing.Final = project_path / _SOURCES_RELATIVE_PATH 

94 source_dir.mkdir(parents=True) 

95 (source_dir / "main.py").write_text(_APPLICATION_SOURCE) 

96 chart_dir: typing.Final = project_path / _CHART_RELATIVE_PATH if chart_path is None else chart_path 

97 chart_dir.mkdir(parents=True) 

98 (chart_dir / "Chart.yaml").write_text("apiVersion: v2\nname: mychart\n") 

99 (chart_dir / "values.yaml").write_text(chart_values) 

100 return source_dir 

101 

102 

103# The last case is the mirror one: a service living in a `build` directory must not skip itself. 

104@pytest.mark.parametrize( 

105 ("project_subpath", "vendored_relative_path"), 

106 [ 

107 (".", ".venv/lib/python3.12/site-packages/celery"), 

108 (".", "venv/celery"), 

109 (".", "node_modules/celery"), 

110 (".", "build/lib/celery"), 

111 ("build/dist/myservice", ".venv/celery"), 

112 ], 

113) 

114def test_dependencies_stay_out_of_the_diagram( 

115 tmp_path: pathlib.Path, 

116 project_subpath: str, 

117 vendored_relative_path: str, 

118) -> None: 

119 project_path: typing.Final = tmp_path / project_subpath 

120 project_path.mkdir(parents=True, exist_ok=True) 

121 (project_path / "service.py").write_text(_OWN_SOURCE) 

122 vendored_dir: typing.Final = project_path / vendored_relative_path 

123 vendored_dir.mkdir(parents=True) 

124 (vendored_dir / "vendored.py").write_text(_VENDORED_SOURCE) 

125 

126 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

127 SettingsForArchdocs( 

128 root_dir=project_path, 

129 service_name="vendor-svc", 

130 kubernetes_dir=diagram_rendering.WITHOUT_MANIFESTS, 

131 ), 

132 ) 

133 

134 assert 'redisdb["redis"]' in rendered_diagram 

135 assert "celery" not in rendered_diagram 

136 assert "uvicorn" not in rendered_diagram 

137 

138 

139# The scanned tree is the user's whole project: one file the process cannot decode or open used 

140# to raise out of the thread pool and answer the route with 500 instead of the rest of the service. 

141@pytest.mark.parametrize("break_one_source", _ALL_UNREADABLE_SOURCES.values(), ids=_ALL_UNREADABLE_SOURCES) 

142def test_unreadable_source_costs_only_itself( 

143 tmp_path: pathlib.Path, 

144 break_one_source: typing.Callable[[pathlib.Path], None], 

145) -> None: 

146 (tmp_path / "cache.py").write_text(_OWN_SOURCE) 

147 break_one_source(tmp_path) 

148 

149 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

150 SettingsForArchdocs( 

151 root_dir=tmp_path, 

152 service_name="unreadable-svc", 

153 kubernetes_dir=diagram_rendering.WITHOUT_MANIFESTS, 

154 ), 

155 ) 

156 

157 assert 'redisdb["redis"]' in rendered_diagram 

158 

159 

160@pytest.mark.parametrize("sources_subpath", [".", "one"]) 

161def test_manifests_are_found_above_the_sources(tmp_path: pathlib.Path, sources_subpath: str) -> None: 

162 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

163 SettingsForArchdocs( 

164 root_dir=_build_charted_project(tmp_path / sources_subpath, chart_path=tmp_path / _CHART_RELATIVE_PATH), 

165 service_name="above-svc", 

166 ), 

167 ) 

168 

169 assert 'above_svc{"above-svc (replicas 4)"}' in rendered_diagram 

170 assert _NEIGHBOUR_HOST_EDGE in rendered_diagram 

171 

172 

173def test_chart_is_found_by_its_templates(tmp_path: pathlib.Path) -> None: 

174 templates_dir: typing.Final = tmp_path / _CHART_RELATIVE_PATH / "templates" 

175 templates_dir.mkdir(parents=True) 

176 (templates_dir / "ingress.yaml").write_text(_RAW_INGRESS_MANIFEST) 

177 

178 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

179 SettingsForArchdocs(root_dir=tmp_path, service_name="templates-svc"), 

180 ) 

181 

182 assert "HTTPS from-templates.example.com" in rendered_diagram 

183 

184 

185# The decoy chart is what the working directory would offer a relative path. 

186@pytest.mark.parametrize( 

187 ("kubernetes_dir", "expected_part", "forbidden_part"), 

188 [ 

189 (diagram_rendering.KUBERNETES_VARIANTS_ROOT / "loadbalancer", "LoadBalancer", "neighbour.example.com"), 

190 ("there-is-no-such-chart", 'config_svc{"config-svc"}', "neighbour.example.com"), 

191 (_CHART_RELATIVE_PATH, _NEIGHBOUR_HOST_EDGE, "decoy.example.com"), 

192 ], 

193) 

194def test_configured_dir_wins_over_the_search( 

195 tmp_path: pathlib.Path, 

196 monkeypatch: pytest.MonkeyPatch, 

197 kubernetes_dir: str | pathlib.Path, 

198 expected_part: str, 

199 forbidden_part: str, 

200) -> None: 

201 decoy_project_path: typing.Final = tmp_path / "elsewhere" 

202 _build_charted_project(decoy_project_path, chart_values=_DECOY_CHART_VALUES) 

203 monkeypatch.chdir(decoy_project_path) 

204 

205 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

206 SettingsForArchdocs( 

207 root_dir=_build_charted_project(tmp_path / "project"), 

208 service_name="config-svc", 

209 kubernetes_dir=kubernetes_dir, 

210 ), 

211 ) 

212 

213 assert expected_part in rendered_diagram 

214 assert forbidden_part not in rendered_diagram 

215 

216 

217# A typo in root_dir is the emptiest possible project, not an error page: the service node 

218# still has to appear, alone. 

219def test_missing_root_dir_draws_the_service_alone(tmp_path: pathlib.Path) -> None: 

220 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

221 SettingsForArchdocs( 

222 root_dir=tmp_path / "never-created", 

223 service_name="missing-svc", 

224 kubernetes_dir=diagram_rendering.WITHOUT_MANIFESTS, 

225 ), 

226 ) 

227 

228 assert 'missing_svc{"missing-svc"}' in rendered_diagram 

229 assert " --> " not in rendered_diagram 

230 

231 

232# Manifests are hunted through the same foreign tree as the sources: a dangling symlink or a 

233# manifest in a legacy encoding next to the chart used to raise out of the walk and answer the 

234# route with 500 instead of the chart. 

235@pytest.mark.parametrize("break_one_manifest", _ALL_UNREADABLE_MANIFESTS.values(), ids=_ALL_UNREADABLE_MANIFESTS) 

236def test_unreadable_manifest_costs_only_itself( 

237 tmp_path: pathlib.Path, 

238 break_one_manifest: typing.Callable[[pathlib.Path], None], 

239) -> None: 

240 source_dir: typing.Final = _build_charted_project(tmp_path) 

241 break_one_manifest(tmp_path / _CHART_RELATIVE_PATH) 

242 

243 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

244 SettingsForArchdocs(root_dir=source_dir, service_name="broken-chart-svc"), 

245 ) 

246 

247 assert 'broken_chart_svc{"broken-chart-svc (replicas 4)"}' in rendered_diagram 

248 assert _NEIGHBOUR_HOST_EDGE in rendered_diagram 

249 

250 

251# Helm reads a subchart's values under the ones of the chart that pulls it in, so the chart the 

252# diagram is drawn from is the outer one — even though `charts/` sorts before `values.yaml`. 

253def test_subchart_values_lose_to_the_chart(tmp_path: pathlib.Path) -> None: 

254 source_dir: typing.Final = _build_charted_project(tmp_path) 

255 subchart_dir: typing.Final = tmp_path / _CHART_RELATIVE_PATH / "charts" / "redis" 

256 subchart_dir.mkdir(parents=True) 

257 (subchart_dir / "Chart.yaml").write_text("apiVersion: v2\nname: redis\n") 

258 (subchart_dir / "values.yaml").write_text(_SUBCHART_VALUES) 

259 

260 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

261 SettingsForArchdocs(root_dir=source_dir, service_name="subchart-svc"), 

262 ) 

263 

264 assert 'subchart_svc{"subchart-svc (replicas 4)"}' in rendered_diagram 

265 assert _NEIGHBOUR_HOST_EDGE in rendered_diagram 

266 

267 

268# A mounted route keeps one engine alive, and a rescan on every request would walk the whole 

269# tree again: sources edited under a running process wait for a restart, see the playground. 

270def test_sources_are_scanned_once_per_engine(tmp_path: pathlib.Path) -> None: 

271 (tmp_path / "service.py").write_text(_OWN_SOURCE) 

272 architecture_engine: typing.Final = ArchitectureParserAndRenderer( 

273 local_settings=SettingsForArchdocs( 

274 root_dir=tmp_path, 

275 service_name="cached-svc", 

276 kubernetes_dir=diagram_rendering.WITHOUT_MANIFESTS, 

277 ), 

278 ) 

279 first_diagram: typing.Final = architecture_engine.render_architecture_diagram() 

280 

281 (tmp_path / "service.py").write_text(_APPLICATION_SOURCE) 

282 second_diagram: typing.Final = architecture_engine.render_architecture_diagram() 

283 

284 assert second_diagram == first_diagram 

285 assert 'redisdb["redis"]' in second_diagram 

286 assert "REST" not in second_diagram 

287 

288 

289@pytest.mark.parametrize(("project_subpath", "repository_marker"), [("project", ".git"), ("one/two/three", "")]) 

290def test_far_away_manifests_are_ignored( 

291 tmp_path: pathlib.Path, 

292 project_subpath: str, 

293 repository_marker: str, 

294) -> None: 

295 project_path: typing.Final = tmp_path / project_subpath 

296 source_dir: typing.Final = _build_charted_project( 

297 project_path, 

298 chart_path=tmp_path / _CHART_RELATIVE_PATH, 

299 chart_values=_DECOY_CHART_VALUES, 

300 ) 

301 if repository_marker: 

302 (project_path / repository_marker).mkdir() 

303 

304 rendered_diagram: typing.Final = diagram_rendering.render_diagram( 

305 SettingsForArchdocs(root_dir=source_dir, service_name="far-svc"), 

306 ) 

307 

308 assert 'far_svc{"far-svc"}' in rendered_diagram 

309 assert "decoy.example.com" not in rendered_diagram