|
8 | 8 | import subprocess |
9 | 9 | import sys |
10 | 10 | from pathlib import Path |
11 | | -from typing import Any |
| 11 | +from typing import Any, cast |
12 | 12 |
|
13 | 13 | SCRIPT_DIR = Path(__file__).resolve().parent |
14 | 14 | if str(SCRIPT_DIR) not in sys.path: |
15 | 15 | sys.path.insert(0, str(SCRIPT_DIR)) |
16 | 16 |
|
17 | | -from project_flow_policy import load_policy |
| 17 | +from project_flow_policy import load_policy # noqa: E402 |
18 | 18 |
|
19 | 19 |
|
20 | 20 | RUNTIME_IGNORED = { |
@@ -216,9 +216,12 @@ def flush() -> None: |
216 | 216 | return candidates |
217 | 217 |
|
218 | 218 |
|
| 219 | +def _dict_section(value: object) -> dict[str, Any]: |
| 220 | + return cast(dict[str, Any], value) if isinstance(value, dict) else {} |
| 221 | + |
| 222 | + |
219 | 223 | def _effective_policy(policy: dict[str, Any]) -> dict[str, Any]: |
220 | | - effective = policy.get("effective") |
221 | | - return effective if isinstance(effective, dict) else {} |
| 224 | + return _dict_section(policy.get("effective")) |
222 | 225 |
|
223 | 226 |
|
224 | 227 | def _runtime_execution(effective_policy: dict[str, Any]) -> dict[str, Any]: |
@@ -277,8 +280,8 @@ def _policy_list(section: dict[str, Any], key: str, fallback: tuple[str, ...]) - |
277 | 280 |
|
278 | 281 | def _branch_cleanup_state(current_branch: str, policy: dict[str, Any]) -> dict[str, Any]: |
279 | 282 | effective = _effective_policy(policy) |
280 | | - branches_policy = effective.get("branches") if isinstance(effective.get("branches"), dict) else {} |
281 | | - cleanup_policy = effective.get("branch_cleanup") if isinstance(effective.get("branch_cleanup"), dict) else {} |
| 283 | + branches_policy = _dict_section(effective.get("branches")) |
| 284 | + cleanup_policy = _dict_section(effective.get("branch_cleanup")) |
282 | 285 | protected_branches = set(PROTECTED_BRANCHES) |
283 | 286 | protected_branches.update(branches_policy.get("protected_branches", []) if isinstance(branches_policy, dict) else []) |
284 | 287 | protected_branches.update(cleanup_policy.get("protected_branches", []) if isinstance(cleanup_policy, dict) else []) |
@@ -421,9 +424,9 @@ def state() -> dict[str, Any]: |
421 | 424 |
|
422 | 425 | project_policy = load_policy(Path.cwd()) |
423 | 426 | effective_policy = _effective_policy(project_policy) |
424 | | - fullrepo_policy = effective_policy.get("fullrepo") if isinstance(effective_policy.get("fullrepo"), dict) else {} |
425 | | - stop_policy = effective_policy.get("stop_hook") if isinstance(effective_policy.get("stop_hook"), dict) else {} |
426 | | - serena_policy = effective_policy.get("serena") if isinstance(effective_policy.get("serena"), dict) else {} |
| 427 | + fullrepo_policy = _dict_section(effective_policy.get("fullrepo")) |
| 428 | + stop_policy = _dict_section(effective_policy.get("stop_hook")) |
| 429 | + serena_policy = _dict_section(effective_policy.get("serena")) |
427 | 430 | fullrepo_mode = str(fullrepo_policy.get("mode", "auto")) |
428 | 431 | runtime_execution = _runtime_execution(effective_policy) |
429 | 432 | execution_mode = str(runtime_execution["execution_mode"]) |
|
0 commit comments