test_framework_plugins.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Tests for muse/plugins/code/_framework.py. |
| 2 | |
| 3 | Coverage |
| 4 | -------- |
| 5 | ImplicitEntryEdge |
| 6 | - Immutable dataclass (frozen=True). |
| 7 | - Equality comparison uses all fields. |
| 8 | |
| 9 | FrameworkPlugin protocol |
| 10 | - All three built-in plugins satisfy the Protocol at runtime. |
| 11 | - _CustomPatternPlugin satisfies the Protocol at runtime. |
| 12 | |
| 13 | _FastAPIPlugin.detect_entry_points |
| 14 | - @router.get / .post / .put / .delete / .patch recognised. |
| 15 | - @router.head / .options / .trace recognised. |
| 16 | - Path extracted from first positional argument. |
| 17 | - Method stored in UPPER CASE. |
| 18 | - No-path decorator still produces an edge (path=""). |
| 19 | - Non-HTTP attribute (@router.dependency) ignored. |
| 20 | - Non-Python source → empty list. |
| 21 | - Syntax error → empty list. |
| 22 | - Symbol not in sym_tree → no edge emitted. |
| 23 | - Class method decorated with @router.get → edge with qualified address. |
| 24 | |
| 25 | _FlaskPlugin.detect_entry_points |
| 26 | - @app.route → kind "http-route", methods from kwarg. |
| 27 | - @app.route with no methods kwarg → defaults to ["GET"]. |
| 28 | - @app.before_request (bare) → kind "lifecycle-hook". |
| 29 | - @app.before_request() (called) → kind "lifecycle-hook". |
| 30 | - @app.errorhandler(404) → kind "error-handler". |
| 31 | - Non-Flask attribute ignored. |
| 32 | |
| 33 | _CeleryPlugin.detect_entry_points |
| 34 | - @shared_task (bare Name) → kind "task". |
| 35 | - @shared_task() (called Name) → kind "task". |
| 36 | - @app.task (bare Attribute) → kind "task". |
| 37 | - @app.task() (called Attribute) → kind "task". |
| 38 | - @app.periodic_task() → kind "periodic-task". |
| 39 | - Unrelated decorator ignored. |
| 40 | |
| 41 | _CustomPatternPlugin.detect_entry_points |
| 42 | - Custom decorator name → edge with provided kind. |
| 43 | - Non-matching decorator → no edge. |
| 44 | - Only Python rules applied. |
| 45 | |
| 46 | build_implicit_edge_graph |
| 47 | - Empty manifest → empty graph. |
| 48 | - Single file with FastAPI handler → address in graph. |
| 49 | - Multiple files, multiple plugins → all edges collected. |
| 50 | - Non-Python files ignored. |
| 51 | - Syntax error in file → skipped gracefully. |
| 52 | - auto_detect=False → empty graph. |
| 53 | - disabled_plugins excludes matching plugin. |
| 54 | - Custom rule via FrameworkConfig applied. |
| 55 | |
| 56 | load_framework_config |
| 57 | - Missing file → FrameworkConfig defaults. |
| 58 | - Valid TOML section → fields parsed. |
| 59 | - Malformed TOML → defaults returned (no crash). |
| 60 | - Missing [framework_detection] section → defaults. |
| 61 | - disabled_plugins stored as frozenset. |
| 62 | - custom_entry_points parsed into _CustomEntryPointRule. |
| 63 | |
| 64 | Integration |
| 65 | - FastAPI handler excluded from dead-code analysis. |
| 66 | - Celery task excluded from dead-code analysis. |
| 67 | """ |
| 68 | |
| 69 | from __future__ import annotations |
| 70 | |
| 71 | import dataclasses |
| 72 | import hashlib |
| 73 | import pathlib |
| 74 | import textwrap |
| 75 | |
| 76 | import pytest |
| 77 | |
| 78 | from muse.core.object_store import write_object |
| 79 | from muse.core._types import Manifest |
| 80 | from muse.plugins.code.ast_parser import SymbolTree, parse_symbols |
| 81 | from muse.plugins.code._framework import ( |
| 82 | BUILTIN_PLUGINS, |
| 83 | FrameworkConfig, |
| 84 | FrameworkPlugin, |
| 85 | ImplicitEntryEdge, |
| 86 | _CeleryPlugin, |
| 87 | _CustomEntryPointRule, |
| 88 | _CustomPatternPlugin, |
| 89 | _FastAPIPlugin, |
| 90 | _FlaskPlugin, |
| 91 | build_implicit_edge_graph, |
| 92 | load_framework_config, |
| 93 | ) |
| 94 | |
| 95 | |
| 96 | # --------------------------------------------------------------------------- |
| 97 | # Helpers |
| 98 | # --------------------------------------------------------------------------- |
| 99 | |
| 100 | |
| 101 | type _SourceMap = dict[str, str] |
| 102 | |
| 103 | |
| 104 | def _write_snapshot(tmp_path: pathlib.Path, files: _SourceMap) -> Manifest: |
| 105 | """Write source files into the object store and return a manifest.""" |
| 106 | manifest: Manifest = {} |
| 107 | for rel_path, source in files.items(): |
| 108 | blob = source.encode() |
| 109 | oid = hashlib.sha256(blob).hexdigest() |
| 110 | write_object(tmp_path, oid, blob) |
| 111 | manifest[rel_path] = oid |
| 112 | return manifest |
| 113 | |
| 114 | |
| 115 | def _sym_tree(source: str, file_path: str) -> SymbolTree: |
| 116 | """Parse a source string into a SymbolTree.""" |
| 117 | return parse_symbols(source.encode(), file_path) |
| 118 | |
| 119 | |
| 120 | # --------------------------------------------------------------------------- |
| 121 | # ImplicitEntryEdge |
| 122 | # --------------------------------------------------------------------------- |
| 123 | |
| 124 | |
| 125 | class TestImplicitEntryEdge: |
| 126 | def test_is_frozen(self) -> None: |
| 127 | edge = ImplicitEntryEdge( |
| 128 | framework_id="fastapi", |
| 129 | symbol_address="app/routes.py::create_item", |
| 130 | kind="http-route", |
| 131 | metadata={"method": "POST", "path": "/items"}, |
| 132 | ) |
| 133 | with pytest.raises((AttributeError, TypeError, dataclasses.FrozenInstanceError)): |
| 134 | setattr(edge, "framework_id", "other") |
| 135 | |
| 136 | def test_equality_uses_all_fields(self) -> None: |
| 137 | e1 = ImplicitEntryEdge("fastapi", "f.py::fn", "http-route", {"method": "GET"}) |
| 138 | e2 = ImplicitEntryEdge("fastapi", "f.py::fn", "http-route", {"method": "GET"}) |
| 139 | e3 = ImplicitEntryEdge("fastapi", "f.py::fn", "http-route", {"method": "POST"}) |
| 140 | assert e1 == e2 |
| 141 | assert e1 != e3 |
| 142 | |
| 143 | def test_default_metadata_is_empty_dict(self) -> None: |
| 144 | edge = ImplicitEntryEdge( |
| 145 | framework_id="celery", |
| 146 | symbol_address="tasks.py::send_email", |
| 147 | kind="task", |
| 148 | ) |
| 149 | assert edge.metadata == {} |
| 150 | |
| 151 | |
| 152 | # --------------------------------------------------------------------------- |
| 153 | # FrameworkPlugin protocol — runtime-checkable |
| 154 | # --------------------------------------------------------------------------- |
| 155 | |
| 156 | |
| 157 | class TestFrameworkPluginProtocol: |
| 158 | def test_builtin_plugins_satisfy_protocol(self) -> None: |
| 159 | for plugin in BUILTIN_PLUGINS: |
| 160 | assert isinstance(plugin, FrameworkPlugin), ( |
| 161 | f"{type(plugin).__name__} does not satisfy FrameworkPlugin protocol" |
| 162 | ) |
| 163 | |
| 164 | def test_custom_pattern_plugin_satisfies_protocol(self) -> None: |
| 165 | plugin = _CustomPatternPlugin([]) |
| 166 | assert isinstance(plugin, FrameworkPlugin) |
| 167 | |
| 168 | |
| 169 | # --------------------------------------------------------------------------- |
| 170 | # _FastAPIPlugin |
| 171 | # --------------------------------------------------------------------------- |
| 172 | |
| 173 | |
| 174 | class TestFastAPIPlugin: |
| 175 | plugin = _FastAPIPlugin() |
| 176 | |
| 177 | def _edges(self, source: str, file_path: str = "app/routes.py") -> list[ImplicitEntryEdge]: |
| 178 | sym_tree = _sym_tree(source, file_path) |
| 179 | return self.plugin.detect_entry_points(file_path, sym_tree, source.encode()) |
| 180 | |
| 181 | def test_get_route(self) -> None: |
| 182 | src = textwrap.dedent("""\ |
| 183 | @router.get("/items") |
| 184 | async def list_items(): |
| 185 | return [] |
| 186 | """) |
| 187 | edges = self._edges(src) |
| 188 | assert len(edges) == 1 |
| 189 | assert edges[0].kind == "http-route" |
| 190 | assert edges[0].metadata["method"] == "GET" |
| 191 | assert edges[0].metadata["path"] == "/items" |
| 192 | assert edges[0].framework_id == "fastapi" |
| 193 | |
| 194 | def test_post_route(self) -> None: |
| 195 | src = textwrap.dedent("""\ |
| 196 | @router.post("/items") |
| 197 | def create_item(): |
| 198 | pass |
| 199 | """) |
| 200 | edges = self._edges(src) |
| 201 | assert len(edges) == 1 |
| 202 | assert edges[0].metadata["method"] == "POST" |
| 203 | |
| 204 | def test_put_delete_patch_routes(self) -> None: |
| 205 | src = textwrap.dedent("""\ |
| 206 | @router.put("/items/{id}") |
| 207 | def update_item(): pass |
| 208 | |
| 209 | @router.delete("/items/{id}") |
| 210 | def delete_item(): pass |
| 211 | |
| 212 | @router.patch("/items/{id}") |
| 213 | def patch_item(): pass |
| 214 | """) |
| 215 | edges = self._edges(src) |
| 216 | methods = {e.metadata["method"] for e in edges} |
| 217 | assert methods == {"PUT", "DELETE", "PATCH"} |
| 218 | |
| 219 | def test_head_options_trace(self) -> None: |
| 220 | src = textwrap.dedent("""\ |
| 221 | @router.head("/ping") |
| 222 | def ping_head(): pass |
| 223 | |
| 224 | @router.options("/ping") |
| 225 | def ping_options(): pass |
| 226 | |
| 227 | @router.trace("/ping") |
| 228 | def ping_trace(): pass |
| 229 | """) |
| 230 | edges = self._edges(src) |
| 231 | methods = {e.metadata["method"] for e in edges} |
| 232 | assert "HEAD" in methods |
| 233 | assert "OPTIONS" in methods |
| 234 | assert "TRACE" in methods |
| 235 | |
| 236 | def test_path_extracted_from_first_arg(self) -> None: |
| 237 | src = textwrap.dedent("""\ |
| 238 | @api_router.get("/api/v1/users", tags=["users"]) |
| 239 | def get_users(): pass |
| 240 | """) |
| 241 | edges = self._edges(src) |
| 242 | assert edges[0].metadata["path"] == "/api/v1/users" |
| 243 | |
| 244 | def test_no_path_arg_gives_empty_string(self) -> None: |
| 245 | src = textwrap.dedent("""\ |
| 246 | @router.get() |
| 247 | def no_path(): pass |
| 248 | """) |
| 249 | edges = self._edges(src) |
| 250 | assert len(edges) == 1 |
| 251 | assert edges[0].metadata["path"] == "" |
| 252 | |
| 253 | def test_non_http_attribute_ignored(self) -> None: |
| 254 | src = textwrap.dedent("""\ |
| 255 | @router.dependency |
| 256 | def get_db(): pass |
| 257 | """) |
| 258 | edges = self._edges(src) |
| 259 | assert edges == [] |
| 260 | |
| 261 | def test_bare_decorator_not_a_call_ignored(self) -> None: |
| 262 | # @router.get without parens — not a Call node in AST |
| 263 | src = textwrap.dedent("""\ |
| 264 | @router.get |
| 265 | def index(): pass |
| 266 | """) |
| 267 | edges = self._edges(src) |
| 268 | assert edges == [] |
| 269 | |
| 270 | def test_syntax_error_returns_empty(self) -> None: |
| 271 | src = "def broken(:\n pass\n" |
| 272 | edges = self.plugin.detect_entry_points("f.py", {}, src.encode()) |
| 273 | assert edges == [] |
| 274 | |
| 275 | def test_symbol_not_in_tree_returns_no_edge(self) -> None: |
| 276 | # Provide an empty sym_tree — _resolve_address will return None. |
| 277 | src = textwrap.dedent("""\ |
| 278 | @router.get("/items") |
| 279 | def list_items(): pass |
| 280 | """) |
| 281 | edges = self.plugin.detect_entry_points("app/routes.py", {}, src.encode()) |
| 282 | assert edges == [] |
| 283 | |
| 284 | def test_multiple_routes_in_one_file(self) -> None: |
| 285 | src = textwrap.dedent("""\ |
| 286 | @router.get("/a") |
| 287 | def get_a(): pass |
| 288 | |
| 289 | @router.post("/b") |
| 290 | def post_b(): pass |
| 291 | """) |
| 292 | edges = self._edges(src) |
| 293 | assert len(edges) == 2 |
| 294 | |
| 295 | def test_method_stored_uppercase(self) -> None: |
| 296 | src = textwrap.dedent("""\ |
| 297 | @router.get("/x") |
| 298 | def get_x(): pass |
| 299 | """) |
| 300 | edges = self._edges(src) |
| 301 | assert edges[0].metadata["method"] == "GET" |
| 302 | |
| 303 | def test_address_uses_file_path(self) -> None: |
| 304 | src = textwrap.dedent("""\ |
| 305 | @router.get("/runs") |
| 306 | async def list_runs(): pass |
| 307 | """) |
| 308 | edges = self._edges(src, file_path="server/app/routers/runs.py") |
| 309 | assert edges[0].symbol_address.startswith("server/app/routers/runs.py::") |
| 310 | |
| 311 | |
| 312 | # --------------------------------------------------------------------------- |
| 313 | # _FlaskPlugin |
| 314 | # --------------------------------------------------------------------------- |
| 315 | |
| 316 | |
| 317 | class TestFlaskPlugin: |
| 318 | plugin = _FlaskPlugin() |
| 319 | |
| 320 | def _edges(self, source: str, file_path: str = "app/views.py") -> list[ImplicitEntryEdge]: |
| 321 | sym_tree = _sym_tree(source, file_path) |
| 322 | return self.plugin.detect_entry_points(file_path, sym_tree, source.encode()) |
| 323 | |
| 324 | def test_route_with_methods_kwarg(self) -> None: |
| 325 | src = textwrap.dedent("""\ |
| 326 | @app.route("/users", methods=["GET", "POST"]) |
| 327 | def users(): pass |
| 328 | """) |
| 329 | edges = self._edges(src) |
| 330 | assert len(edges) == 1 |
| 331 | assert edges[0].kind == "http-route" |
| 332 | assert "GET" in edges[0].metadata["method"] |
| 333 | assert "POST" in edges[0].metadata["method"] |
| 334 | |
| 335 | def test_route_no_methods_defaults_to_get(self) -> None: |
| 336 | src = textwrap.dedent("""\ |
| 337 | @app.route("/index") |
| 338 | def index(): pass |
| 339 | """) |
| 340 | edges = self._edges(src) |
| 341 | assert len(edges) == 1 |
| 342 | assert "GET" in edges[0].metadata["method"] |
| 343 | |
| 344 | def test_route_path_extracted(self) -> None: |
| 345 | src = textwrap.dedent("""\ |
| 346 | @bp.route("/dashboard") |
| 347 | def dashboard(): pass |
| 348 | """) |
| 349 | edges = self._edges(src) |
| 350 | assert edges[0].metadata["path"] == "/dashboard" |
| 351 | |
| 352 | def test_before_request_bare(self) -> None: |
| 353 | src = textwrap.dedent("""\ |
| 354 | @app.before_request |
| 355 | def setup(): pass |
| 356 | """) |
| 357 | edges = self._edges(src) |
| 358 | assert len(edges) == 1 |
| 359 | assert edges[0].kind == "lifecycle-hook" |
| 360 | assert edges[0].metadata.get("hook") == "before_request" |
| 361 | |
| 362 | def test_before_request_called(self) -> None: |
| 363 | src = textwrap.dedent("""\ |
| 364 | @app.before_request() |
| 365 | def setup(): pass |
| 366 | """) |
| 367 | edges = self._edges(src) |
| 368 | assert len(edges) == 1 |
| 369 | assert edges[0].kind == "lifecycle-hook" |
| 370 | |
| 371 | def test_after_request(self) -> None: |
| 372 | src = textwrap.dedent("""\ |
| 373 | @app.after_request |
| 374 | def add_headers(response): pass |
| 375 | """) |
| 376 | edges = self._edges(src) |
| 377 | assert any(e.kind == "lifecycle-hook" for e in edges) |
| 378 | |
| 379 | def test_errorhandler(self) -> None: |
| 380 | src = textwrap.dedent("""\ |
| 381 | @app.errorhandler(404) |
| 382 | def not_found(e): pass |
| 383 | """) |
| 384 | edges = self._edges(src) |
| 385 | assert len(edges) == 1 |
| 386 | assert edges[0].kind == "error-handler" |
| 387 | |
| 388 | def test_non_flask_attribute_ignored(self) -> None: |
| 389 | src = textwrap.dedent("""\ |
| 390 | @app.middleware |
| 391 | def mw(): pass |
| 392 | """) |
| 393 | edges = self._edges(src) |
| 394 | assert edges == [] |
| 395 | |
| 396 | def test_syntax_error_returns_empty(self) -> None: |
| 397 | src = "def bad(:\n pass\n" |
| 398 | edges = self.plugin.detect_entry_points("v.py", {}, src.encode()) |
| 399 | assert edges == [] |
| 400 | |
| 401 | def test_framework_id_is_flask(self) -> None: |
| 402 | src = textwrap.dedent("""\ |
| 403 | @app.route("/") |
| 404 | def root(): pass |
| 405 | """) |
| 406 | edges = self._edges(src) |
| 407 | assert edges[0].framework_id == "flask" |
| 408 | |
| 409 | |
| 410 | # --------------------------------------------------------------------------- |
| 411 | # _CeleryPlugin |
| 412 | # --------------------------------------------------------------------------- |
| 413 | |
| 414 | |
| 415 | class TestCeleryPlugin: |
| 416 | plugin = _CeleryPlugin() |
| 417 | |
| 418 | def _edges(self, source: str, file_path: str = "tasks/email.py") -> list[ImplicitEntryEdge]: |
| 419 | sym_tree = _sym_tree(source, file_path) |
| 420 | return self.plugin.detect_entry_points(file_path, sym_tree, source.encode()) |
| 421 | |
| 422 | def test_shared_task_bare_name(self) -> None: |
| 423 | src = textwrap.dedent("""\ |
| 424 | @shared_task |
| 425 | def send_email(): pass |
| 426 | """) |
| 427 | edges = self._edges(src) |
| 428 | assert len(edges) == 1 |
| 429 | assert edges[0].kind == "task" |
| 430 | assert edges[0].framework_id == "celery" |
| 431 | |
| 432 | def test_shared_task_called(self) -> None: |
| 433 | src = textwrap.dedent("""\ |
| 434 | @shared_task(bind=True) |
| 435 | def send_email(self): pass |
| 436 | """) |
| 437 | edges = self._edges(src) |
| 438 | assert len(edges) == 1 |
| 439 | assert edges[0].kind == "task" |
| 440 | |
| 441 | def test_app_task_bare_attribute(self) -> None: |
| 442 | src = textwrap.dedent("""\ |
| 443 | @celery.task |
| 444 | def process(): pass |
| 445 | """) |
| 446 | edges = self._edges(src) |
| 447 | assert len(edges) == 1 |
| 448 | assert edges[0].kind == "task" |
| 449 | |
| 450 | def test_app_task_called_attribute(self) -> None: |
| 451 | src = textwrap.dedent("""\ |
| 452 | @app.task(queue="high") |
| 453 | def urgent(): pass |
| 454 | """) |
| 455 | edges = self._edges(src) |
| 456 | assert len(edges) == 1 |
| 457 | assert edges[0].kind == "task" |
| 458 | |
| 459 | def test_periodic_task_called(self) -> None: |
| 460 | src = textwrap.dedent("""\ |
| 461 | @app.periodic_task(run_every=60) |
| 462 | def heartbeat(): pass |
| 463 | """) |
| 464 | edges = self._edges(src) |
| 465 | assert len(edges) == 1 |
| 466 | assert edges[0].kind == "periodic-task" |
| 467 | |
| 468 | def test_periodic_task_bare(self) -> None: |
| 469 | src = textwrap.dedent("""\ |
| 470 | @app.periodic_task |
| 471 | def tick(): pass |
| 472 | """) |
| 473 | edges = self._edges(src) |
| 474 | assert len(edges) == 1 |
| 475 | assert edges[0].kind == "periodic-task" |
| 476 | |
| 477 | def test_unrelated_decorator_ignored(self) -> None: |
| 478 | src = textwrap.dedent("""\ |
| 479 | @login_required |
| 480 | def secure(): pass |
| 481 | """) |
| 482 | edges = self._edges(src) |
| 483 | assert edges == [] |
| 484 | |
| 485 | def test_syntax_error_returns_empty(self) -> None: |
| 486 | edges = self.plugin.detect_entry_points("t.py", {}, b"def broken(:\n pass\n") |
| 487 | assert edges == [] |
| 488 | |
| 489 | def test_framework_id_is_celery(self) -> None: |
| 490 | src = textwrap.dedent("""\ |
| 491 | @shared_task |
| 492 | def work(): pass |
| 493 | """) |
| 494 | edges = self._edges(src) |
| 495 | assert edges[0].framework_id == "celery" |
| 496 | |
| 497 | |
| 498 | # --------------------------------------------------------------------------- |
| 499 | # _CustomPatternPlugin |
| 500 | # --------------------------------------------------------------------------- |
| 501 | |
| 502 | |
| 503 | class TestCustomPatternPlugin: |
| 504 | def _make_plugin(self, rules: list[_CustomEntryPointRule]) -> _CustomPatternPlugin: |
| 505 | return _CustomPatternPlugin(rules) |
| 506 | |
| 507 | def test_custom_decorator_produces_edge(self) -> None: |
| 508 | rule = _CustomEntryPointRule( |
| 509 | language="Python", |
| 510 | kind="rpc-handler", |
| 511 | decorator_names=["rpc_handler"], |
| 512 | ) |
| 513 | plugin = self._make_plugin([rule]) |
| 514 | src = textwrap.dedent("""\ |
| 515 | @rpc_handler |
| 516 | def handle_request(): pass |
| 517 | """) |
| 518 | sym_tree = _sym_tree(src, "svc/rpc.py") |
| 519 | edges = plugin.detect_entry_points("svc/rpc.py", sym_tree, src.encode()) |
| 520 | assert len(edges) == 1 |
| 521 | assert edges[0].kind == "rpc-handler" |
| 522 | assert edges[0].framework_id == "custom" |
| 523 | |
| 524 | def test_called_form_also_recognised(self) -> None: |
| 525 | rule = _CustomEntryPointRule( |
| 526 | language="Python", |
| 527 | kind="webhook", |
| 528 | decorator_names=["webhook_handler"], |
| 529 | ) |
| 530 | plugin = self._make_plugin([rule]) |
| 531 | src = textwrap.dedent("""\ |
| 532 | @webhook_handler(event="push") |
| 533 | def on_push(): pass |
| 534 | """) |
| 535 | sym_tree = _sym_tree(src, "hooks.py") |
| 536 | edges = plugin.detect_entry_points("hooks.py", sym_tree, src.encode()) |
| 537 | assert len(edges) == 1 |
| 538 | assert edges[0].kind == "webhook" |
| 539 | |
| 540 | def test_non_matching_decorator_ignored(self) -> None: |
| 541 | rule = _CustomEntryPointRule( |
| 542 | language="Python", |
| 543 | kind="grpc", |
| 544 | decorator_names=["grpc_handler"], |
| 545 | ) |
| 546 | plugin = self._make_plugin([rule]) |
| 547 | src = textwrap.dedent("""\ |
| 548 | @login_required |
| 549 | def view(): pass |
| 550 | """) |
| 551 | sym_tree = _sym_tree(src, "views.py") |
| 552 | edges = plugin.detect_entry_points("views.py", sym_tree, src.encode()) |
| 553 | assert edges == [] |
| 554 | |
| 555 | def test_non_python_rules_excluded(self) -> None: |
| 556 | rule = _CustomEntryPointRule( |
| 557 | language="TypeScript", |
| 558 | kind="ts-handler", |
| 559 | decorator_names=["Handler"], |
| 560 | ) |
| 561 | plugin = self._make_plugin([rule]) |
| 562 | # Plugin should have no Python rules, so no edges. |
| 563 | src = textwrap.dedent("""\ |
| 564 | @Handler |
| 565 | def fn(): pass |
| 566 | """) |
| 567 | sym_tree = _sym_tree(src, "f.py") |
| 568 | edges = plugin.detect_entry_points("f.py", sym_tree, src.encode()) |
| 569 | assert edges == [] |
| 570 | |
| 571 | def test_empty_rules_returns_empty(self) -> None: |
| 572 | plugin = self._make_plugin([]) |
| 573 | src = textwrap.dedent("""\ |
| 574 | @anything |
| 575 | def fn(): pass |
| 576 | """) |
| 577 | sym_tree = _sym_tree(src, "f.py") |
| 578 | edges = plugin.detect_entry_points("f.py", sym_tree, src.encode()) |
| 579 | assert edges == [] |
| 580 | |
| 581 | def test_metadata_includes_decorator_name(self) -> None: |
| 582 | rule = _CustomEntryPointRule( |
| 583 | language="Python", |
| 584 | kind="job", |
| 585 | decorator_names=["schedule_job"], |
| 586 | ) |
| 587 | plugin = self._make_plugin([rule]) |
| 588 | src = textwrap.dedent("""\ |
| 589 | @schedule_job |
| 590 | def nightly(): pass |
| 591 | """) |
| 592 | sym_tree = _sym_tree(src, "jobs.py") |
| 593 | edges = plugin.detect_entry_points("jobs.py", sym_tree, src.encode()) |
| 594 | assert edges[0].metadata.get("decorator") == "schedule_job" |
| 595 | |
| 596 | |
| 597 | # --------------------------------------------------------------------------- |
| 598 | # build_implicit_edge_graph |
| 599 | # --------------------------------------------------------------------------- |
| 600 | |
| 601 | |
| 602 | class TestBuildImplicitEdgeGraph: |
| 603 | def test_empty_manifest_returns_empty(self, tmp_path: pathlib.Path) -> None: |
| 604 | config = FrameworkConfig() |
| 605 | result = build_implicit_edge_graph(tmp_path, {}, config=config) |
| 606 | assert result == {} |
| 607 | |
| 608 | def test_fastapi_handler_detected(self, tmp_path: pathlib.Path) -> None: |
| 609 | src = textwrap.dedent("""\ |
| 610 | @router.get("/runs") |
| 611 | async def list_runs(): pass |
| 612 | """) |
| 613 | manifest = _write_snapshot(tmp_path, {"server/routers/runs.py": src}) |
| 614 | config = FrameworkConfig() |
| 615 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 616 | assert any("list_runs" in addr for addr in result) |
| 617 | |
| 618 | def test_celery_task_detected(self, tmp_path: pathlib.Path) -> None: |
| 619 | src = textwrap.dedent("""\ |
| 620 | @shared_task |
| 621 | def send_notification(): pass |
| 622 | """) |
| 623 | manifest = _write_snapshot(tmp_path, {"tasks/notify.py": src}) |
| 624 | config = FrameworkConfig() |
| 625 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 626 | assert any("send_notification" in addr for addr in result) |
| 627 | |
| 628 | def test_flask_handler_detected(self, tmp_path: pathlib.Path) -> None: |
| 629 | src = textwrap.dedent("""\ |
| 630 | @app.route("/dashboard") |
| 631 | def dashboard(): pass |
| 632 | """) |
| 633 | manifest = _write_snapshot(tmp_path, {"web/views.py": src}) |
| 634 | config = FrameworkConfig() |
| 635 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 636 | assert any("dashboard" in addr for addr in result) |
| 637 | |
| 638 | def test_non_python_file_ignored(self, tmp_path: pathlib.Path) -> None: |
| 639 | go_src = "func handler() {}\n" |
| 640 | manifest = _write_snapshot(tmp_path, {"main.go": go_src}) |
| 641 | config = FrameworkConfig() |
| 642 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 643 | assert result == {} |
| 644 | |
| 645 | def test_syntax_error_file_skipped_gracefully(self, tmp_path: pathlib.Path) -> None: |
| 646 | bad_src = "def broken(:\n pass\n" |
| 647 | manifest = _write_snapshot(tmp_path, {"bad.py": bad_src}) |
| 648 | config = FrameworkConfig() |
| 649 | # Must not raise. |
| 650 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 651 | assert isinstance(result, dict) |
| 652 | |
| 653 | def test_auto_detect_false_returns_empty(self, tmp_path: pathlib.Path) -> None: |
| 654 | src = textwrap.dedent("""\ |
| 655 | @router.get("/x") |
| 656 | def get_x(): pass |
| 657 | """) |
| 658 | manifest = _write_snapshot(tmp_path, {"routes.py": src}) |
| 659 | config = FrameworkConfig(auto_detect=False) |
| 660 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 661 | assert result == {} |
| 662 | |
| 663 | def test_disabled_plugins_excluded(self, tmp_path: pathlib.Path) -> None: |
| 664 | src = textwrap.dedent("""\ |
| 665 | @router.get("/items") |
| 666 | async def list_items(): pass |
| 667 | """) |
| 668 | manifest = _write_snapshot(tmp_path, {"routes.py": src}) |
| 669 | # Disable FastAPI; FlaskPlugin and CeleryPlugin won't match this source. |
| 670 | config = FrameworkConfig(disabled_plugins=frozenset({"fastapi"})) |
| 671 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 672 | # No edge should appear since fastapi is the only matching plugin. |
| 673 | assert not any("list_items" in addr for addr in result) |
| 674 | |
| 675 | def test_multiple_files_multiple_plugins(self, tmp_path: pathlib.Path) -> None: |
| 676 | fastapi_src = textwrap.dedent("""\ |
| 677 | @router.post("/orders") |
| 678 | def create_order(): pass |
| 679 | """) |
| 680 | celery_src = textwrap.dedent("""\ |
| 681 | @shared_task |
| 682 | def process_order(): pass |
| 683 | """) |
| 684 | manifest = _write_snapshot(tmp_path, { |
| 685 | "api/routes.py": fastapi_src, |
| 686 | "workers/tasks.py": celery_src, |
| 687 | }) |
| 688 | config = FrameworkConfig() |
| 689 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 690 | addrs = set(result.keys()) |
| 691 | assert any("create_order" in a for a in addrs) |
| 692 | assert any("process_order" in a for a in addrs) |
| 693 | |
| 694 | def test_custom_rule_in_config(self, tmp_path: pathlib.Path) -> None: |
| 695 | src = textwrap.dedent("""\ |
| 696 | @grpc_handler |
| 697 | def serve_request(): pass |
| 698 | """) |
| 699 | manifest = _write_snapshot(tmp_path, {"svc/grpc.py": src}) |
| 700 | rule = _CustomEntryPointRule( |
| 701 | language="Python", kind="grpc", decorator_names=["grpc_handler"] |
| 702 | ) |
| 703 | config = FrameworkConfig(custom_entry_points=[rule]) |
| 704 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 705 | assert any("serve_request" in addr for addr in result) |
| 706 | # Kind should be grpc |
| 707 | edges = [e for edges in result.values() for e in edges] |
| 708 | assert any(e.kind == "grpc" for e in edges) |
| 709 | |
| 710 | def test_all_disabled_returns_empty(self, tmp_path: pathlib.Path) -> None: |
| 711 | src = textwrap.dedent("""\ |
| 712 | @router.get("/x") |
| 713 | async def get_x(): pass |
| 714 | """) |
| 715 | manifest = _write_snapshot(tmp_path, {"r.py": src}) |
| 716 | config = FrameworkConfig(disabled_plugins=frozenset({"fastapi", "flask", "celery"})) |
| 717 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 718 | assert result == {} |
| 719 | |
| 720 | def test_graph_maps_address_to_edges(self, tmp_path: pathlib.Path) -> None: |
| 721 | src = textwrap.dedent("""\ |
| 722 | @router.get("/x") |
| 723 | async def get_x(): pass |
| 724 | """) |
| 725 | manifest = _write_snapshot(tmp_path, {"routes.py": src}) |
| 726 | config = FrameworkConfig() |
| 727 | result = build_implicit_edge_graph(tmp_path, manifest, config=config) |
| 728 | for addr, edges in result.items(): |
| 729 | assert isinstance(addr, str) |
| 730 | assert isinstance(edges, list) |
| 731 | for e in edges: |
| 732 | assert isinstance(e, ImplicitEntryEdge) |
| 733 | |
| 734 | |
| 735 | # --------------------------------------------------------------------------- |
| 736 | # load_framework_config |
| 737 | # --------------------------------------------------------------------------- |
| 738 | |
| 739 | |
| 740 | class TestLoadFrameworkConfig: |
| 741 | def test_missing_file_returns_defaults(self, tmp_path: pathlib.Path) -> None: |
| 742 | config = load_framework_config(tmp_path) |
| 743 | assert config.auto_detect is True |
| 744 | assert config.disabled_plugins == frozenset() |
| 745 | assert config.custom_entry_points == [] |
| 746 | |
| 747 | def test_valid_toml_parsed(self, tmp_path: pathlib.Path) -> None: |
| 748 | muse_dir = tmp_path / ".muse" |
| 749 | muse_dir.mkdir() |
| 750 | toml_content = textwrap.dedent("""\ |
| 751 | [framework_detection] |
| 752 | auto_detect = true |
| 753 | disabled_plugins = ["celery"] |
| 754 | |
| 755 | [[framework_detection.custom_entry_points]] |
| 756 | language = "Python" |
| 757 | kind = "grpc" |
| 758 | decorator_names = ["grpc_method"] |
| 759 | """) |
| 760 | (muse_dir / "code_config.toml").write_text(toml_content) |
| 761 | config = load_framework_config(tmp_path) |
| 762 | assert config.auto_detect is True |
| 763 | assert "celery" in config.disabled_plugins |
| 764 | assert len(config.custom_entry_points) == 1 |
| 765 | assert config.custom_entry_points[0].kind == "grpc" |
| 766 | assert "grpc_method" in config.custom_entry_points[0].decorator_names |
| 767 | |
| 768 | def test_auto_detect_false_parsed(self, tmp_path: pathlib.Path) -> None: |
| 769 | muse_dir = tmp_path / ".muse" |
| 770 | muse_dir.mkdir() |
| 771 | (muse_dir / "code_config.toml").write_text( |
| 772 | "[framework_detection]\nauto_detect = false\n" |
| 773 | ) |
| 774 | config = load_framework_config(tmp_path) |
| 775 | assert config.auto_detect is False |
| 776 | |
| 777 | def test_missing_section_returns_defaults(self, tmp_path: pathlib.Path) -> None: |
| 778 | muse_dir = tmp_path / ".muse" |
| 779 | muse_dir.mkdir() |
| 780 | (muse_dir / "code_config.toml").write_text("[other_section]\nkey = 1\n") |
| 781 | config = load_framework_config(tmp_path) |
| 782 | assert config.auto_detect is True |
| 783 | assert config.disabled_plugins == frozenset() |
| 784 | |
| 785 | def test_malformed_toml_returns_defaults(self, tmp_path: pathlib.Path) -> None: |
| 786 | muse_dir = tmp_path / ".muse" |
| 787 | muse_dir.mkdir() |
| 788 | (muse_dir / "code_config.toml").write_bytes(b"\xff\xfe invalid toml !!!") |
| 789 | # Must not raise — returns defaults. |
| 790 | config = load_framework_config(tmp_path) |
| 791 | assert isinstance(config, FrameworkConfig) |
| 792 | |
| 793 | def test_disabled_plugins_stored_as_frozenset(self, tmp_path: pathlib.Path) -> None: |
| 794 | muse_dir = tmp_path / ".muse" |
| 795 | muse_dir.mkdir() |
| 796 | (muse_dir / "code_config.toml").write_text( |
| 797 | '[framework_detection]\ndisabled_plugins = ["flask", "celery"]\n' |
| 798 | ) |
| 799 | config = load_framework_config(tmp_path) |
| 800 | assert isinstance(config.disabled_plugins, frozenset) |
| 801 | assert "flask" in config.disabled_plugins |
| 802 | assert "celery" in config.disabled_plugins |
| 803 | |
| 804 | def test_multiple_custom_rules(self, tmp_path: pathlib.Path) -> None: |
| 805 | muse_dir = tmp_path / ".muse" |
| 806 | muse_dir.mkdir() |
| 807 | toml = textwrap.dedent("""\ |
| 808 | [framework_detection] |
| 809 | |
| 810 | [[framework_detection.custom_entry_points]] |
| 811 | language = "Python" |
| 812 | kind = "grpc" |
| 813 | decorator_names = ["grpc_method"] |
| 814 | |
| 815 | [[framework_detection.custom_entry_points]] |
| 816 | language = "Python" |
| 817 | kind = "webhook" |
| 818 | decorator_names = ["webhook_handler", "on_event"] |
| 819 | """) |
| 820 | (muse_dir / "code_config.toml").write_text(toml) |
| 821 | config = load_framework_config(tmp_path) |
| 822 | assert len(config.custom_entry_points) == 2 |
| 823 | kinds = {r.kind for r in config.custom_entry_points} |
| 824 | assert kinds == {"grpc", "webhook"} |
| 825 | |
| 826 | |
| 827 | # --------------------------------------------------------------------------- |
| 828 | # Integration — dead-code analysis excludes entry points |
| 829 | # --------------------------------------------------------------------------- |
| 830 | |
| 831 | |
| 832 | class TestDeadCodeIntegration: |
| 833 | """Verify that framework-wired symbols are not reported as dead code.""" |
| 834 | |
| 835 | def test_fastapi_handler_excluded_from_dead(self, tmp_path: pathlib.Path) -> None: |
| 836 | """A FastAPI route handler with no explicit callers is not dead code.""" |
| 837 | from muse.plugins.code._framework import build_implicit_edge_graph |
| 838 | |
| 839 | src = textwrap.dedent("""\ |
| 840 | @router.get("/items") |
| 841 | async def list_items(): pass |
| 842 | """) |
| 843 | manifest = _write_snapshot(tmp_path, {"api/routes.py": src}) |
| 844 | implicit = build_implicit_edge_graph(tmp_path, manifest) |
| 845 | entry_point_addresses = frozenset(implicit.keys()) |
| 846 | # The handler must be recognised as an entry point. |
| 847 | assert any("list_items" in addr for addr in entry_point_addresses) |
| 848 | |
| 849 | def test_celery_task_excluded_from_dead(self, tmp_path: pathlib.Path) -> None: |
| 850 | from muse.plugins.code._framework import build_implicit_edge_graph |
| 851 | |
| 852 | src = textwrap.dedent("""\ |
| 853 | @shared_task |
| 854 | def send_report(): pass |
| 855 | """) |
| 856 | manifest = _write_snapshot(tmp_path, {"tasks/report.py": src}) |
| 857 | implicit = build_implicit_edge_graph(tmp_path, manifest) |
| 858 | entry_point_addresses = frozenset(implicit.keys()) |
| 859 | assert any("send_report" in addr for addr in entry_point_addresses) |
| 860 | |
| 861 | def test_plain_function_not_mistaken_for_entry_point( |
| 862 | self, tmp_path: pathlib.Path |
| 863 | ) -> None: |
| 864 | from muse.plugins.code._framework import build_implicit_edge_graph |
| 865 | |
| 866 | src = textwrap.dedent("""\ |
| 867 | def compute_total(items): pass |
| 868 | """) |
| 869 | manifest = _write_snapshot(tmp_path, {"billing.py": src}) |
| 870 | implicit = build_implicit_edge_graph(tmp_path, manifest) |
| 871 | entry_point_addresses = frozenset(implicit.keys()) |
| 872 | assert not any("compute_total" in addr for addr in entry_point_addresses) |
File History
4 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e
Merge branch 'dev' into main
Human
29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce
Merge branch 'dev' into main
Human
48 days ago