sysadmin
4a63578003
Address REQUEST_CHANGES on PR #688: - migrate_profiles: emit fully canonical reconciler/author/reviewer defaults; canonicalize explicit ops; fail if reconciler loses required pr.close/read - gitea_delete_branch: deny reconciler (and non-author roles); refuse preservation/protected branches before any API call - gitea_cleanup_merged_pr_branch: require reconciler role only; block preservation/evidence branches via branch_cleanup_guard - docs: end-to-end operator runbook for profile migrate/apply/reconnect/cleanup - tests: migration canonicalization, idempotency, raw-delete denial, guarded cleanup success and unmerged/preserve rejections Closes #687.
482 lines
19 KiB
Python
482 lines
19 KiB
Python
"""Unit tests for migrate_profiles.py migration helper."""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import unittest
|
|
import tempfile
|
|
import shutil
|
|
from unittest.mock import patch
|
|
from io import StringIO
|
|
|
|
from _pytest.monkeypatch import MonkeyPatch
|
|
|
|
# Add project root to sys.path
|
|
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if PROJECT_ROOT not in sys.path:
|
|
sys.path.insert(0, PROJECT_ROOT)
|
|
|
|
import migrate_profiles
|
|
|
|
|
|
class TestMigrateProfiles(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.v1_content = {
|
|
"version": 1,
|
|
"profiles": {
|
|
"prgs": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "jcwalker3",
|
|
"auth": {
|
|
"type": "keychain",
|
|
"id": "redacted-author-ref"
|
|
},
|
|
"default_owner": "Scaled-Tech-Consulting",
|
|
"execution_profile": "personal-prgs",
|
|
"allowed_operations": ["read", "comment"],
|
|
"forbidden_operations": ["approve", "merge"]
|
|
},
|
|
"mdcps": {
|
|
"base_url": "redacted-mdcps-service",
|
|
"username": "913443",
|
|
"auth": {
|
|
"type": "keychain",
|
|
"id": "redacted-mdcps-ref"
|
|
},
|
|
"default_owner": "Contractor",
|
|
"execution_profile": "mdcps",
|
|
"allowed_operations": ["read"],
|
|
"forbidden_operations": ["merge"]
|
|
},
|
|
"prgs-reviewer": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "sysadmin",
|
|
"auth": {
|
|
"type": "keychain",
|
|
"id": "redacted-reviewer-ref"
|
|
},
|
|
"default_owner": "Scaled-Tech-Consulting",
|
|
"execution_profile": "prgs-reviewer",
|
|
"allowed_operations": [
|
|
"read", "review", "comment", "approve",
|
|
"request_changes", "merge"
|
|
],
|
|
"forbidden_operations": ["branch", "commit", "push", "open_pr"]
|
|
}
|
|
}
|
|
}
|
|
self.input_file = os.path.join(self.temp_dir, "profiles.json")
|
|
with open(self.input_file, "w") as f:
|
|
json.dump(self.v1_content, f)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_migration_logic(self):
|
|
"""Test the structural transformation and capability mapping."""
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
|
self.assertEqual(v2_data["version"], 2)
|
|
|
|
# Check environment structure
|
|
envs = v2_data["environments"]
|
|
self.assertIn("prgs", envs)
|
|
self.assertIn("mdcps", envs)
|
|
|
|
# Check service and identity structure
|
|
prgs_gitea = envs["prgs"]["services"]["gitea"]
|
|
self.assertEqual(prgs_gitea["base_url"], "redacted-prgs-service")
|
|
self.assertEqual(prgs_gitea["default_owner"], "Scaled-Tech-Consulting")
|
|
|
|
author = prgs_gitea["identities"]["author"]
|
|
self.assertEqual(author["username"], "jcwalker3")
|
|
self.assertEqual(author["auth"]["id"], "redacted-author-ref")
|
|
self.assertEqual(
|
|
author["allowed_operations"], ["gitea.read", "gitea.pr.comment"]
|
|
)
|
|
self.assertEqual(
|
|
author["forbidden_operations"],
|
|
["gitea.pr.approve", "gitea.pr.merge"],
|
|
)
|
|
|
|
reviewer = prgs_gitea["identities"]["reviewer"]
|
|
self.assertEqual(reviewer["role"], "reviewer")
|
|
self.assertEqual(reviewer["username"], "sysadmin")
|
|
self.assertEqual(reviewer["auth"]["id"], "redacted-reviewer-ref")
|
|
self.assertIn("gitea.pr.merge", reviewer["allowed_operations"])
|
|
|
|
def test_alias_generation(self):
|
|
"""Test that aliases are correctly generated to support old profile names."""
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
|
aliases = v2_data["aliases"]
|
|
|
|
self.assertEqual(aliases["prgs"], "prgs.gitea.author")
|
|
self.assertEqual(aliases["prgs-author"], "prgs.gitea.author")
|
|
self.assertEqual(aliases["prgs-reviewer"], "prgs.gitea.reviewer")
|
|
self.assertEqual(aliases["mdcps"], "mdcps.gitea.author")
|
|
|
|
def test_no_secret_behavior(self):
|
|
"""Ensure secrets are never extracted, printed, or processed."""
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
|
# Check that auth structures only contain keychain references, not credentials
|
|
for env in v2_data["environments"].values():
|
|
for svc in env["services"].values():
|
|
for ident in svc["identities"].values():
|
|
auth = ident["auth"]
|
|
self.assertEqual(auth["type"], "keychain")
|
|
self.assertIn("id", auth)
|
|
self.assertNotIn("token", auth)
|
|
self.assertNotIn("password", auth)
|
|
|
|
def test_validation(self):
|
|
"""Test that the generated v2 configuration validates against Gitea-Tools v2 loader."""
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
|
self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
|
|
|
|
def test_dry_run_default(self):
|
|
"""Verify that running without -w prints generated config without modifying files."""
|
|
monkeypatch = MonkeyPatch()
|
|
mock_stdout = StringIO()
|
|
monkeypatch.setattr(sys, "stdout", mock_stdout)
|
|
try:
|
|
output_file = os.path.join(self.temp_dir, "migrated_dry.json")
|
|
test_args = [
|
|
"migrate_profiles.py",
|
|
"-i", self.input_file,
|
|
"-o", output_file
|
|
]
|
|
with patch.object(sys, "argv", test_args):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
migrate_profiles.main()
|
|
self.assertEqual(cm.exception.code, 0)
|
|
|
|
self.assertFalse(os.path.exists(output_file))
|
|
self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
|
|
|
|
stdout_output = mock_stdout.getvalue()
|
|
finally:
|
|
monkeypatch.undo()
|
|
self.assertIn("DRY-RUN MODE", stdout_output)
|
|
self.assertIn("version", stdout_output)
|
|
self.assertIn("identities", stdout_output)
|
|
self.assertIn("aliases", stdout_output)
|
|
self.assertNotIn("redacted-prgs-service", stdout_output)
|
|
self.assertNotIn("redacted-mdcps-service", stdout_output)
|
|
self.assertNotIn("redacted-author-ref", stdout_output)
|
|
self.assertNotIn("redacted-mdcps-ref", stdout_output)
|
|
self.assertNotIn("redacted-reviewer-ref", stdout_output)
|
|
self.assertNotIn("keychain", stdout_output)
|
|
self.assertNotIn("auth", stdout_output)
|
|
|
|
def test_dry_run_hides_token_like_values(self):
|
|
"""Verify dry-run summary does not expose token-like auth metadata."""
|
|
sensitive = json.loads(json.dumps(self.v1_content))
|
|
sensitive["profiles"]["prgs"]["auth"]["id"] = "super-secret-token-value"
|
|
with open(self.input_file, "w") as f:
|
|
json.dump(sensitive, f)
|
|
|
|
test_args = ["migrate_profiles.py", "-i", self.input_file]
|
|
monkeypatch = MonkeyPatch()
|
|
mock_stdout = StringIO()
|
|
monkeypatch.setattr(sys, "stdout", mock_stdout)
|
|
try:
|
|
with patch.object(sys, "argv", test_args):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
migrate_profiles.main()
|
|
self.assertEqual(cm.exception.code, 0)
|
|
|
|
stdout_output = mock_stdout.getvalue()
|
|
finally:
|
|
monkeypatch.undo()
|
|
self.assertNotIn("super-secret-token-value", stdout_output)
|
|
self.assertNotIn("token", stdout_output.lower())
|
|
|
|
def test_explicit_operations_are_preserved(self):
|
|
"""Explicit v1 permissions are canonicalized, not replaced by role defaults."""
|
|
v1_data = json.loads(json.dumps(self.v1_content))
|
|
v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"]
|
|
v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"]
|
|
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
reviewer = (
|
|
v2_data["environments"]["prgs"]["services"]["gitea"]
|
|
["identities"]["reviewer"]
|
|
)
|
|
self.assertEqual(reviewer["allowed_operations"], ["gitea.read"])
|
|
self.assertEqual(reviewer["forbidden_operations"], ["gitea.pr.merge"])
|
|
|
|
def test_inferred_role_defaults_only_when_unambiguous(self):
|
|
"""Role defaults are allowed only for clear author/reviewer profiles."""
|
|
v1_data = {
|
|
"version": 1,
|
|
"profiles": {
|
|
"prgs-author": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "jcwalker3",
|
|
"auth": {"type": "keychain", "id": "hidden-author-ref"},
|
|
"execution_profile": "prgs-author",
|
|
}
|
|
}
|
|
}
|
|
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
author = (
|
|
v2_data["environments"]["prgs"]["services"]["gitea"]
|
|
["identities"]["author"]
|
|
)
|
|
self.assertEqual(
|
|
author["allowed_operations"],
|
|
migrate_profiles.AUTHOR_DEFAULT_ALLOWED,
|
|
)
|
|
self.assertEqual(
|
|
author["forbidden_operations"],
|
|
migrate_profiles.AUTHOR_DEFAULT_FORBIDDEN,
|
|
)
|
|
|
|
def test_ambiguous_permission_source_fails_closed(self):
|
|
"""A profile without explicit permissions or clear role must not widen."""
|
|
v1_data = {
|
|
"version": 1,
|
|
"profiles": {
|
|
"prgs": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "jcwalker3",
|
|
"auth": {"type": "keychain", "id": "hidden-author-ref"},
|
|
"execution_profile": "personal-prgs",
|
|
}
|
|
}
|
|
}
|
|
|
|
with self.assertRaisesRegex(ValueError, "fail closed"):
|
|
migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
|
|
def test_partial_permission_source_fails_closed(self):
|
|
"""Allowed without forbidden, or vice versa, is ambiguous."""
|
|
v1_data = json.loads(json.dumps(self.v1_content))
|
|
del v1_data["profiles"]["prgs"]["forbidden_operations"]
|
|
|
|
with self.assertRaisesRegex(ValueError, "fail closed"):
|
|
migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
|
|
def test_write_mode_and_backup(self):
|
|
"""Verify that write mode creates a backup and correctly saves the validated config."""
|
|
output_file = os.path.join(self.temp_dir, "migrated.json")
|
|
backup_file = os.path.join(self.temp_dir, "profiles_backup.json.bak")
|
|
|
|
test_args = [
|
|
"migrate_profiles.py",
|
|
"-i", self.input_file,
|
|
"-o", output_file,
|
|
"--backup", backup_file,
|
|
"-w"
|
|
]
|
|
with patch.object(sys, "argv", test_args):
|
|
migrate_profiles.main()
|
|
|
|
# Verify backup exists and matches original v1 config
|
|
self.assertTrue(os.path.exists(backup_file))
|
|
with open(backup_file, "r") as f:
|
|
backup_data = json.load(f)
|
|
self.assertEqual(backup_data["version"], 1)
|
|
self.assertIn("prgs", backup_data["profiles"])
|
|
|
|
# Verify migrated v2 config exists and validates
|
|
self.assertTrue(os.path.exists(output_file))
|
|
with open(output_file, "r") as f:
|
|
v2_data = json.load(f)
|
|
self.assertEqual(v2_data["version"], 2)
|
|
self.assertIn("environments", v2_data)
|
|
self.assertEqual(v2_data["aliases"]["prgs"], "prgs.gitea.author")
|
|
|
|
def test_malformed_input_fails_safely(self):
|
|
"""Test that malformed JSON or invalid version numbers cause a clean exit with code 1."""
|
|
bad_json_file = os.path.join(self.temp_dir, "bad.json")
|
|
with open(bad_json_file, "w") as f:
|
|
f.write("{invalid-json}")
|
|
|
|
test_args = ["migrate_profiles.py", "-i", bad_json_file]
|
|
with patch.object(sys, "argv", test_args):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
migrate_profiles.main()
|
|
self.assertEqual(cm.exception.code, 1)
|
|
|
|
bad_version_file = os.path.join(self.temp_dir, "bad_version.json")
|
|
with open(bad_version_file, "w") as f:
|
|
json.dump({"version": 3, "profiles": {}}, f)
|
|
|
|
test_args = ["migrate_profiles.py", "-i", bad_version_file]
|
|
with patch.object(sys, "argv", test_args):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
migrate_profiles.main()
|
|
self.assertEqual(cm.exception.code, 1)
|
|
|
|
def test_reconciler_profile_migration(self):
|
|
"""Legacy reconciler shorthands migrate to valid canonical operations."""
|
|
import gitea_config
|
|
import reconciler_profile
|
|
|
|
v1_data = {
|
|
"version": 1,
|
|
"profiles": {
|
|
"prgs-reconciler": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "reconciler-agent",
|
|
"auth": {"type": "keychain", "id": "reconciler-ref"},
|
|
"execution_profile": "prgs-reconciler",
|
|
"allowed_operations": [
|
|
"read",
|
|
"pr.close",
|
|
"pr.comment",
|
|
"issue.comment",
|
|
"issue.close",
|
|
"gitea.branch.delete",
|
|
],
|
|
"forbidden_operations": [
|
|
"merge",
|
|
"approve",
|
|
"review",
|
|
"pr.create",
|
|
"branch.push",
|
|
"commit",
|
|
],
|
|
}
|
|
}
|
|
}
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
reconciler = (
|
|
v2_data["environments"]["prgs"]["services"]["gitea"]
|
|
["identities"]["reconciler"]
|
|
)
|
|
self.assertEqual(reconciler["role"], "reconciler")
|
|
allowed = reconciler["allowed_operations"]
|
|
forbidden = reconciler["forbidden_operations"]
|
|
# No invalid shorthand remains
|
|
for bad in ("pr.close", "pr.comment", "issue.close", "read", "merge"):
|
|
self.assertNotIn(bad, allowed)
|
|
self.assertNotIn(bad, forbidden)
|
|
for required in (
|
|
"gitea.read",
|
|
"gitea.pr.close",
|
|
"gitea.pr.comment",
|
|
"gitea.issue.comment",
|
|
"gitea.issue.close",
|
|
"gitea.branch.delete",
|
|
):
|
|
self.assertIn(required, allowed)
|
|
# Production loader accepts every allowed op
|
|
self.assertEqual(
|
|
gitea_config.normalize_operation(required), required
|
|
)
|
|
self.assertEqual(v2_data["aliases"]["prgs-reconciler"], "prgs.gitea.reconciler")
|
|
assessment = reconciler_profile.assess_reconciler_profile(allowed, forbidden)
|
|
self.assertTrue(assessment["valid"])
|
|
self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
|
|
|
|
def test_reconciler_profile_defaults(self):
|
|
"""Reconciler defaults are fully canonical and loader-valid."""
|
|
import gitea_config
|
|
import reconciler_profile
|
|
|
|
v1_data = {
|
|
"version": 1,
|
|
"profiles": {
|
|
"prgs-reconciler": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "reconciler-agent",
|
|
"auth": {"type": "keychain", "id": "reconciler-ref"},
|
|
"execution_profile": "prgs-reconciler",
|
|
}
|
|
}
|
|
}
|
|
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
reconciler = (
|
|
v2_data["environments"]["prgs"]["services"]["gitea"]
|
|
["identities"]["reconciler"]
|
|
)
|
|
self.assertEqual(reconciler["role"], "reconciler")
|
|
self.assertEqual(
|
|
reconciler["allowed_operations"],
|
|
migrate_profiles.RECONCILER_DEFAULT_ALLOWED,
|
|
)
|
|
self.assertEqual(
|
|
reconciler["forbidden_operations"],
|
|
migrate_profiles.RECONCILER_DEFAULT_FORBIDDEN,
|
|
)
|
|
for op in reconciler["allowed_operations"]:
|
|
self.assertEqual(gitea_config.normalize_operation(op), op)
|
|
self.assertTrue(op.startswith("gitea."))
|
|
assessment = reconciler_profile.assess_reconciler_profile(
|
|
reconciler["allowed_operations"],
|
|
reconciler["forbidden_operations"],
|
|
)
|
|
self.assertTrue(assessment["valid"])
|
|
self.assertNotIn(
|
|
"gitea.branch.delete", assessment["missing_recommended_operations"]
|
|
)
|
|
|
|
def test_reconciler_migration_idempotent_canonicalize(self):
|
|
"""Second canonicalize of already-canonical ops is a no-op."""
|
|
first = migrate_profiles.canonicalize_operations(
|
|
list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED)
|
|
)
|
|
second = migrate_profiles.canonicalize_operations(first)
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(first, list(migrate_profiles.RECONCILER_DEFAULT_ALLOWED))
|
|
|
|
def test_reconciler_missing_required_fails_visibly(self):
|
|
"""Missing gitea.pr.close after migration fails closed (not silent drop)."""
|
|
v1_data = {
|
|
"version": 1,
|
|
"profiles": {
|
|
"prgs-reconciler": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "reconciler-agent",
|
|
"auth": {"type": "keychain", "id": "reconciler-ref"},
|
|
"execution_profile": "prgs-reconciler",
|
|
"allowed_operations": ["read", "gitea.branch.delete"],
|
|
"forbidden_operations": ["merge"],
|
|
}
|
|
},
|
|
}
|
|
with self.assertRaisesRegex(ValueError, "missing required"):
|
|
migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
|
|
def test_unknown_operation_fails_visibly(self):
|
|
v1_data = {
|
|
"version": 1,
|
|
"profiles": {
|
|
"prgs-author": {
|
|
"base_url": "redacted-prgs-service",
|
|
"username": "jcwalker3",
|
|
"auth": {"type": "keychain", "id": "hidden-author-ref"},
|
|
"execution_profile": "prgs-author",
|
|
"allowed_operations": ["read", "not.a.real.op"],
|
|
"forbidden_operations": ["merge"],
|
|
}
|
|
},
|
|
}
|
|
with self.assertRaisesRegex(ValueError, "cannot be canonicalized"):
|
|
migrate_profiles.migrate_v1_to_v2(v1_data)
|
|
|
|
def test_role_inference_author_reviewer_merger_reconciler(self):
|
|
self.assertEqual(
|
|
migrate_profiles.infer_role("prgs-author", "prgs-author"), "author"
|
|
)
|
|
self.assertEqual(
|
|
migrate_profiles.infer_role("prgs-reviewer", "prgs-reviewer"),
|
|
"reviewer",
|
|
)
|
|
self.assertEqual(
|
|
migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
|
|
"reconciler",
|
|
)
|
|
self.assertIsNone(
|
|
migrate_profiles.infer_role("prgs-merger", "prgs-merger")
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|