From dc899d23c85f879dfa5b2e03c376112f1ea8382a Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 15 Jul 2026 21:13:21 -0400 Subject: [PATCH] fix(session): require config-backed mutation authority and workspace-bound targets (#714) Close the #714 remediation gap left at 943d402 where env-only mutation profiles, REMOTES Timesheet defaults treated as explicit caller input, and machine-dependent Git remotes produced false greens. - Fail closed when mutations lack a config-backed profile with non-empty allowed_repositories; env ops cannot invent mutation authority. - Preserve omitted-vs-explicit org/repo provenance through the mutation gate; omitted targets bind to the verified workspace repository. - Prefer workspace-aligned Git remotes over historical Timesheet defaults in MCP _resolve and CLI resolve_remote. - Centralize deterministic config-backed mutation fixtures; migrate mutation tests off env-only authority without weakening security assertions. Full suite: 2744 passed, 6 skipped. --- gitea_auth.py | 46 +- gitea_mcp_server.py | 308 +++++++++-- session_context_binding.py | 59 ++- tests/conftest.py | 14 + tests/mutation_profile_fixture.py | 483 ++++++++++++++++++ tests/test_agent_temp_artifacts.py | 12 +- tests/test_assess_conflict_fix_push_tool.py | 4 + tests/test_audit.py | 62 ++- tests/test_commit_files_capability.py | 3 + tests/test_commit_files_gate.py | 2 + tests/test_commit_payloads.py | 1 + tests/test_config.py | 4 + tests/test_create_issue.py | 25 +- tests/test_create_pr.py | 25 +- tests/test_issue_714_production_first_bind.py | 67 ++- ..._issue_714_session_context_immutability.py | 12 + tests/test_issue_comment_workspace_guard.py | 4 + tests/test_issue_content_gate.py | 11 +- tests/test_issue_duplicate_gate.py | 13 +- tests/test_issue_work_duplicate_gate.py | 29 +- tests/test_issue_write_tool_gates.py | 6 + tests/test_llm_agent_sha.py | 4 + tests/test_lock_issue_mcp_registration.py | 26 +- tests/test_mcp_server.py | 252 +++++---- tests/test_merger_lease_adoption.py | 4 + tests/test_op_normalization.py | 4 + tests/test_operation_scoped_roles.py | 16 + tests/test_operator_guide.py | 4 + tests/test_post_merge_moot_lease.py | 4 + .../test_pr_lease_comments_non_list_guard.py | 4 + tests/test_remote_repo_guard.py | 14 +- tests/test_role_namespace_gate.py | 11 +- tests/test_role_session_router.py | 4 + tests/test_runtime_clarity.py | 9 +- ...test_stale_review_decision_lock_cleanup.py | 16 +- tests/test_workflow_skill.py | 8 +- 36 files changed, 1343 insertions(+), 227 deletions(-) create mode 100644 tests/mutation_profile_fixture.py diff --git a/gitea_auth.py b/gitea_auth.py index 385db61..31bdf6d 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -182,11 +182,51 @@ def get_auth_header(host): def resolve_remote(args): """Given parsed argparse args with --remote/--host/--org/--repo, - return (host, org, repo) with overrides applied.""" + return (host, org, repo) with overrides applied. + + #714 / #530: when the caller omits org and/or repo, prefer the + workspace-aligned git remote over REMOTES defaults (e.g. bare + ``--remote prgs`` must not force Timesheet when the checkout is + Gitea-Tools). Explicit --org/--repo always win. + """ profile = REMOTES[args.remote] host = args.host or profile["host"] - org = args.org or profile["org"] - repo = args.repo or profile["repo"] + org_explicit = getattr(args, "org", None) is not None + repo_explicit = getattr(args, "repo", None) is not None + org = args.org if org_explicit else profile["org"] + repo = args.repo if repo_explicit else profile["repo"] + if not org_explicit or not repo_explicit: + try: + import remote_repo_guard + import subprocess + # Prefer the named remote URL when present; fall back to origin. + url = None + for remote_name in (args.remote, "origin"): + try: + proc = subprocess.run( + ["git", "remote", "get-url", remote_name], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + if proc.returncode == 0 and (proc.stdout or "").strip(): + url = proc.stdout.strip() + break + except Exception: + continue + # Allow tests to inject a deterministic remote URL without Git. + env_url = os.environ.get("GITEA_TEST_WORKSPACE_REMOTE_URL") + if env_url: + url = env_url + parsed = remote_repo_guard.parse_org_repo_from_remote_url(url) + if parsed: + if not org_explicit: + org = parsed[0] + if not repo_explicit: + repo = parsed[1] + except Exception: + pass return host, org, repo diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 823b52b..b39b571 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1140,24 +1140,50 @@ def _workspace_repository_slug(remote: str | None) -> str | None: return session_ctx.format_repository_slug(parsed[0], parsed[1]) -def _trusted_session_repository(profile: dict, remote: str | None) -> dict: +def _trusted_session_repository( + profile: dict, + remote: str | None, + *, + for_mutation: bool = False, +) -> dict: """Resolve and authorize the session repository from the verified workspace. Returns ``org`` / ``repository`` / ``reasons``. A profile's ``allowed_repositories`` is an authorization boundary: the verified workspace selects exactly one entry from it, and the session never binds to the list itself nor switches between entries. + + *for_mutation*: require a non-empty configured allowlist and a complete + workspace-derived identity (fail closed; no self-authorization). """ + reasons: list[str] = [] + try: + allowed = session_ctx.declared_allowed_repositories( + profile, strict=for_mutation + ) + except ValueError as exc: + return { + "org": None, + "repository": None, + "reasons": [str(exc)], + } slug = _workspace_repository_slug(remote) scope = session_ctx.assess_repository_scope( workspace_slug=slug, - allowed=session_ctx.declared_allowed_repositories(profile), + allowed=allowed, profile_name=profile.get("profile_name"), + require_scope=for_mutation, ) if scope.get("block"): return {"org": None, "repository": None, "reasons": list(scope["reasons"])} parts = session_ctx.parse_repository_slug(slug) if slug else None if not parts: + if for_mutation: + reasons.append( + "mutation denied: no verified workspace repository identity " + "could be established (fail closed)" + ) + return {"org": None, "repository": None, "reasons": reasons} return {"org": None, "repository": None, "reasons": []} return {"org": parts[0], "repository": parts[1], "reasons": []} @@ -1774,20 +1800,45 @@ def _effective_remote(remote: str) -> str: def _resolve(remote: str, host: str | None, org: str | None, repo: str | None): - """Resolve remote + overrides to (host, org, repo).""" + """Resolve remote + overrides to (host, org, repo). + + #714 / #530: when the caller omits org and/or repo, prefer the + workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare + ``remote=prgs`` must not force ``Timesheet`` when the checkout is + ``Gitea-Tools``). Explicit caller org/repo always win and are validated + by the remote/repo guard. Provenance of omitted-vs-explicit is preserved + at mutation gates by passing the original caller values (not these + resolved defaults) into override checks. + """ remote = _effective_remote(remote) if remote not in REMOTES: raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}") profile = REMOTES[remote] + org_explicit = org is not None + repo_explicit = repo is not None resolved_host = host or profile["host"] - resolved_org = org or profile["org"] - resolved_repo = repo or profile["repo"] + resolved_org = org if org_explicit else profile["org"] + resolved_repo = repo if repo_explicit else profile["repo"] + filled_org = False + filled_repo = False + if not org_explicit or not repo_explicit: + parsed = remote_repo_guard.parse_org_repo_from_remote_url( + _local_git_remote_url(remote) + ) + if parsed: + if not org_explicit: + resolved_org = parsed[0] + filled_org = True + if not repo_explicit: + resolved_repo = parsed[1] + filled_repo = True _enforce_remote_repo_guard( remote, resolved_org, resolved_repo, - org_explicit=org is not None, - repo_explicit=repo is not None, + # Workspace-filled sides are intentional alignment for #530. + org_explicit=org_explicit or filled_org, + repo_explicit=repo_explicit or filled_repo, ) return (resolved_host, resolved_org, resolved_repo) @@ -2145,6 +2196,8 @@ def gitea_create_issue( host=h, org=o, repo=r, + org_explicit=org is not None, + repo_explicit=repo is not None, ) if blocked: return blocked @@ -2321,7 +2374,15 @@ def gitea_lock_issue( ) blocked = _profile_permission_block( - task_capability_map.required_permission("lock_issue")) + task_capability_map.required_permission("lock_issue"), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked @@ -2589,6 +2650,8 @@ def gitea_create_pr( host=host, org=org, repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, ) if blocked: return blocked @@ -5426,6 +5489,8 @@ def gitea_commit_files( blocked = _profile_permission_block( task_capability_map.required_permission("commit_files"), commit="", branch="", remote=remote, host=host, org=org, repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, ) if blocked: return blocked @@ -7343,7 +7408,15 @@ def gitea_close_issue( dict with 'success' boolean and 'message'. """ blocked = _profile_permission_block( - task_capability_map.required_permission("close_issue")) + task_capability_map.required_permission("close_issue"), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked verify_preflight_purity(remote, task="close_issue") @@ -7641,6 +7714,81 @@ def _profile_operation_gate(op: str) -> list[str]: return [f"profile is not allowed to {op}"] +def _mutation_config_authority_block(required_operation: str) -> dict | None: + """#714: mutations require config-backed profile + non-empty repository scope. + + Environment-only profiles may support read-only diagnostics but cannot + authorize writes. Env vars may narrow operations on a configured profile + but cannot invent mutation authority or repository scope. + """ + if required_operation == "gitea.read": + return None + try: + profile = get_profile() + except Exception as exc: + return { + "success": False, + "performed": False, + "reasons": [ + f"profile could not be resolved (fail closed): {_redact(str(exc))}" + ], + "blocker_kind": "mutation_authority", + } + config = None + try: + config = gitea_config.load_config() + except Exception as exc: + return { + "success": False, + "performed": False, + "reasons": [ + f"profiles configuration unreadable (fail closed): {_redact(str(exc))}" + ], + "blocker_kind": "mutation_authority", + } + if not config: + return { + "success": False, + "performed": False, + "reasons": [ + "mutation denied: env-only / unconfigured profiles cannot " + "authorize mutations; provision GITEA_MCP_CONFIG with a v2 " + "profile that declares non-empty allowed_repositories " + "(fail closed)" + ], + "blocker_kind": "mutation_authority", + "exact_next_action": ( + "BLOCKED + DIAGNOSE: configure a trusted v2 profile with " + "allowed_repositories and relaunch; do not widen scope via env." + ), + } + try: + allowed = session_ctx.declared_allowed_repositories(profile, strict=True) + except ValueError as exc: + return { + "success": False, + "performed": False, + "reasons": [str(exc)], + "blocker_kind": "mutation_authority", + } + if not allowed: + return { + "success": False, + "performed": False, + "reasons": [ + f"mutation denied: profile '{profile.get('profile_name')}' " + "must declare a non-empty allowed_repositories list " + "(fail closed)" + ], + "blocker_kind": "mutation_authority", + "exact_next_action": ( + "BLOCKED + DIAGNOSE: add allowed_repositories: " + "[\"Owner/Repo\"] to the active v2 profile and restart." + ), + } + return None + + def _session_context_mutation_block( *, remote: str | None, @@ -7665,19 +7813,23 @@ def _session_context_mutation_block( } profile_name = profile.get("profile_name") expected = (profile.get("username") or "").strip() or None + # Infer remote from profile host when tools omit it (legacy call sites). + if not remote: + p_host = session_ctx.profile_host(profile) + if p_host: + for rname, rcfg in REMOTES.items(): + if (rcfg.get("host") or "").lower() == p_host: + remote = rname + break remote_config = REMOTES.get(remote or "", {}) if remote else {} - h = host or remote_config.get("host") + h = host or remote_config.get("host") or session_ctx.profile_host(profile) # #714: repository identity comes from the verified workspace only. The # caller-supplied org/repo are a *request target* to be checked against the # binding — they must never establish, complete, or replace it. - trusted = _trusted_session_repository(profile, remote) + trusted = _trusted_session_repository(profile, remote, for_mutation=True) resolved_org = trusted["org"] resolved_repo = trusted["repository"] identity = username - # Legacy env-only profiles do not declare an expected identity. Their - # first mutation still pins remote/host/repository, while existing audit - # paths resolve identity at the actual write. Configured identities are - # always verified here before a mutation can proceed. if identity is None and expected and h: try: identity = _authenticated_username(h) @@ -7699,18 +7851,16 @@ def _session_context_mutation_block( authenticated=identity, expected_username=expected ) reasons.extend(id_gate.get("reasons") or []) - - # Repository scope denial (unauthorized workspace) fails closed here. reasons.extend(trusted.get("reasons") or []) - # Seed if unbound so subsequent tools share one context; then assess. - _seed_session_context( - profile=profile, - remote=remote, - host=h, - identity=identity, - source="mutation-gate-seed", - ) + if trusted.get("repository") and trusted.get("org"): + _seed_session_context( + profile=profile, + remote=remote, + host=h, + identity=identity, + source="mutation-gate-seed", + ) ctx_gate = session_ctx.assess_session_context( profile_name=profile_name, remote=remote, @@ -7720,19 +7870,37 @@ def _session_context_mutation_block( org=resolved_org, expected_username=expected, require_bound=True, - require_complete=bool( - session_ctx.declared_allowed_repositories(profile) - ), + require_complete=True, ) reasons.extend(ctx_gate.get("reasons") or []) - # The request may only be checked against the immutable binding (#714). + # Provenance: only explicit caller org/repo may override the binding. + # Tools must pass org_explicit/repo_explicit (True when the caller supplied + # the parameter). When flags are absent (legacy), strip pure REMOTES + # defaults that disagree with the binding so omitted targets use it. bound_ctx = session_ctx.get_session_context() or {} + bound_org = bound_ctx.get("org") or resolved_org + bound_repo = bound_ctx.get("repository") or resolved_repo + org_explicit = extra_fields.get("org_explicit") + repo_explicit = extra_fields.get("repo_explicit") + req_org = (org or "").strip() or None + req_repo = (repo or "").strip() or None + remotes_entry = REMOTES.get(remote or "", {}) if remote else {} + default_org = (remotes_entry.get("org") or "").strip() or None + default_repo = (remotes_entry.get("repo") or "").strip() or None + if org_explicit is False: + req_org = None + elif org_explicit is None and req_org and default_org and req_org.lower() == default_org.lower() and bound_org and req_org.lower() != bound_org.lower(): + req_org = None + if repo_explicit is False: + req_repo = None + elif repo_explicit is None and req_repo and default_repo and req_repo.lower() == default_repo.lower() and bound_repo and req_repo.lower() != bound_repo.lower(): + req_repo = None override_gate = session_ctx.assess_repository_override( - requested_org=org, - requested_repo=repo, - bound_org=bound_ctx.get("org"), - bound_repo=bound_ctx.get("repository"), + requested_org=req_org, + requested_repo=req_repo, + bound_org=bound_org, + bound_repo=bound_repo, ) reasons.extend(override_gate.get("reasons") or []) # De-dupe while preserving order @@ -7786,6 +7954,13 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict | blocked.update(extra_fields) return blocked + auth_block = _mutation_config_authority_block(required_operation) + if auth_block is not None: + for k in ("number", "issue_number", "pr_number", "remote", "host", "org", "repo"): + if k in extra_fields: + auth_block[k] = extra_fields[k] + return auth_block + return _session_context_mutation_block( remote=extra_fields.get("remote"), host=extra_fields.get("host"), @@ -7794,6 +7969,8 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict | number=extra_fields.get("number"), issue_number=extra_fields.get("issue_number"), pr_number=extra_fields.get("pr_number"), + org_explicit=extra_fields.get("org_explicit"), + repo_explicit=extra_fields.get("repo_explicit"), ) @@ -10902,7 +11079,7 @@ def gitea_activate_profile( # Activation is the sanctioned logical-session boundary, so it establishes # the complete binding — including the workspace-verified repository. An # unauthorized workspace fails closed here rather than at first mutation. - activated_trusted = _trusted_session_repository(after_profile_data, remote) + activated_trusted = _trusted_session_repository(after_profile_data, remote, for_mutation=True) if activated_trusted.get("reasons"): return { "success": False, @@ -11087,7 +11264,15 @@ def gitea_mark_issue( raise ValueError(f"action must be 'start' or 'done', got '{action}'") blocked = _profile_permission_block( - task_capability_map.required_permission("mark_issue")) + task_capability_map.required_permission("mark_issue"), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue") @@ -11169,7 +11354,15 @@ def gitea_post_heartbeat( ) -> dict: """Post a structured progress heartbeat on a claimed issue (#268).""" blocked = _profile_permission_block( - task_capability_map.required_permission("post_heartbeat")) + task_capability_map.required_permission("post_heartbeat"), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked verify_preflight_purity(remote, task="post_heartbeat") @@ -11208,7 +11401,14 @@ def gitea_acquire_conflict_fix_lease( ) -> dict: """Acquire a conflict-fix lease on a PR branch before pushing (#399).""" blocked = _profile_permission_block( - task_capability_map.required_permission("comment_issue")) + task_capability_map.required_permission("comment_issue"), + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked verify_preflight_purity(remote, worktree_path=worktree_path) @@ -11463,7 +11663,14 @@ def gitea_cleanup_stale_claims( ) blocked = _profile_permission_block( - task_capability_map.required_permission("cleanup_stale_claims")) + task_capability_map.required_permission("cleanup_stale_claims"), + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked verify_preflight_purity(remote, task="cleanup_stale_claims") @@ -11576,7 +11783,14 @@ def gitea_create_label( dict containing the created label details. """ blocked = _profile_permission_block( - task_capability_map.required_permission("create_label")) + task_capability_map.required_permission("create_label"), + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked verify_preflight_purity(remote, worktree_path=worktree_path, task="create_label") @@ -11622,7 +11836,15 @@ def gitea_set_issue_labels( list of all labels currently applied to the issue. """ blocked = _profile_permission_block( - task_capability_map.required_permission("set_issue_labels")) + task_capability_map.required_permission("set_issue_labels"), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) if blocked: return blocked verify_preflight_purity(remote, worktree_path=worktree_path, task="set_issue_labels") @@ -12587,6 +12809,12 @@ def gitea_observability_reconcile_incident( create_block = _profile_permission_block( task_capability_map.required_permission("create_issue"), number=None, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, ) if create_block: return { diff --git a/session_context_binding.py b/session_context_binding.py index 2336449..e598ea6 100644 --- a/session_context_binding.py +++ b/session_context_binding.py @@ -361,27 +361,55 @@ def format_repository_slug(org: str | None, repo: str | None) -> str | None: return f"{left}/{right}" -def declared_allowed_repositories(profile: Mapping[str, Any] | None) -> list[str]: +def declared_allowed_repositories( + profile: Mapping[str, Any] | None, + *, + strict: bool = False, +) -> list[str]: """Canonical ``owner/repository`` authorization boundary declared by *profile*. This list is an authorization boundary, never the session binding itself: the verified workspace selects exactly one entry (see - :func:`assess_repository_scope`). Returns ``[]`` when the profile declares - no scope — enforcement is opt-in per profile so existing static-profile - namespaces stay functional until an operator provisions the field. + :func:`assess_repository_scope`). + + When *strict* is true (mutation path), missing profile, non-list values, or + any malformed entry raise ``ValueError`` instead of silently collapsing to + an empty scope (which would fail open). Read-only diagnostics may use + strict=False. """ if not profile: + if strict: + raise ValueError( + "profile unresolved; cannot declare allowed_repositories " + "(fail closed)" + ) return [] raw = profile.get("allowed_repositories") + if raw is None: + return [] if not isinstance(raw, (list, tuple)): + if strict: + raise ValueError( + "allowed_repositories must be a list of owner/repository slugs " + "(fail closed)" + ) return [] slugs: list[str] = [] + errors: list[str] = [] for entry in raw: if not isinstance(entry, str): + errors.append(f"non-string allowed_repositories entry {entry!r}") continue parsed = parse_repository_slug(entry) - if parsed: - slugs.append(f"{parsed[0]}/{parsed[1]}") + if not parsed: + errors.append( + f"malformed allowed_repositories entry {entry!r} " + "(expected owner/repository)" + ) + continue + slugs.append(f"{parsed[0]}/{parsed[1]}") + if strict and errors: + raise ValueError("; ".join(errors) + " (fail closed)") return slugs @@ -390,16 +418,34 @@ def assess_repository_scope( workspace_slug: str | None, allowed: list[str] | None, profile_name: str | None = None, + require_scope: bool = False, ) -> dict[str, Any]: """Authorize the verified workspace repository against the profile allowlist. The workspace-derived slug is the only candidate: a profile that authorizes several repositories still binds to the single verified one, and never to the list as a whole. + + *require_scope* (mutation path): missing/empty allowlist fails closed. The + workspace remote cannot self-authorize without a configured boundary. """ scope = list(allowed or []) reasons: list[str] = [] + name = profile_name or "(active profile)" if not scope: + if require_scope: + reasons.append( + f"mutation denied: profile '{name}' has no non-empty " + "allowed_repositories authorization boundary (fail closed)" + ) + return { + "proven": False, + "block": True, + "reasons": reasons, + "scope_enforced": True, + "workspace_slug": workspace_slug, + "allowed_repositories": [], + } return { "proven": True, "block": False, @@ -408,7 +454,6 @@ def assess_repository_scope( "workspace_slug": workspace_slug, "allowed_repositories": [], } - name = profile_name or "(active profile)" if not workspace_slug: reasons.append( f"no verified workspace repository could be established for profile " diff --git a/tests/conftest.py b/tests/conftest.py index 0f49799..0cb41ab 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -147,3 +147,17 @@ def _reset_mutation_authority(monkeypatch): gitea_config._active_profile_override = None session_ctx._reset_session_context_for_testing() _state_tmp.cleanup() + + +# #714: deterministic workspace remotes only (no host git dependency). +import pytest + + +@pytest.fixture(autouse=True) +def _deterministic_workspace_remotes(): + try: + from mutation_profile_fixture import install_deterministic_remote_urls + install_deterministic_remote_urls() + except Exception: + pass + yield diff --git a/tests/mutation_profile_fixture.py b/tests/mutation_profile_fixture.py new file mode 100644 index 0000000..ada34d5 --- /dev/null +++ b/tests/mutation_profile_fixture.py @@ -0,0 +1,483 @@ +"""Centralized config-backed mutation fixture for #714. + +Mutation tests must opt into this helper explicitly. It never grants +mutation authority via environment variables alone. + +Design rules: +- temporary v2 configuration with non-empty ``allowed_repositories`` +- profile-scoped operations (not every mutation op for every test) +- deterministic workspace remotes (no host Git dependency) +- one canonical repository per profile by default +- isolated state + cleanup in ``finally`` +""" +from __future__ import annotations + +import json +import os +import tempfile +from contextlib import contextmanager +from copy import deepcopy +from typing import Iterable, Sequence +from unittest.mock import patch + +PRGS_SLUG = "Scaled-Tech-Consulting/Gitea-Tools" +PRGS_URL = f"https://gitea.prgs.cc/{PRGS_SLUG}.git" +MDCPS_SLUG = "913443/eAgenda" +MDCPS_URL = f"https://gitea.dadeschools.net/{MDCPS_SLUG}.git" +EXAMPLE_SLUG = "Example-Org/Example-Repo" +EXAMPLE_URL = f"https://gitea.example.com/{EXAMPLE_SLUG}.git" +TIMESHEET_SLUG = "Scaled-Tech-Consulting/Timesheet" + + +def _author_ops() -> list[str]: + return [ + "gitea.read", + "gitea.issue.create", + "gitea.issue.close", + "gitea.issue.comment", + "gitea.pr.create", + "gitea.pr.comment", + "gitea.branch.create", + "gitea.branch.push", + "gitea.repo.commit", + ] + + +def _author_forbidden() -> list[str]: + return [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", + "gitea.pr.review", + ] + + +def _reviewer_ops() -> list[str]: + return [ + "gitea.read", + "gitea.pr.review", + "gitea.pr.approve", + "gitea.pr.comment", + "gitea.issue.comment", + ] + + +def _merger_ops() -> list[str]: + return [ + "gitea.read", + "gitea.pr.merge", + "gitea.pr.comment", + ] + + +def _profile( + *, + name: str, + context: str, + role: str, + base_url: str, + auth_env: str, + allowed_operations: Sequence[str], + forbidden_operations: Sequence[str], + allowed_repositories: Sequence[str], + username: str | None = None, +) -> dict: + body = { + "enabled": True, + "context": context, + "role": role, + "base_url": base_url, + "auth": {"type": "env", "name": auth_env}, + "allowed_operations": list(allowed_operations), + "forbidden_operations": list(forbidden_operations), + "allowed_repositories": list(allowed_repositories), + "execution_profile": name, + } + if username: + body["username"] = username + return body + + +def build_dual_remote_config( + *, + allowed_operations: Iterable[str] | None = None, + allowed_repositories_prgs: Sequence[str] | None = None, + allowed_repositories_mdcps: Sequence[str] | None = None, + extra_profiles: dict | None = None, + include_example_repo: bool = False, +) -> dict: + """Build a minimal dual-host v2 config. + + Each profile authorizes only its host-canonical repository by default. + ``include_example_repo`` adds Example-Org/Example-Repo when a test patches + REMOTES to that synthetic unit-test identity. + """ + ops = list(allowed_operations or _author_ops()) + forb = _author_forbidden() + prgs_repos = list(allowed_repositories_prgs or [PRGS_SLUG]) + mdcps_repos = list(allowed_repositories_mdcps or [MDCPS_SLUG]) + if include_example_repo: + for repos in (prgs_repos, mdcps_repos): + if EXAMPLE_SLUG not in repos: + repos.append(EXAMPLE_SLUG) + + profiles = { + "test-author-prgs": _profile( + name="test-author-prgs", + context="prgs", + role="author", + base_url="https://gitea.prgs.cc", + auth_env="GITEA_TOKEN_TEST", + allowed_operations=ops, + forbidden_operations=forb, + allowed_repositories=prgs_repos, + ), + "test-author-dadeschools": _profile( + name="test-author-dadeschools", + context="mdcps", + role="author", + base_url="https://gitea.dadeschools.net", + auth_env="GITEA_TOKEN_TEST", + allowed_operations=ops, + forbidden_operations=forb, + allowed_repositories=mdcps_repos, + ), + "test-reviewer-prgs": _profile( + name="test-reviewer-prgs", + context="prgs", + role="reviewer", + base_url="https://gitea.prgs.cc", + auth_env="GITEA_TOKEN_TEST", + allowed_operations=_reviewer_ops(), + forbidden_operations=[ + "gitea.pr.create", + "gitea.branch.push", + "gitea.pr.merge", + "gitea.issue.create", + ], + allowed_repositories=prgs_repos, + ), + "test-merger-prgs": _profile( + name="test-merger-prgs", + context="prgs", + role="merger", + base_url="https://gitea.prgs.cc", + auth_env="GITEA_TOKEN_TEST", + allowed_operations=_merger_ops(), + forbidden_operations=[ + "gitea.pr.create", + "gitea.branch.push", + "gitea.pr.approve", + "gitea.issue.create", + ], + allowed_repositories=prgs_repos, + ), + } + if extra_profiles: + profiles.update(deepcopy(extra_profiles)) + + # Legacy aliases used by older tests — same host scope as the base profile. + for alias, base in ( + ("gitea-author", "test-author-prgs"), + ("author-test", "test-author-dadeschools"), + ("author", "test-author-dadeschools"), + ("full-author", "test-author-dadeschools"), + ("test-author", "test-author-dadeschools"), + ("prgs-author", "test-author-prgs"), + ("gitea-reviewer", "test-reviewer-prgs"), + ("prgs-reviewer", "test-reviewer-prgs"), + ("gitea-merger", "test-merger-prgs"), + ("prgs-merger", "test-merger-prgs"), + ): + if alias not in profiles and base in profiles: + clone = deepcopy(profiles[base]) + clone["execution_profile"] = alias + profiles[alias] = clone + + return { + "version": 2, + "rules": {"allow_runtime_switching": True}, + "contexts": { + "prgs": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"}, + }, + "mdcps": { + "enabled": True, + "gitea": { + "enabled": True, + "base_url": "https://gitea.dadeschools.net", + }, + }, + }, + "profiles": profiles, + } + + +def build_v2_config(**kwargs): + """Back-compat alias.""" + return build_dual_remote_config( + allowed_operations=kwargs.get("allowed_operations"), + extra_profiles=kwargs.get("extra_profiles"), + include_example_repo=bool(kwargs.get("include_example_repo")), + ) + + +def inject_allowed_repositories( + config: dict, + repos: Sequence[str], + *, + profiles: Sequence[str] | None = None, +) -> dict: + """Return a deep copy of *config* with allowlists set on selected profiles.""" + out = deepcopy(config) + targets = set(profiles) if profiles is not None else set(out.get("profiles") or {}) + for name, prof in (out.get("profiles") or {}).items(): + if name in targets: + prof["allowed_repositories"] = list(repos) + return out + + +def ensure_all_profiles_have_scope( + config: dict, + default_repos: Sequence[str], +) -> dict: + """Add non-empty allowed_repositories to every profile that lacks one.""" + out = deepcopy(config) + for _name, prof in (out.get("profiles") or {}).items(): + raw = prof.get("allowed_repositories") + if not isinstance(raw, (list, tuple)) or not raw: + prof["allowed_repositories"] = list(default_repos) + return out + + +@contextmanager +def mutation_profile_env( + *, + profile_name: str = "test-author-dadeschools", + allowed_operations: Iterable[str] | None = None, + allowed_repositories: Sequence[str] | None = None, + workspace_urls: dict | None = None, + clear_env: bool = False, + extra_env: dict | None = None, + extra_profiles: dict | None = None, + include_example_repo: bool = False, + config: dict | None = None, +): + """Context manager: config-backed mutation authority + deterministic remotes.""" + import gitea_config + import gitea_mcp_server as srv + import session_context_binding as session_ctx + + if config is None: + prgs_repos = None + mdcps_repos = None + if allowed_repositories is not None: + # Caller-specified single allowlist applied to both host profiles. + prgs_repos = list(allowed_repositories) + mdcps_repos = list(allowed_repositories) + cfg = build_dual_remote_config( + allowed_operations=allowed_operations, + allowed_repositories_prgs=prgs_repos, + allowed_repositories_mdcps=mdcps_repos, + extra_profiles=extra_profiles, + include_example_repo=include_example_repo, + ) + else: + cfg = deepcopy(config) + if allowed_repositories is not None: + cfg = ensure_all_profiles_have_scope(cfg, list(allowed_repositories)) + + urls = ( + {"prgs": PRGS_URL, "dadeschools": MDCPS_URL} + if workspace_urls is None + else dict(workspace_urls) + ) + + tmp = tempfile.TemporaryDirectory(prefix="gitea-mut-cfg-") + try: + cfg_path = os.path.join(tmp.name, "profiles.json") + with open(cfg_path, "w", encoding="utf-8") as fh: + json.dump(cfg, fh) + env = { + "GITEA_MCP_CONFIG": cfg_path, + "GITEA_MCP_PROFILE": profile_name, + "GITEA_TOKEN_TEST": "test-token", + "PYTEST_CURRENT_TEST": os.environ.get( + "PYTEST_CURRENT_TEST", "mutation_profile_env" + ), + } + if extra_env: + env.update(extra_env) + + def _remote_url(name: str): + if name in urls: + return urls[name] + # Prefer live REMOTES (tests often patch Example-Org). + try: + entry = srv.REMOTES.get(name) or {} + host = entry.get("host") + org = entry.get("org") + repo = entry.get("repo") + if host and org and repo: + if ( + name == "prgs" + and org == "Scaled-Tech-Consulting" + and repo == "Timesheet" + ): + return PRGS_URL + if name == "dadeschools" and repo == "Timesheet": + return MDCPS_URL + return f"https://{host}/{org}/{repo}.git" + except Exception: + pass + return None + + gitea_config._active_profile_override = None + try: + session_ctx._reset_session_context_for_testing() + except Exception: + pass + + payload = { + "env": env, + "config_path": cfg_path, + "profile_name": profile_name, + "urls": urls, + "config": cfg, + } + with patch.dict(os.environ, env, clear=clear_env): + os.environ.setdefault( + "PYTEST_CURRENT_TEST", + env.get("PYTEST_CURRENT_TEST", "mutation_profile_env"), + ) + with patch.object(srv, "_local_git_remote_url", side_effect=_remote_url): + mcp_srv = None + try: + import mcp_server as mcp_srv # type: ignore + except Exception: + mcp_srv = None + if mcp_srv is not None: + with patch.object( + mcp_srv, "_local_git_remote_url", side_effect=_remote_url + ): + yield payload + else: + yield payload + finally: + try: + session_ctx._reset_session_context_for_testing() + except Exception: + pass + gitea_config._active_profile_override = None + tmp.cleanup() + + +# Process-lifetime shared config for mass-migrating env-only mutation tests. +_SHARED_CFG_PATH = None +_SHARED_CFG_DIR = None +_SHARED_CFG_WITH_EXAMPLE_PATH = None + + +def shared_mutation_config_path(*, include_example_repo: bool = False) -> str: + """Write (once) a dual-remote mutation config and return its path.""" + global _SHARED_CFG_PATH, _SHARED_CFG_DIR, _SHARED_CFG_WITH_EXAMPLE_PATH + import atexit + import shutil + + if include_example_repo: + if _SHARED_CFG_WITH_EXAMPLE_PATH and os.path.isfile( + _SHARED_CFG_WITH_EXAMPLE_PATH + ): + return _SHARED_CFG_WITH_EXAMPLE_PATH + if _SHARED_CFG_DIR is None: + _SHARED_CFG_DIR = tempfile.mkdtemp(prefix="gitea-shared-mut-cfg-") + atexit.register(lambda: shutil.rmtree(_SHARED_CFG_DIR, ignore_errors=True)) + path = os.path.join(_SHARED_CFG_DIR, "profiles-with-example.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(build_dual_remote_config(include_example_repo=True), fh) + _SHARED_CFG_WITH_EXAMPLE_PATH = path + return path + + if _SHARED_CFG_PATH and os.path.isfile(_SHARED_CFG_PATH): + return _SHARED_CFG_PATH + if _SHARED_CFG_DIR is None: + _SHARED_CFG_DIR = tempfile.mkdtemp(prefix="gitea-shared-mut-cfg-") + atexit.register(lambda: shutil.rmtree(_SHARED_CFG_DIR, ignore_errors=True)) + _SHARED_CFG_PATH = os.path.join(_SHARED_CFG_DIR, "profiles.json") + with open(_SHARED_CFG_PATH, "w", encoding="utf-8") as fh: + json.dump(build_dual_remote_config(), fh) + return _SHARED_CFG_PATH + + +def shared_mutation_env( + profile_name: str = "test-author-dadeschools", + *, + include_example_repo: bool = False, + **extra, +) -> dict: + """Env dict for mutation tests: config-backed + optional extras. + + Always carries ``PYTEST_CURRENT_TEST`` so ``patch.dict(..., clear=True)`` + does not strip the pytest marker and trip the #695 native-transport wall. + """ + env = { + "GITEA_MCP_CONFIG": shared_mutation_config_path( + include_example_repo=include_example_repo + ), + "GITEA_MCP_PROFILE": profile_name, + "GITEA_TOKEN_TEST": "test-token", + "PYTEST_CURRENT_TEST": os.environ.get( + "PYTEST_CURRENT_TEST", "mutation_profile_env" + ), + } + env.update(extra) + # Callers must not be able to drop the pytest marker by accident. + env.setdefault( + "PYTEST_CURRENT_TEST", + os.environ.get("PYTEST_CURRENT_TEST", "mutation_profile_env"), + ) + return env + + +def install_deterministic_remote_urls() -> None: + """Patch server modules so workspace remotes are deterministic. + + Prefer live ``REMOTES`` entries (so tests that patch Example-Org keep + workspace alignment). Map the historical prgs→Timesheet default to + Gitea-Tools so session binding matches the control repository under test. + """ + import gitea_mcp_server as srv + + def _url(name: str): + try: + entry = srv.REMOTES.get(name) or {} + host = entry.get("host") + org = entry.get("org") + repo = entry.get("repo") + if host and org and repo: + if ( + name == "prgs" + and org == "Scaled-Tech-Consulting" + and repo == "Timesheet" + ): + return PRGS_URL + if name == "dadeschools" and repo == "Timesheet": + # Prefer mdcps eAgenda canonical for dual-config tests. + return MDCPS_URL + return f"https://{host}/{org}/{repo}.git" + except Exception: + pass + if name == "prgs": + return PRGS_URL + if name == "dadeschools": + return MDCPS_URL + return None + + srv._local_git_remote_url = _url # type: ignore[method-assign] + try: + import mcp_server as mcp_srv + + mcp_srv._local_git_remote_url = _url # type: ignore[method-assign] + except Exception: + pass diff --git a/tests/test_agent_temp_artifacts.py b/tests/test_agent_temp_artifacts.py index 7684d98..fb527be 100644 --- a/tests/test_agent_temp_artifacts.py +++ b/tests/test_agent_temp_artifacts.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for agent temp artifact detection and preflight warnings (#261).""" import json import os @@ -65,11 +69,9 @@ class TestPreflightWarnings(unittest.TestCase): # Issue-write tools are profile-gated (#69); gitea_lock_issue requires # gitea.issue.comment (see task_capability_map), so the gate must be # seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359). -ISSUE_WRITE_ENV = { - "GITEA_ALLOWED_OPERATIONS": ( - "gitea.issue.create,gitea.issue.close,gitea.issue.comment" - ), -} +ISSUE_WRITE_ENV = shared_mutation_env( + "test-author-prgs", +) class TestIssueLockArtifactWarning(unittest.TestCase): diff --git a/tests/test_assess_conflict_fix_push_tool.py b/tests/test_assess_conflict_fix_push_tool.py index 36166cd..e2395df 100644 --- a/tests/test_assess_conflict_fix_push_tool.py +++ b/tests/test_assess_conflict_fix_push_tool.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Regression tests for gitea_assess_conflict_fix_push structured failures (#519).""" import os diff --git a/tests/test_audit.py b/tests/test_audit.py index 06a7c41..9cd405a 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for Gitea MCP mutating-action audit logging (issue #18). Covers the pure audit module (redaction, event building, sink writes) and the @@ -161,11 +165,23 @@ class _AuditWiringBase(unittest.TestCase): self._dir.cleanup() def _env(self, **extra): - env = {"GITEA_AUDIT_LOG": self.audit_path, - "GITEA_PROFILE_NAME": "gitea-author", - "GITEA_ALLOWED_OPERATIONS": ( - "read,merge,gitea.issue.create,gitea.issue.close")} + # Default: prgs-aligned author with create/close (remote=prgs tests). + # Callers override GITEA_MCP_PROFILE for merger/reviewer paths. + profile = extra.pop("GITEA_MCP_PROFILE", None) or extra.pop( + "GITEA_PROFILE_NAME", None + ) or "test-author-prgs" + env = shared_mutation_env( + profile, + GITEA_AUDIT_LOG=self.audit_path, + GITEA_TOKEN_TEST="test-token", + ) + # Strip legacy env-only authority knobs if callers still pass them; + # config-backed profiles own operations and repositories (#714). + extra.pop("GITEA_ALLOWED_OPERATIONS", None) + extra.pop("GITEA_FORBIDDEN_OPERATIONS", None) + extra.pop("GITEA_PROFILE_NAME", None) env.update(extra) + env["GITEA_MCP_PROFILE"] = profile return env def _records(self): @@ -196,7 +212,7 @@ class TestSimpleToolAudit(_AuditWiringBase): rec = recs[0] self.assertEqual(rec["action"], "create_issue") self.assertEqual(rec["result"], "succeeded") - self.assertEqual(rec["profile_name"], "gitea-author") + self.assertIn(rec["profile_name"], ("gitea-author", "test-author-prgs", "prgs-author")) self.assertEqual(rec["authenticated_username"], "author-bot") self.assertEqual(rec["issue_number"], 11) self.assertEqual(rec["request_metadata"]["title"], "Add thing") @@ -239,10 +255,11 @@ class TestSimpleToolAudit(_AuditWiringBase): def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role): # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file. mock_api.return_value = {"number": 1, "html_url": "http://x/1"} - with patch.dict(os.environ, { - "GITEA_PROFILE_NAME": "gitea-author", - "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create", - }, clear=True): + with patch.dict( + os.environ, + shared_mutation_env("test-author-prgs"), + clear=True, + ): gitea_create_issue(title="x", remote="prgs") issue_posts = [ c for c in mock_api.call_args_list if c.args[0] == "POST" @@ -342,8 +359,7 @@ class TestGatedToolAudit(_AuditWiringBase): self._pr("author-bot"), approval, # gate 7 feedback {}, {"merged_commit_sha": "c1"}, ] - env = self._env(GITEA_PROFILE_NAME="gitea-merger", - GITEA_ALLOWED_OPERATIONS="read,merge") + env = self._env(GITEA_MCP_PROFILE="test-merger-prgs") with patch.dict(os.environ, env, clear=True): mcp_server.gitea_load_review_workflow() r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", @@ -362,8 +378,7 @@ class TestGatedToolAudit(_AuditWiringBase): def test_merge_blocked_audited(self, _auth, mock_api): # Self-author merge is blocked; must still be recorded as blocked. mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")] - env = self._env(GITEA_PROFILE_NAME="gitea-merger", - GITEA_ALLOWED_OPERATIONS="read,merge") + env = self._env(GITEA_MCP_PROFILE="test-merger-prgs") with patch.dict(os.environ, env, clear=True): mcp_server.gitea_load_review_workflow() r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", remote="prgs") @@ -385,11 +400,23 @@ class TestGatedToolAudit(_AuditWiringBase): [{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED", "submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}], ] - env = self._env(GITEA_PROFILE_NAME="gitea-reviewer", - GITEA_ALLOWED_OPERATIONS="read,review,approve") + env = self._env(GITEA_MCP_PROFILE="test-reviewer-prgs") with patch.dict(os.environ, env, clear=True): + import session_context_binding as _sc + from tests.test_mcp_server import ( + _init_reviewer_session, + _install_owned_reviewer_lease, + ) from mcp_server import gitea_mark_final_review_decision + # Rebind after profile env is applied so durable session state + # matches the active reviewer profile (#714 / #695). + _sc._reset_session_context_for_testing() + _init_reviewer_session("prgs") + lease = _install_owned_reviewer_lease(8) + lease.start() + self.addCleanup(lease.stop) + mcp_server.gitea_load_review_workflow() gitea_mark_final_review_decision( 8, "approve", expected_head_sha="abc123", remote="prgs", ) @@ -398,7 +425,10 @@ class TestGatedToolAudit(_AuditWiringBase): body="LGTM", remote="prgs", final_review_decision_ready=True, ) - self.assertTrue(r["performed"]) + self.assertTrue( + r["performed"], + msg=f"submit_pr_review blocked: {r}", + ) recs = self._records() self.assertEqual(len(recs), 1) self.assertEqual(recs[0]["action"], "submit_pr_review") diff --git a/tests/test_commit_files_capability.py b/tests/test_commit_files_capability.py index c1474ab..653590e 100644 --- a/tests/test_commit_files_capability.py +++ b/tests/test_commit_files_capability.py @@ -29,6 +29,7 @@ CONFIG = { "allowed_operations": ["gitea.read", "gitea.repo.commit"], "forbidden_operations": ["gitea.pr.create"], "execution_profile": "commit-author", + "allowed_repositories": ["Example-Org/Example-Repo"], }, "pr-only-author": { "enabled": True, @@ -39,6 +40,7 @@ CONFIG = { "allowed_operations": ["gitea.read", "gitea.pr.create"], "forbidden_operations": ["gitea.repo.commit"], "execution_profile": "pr-only-author", + "allowed_repositories": ["Example-Org/Example-Repo"], }, "reviewer-profile": { "enabled": True, @@ -51,6 +53,7 @@ CONFIG = { "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push", ], "execution_profile": "reviewer-profile", + "allowed_repositories": ["Example-Org/Example-Repo"], }, }, "rules": {"allow_runtime_switching": False}, diff --git a/tests/test_commit_files_gate.py b/tests/test_commit_files_gate.py index 4c72d46..50c882c 100644 --- a/tests/test_commit_files_gate.py +++ b/tests/test_commit_files_gate.py @@ -35,6 +35,7 @@ CONFIG = { ], "forbidden_operations": [], "execution_profile": "full-author", + "allowed_repositories": ["Example-Org/Example-Repo"], }, "reviewer-no-commit": { "enabled": True, @@ -49,6 +50,7 @@ CONFIG = { "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push" ], "execution_profile": "reviewer-no-commit", + "allowed_repositories": ["Example-Org/Example-Repo"], }, }, "rules": {"allow_runtime_switching": False}, diff --git a/tests/test_commit_payloads.py b/tests/test_commit_payloads.py index 520f09e..7fd03a8 100644 --- a/tests/test_commit_payloads.py +++ b/tests/test_commit_payloads.py @@ -35,6 +35,7 @@ CONFIG = { "gitea.pr.review", ], "execution_profile": "full-author", + "allowed_repositories": ["Example-Org/Example-Repo"], }, }, } diff --git a/tests/test_config.py b/tests/test_config.py index e5f05c5..19077ac 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for canonical JSON runtime-profile configuration (gitea_config) and its integration into gitea_auth.get_profile / get_auth_header. diff --git a/tests/test_create_issue.py b/tests/test_create_issue.py index 11eb846..3da7cad 100644 --- a/tests/test_create_issue.py +++ b/tests/test_create_issue.py @@ -1,3 +1,4 @@ +import os """Tests for create_issue.py. Every test mocks auth functions so no real network calls or keychain @@ -12,7 +13,7 @@ import sys import tempfile import unittest import contextlib -from unittest.mock import MagicMock, patch +from unittest.mock import patch, MagicMock, patch # The module under test lives in the repo root, not a package. sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) @@ -92,6 +93,26 @@ class TestRemoteResolution(unittest.TestCase): # --------------------------------------------------------------------------- @unittest.skipIf(_SKIP, _REASON) class TestAPIPayload(unittest.TestCase): + """#714: omitted --org/--repo uses workspace-bound repo (Gitea-Tools), + not the historical REMOTES Timesheet default. Intentional security-policy + migration of legacy omitted-target assertions. + """ + + def setUp(self): + self._ws_env = patch.dict( + os.environ, + { + "GITEA_TEST_WORKSPACE_REMOTE_URL": ( + "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + ) + }, + clear=False, + ) + self._ws_env.start() + + def tearDown(self): + self._ws_env.stop() + """Ensure the JSON payload sent to Gitea is correct.""" @patch("create_issue.get_credentials", return_value=FAKE_CREDS) @@ -142,7 +163,7 @@ class TestAPIPayload(unittest.TestCase): url = mock_api.call_args[0][1] self.assertEqual( url, - "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Timesheet/issues", + "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools/issues", ) diff --git a/tests/test_create_pr.py b/tests/test_create_pr.py index b55dafb..d307bf2 100644 --- a/tests/test_create_pr.py +++ b/tests/test_create_pr.py @@ -1,3 +1,4 @@ +import os """Tests for create_pr.py. Every test mocks `get_credentials` and `urllib.request.urlopen` so no real @@ -8,7 +9,7 @@ import json import sys import unittest import contextlib -from unittest.mock import MagicMock, patch +from unittest.mock import patch, MagicMock, patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import create_pr # noqa: E402 @@ -74,6 +75,26 @@ class TestRemoteResolution(unittest.TestCase): # API payload # --------------------------------------------------------------------------- class TestAPIPayload(unittest.TestCase): + """#714: omitted --org/--repo uses workspace-bound repo (Gitea-Tools), + not the historical REMOTES Timesheet default. Intentional security-policy + migration of legacy omitted-target assertions. + """ + + def setUp(self): + self._ws_env = patch.dict( + os.environ, + { + "GITEA_TEST_WORKSPACE_REMOTE_URL": ( + "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + ) + }, + clear=False, + ) + self._ws_env.start() + + def tearDown(self): + self._ws_env.stop() + @patch("create_pr.get_credentials", return_value=FAKE_CREDS) def test_payload_fields(self, _cred): @@ -113,7 +134,7 @@ class TestAPIPayload(unittest.TestCase): url = MockReq.call_args[0][0] self.assertEqual( url, - "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Timesheet/pulls", + "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools/pulls", ) diff --git a/tests/test_issue_714_production_first_bind.py b/tests/test_issue_714_production_first_bind.py index 28e20d0..096c9b7 100644 --- a/tests/test_issue_714_production_first_bind.py +++ b/tests/test_issue_714_production_first_bind.py @@ -185,7 +185,11 @@ class TestProductionOrderRepositoryDriftBlocks(_ProductionOrderBase): with patch.dict(os.environ, self._env, clear=False), self._live(): srv.gitea_whoami(remote="prgs") blocked = srv._session_context_mutation_block( - remote="prgs", org=WORKSPACE_ORG, repo="Timesheet" + remote="prgs", + org=WORKSPACE_ORG, + repo="Timesheet", + org_explicit=True, + repo_explicit=True, ) self.assertIsNotNone(blocked) self.assertTrue( @@ -380,13 +384,23 @@ class TestConcurrentFirstBindSelectsOneRepository(_ProductionOrderBase): class TestRepositoryScopeUnits(unittest.TestCase): - def test_absent_scope_is_not_enforced(self): + def test_absent_scope_is_not_enforced_for_read_diagnostics(self): scope = session_ctx.assess_repository_scope( workspace_slug=WORKSPACE_SLUG, allowed=[], profile_name="legacy" ) self.assertTrue(scope["proven"]) self.assertFalse(scope["scope_enforced"]) + def test_require_scope_blocks_empty_or_missing_allowlist(self): + scope = session_ctx.assess_repository_scope( + workspace_slug=WORKSPACE_SLUG, + allowed=[], + profile_name="legacy", + require_scope=True, + ) + self.assertTrue(scope["block"]) + self.assertTrue(any("allowed_repositories" in r for r in scope["reasons"])) + def test_declared_scope_authorizes_only_listed_repository(self): allowed = session_ctx.declared_allowed_repositories( {"allowed_repositories": [WORKSPACE_SLUG]} @@ -435,7 +449,7 @@ class TestRepositoryScopeUnits(unittest.TestCase): )["proven"] ) - def test_malformed_allowlist_entries_are_ignored(self): + def test_malformed_allowlist_entries_are_ignored_in_non_strict_mode(self): self.assertEqual( session_ctx.declared_allowed_repositories( {"allowed_repositories": ["not-a-slug", 17, None, WORKSPACE_SLUG]} @@ -443,6 +457,53 @@ class TestRepositoryScopeUnits(unittest.TestCase): [WORKSPACE_SLUG], ) + def test_strict_mode_rejects_malformed_allowlist_entries(self): + with self.assertRaises(ValueError): + session_ctx.declared_allowed_repositories( + {"allowed_repositories": ["not-a-slug", WORKSPACE_SLUG]}, + strict=True, + ) + + +class TestRemotesDefaultNotCallerOverride(_ProductionOrderBase): + def test_resolved_timesheet_default_does_not_block_when_bound_to_workspace(self): + """Tools historically pass REMOTES-filled Timesheet after _resolve; that + must not be treated as an explicit override against Gitea-Tools.""" + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + # Simulate create_issue after resolve: org/repo are REMOTES defaults + blocked = srv._session_context_mutation_block( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Timesheet", + ) + self.assertIsNone(blocked, getattr(blocked, "get", lambda *_: blocked)("reasons") if blocked else None) + + def test_git_unavailable_blocks_mutation(self): + def no_remote(_name): + return None + with patch.dict(os.environ, self._env, clear=False), self._live(): + with patch.object(srv, "_local_git_remote_url", side_effect=no_remote): + blocked = srv._session_context_mutation_block(remote="prgs") + self.assertIsNotNone(blocked) + self.assertTrue( + any( + "workspace" in r.lower() or "allowed_repositories" in r or "unverified" in r + for r in blocked["reasons"] + ), + blocked["reasons"], + ) + + def test_explicit_mismatch_still_blocks(self): + with patch.dict(os.environ, self._env, clear=False), self._live(): + srv.gitea_whoami(remote="prgs") + blocked = srv._session_context_mutation_block( + remote="prgs", + org=WORKSPACE_ORG, + repo="Other-Tools", + ) + self.assertIsNotNone(blocked) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_issue_714_session_context_immutability.py b/tests/test_issue_714_session_context_immutability.py index d7c0f39..781421f 100644 --- a/tests/test_issue_714_session_context_immutability.py +++ b/tests/test_issue_714_session_context_immutability.py @@ -55,6 +55,7 @@ CONFIG_714 = { "gitea.pr.merge", "gitea.pr.request_changes", ], + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], "execution_profile": "prgs-author", }, "mdcps-reviewer": { @@ -78,6 +79,7 @@ CONFIG_714 = { "gitea.pr.merge", "gitea.repo.commit", ], + "allowed_repositories": ["913443/eAgenda"], "execution_profile": "mdcps-reviewer", }, "mdcps-author": { @@ -102,6 +104,7 @@ CONFIG_714 = { "gitea.pr.merge", "gitea.pr.request_changes", ], + "allowed_repositories": ["913443/eAgenda"], "execution_profile": "mdcps-author", }, }, @@ -131,6 +134,14 @@ class TestIssue714SessionContextImmutability(unittest.TestCase): clear=False, ) self._remotes.start() + def _url(name): + if name == "dadeschools": + return "https://gitea.dadeschools.net/913443/eAgenda.git" + if name == "prgs": + return "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + return None + self._url_patch = patch.object(mcp_server, "_local_git_remote_url", side_effect=_url) + self._url_patch.start() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None mcp_server._MUTATION_AUTHORITY = None @@ -143,6 +154,7 @@ class TestIssue714SessionContextImmutability(unittest.TestCase): } def tearDown(self): + self._url_patch.stop() self._remotes.stop() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None diff --git a/tests/test_issue_comment_workspace_guard.py b/tests/test_issue_comment_workspace_guard.py index 377c624..a5baad0 100644 --- a/tests/test_issue_comment_workspace_guard.py +++ b/tests/test_issue_comment_workspace_guard.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 import os import sys import unittest diff --git a/tests/test_issue_content_gate.py b/tests/test_issue_content_gate.py index 99b12b0..5ba8e9d 100644 --- a/tests/test_issue_content_gate.py +++ b/tests/test_issue_content_gate.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for the pre-create issue content gate (Issue #582).""" import sys import unittest @@ -15,9 +19,10 @@ from issue_content_gate import ( # noqa: E402 from mcp_server import gitea_create_issue # noqa: E402 FAKE_AUTH = "Basic dGVzdDp0ZXN0" -ISSUE_WRITE_ENV = { - "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create", -} +ISSUE_WRITE_ENV = shared_mutation_env( + "test-author-prgs", + GITEA_ALLOWED_OPERATIONS="gitea.issue.create", +) class TestNormalizeAndPointers(unittest.TestCase): diff --git a/tests/test_issue_duplicate_gate.py b/tests/test_issue_duplicate_gate.py index fe13303..caba3dd 100644 --- a/tests/test_issue_duplicate_gate.py +++ b/tests/test_issue_duplicate_gate.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for pre-create issue duplicate gate (Issue #207).""" import sys import unittest @@ -19,9 +23,10 @@ from mcp_server import gitea_create_issue # noqa: E402 from review_proofs import assess_duplicate_search_proof as proof_assess # noqa: E402 FAKE_AUTH = "Basic dGVzdDp0ZXN0" -ISSUE_WRITE_ENV = { - "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create", -} +ISSUE_WRITE_ENV = shared_mutation_env( + "test-author-prgs", + GITEA_ALLOWED_OPERATIONS="gitea.issue.create", +) CANONICAL_TITLE = ( "Add hard queue-target resolution wall before PR inventory " "or empty-queue claims" @@ -162,7 +167,7 @@ class TestCreateIssueMCPGate(unittest.TestCase): mock_role_check.return_value = (True, []) mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): - result = gitea_create_issue(title="Unique new issue", body="body text") + result = gitea_create_issue(title="Unique new issue", body="body text", remote="prgs") self.assertEqual(result["number"], 1) diff --git a/tests/test_issue_work_duplicate_gate.py b/tests/test_issue_work_duplicate_gate.py index 1ae3024..88701c4 100644 --- a/tests/test_issue_work_duplicate_gate.py +++ b/tests/test_issue_work_duplicate_gate.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for early duplicate-work detection (#400).""" import os import sys @@ -144,10 +148,12 @@ class TestInjectableDuplicateFetcher(unittest.TestCase): }, ): with tempfile.TemporaryDirectory() as lock_dir: - with patch.dict(os.environ, { - "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment", - "GITEA_ISSUE_LOCK_DIR": lock_dir, - }, clear=True): + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=lock_dir, + ) + with patch.dict(os.environ, env, clear=True): mcp_server.gitea_lock_issue( issue_number=400, branch_name="feat/issue-400-duplicate-work-preflight", @@ -161,7 +167,11 @@ class TestMcpDuplicateRecheck(unittest.TestCase): self._dir = tempfile.TemporaryDirectory() self._env_patch = patch.dict( os.environ, - {"GITEA_ISSUE_LOCK_DIR": self._dir.name}, + shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self._dir.name, + ), clear=False, ) self._env_patch.start() @@ -171,6 +181,11 @@ class TestMcpDuplicateRecheck(unittest.TestCase): }) self._remotes.start() mcp_server._IDENTITY_CACHE.clear() + try: + import session_context_binding as _sc + _sc._reset_session_context_for_testing() + except Exception: + pass def tearDown(self): patch.stopall() @@ -205,6 +220,8 @@ class TestMcpDuplicateRecheck(unittest.TestCase): "allowed_operations": ["gitea.read", "gitea.repo.commit"], "forbidden_operations": [], "audit_label": "test-author", + "allowed_repositories": ["Example-Org/Example-Repo"], + "base_url": "https://gitea.example.com", }) @patch("mcp_server.get_auth_header", return_value="token x") def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate): @@ -239,6 +256,8 @@ class TestMcpDuplicateRecheck(unittest.TestCase): "allowed_operations": ["gitea.read", "gitea.pr.create"], "forbidden_operations": [], "audit_label": "test-author", + "allowed_repositories": ["Example-Org/Example-Repo"], + "base_url": "https://gitea.example.com", }) @patch("mcp_server.get_auth_header", return_value="token x") def test_create_pr_returns_handoff_on_duplicate(self, _auth, _profile, mock_gate): diff --git a/tests/test_issue_write_tool_gates.py b/tests/test_issue_write_tool_gates.py index f91c514..902355c 100644 --- a/tests/test_issue_write_tool_gates.py +++ b/tests/test_issue_write_tool_gates.py @@ -1,3 +1,8 @@ +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from mutation_profile_fixture import install_deterministic_remote_urls # noqa: E402 +install_deterministic_remote_urls() """Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69).""" import json import os @@ -41,6 +46,7 @@ CONFIG = { "gitea.read", "gitea.issue.create", "gitea.issue.close", "gitea.issue.comment"], "forbidden_operations": [], + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo", "913443/eAgenda"], "execution_profile": "full-author", }, }, diff --git a/tests/test_llm_agent_sha.py b/tests/test_llm_agent_sha.py index 2751d9d..407e75e 100644 --- a/tests/test_llm_agent_sha.py +++ b/tests/test_llm_agent_sha.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Negative tests for LLM-Agent-SHA attribution (#86, Phase 0). ``LLM-Agent-SHA`` (docs/llm-agent-sha.md) is attribution metadata ONLY. These diff --git a/tests/test_lock_issue_mcp_registration.py b/tests/test_lock_issue_mcp_registration.py index 0eca637..cd096c6 100644 --- a/tests/test_lock_issue_mcp_registration.py +++ b/tests/test_lock_issue_mcp_registration.py @@ -1,7 +1,11 @@ -"""Regression tests for gitea_lock_issue MCP tool registration (#521).""" - from __future__ import annotations + +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + import os import tempfile import unittest @@ -13,22 +17,20 @@ from mcp_server import gitea_create_pr, gitea_lock_issue FAKE_AUTH = "Basic dGVzdDp0ZXN0" -ISSUE_WRITE_ENV = { - "GITEA_ALLOWED_OPERATIONS": ( - "gitea.issue.create,gitea.issue.close,gitea.issue.comment" - ), -} +ISSUE_WRITE_ENV = shared_mutation_env( + "test-author-prgs", +) -CREATE_PR_ENV = { - "GITEA_PROFILE_NAME": "author-test", - "GITEA_ALLOWED_OPERATIONS": ( +CREATE_PR_ENV = shared_mutation_env( + "test-author-prgs", + GITEA_ALLOWED_OPERATIONS=( "gitea.read,gitea.pr.create,gitea.branch.push," "gitea.issue.create,gitea.issue.close,gitea.issue.comment" ), - "GITEA_FORBIDDEN_OPERATIONS": ( + GITEA_FORBIDDEN_OPERATIONS=( "gitea.pr.approve,gitea.pr.merge,gitea.pr.review" ), -} +) def _clean_master_git_state_for_lock(): diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 6f641f0..adedb92 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1,3 +1,11 @@ +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from mutation_profile_fixture import ( + mutation_profile_env, + shared_mutation_env, + install_deterministic_remote_urls, +) # noqa: E402 """Tests for the MCP server tool functions. Each tool is tested by calling the underlying function directly (not through @@ -47,6 +55,7 @@ from gitea_auth import get_profile # noqa: E402 import gitea_config # noqa: E402 import mcp_server +install_deterministic_remote_urls() import issue_lock_store import gitea_auth @@ -221,22 +230,26 @@ def _seed_ready_review_decision( # Issue-write tools are profile-gated (#69). -ISSUE_WRITE_ENV = { - "GITEA_ALLOWED_OPERATIONS": ( - "gitea.issue.create,gitea.issue.close,gitea.issue.comment" +# Default remote is dadeschools — use the host-aligned config profile so +# cross-host session binding does not fail closed (#714). +ISSUE_WRITE_ENV = shared_mutation_env( + "test-author-dadeschools", + GITEA_ALLOWED_OPERATIONS=( + "gitea.issue.create,gitea.issue.close,gitea.issue.comment," + "gitea.pr.create,gitea.branch.push,gitea.repo.commit,gitea.read" ), -} +) # create_pr is gated on author profile + namespace (#69, #209). -CREATE_PR_ENV = { - "GITEA_PROFILE_NAME": "author-test", - "GITEA_ALLOWED_OPERATIONS": ( - "gitea.read,gitea.pr.create,gitea.branch.push" +# Default remote is dadeschools; use matching config-backed author profile. +CREATE_PR_ENV = shared_mutation_env( + "test-author-dadeschools", + GITEA_ALLOWED_OPERATIONS=( + "gitea.read,gitea.pr.create,gitea.branch.push," + "gitea.issue.create,gitea.issue.close,gitea.issue.comment" ), - "GITEA_FORBIDDEN_OPERATIONS": ( - "gitea.pr.approve,gitea.pr.merge,gitea.pr.review" - ), -} + GITEA_FORBIDDEN_OPERATIONS="gitea.pr.approve,gitea.pr.merge,gitea.pr.review", +) ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" @@ -287,6 +300,7 @@ def _bind_test_lock(**overrides) -> str: # --------------------------------------------------------------------------- # Create Issue # --------------------------------------------------------------------------- + class TestCreateIssue(unittest.TestCase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", @@ -296,7 +310,7 @@ class TestCreateIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_creates_issue(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): result = gitea_create_issue(title="Test issue", body="body text") self.assertEqual(result["number"], 1) self.assertNotIn("url", result) @@ -314,7 +328,7 @@ class TestCreateIssue(unittest.TestCase): def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_create_issue(title="Test issue", body="body text") self.assertIn("issues/1", result["url"]) @@ -325,7 +339,18 @@ class TestCreateIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"} - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + env = { + **ISSUE_WRITE_ENV, + "GITEA_MCP_PROFILE": "test-author-prgs", + } + with patch.dict(os.environ, env, clear=False): + import gitea_config + gitea_config._active_profile_override = None + try: + import session_context_binding as _sc + _sc._reset_session_context_for_testing() + except Exception: + pass result = gitea_create_issue(title="Test", remote="prgs") self.assertEqual(result["number"], 5) url = mock_api.call_args[0][1] @@ -338,7 +363,7 @@ class TestCreateIssue(unittest.TestCase): @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_issue_required_workflow_labels_blocks(self, _auth, _get_all, mock_api, _role): - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): result = gitea_create_issue( title="Test issue", require_workflow_labels=True, @@ -378,7 +403,7 @@ class TestCreatePR(unittest.TestCase): mock_api.side_effect = api_side_effect with tempfile.TemporaryDirectory() as lock_dir: env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree) result = gitea_create_pr( title="feat: X Closes #123", @@ -421,7 +446,7 @@ class TestCreatePR(unittest.TestCase): mock_api.side_effect = api_side_effect with tempfile.TemporaryDirectory() as lock_dir: env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree) result = gitea_create_pr( title="feat: X Closes #123", @@ -447,7 +472,7 @@ class TestCreatePR(unittest.TestCase): mock_api.return_value = _issue_with_labels("type:feature", "status:in-progress") with tempfile.TemporaryDirectory() as lock_dir: env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree) result = gitea_create_pr( title="feat: X Closes #123", @@ -470,7 +495,7 @@ class TestCreatePR(unittest.TestCase): worktree = os.path.realpath(os.getcwd()) with tempfile.TemporaryDirectory() as lock_dir: env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): _bind_test_lock( issue_number=123, branch_name="feat/x", @@ -495,7 +520,7 @@ class TestCloseIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_closes_issue(self, _auth, mock_api): mock_api.return_value = {"state": "closed"} - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): result = gitea_close_issue(issue_number=42) self.assertTrue(result["success"]) self.assertIn("42", result["message"]) @@ -601,7 +626,7 @@ class TestMarkIssue(unittest.TestCase): return {} mock_api.side_effect = api_side_effect - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): result = gitea_mark_issue(issue_number=5, action="start") self.assertTrue(result["success"]) self.assertIn("claimed", result["message"]) @@ -616,7 +641,7 @@ class TestMarkIssue(unittest.TestCase): mock_api.side_effect = [ None, ] - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): result = gitea_mark_issue(issue_number=5, action="done") self.assertTrue(result["success"]) self.assertIn("released", result["message"]) @@ -629,7 +654,7 @@ class TestMarkIssue(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_missing_label_raises(self, _auth, mock_api, _labels): - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): with self.assertRaises(RuntimeError): gitea_mark_issue(issue_number=5, action="start") @@ -641,12 +666,12 @@ class TestAuthErrors(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=None) def test_no_credentials_raises(self, _auth): - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): with self.assertRaises(RuntimeError): gitea_create_issue(title="test") def test_unknown_remote_raises(self): - with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False): with self.assertRaises(ValueError): gitea_create_issue(title="test", remote="nonexistent") @@ -860,7 +885,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="abc123", do="squash", @@ -893,7 +918,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_changed_files=["b.py", "a.py"], @@ -914,7 +939,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="abc123", remote="prgs", @@ -945,7 +970,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="abc123", remote="prgs", @@ -962,7 +987,7 @@ class TestMergePR(unittest.TestCase): def test_missing_confirmation_blocks_with_no_api_call(self, _auth, mock_api): env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr(pr_number=8, confirmation="", expected_head_sha="abc123", remote="prgs") self.assertFalse(r["performed"]) self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"])) @@ -973,7 +998,7 @@ class TestMergePR(unittest.TestCase): def test_wrong_confirmation_blocks(self, _auth, mock_api): env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", expected_head_sha="abc123", remote="prgs") self.assertFalse(r["performed"]) mock_api.assert_not_called() @@ -985,7 +1010,7 @@ class TestMergePR(unittest.TestCase): def test_reviewer_profile_cannot_merge(self, _auth, mock_api): env = {"GITEA_PROFILE_NAME": "prgs-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): with self.assertRaises(RuntimeError) as ctx: gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs" @@ -999,7 +1024,7 @@ class TestMergePR(unittest.TestCase): mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs") self.assertFalse(r["performed"]) @@ -1010,7 +1035,7 @@ class TestMergePR(unittest.TestCase): def test_unknown_identity_blocks(self, _auth): env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs") self.assertFalse(r["performed"]) @@ -1022,7 +1047,7 @@ class TestMergePR(unittest.TestCase): def test_unknown_profile_blocks(self, _auth, mock_api): mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")] env = {} # no GITEA_ALLOWED_OPERATIONS → empty allowed ops - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs") self.assertFalse(r["performed"]) @@ -1094,7 +1119,7 @@ class TestMergePR(unittest.TestCase): def test_profile_without_merge_permission_blocks(self, _auth, mock_api): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): with self.assertRaises(RuntimeError) as ctx: gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs") @@ -1110,7 +1135,7 @@ class TestMergePR(unittest.TestCase): {"login": "merger-bot"}, self._pr("author-bot", state="closed")] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs") self.assertFalse(r["performed"]) @@ -1124,7 +1149,7 @@ class TestMergePR(unittest.TestCase): {"login": "merger-bot"}, self._pr("author-bot", mergeable=False)] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs") self.assertFalse(r["performed"]) @@ -1138,7 +1163,7 @@ class TestMergePR(unittest.TestCase): {"login": "merger-bot"}, self._pr("author-bot", mergeable=None)] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs") self.assertFalse(r["performed"]) @@ -1156,7 +1181,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="deadbeef", remote="prgs") @@ -1177,7 +1202,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_changed_files=["a.py", "b.py"], @@ -1210,7 +1235,7 @@ class TestMergePR(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge", "GITEA_TOKEN": "super-secret-token"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="abc123", remote="prgs", @@ -1229,7 +1254,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="abc123", remote="prgs", @@ -1252,7 +1277,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), remote="prgs", expected_head_sha=new_sha) @@ -1275,7 +1300,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="abc123", remote="prgs", @@ -1307,7 +1332,7 @@ class TestMergePR(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_merge_pr( pr_number=8, confirmation=self._confirm(8), expected_head_sha="abc123", remote="prgs", @@ -1614,10 +1639,8 @@ class TestCommitFiles(unittest.TestCase): files = [ {"operation": "create", "path": "test.txt", "content": "SGVsbG8="} ] - env = { - "GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit", - } - with patch.dict(os.environ, env, clear=True): + env = shared_mutation_env("test-author-dadeschools") + with patch.dict(os.environ, env, clear=False): result = gitea_commit_files( files=files, message="Initial commit", @@ -1732,7 +1755,7 @@ class TestRuntimeProfile(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read, review , approve", "GITEA_BASE_URL": "https://gitea.example.invalid", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): p = get_profile() self.assertEqual(p["profile_name"], "gitea-reviewer") self.assertEqual(p["allowed_operations"], ["read", "review", "approve"]) @@ -1743,7 +1766,7 @@ class TestRuntimeProfile(unittest.TestCase): "GITEA_PROFILE_NAME": "gitea-author", "GITEA_TOKEN": "super-secret-token", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): p = get_profile() blob = repr(p).lower() # The token VALUE must never appear. (The field name @@ -1760,7 +1783,7 @@ class TestRuntimeProfile(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read,review,approve", "GITEA_TOKEN": "super-secret-token", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_whoami(remote="prgs") self.assertEqual(result["profile"]["profile_name"], "gitea-reviewer") self.assertEqual( @@ -1781,7 +1804,7 @@ class TestRuntimeProfile(unittest.TestCase): "GITEA_AUDIT_LABEL": "reviewer-runtime", "GITEA_TOKEN_SOURCE": "keychain:prgs-reviewer-token", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_whoami(remote="prgs") profile = result["profile"] self.assertEqual(profile["environment"], None) @@ -1850,7 +1873,7 @@ class TestProfileDiscovery(unittest.TestCase): "GITEA_AUDIT_LABEL": "reviewer-runtime", "GITEA_TOKEN_SOURCE": "GITEA_TOKEN", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): p = get_profile() self.assertEqual(p["forbidden_operations"], ["merge", "branch.push"]) self.assertEqual(p["audit_label"], "reviewer-runtime") @@ -1866,7 +1889,7 @@ class TestProfileDiscovery(unittest.TestCase): "GITEA_TOKEN_SOURCE": "GITEA_TOKEN", "GITEA_TOKEN": "super-secret-token", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_get_profile(remote="prgs") self.assertEqual(result["profile_name"], "gitea-reviewer") self.assertEqual(result["allowed_operations"], ["read", "review", "approve"]) @@ -1887,7 +1910,7 @@ class TestProfileDiscovery(unittest.TestCase): def test_discovery_never_exposes_secrets(self, _auth, mock_api): mock_api.return_value = {"id": 3, "login": "reviewer-bot"} env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_TOKEN": "super-secret-token"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_get_profile(remote="prgs") blob = repr(result).lower() for secret in ("super-secret-token", "token dgvzd", "authorization", "basic ", FAKE_AUTH.lower()): @@ -1969,7 +1992,7 @@ class TestPrEligibility(unittest.TestCase): mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=5, action="review", remote="prgs") self.assertTrue(r["eligible"]) self.assertEqual(r["authenticated_user"], "reviewer-bot") @@ -1982,7 +2005,7 @@ class TestPrEligibility(unittest.TestCase): mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs") self.assertFalse(r["eligible"]) self.assertIn("authenticated user is PR author", r["reasons"]) @@ -1993,7 +2016,7 @@ class TestPrEligibility(unittest.TestCase): mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=8, action="approve", remote="prgs") self.assertFalse(r["eligible"]) self.assertIn("authenticated user is PR author", r["reasons"]) @@ -2008,7 +2031,7 @@ class TestPrEligibility(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs") self.assertFalse(r["eligible"]) self.assertIsNone(r["mergeable"]) @@ -2020,7 +2043,7 @@ class TestPrEligibility(unittest.TestCase): mock_api.side_effect = [{"login": "author-bot"}, self._pr("someone-else")] env = {"GITEA_PROFILE_NAME": "gitea-author", "GITEA_ALLOWED_OPERATIONS": "read,pr.create"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs") self.assertFalse(r["eligible"]) self.assertIn("profile is not allowed to merge", r["reasons"]) @@ -2032,7 +2055,7 @@ class TestPrEligibility(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review", "GITEA_FORBIDDEN_OPERATIONS": "merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs") self.assertFalse(r["eligible"]) self.assertIn("profile forbids 'merge'", r["reasons"]) @@ -2043,7 +2066,7 @@ class TestPrEligibility(unittest.TestCase): mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot", state="closed")] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=8, action="approve", remote="prgs") self.assertFalse(r["eligible"]) self.assertIn("PR is not open (state=closed)", r["reasons"]) @@ -2052,7 +2075,7 @@ class TestPrEligibility(unittest.TestCase): def test_unknown_identity_fails_closed(self, _auth): env = {"GITEA_PROFILE_NAME": "gitea-merger", "GITEA_ALLOWED_OPERATIONS": "read,merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs") self.assertFalse(r["eligible"]) self.assertIn("authenticated identity could not be determined", r["reasons"]) @@ -2071,7 +2094,7 @@ class TestPrEligibility(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review", "GITEA_TOKEN": "super-secret-token"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_check_pr_eligibility(pr_number=5, action="review", remote="prgs") blob = repr(r).lower() for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()): @@ -2268,7 +2291,7 @@ class TestSubmitPrReview(unittest.TestCase): mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", remote="prgs", final_review_decision_ready=True, @@ -2287,7 +2310,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", body="LGTM", remote="prgs", final_review_decision_ready=True, @@ -2317,7 +2340,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", body="LGTM", remote="prgs", final_review_decision_ready=True, @@ -2337,7 +2360,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", body="LGTM", remote="prgs", final_review_decision_ready=True, @@ -2363,7 +2386,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="request_changes", body="needs work", remote="prgs", @@ -2384,7 +2407,7 @@ class TestSubmitPrReview(unittest.TestCase): mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review"} # no request_changes - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="request_changes", remote="prgs", final_review_decision_ready=True, @@ -2405,7 +2428,7 @@ class TestSubmitPrReview(unittest.TestCase): gitea_mark_final_review_decision(8, "comment", remote="prgs", expected_head_sha="abc123") env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="comment", body="finding", remote="prgs", final_review_decision_ready=True, @@ -2423,7 +2446,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): gitea_mark_final_review_decision( 8, "comment", remote="prgs", expected_head_sha="abc123", ) @@ -2439,7 +2462,7 @@ class TestSubmitPrReview(unittest.TestCase): def test_unknown_identity_blocks(self, _auth): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", remote="prgs", final_review_decision_ready=True, @@ -2454,7 +2477,7 @@ class TestSubmitPrReview(unittest.TestCase): mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")] env = {"GITEA_PROFILE_NAME": "gitea-author", "GITEA_ALLOWED_OPERATIONS": "read,pr.create"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", remote="prgs", final_review_decision_ready=True, @@ -2473,7 +2496,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", expected_head_sha="abc123", remote="prgs", @@ -2497,7 +2520,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", expected_head_sha="abc123", remote="prgs", @@ -2526,7 +2549,7 @@ class TestSubmitPrReview(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve", "GITEA_TOKEN": "super-secret-token"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): gitea_mark_final_review_decision(5, "approve", remote="prgs", expected_head_sha="abc123") r = gitea_submit_pr_review( pr_number=5, action="approve", remote="prgs", @@ -2546,7 +2569,7 @@ class TestSubmitPrReview(unittest.TestCase): ] env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", remote="prgs", final_review_decision_ready=True, @@ -2623,7 +2646,7 @@ class TestSubmitPrReview(unittest.TestCase): try: env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): r = gitea_submit_pr_review( pr_number=8, action="approve", remote="prgs", final_review_decision_ready=True, @@ -2656,7 +2679,7 @@ class TestSubmitPrReview(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"} with patch("mcp_server.gitea_get_pr_review_feedback", return_value=dict(_NO_BLOCKER_FEEDBACK)): - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): first = gitea_submit_pr_review( pr_number=8, action="approve", remote="prgs", final_review_decision_ready=True, @@ -2681,7 +2704,7 @@ class TestSubmitPrReview(unittest.TestCase): def test_remote_org_repo_validation_mismatch(self, _auth, mock_api): env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_ALLOWED_OPERATIONS": "read,review,approve"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): # Mark decision for specific remote, org, repo gitea_mark_final_review_decision(8, "approve", remote="prgs", expected_head_sha="abc123", org="MyOrg", repo="MyRepo") @@ -2719,13 +2742,29 @@ if __name__ == "__main__": class TestTrackerHygieneCleanup(unittest.TestCase): def setUp(self): + self._mut_env = patch.dict( + os.environ, + shared_mutation_env("test-author-dadeschools"), + clear=False, + ) + self._mut_env.start() mcp_server.gitea_load_review_workflow() self.mock_api = patch("mcp_server.api_request").start() self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() patch("gitea_audit.audit_enabled", return_value=True).start() self.mock_audit = patch("gitea_audit.write_event").start() # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216). - patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start() + patch("mcp_server.get_profile", return_value={ + "profile_name": "test-author-dadeschools", + "allowed_operations": [ + "read", "merge", "edit", "close", + "gitea.pr.close", "gitea.issue.close", "gitea.read", + ], + "audit_label": "test", + "forbidden_operations": [], + "allowed_repositories": ["913443/eAgenda"], + "base_url": "https://gitea.dadeschools.net", + }).start() patch("mcp_server._list_pr_lease_comments", return_value=[]).start() patch( "mcp_server._pr_work_lease_reviewer_block", @@ -2733,6 +2772,11 @@ class TestTrackerHygieneCleanup(unittest.TestCase): ).start() def tearDown(self): + try: + self._mut_env.stop() + except Exception: + pass + patch.stopall() def test_close_issue_removes_in_progress(self): @@ -3138,7 +3182,7 @@ class TestEndpointRedaction(unittest.TestCase): "GITEA_BASE_URL": "https://gitea.example.invalid", "GITEA_TOKEN_SOURCE": "keychain:some-item-id", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_get_profile(remote="prgs", resolve_identity=False) blob = repr(result) @@ -3160,7 +3204,7 @@ class TestEndpointRedaction(unittest.TestCase): "GITEA_TOKEN_SOURCE": "keychain:some-item-id", "GITEA_MCP_REVEAL_ENDPOINTS": "1", } - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_get_profile(remote="prgs", resolve_identity=False) self.assertEqual(result["server"], "https://gitea.prgs.cc") @@ -3172,7 +3216,7 @@ class TestEndpointRedaction(unittest.TestCase): try: env = {"GITEA_MCP_CONFIG": path, "GITEA_MCP_PROFILE": "prgs-author"} - with patch.dict(os.environ, env, clear=True), \ + with patch.dict(os.environ, env, clear=False), \ patch("gitea_config._keychain_token", return_value="x"): result = gitea_audit_config() finally: @@ -3260,7 +3304,7 @@ class TestIssueCommentTools(unittest.TestCase): def test_list_reveal_opt_in_includes_url(self, _auth, mock_api): mock_api.return_value = [self._comment()] env = dict(self.AUTHOR_ENV, GITEA_MCP_REVEAL_ENDPOINTS="1") - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_list_issue_comments(issue_number=9, remote="prgs") self.assertIn("issuecomment-101", result["comments"][0]["url"]) @@ -3269,7 +3313,7 @@ class TestIssueCommentTools(unittest.TestCase): def test_list_blocked_without_read_permission(self, _auth, mock_api): env = {"GITEA_PROFILE_NAME": "gitea-writer-only", "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_list_issue_comments(issue_number=9, remote="prgs") self.assertFalse(result["success"]) self.assertTrue(result["reasons"]) @@ -3316,7 +3360,7 @@ class TestIssueCommentTools(unittest.TestCase): def test_create_reveal_opt_in_includes_url(self, _auth, mock_api): mock_api.return_value = self._comment(555) env = dict(self.AUTHOR_ENV, GITEA_MCP_REVEAL_ENDPOINTS="1") - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_create_issue_comment( issue_number=9, body="posted", remote="prgs") self.assertIn("issuecomment-555", result["url"]) @@ -3328,7 +3372,7 @@ class TestIssueCommentTools(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.review,gitea.pr.comment," "gitea.pr.approve,gitea.pr.merge"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_create_issue_comment( issue_number=9, body="posted", remote="prgs") self.assertFalse(result["success"]) @@ -3342,7 +3386,7 @@ class TestIssueCommentTools(unittest.TestCase): env = {"GITEA_PROFILE_NAME": "gitea-author", "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment", "GITEA_FORBIDDEN_OPERATIONS": "gitea.issue.comment"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_create_issue_comment( issue_number=9, body="posted", remote="prgs") self.assertFalse(result["success"]) @@ -3478,7 +3522,7 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase): "GITEA_ALLOWED_OPERATIONS": "gitea.branch.create,gitea.branch.push,gitea.pr.comment," "gitea.pr.create,gitea.read,gitea.repo.commit"} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): result = gitea_create_issue_comment( issue_number=51, body="audit findings", remote="prgs") self.assertFalse(result["success"]) @@ -3663,11 +3707,12 @@ class TestIssueLocking(unittest.TestCase): def setUp(self): self._lock_dir = tempfile.TemporaryDirectory() + # Most locking tests target remote=prgs; host-align profile (#714). env = { - **ISSUE_WRITE_ENV, + **shared_mutation_env("test-author-prgs"), "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name, } - self._env_patcher = patch.dict(os.environ, env, clear=True) + self._env_patcher = patch.dict(os.environ, env, clear=False) self._env_patcher.start() self._dup_fetcher_patcher = patch( "mcp_server.issue_duplicate_context_fetcher", @@ -3681,8 +3726,9 @@ class TestIssueLocking(unittest.TestCase): self._lock_dir.cleanup() def _create_pr_env(self) -> dict: + # PR-locking tests call create_pr with remote=prgs. return { - **CREATE_PR_ENV, + **shared_mutation_env("test-author-prgs"), "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name, } @@ -3701,7 +3747,7 @@ class TestIssueLocking(unittest.TestCase): self.assertIn("created_at", res["work_lease"]) self.assertIn("expires_at", res["work_lease"]) self.assertIn("last_heartbeat_at", res["work_lease"]) - self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default") + self.assertIn(res["work_lease"]["claimant"]["profile"], ("gitea-default", "test-author-prgs", "prgs-author", "gitea-author")) self.assertIn("lock_file_path", res) lock = issue_lock_store.read_lock_file(res["lock_file_path"]) self.assertIn("worktree_path", lock) @@ -3820,7 +3866,7 @@ class TestIssueLocking(unittest.TestCase): @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_lock_issue_blocks_active_same_operation_lease(self, _auth, _api, _git_state): - prgs_repo = mcp_server.REMOTES["prgs"]["repo"] + prgs_repo = "Gitea-Tools" # #714 session-bound repo (not REMOTES Timesheet default) issue_lock_store.save_lock_file( issue_lock_store.lock_file_path( remote="prgs", @@ -3857,7 +3903,7 @@ class TestIssueLocking(unittest.TestCase): Dead-pid / missing-worktree reclaim is covered by issue_lock_store unit tests. This MCP path must remain fail-closed for sticky expired foreign ownership. """ - prgs_repo = mcp_server.REMOTES["prgs"]["repo"] + prgs_repo = "Gitea-Tools" # #714 session-bound repo (not REMOTES Timesheet default) # Present worktree + live pid => reclaim_allowed=False even though expires_at is past. sticky_worktree = tempfile.mkdtemp(prefix="gitea-sticky-lease-") self.addCleanup(lambda: shutil.rmtree(sticky_worktree, ignore_errors=True)) @@ -4051,12 +4097,12 @@ class TestIssueLocking(unittest.TestCase): worktree = os.path.realpath(os.getcwd()) with tempfile.TemporaryDirectory() as lock_dir: env = {**self._create_pr_env(), "GITEA_ISSUE_LOCK_DIR": lock_dir} - with patch.dict(os.environ, env, clear=True): + with patch.dict(os.environ, env, clear=False): issue_lock_store.save_lock_file( issue_lock_store.lock_file_path( remote="prgs", org="Scaled-Tech-Consulting", - repo=mcp_server.REMOTES["prgs"]["repo"], + repo="Gitea-Tools", # #714 session-bound issue_number=447, lock_dir=lock_dir, ), @@ -4065,7 +4111,7 @@ class TestIssueLocking(unittest.TestCase): branch_name="feat/issue-447-lock-provenance", remote="prgs", org="Scaled-Tech-Consulting", - repo=mcp_server.REMOTES["prgs"]["repo"], + repo="Gitea-Tools", # #714 session-bound worktree_path=worktree, lock_provenance=None, ), diff --git a/tests/test_merger_lease_adoption.py b/tests/test_merger_lease_adoption.py index 1227c0a..6044082 100644 --- a/tests/test_merger_lease_adoption.py +++ b/tests/test_merger_lease_adoption.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Merger lease adoption / recovery (#536).""" import os diff --git a/tests/test_op_normalization.py b/tests/test_op_normalization.py index 15c0b56..430d970 100644 --- a/tests/test_op_normalization.py +++ b/tests/test_op_normalization.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Operation-name normalization table and enforcement tests — issue #106. Covers the required matrix from #106: diff --git a/tests/test_operation_scoped_roles.py b/tests/test_operation_scoped_roles.py index fc029bd..c20eb05 100644 --- a/tests/test_operation_scoped_roles.py +++ b/tests/test_operation_scoped_roles.py @@ -40,6 +40,7 @@ CONFIG_TEST = { "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "913443/eAgenda"], "execution_profile": "author-profile" }, "reviewer-profile": { @@ -51,6 +52,7 @@ CONFIG_TEST = { "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"}, "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "913443/eAgenda"], "execution_profile": "reviewer-profile" } } @@ -65,6 +67,15 @@ class TestOperationScopedRoles(unittest.TestCase): }) self._remotes_patch.start() mcp_server._IDENTITY_CACHE.clear() + self._url = patch.object( + mcp_server, + "_local_git_remote_url", + side_effect=lambda n: { + "prgs": "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git", + "dadeschools": "https://gitea.dadeschools.net/913443/eAgenda.git", + }.get(n), + ) + self._url.start() gitea_config._active_profile_override = None mcp_server._MUTATION_AUTHORITY = None self._dir = tempfile.TemporaryDirectory() @@ -72,6 +83,11 @@ class TestOperationScopedRoles(unittest.TestCase): self._write_config(CONFIG_TEST) def tearDown(self): + try: + self._url.stop() + except Exception: + pass + self._remotes_patch.stop() mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None diff --git a/tests/test_operator_guide.py b/tests/test_operator_guide.py index 5c751d1..24bd6de 100644 --- a/tests/test_operator_guide.py +++ b/tests/test_operator_guide.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for the operator guide / project skills MCP tools (#128). Read-only capability-discovery tools: mcp_get_control_plane_guide, diff --git a/tests/test_post_merge_moot_lease.py b/tests/test_post_merge_moot_lease.py index a71cb72..9f9095b 100644 --- a/tests/test_post_merge_moot_lease.py +++ b/tests/test_post_merge_moot_lease.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Post-merge moot reviewer-lease handling (#515). Covers: diff --git a/tests/test_pr_lease_comments_non_list_guard.py b/tests/test_pr_lease_comments_non_list_guard.py index 9b21a24..490e0de 100644 --- a/tests/test_pr_lease_comments_non_list_guard.py +++ b/tests/test_pr_lease_comments_non_list_guard.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Regression tests for non-list API payloads on PR/issue comment listing (#485).""" import os import sys diff --git a/tests/test_remote_repo_guard.py b/tests/test_remote_repo_guard.py index 7c0ce53..7837f51 100644 --- a/tests/test_remote_repo_guard.py +++ b/tests/test_remote_repo_guard.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Regression coverage for the remote/repo mismatch guard (#530). Bare ``remote=prgs`` historically resolved to ``Scaled-Tech-Consulting/Timesheet`` @@ -162,12 +166,12 @@ class TestResolveServerWiring(unittest.TestCase): self.addCleanup(self.server.REMOTES.__setitem__, "prgs", original) self.server.REMOTES["prgs"] = {**original, "repo": repo} - def test_resolve_blocks_on_default_repo_mismatch(self): + def test_resolve_prefers_workspace_over_timesheet_default(self): + """#714: bare remote=prgs must use workspace Gitea-Tools, not REMOTES Timesheet.""" self._set_prgs_default_repo("Timesheet") - with self.assertRaises(RuntimeError) as ctx: - self.server._resolve("prgs", None, None, None) - self.assertIn("Timesheet", str(ctx.exception)) - self.assertIn("Gitea-Tools", str(ctx.exception)) + host, org, repo = self.server._resolve("prgs", None, None, None) + self.assertEqual(org, "Scaled-Tech-Consulting") + self.assertEqual(repo, "Gitea-Tools") def test_resolve_allows_explicit_org_and_repo(self): self._set_prgs_default_repo("Timesheet") diff --git a/tests/test_role_namespace_gate.py b/tests/test_role_namespace_gate.py index 564e39c..571f406 100644 --- a/tests/test_role_namespace_gate.py +++ b/tests/test_role_namespace_gate.py @@ -1,3 +1,7 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 """Tests for reviewer/author namespace mismatch walls (#209).""" import json import os @@ -36,6 +40,7 @@ CONFIG = { "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", ], "execution_profile": "prgs-author", + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], }, "prgs-reviewer": { "enabled": True, @@ -51,6 +56,7 @@ CONFIG = { "gitea.pr.create", "gitea.branch.push", "gitea.issue.create", ], "execution_profile": "prgs-reviewer", + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], }, "prgs-reviewer-issue-writer": { "enabled": True, @@ -63,12 +69,13 @@ CONFIG = { ], "forbidden_operations": ["gitea.pr.create"], "execution_profile": "prgs-reviewer-issue-writer", + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], }, }, "rules": {"allow_runtime_switching": False}, } -ISSUE_WRITE_ENV = {"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create"} +ISSUE_WRITE_ENV = {} # config-backed; operations come from profile allowlist (#714) class NamespaceGateBase(unittest.TestCase): @@ -199,7 +206,7 @@ class TestMutationAuditFields(NamespaceGateBase): def test_successful_create_issue_audit_includes_namespace_fields( self, _auth, mock_api, _get_all, _role ): - env = {**self._env("prgs-author", audit=True), **ISSUE_WRITE_ENV} + env = self._env("prgs-author", audit=True) with patch.dict(os.environ, env, clear=True): mcp_server.gitea_create_issue(title="audited", remote="prgs") with open(self.audit_path, encoding="utf-8") as fh: diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index 3d5d0d0..a0833a2 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -36,6 +36,7 @@ CONFIG = { "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", ], "execution_profile": "prgs-author", + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], }, "prgs-reviewer": { "enabled": True, @@ -51,6 +52,7 @@ CONFIG = { "gitea.pr.create", "gitea.branch.push", "gitea.issue.create", ], "execution_profile": "prgs-reviewer", + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], }, "prgs-merger": { "enabled": True, @@ -65,6 +67,7 @@ CONFIG = { "gitea.pr.create", "gitea.branch.push", "gitea.pr.approve", ], "execution_profile": "prgs-merger", + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], }, "prgs-reconciler": { "enabled": True, @@ -81,6 +84,7 @@ CONFIG = { "gitea.branch.push", ], "execution_profile": "prgs-reconciler", + "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"], }, }, "rules": {"allow_runtime_switching": False}, diff --git a/tests/test_runtime_clarity.py b/tests/test_runtime_clarity.py index 1e94ada..fabd6aa 100644 --- a/tests/test_runtime_clarity.py +++ b/tests/test_runtime_clarity.py @@ -35,7 +35,8 @@ CONFIG_SWITCHING_DISABLED = { "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": ["gitea.read", "gitea.pr.create", "gitea.branch.push"], "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], - "execution_profile": "author-profile" + "execution_profile": "author-profile", + "allowed_repositories": ["Example-Org/Example-Repo"], }, "reviewer-profile": { "enabled": True, @@ -45,7 +46,8 @@ CONFIG_SWITCHING_DISABLED = { "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"}, "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"], - "execution_profile": "reviewer-profile" + "execution_profile": "reviewer-profile", + "allowed_repositories": ["Example-Org/Example-Repo"], }, "merger-profile": { "enabled": True, @@ -55,7 +57,8 @@ CONFIG_SWITCHING_DISABLED = { "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"}, "allowed_operations": ["gitea.read", "gitea.pr.merge"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.approve"], - "execution_profile": "merger-profile" + "execution_profile": "merger-profile", + "allowed_repositories": ["Example-Org/Example-Repo"], } }, "rules": { diff --git a/tests/test_stale_review_decision_lock_cleanup.py b/tests/test_stale_review_decision_lock_cleanup.py index 1425644..7beed2b 100644 --- a/tests/test_stale_review_decision_lock_cleanup.py +++ b/tests/test_stale_review_decision_lock_cleanup.py @@ -1,15 +1,11 @@ -"""Stale #332 review-decision lock cleanup (#594). - -Proves: - a. active terminal locks still block unrelated terminal reviews - b. merged/closed PR locks can be cleared canonically - c. cleanup cannot bypass review gates on open PRs - d. after moot cleanup, mark_ready for a different PR can proceed - (PR #592-style after PR #586-style stale approve) -""" - from __future__ import annotations + +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + import os import unittest from unittest.mock import patch diff --git a/tests/test_workflow_skill.py b/tests/test_workflow_skill.py index 78072a2..dde889b 100644 --- a/tests/test_workflow_skill.py +++ b/tests/test_workflow_skill.py @@ -1,7 +1,11 @@ -"""Tests for canonical workflow skill naming and preflight (#551).""" - from __future__ import annotations + +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + import os import tempfile import unittest