Bmad Lens Migrate

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml>=6.0"]
# ///
"""Migration operations — scan legacy LENS v3 branches, check conflicts, migrate features.

Transitions features from the old domain-service-feature-milestone branch topology
to the Lens Next 2-branch model ({featureId} + {featureId}-plan).
"""

import argparse
import json
import os
import re
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path

import yaml

# Sanitization pattern for path-constructing identifiers
SAFE_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")

# Default legacy branch pattern
DEFAULT_BRANCH_PATTERN = r"^([a-z0-9-]+)-([a-z0-9-]+)-([a-z0-9-]+)(?:-([a-z0-9-]+))?$"

# Phase ordering for state derivation (earliest to latest)
PHASE_ORDER = ["planning", "businessplan", "techplan", "sprintplan", "dev", "complete"]


def validate_identifier(value: str, field_name: str) -> str | None:
    """Validate that a path-constructing identifier is safe. Returns error message or None."""
    if not SAFE_ID_PATTERN.match(value):
        return (
            f"Invalid {field_name}: '{value}'. "
            f"Must match [a-z0-9][a-z0-9._-]{{0,63}} (lowercase alphanumeric, dots, hyphens, underscores)."
        )
    return None


def now_iso() -> str:
    """Return current UTC time as ISO 8601 string."""
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def atomic_write_yaml(path: Path, data: dict) -> None:
    """Write YAML atomically via temp file + rename to prevent corruption."""
    dir_path = path.parent
    fd, tmp_path = tempfile.mkstemp(dir=str(dir_path), suffix=".yaml.tmp")
    try:
        with os.fdopen(fd, "w") as f:
            yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
        os.replace(tmp_path, str(path))
    except Exception:
        os.unlink(tmp_path)
        raise


def group_legacy_branches(names: list[str]) -> dict[str, dict]:
    """Group branch names into base branches with their milestones.

    Uses prefix-matching: if name B starts with name A + "-", then B is a
    milestone branch of A, and B's suffix (after A + "-") is the milestone label.
    Names not identified as milestones of any other name are treated as base branches.
    """
    sorted_names = sorted(names)
    milestone_map: dict[str, list[str]] = {}
    is_milestone: set[str] = set()

    for name in sorted_names:
        for base in sorted_names:
            if base == name:
                continue
            prefix = base + "-"
            if name.startswith(prefix):
                milestone = name[len(prefix):]
                if base not in milestone_map:
                    milestone_map[base] = []
                milestone_map[base].append(milestone)
                is_milestone.add(name)

    features = {}
    for name in sorted_names:
        if name in is_milestone:
            continue
        parts = name.split("-")
        if len(parts) < 3:
            continue
        domain = parts[0]
        service = parts[1]
        feature_id = "-".join(parts[2:])
        features[name] = {
            "old_id": name,
            "derived_domain": domain,
            "derived_service": service,
            "feature_id": feature_id,
            "milestones": milestone_map.get(name, []),
        }

    return features


def derive_state(milestones: list[str]) -> str:
    """Derive current state from list of discovered milestone labels."""
    for phase in reversed(PHASE_ORDER):
        if phase in milestones:
            return phase
    return "preplan" if not milestones else "planning"


def cmd_scan(args: argparse.Namespace) -> dict:
    """Detect legacy branches and build migration plan."""
    governance_repo = Path(args.governance_repo)
    if not governance_repo.exists():
        print(f"Error: Governance repo not found: {governance_repo}", file=sys.stderr)
        sys.exit(1)

    branches_dir = governance_repo / "branches"
    if not branches_dir.exists():
        return {"status": "pass", "legacy_features": [], "total": 0, "conflicts": []}

    pattern_str = args.branch_pattern or DEFAULT_BRANCH_PATTERN
    try:
        pattern = re.compile(pattern_str)
    except re.error as e:
        return {"status": "fail", "error": f"Invalid branch pattern: {e}"}

    candidate_names = []
    for entry in branches_dir.iterdir():
        if entry.is_dir() and pattern.match(entry.name):
            candidate_names.append(entry.name)

    if not candidate_names:
        return {"status": "pass", "legacy_features": [], "total": 0, "conflicts": []}

    grouped = group_legacy_branches(candidate_names)

    legacy_features = []
    conflicts = []

    for base_name, info in sorted(grouped.items()):
        feature_id = info["feature_id"]
        domain = info["derived_domain"]
        service = info["derived_service"]
        milestones = info["milestones"]
        state = derive_state(milestones)

        new_feature_path = governance_repo / "features" / domain / service / feature_id / "feature.yaml"
        if new_feature_path.exists():
            conflicts.append({
                "old_id": base_name,
                "feature_id": feature_id,
                "conflict_path": str(new_feature_path),
            })

        legacy_features.append({
            "old_id": base_name,
            "derived_domain": domain,
            "derived_service": service,
            "feature_id": feature_id,
            "milestones": milestones,
            "proposed": {
                "base_branch": feature_id,
                "plan_branch": f"{feature_id}-plan",
            },
            "state": state,
        })

    return {
        "status": "pass",
        "legacy_features": legacy_features,
        "total": len(legacy_features),
        "conflicts": conflicts,
    }


def cmd_migrate_feature(args: argparse.Namespace) -> dict:
    """Execute migration for a single feature."""
    governance_repo = Path(args.governance_repo)
    if not governance_repo.exists():
        print(f"Error: Governance repo not found: {governance_repo}", file=sys.stderr)
        sys.exit(1)

    for field_name, value in [
        ("feature-id", args.feature_id),
        ("domain", args.domain),
        ("service", args.service),
    ]:
        err = validate_identifier(value, field_name)
        if err:
            return {"status": "fail", "error": err}

    dry_run = args.dry_run
    feature_id = args.feature_id
    domain = args.domain
    service = args.service
    old_id = args.old_id
    username = args.username or "unknown"
    timestamp = now_iso()

    feature_dir = governance_repo / "features" / domain / service / feature_id
    feature_path = feature_dir / "feature.yaml"
    index_path = governance_repo / "feature-index.yaml"
    summary_path = governance_repo / "summaries" / f"{feature_id}.md"

    if dry_run:
        return {
            "status": "pass",
            "feature_id": feature_id,
            "dry_run": True,
            "planned_actions": [
                f"Create feature.yaml at {feature_path}",
                f"Update feature-index.yaml at {index_path}",
                f"Create summary stub at {summary_path}",
            ],
            "feature_yaml_created": False,
            "index_updated": False,
        }

    # Create feature.yaml
    feature_yaml_created = False
    if not feature_path.exists():
        feature_data = {
            "featureId": feature_id,
            "name": feature_id.replace("-", " ").title(),
            "description": f"Migrated from legacy branch: {old_id}",
            "domain": domain,
            "service": service,
            "phase": "preplan",
            "track": "full",
            "priority": "medium",
            "created": timestamp,
            "updated": timestamp,
            "team": [{"username": username, "role": "lead"}],
            "phase_transitions": [{"phase": "preplan", "timestamp": timestamp, "user": username}],
            "migrated_from": old_id,
        }
        try:
            feature_dir.mkdir(parents=True, exist_ok=True)
            atomic_write_yaml(feature_path, feature_data)
            feature_yaml_created = True
        except OSError as e:
            return {"status": "fail", "error": f"Failed to create feature.yaml: {e}"}

    # Update feature-index.yaml
    index_updated = False
    try:
        if index_path.exists():
            with open(index_path) as f:
                index_data = yaml.safe_load(f) or {}
        else:
            index_data = {"features": []}

        if "features" not in index_data:
            index_data["features"] = []

        existing_ids = [e.get("featureId") for e in index_data["features"]]
        if feature_id not in existing_ids:
            index_data["features"].append({
                "featureId": feature_id,
                "domain": domain,
                "service": service,
                "migrated_from": old_id,
                "added": timestamp,
            })
            index_path.parent.mkdir(parents=True, exist_ok=True)
            atomic_write_yaml(index_path, index_data)
            index_updated = True
    except (OSError, yaml.YAMLError) as e:
        return {"status": "fail", "error": f"Failed to update feature-index.yaml: {e}"}

    # Create summary stub
    summary_created = False
    try:
        if not summary_path.exists():
            summary_path.parent.mkdir(parents=True, exist_ok=True)
            with open(summary_path, "w") as f:
                f.write(f"# {feature_id.replace('-', ' ').title()}\n\n")
                f.write(f"**Feature ID:** {feature_id}  \n")
                f.write(f"**Domain:** {domain}  \n")
                f.write(f"**Service:** {service}  \n")
                f.write(f"**Migrated from:** {old_id}  \n")
                f.write(f"**Migration date:** {timestamp}  \n\n")
                f.write("## Summary\n\n_To be filled in._\n")
            summary_created = True
    except OSError as e:
        return {"status": "fail", "error": f"Failed to create summary.md: {e}"}

    return {
        "status": "pass",
        "feature_id": feature_id,
        "dry_run": False,
        "feature_yaml_created": feature_yaml_created,
        "index_updated": index_updated,
        "summary_created": summary_created,
    }


def cmd_check_conflicts(args: argparse.Namespace) -> dict:
    """Check for naming conflicts before migration."""
    governance_repo = Path(args.governance_repo)
    if not governance_repo.exists():
        print(f"Error: Governance repo not found: {governance_repo}", file=sys.stderr)
        sys.exit(1)

    feature_id = args.feature_id
    domain = args.domain
    service = args.service

    target_path = governance_repo / "features" / domain / service / feature_id / "feature.yaml"

    if target_path.exists():
        return {
            "status": "conflict",
            "conflict": True,
            "existing_path": str(target_path),
        }

    return {
        "status": "pass",
        "conflict": False,
        "existing_path": None,
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Migration operations — scan legacy LENS v3 branches and migrate to Lens Next model.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s scan --governance-repo /path/to/repo
  %(prog)s scan --governance-repo /path/to/repo --branch-pattern "^custom-.*$"

  %(prog)s check-conflicts --governance-repo /path/to/repo \\
    --feature-id auth-login --domain platform --service identity

  %(prog)s migrate-feature --governance-repo /path/to/repo \\
    --old-id platform-identity-auth-login --feature-id auth-login \\
    --domain platform --service identity --username cweber --dry-run

  %(prog)s migrate-feature --governance-repo /path/to/repo \\
    --old-id platform-identity-auth-login --feature-id auth-login \\
    --domain platform --service identity --username cweber
""",
    )

    subparsers = parser.add_subparsers(dest="command", required=True)

    # scan
    scan_p = subparsers.add_parser("scan", help="Detect legacy branches and build migration plan")
    scan_p.add_argument("--governance-repo", required=True, help="Path to governance repo root")
    scan_p.add_argument("--branch-pattern", help="Optional regex override for branch pattern detection")

    # migrate-feature
    mig_p = subparsers.add_parser("migrate-feature", help="Execute migration for a single feature")
    mig_p.add_argument("--governance-repo", required=True, help="Path to governance repo root")
    mig_p.add_argument("--old-id", required=True, help="Old branch name (legacy ID)")
    mig_p.add_argument("--feature-id", required=True, help="New feature ID (kebab-case)")
    mig_p.add_argument("--domain", required=True, help="Domain name")
    mig_p.add_argument("--service", required=True, help="Service name")
    mig_p.add_argument("--username", default="unknown", help="Username performing the migration")
    mig_p.add_argument("--dry-run", action="store_true", help="Preview without making changes")

    # check-conflicts
    cc_p = subparsers.add_parser("check-conflicts", help="Check for naming conflicts before migration")
    cc_p.add_argument("--governance-repo", required=True, help="Path to governance repo root")
    cc_p.add_argument("--feature-id", required=True, help="Target feature ID")
    cc_p.add_argument("--domain", required=True, help="Domain name")
    cc_p.add_argument("--service", required=True, help="Service name")

    return parser


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()

    commands = {
        "scan": cmd_scan,
        "migrate-feature": cmd_migrate_feature,
        "check-conflicts": cmd_check_conflicts,
    }

    result = commands[args.command](args)
    json.dump(result, sys.stdout, indent=2, default=str)
    print()

    status = result.get("status", "fail")
    sys.exit(0 if status in ("pass", "conflict") else 1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml>=6.0"]
# ///
"""Tests for migrate-ops.py."""

import json
import subprocess
import sys
import tempfile
from pathlib import Path

import yaml

SCRIPT = str(Path(__file__).parent.parent / "migrate-ops.py")
PASS = 0
FAIL = 0


def run(args: list[str]) -> tuple[dict, int]:
    """Run the script and return parsed JSON output."""
    result = subprocess.run(
        [sys.executable, SCRIPT] + args,
        capture_output=True,
        text=True,
    )
    try:
        return json.loads(result.stdout), result.returncode
    except json.JSONDecodeError:
        return {"error": result.stderr, "stdout": result.stdout}, result.returncode


def assert_eq(name: str, actual, expected):
    global PASS, FAIL
    if actual == expected:
        PASS += 1
        print(f"  ✓ {name}", file=sys.stderr)
    else:
        FAIL += 1
        print(f"  ✗ {name}: expected {expected!r}, got {actual!r}", file=sys.stderr)


def assert_true(name: str, actual):
    assert_eq(name, bool(actual), True)


def make_branch_dir(tmp: str, branch_name: str) -> Path:
    """Create a legacy branch directory in the branches/ folder."""
    d = Path(tmp) / "branches" / branch_name
    d.mkdir(parents=True, exist_ok=True)
    return d


def test_scan_detects_legacy():
    """scan detects legacy feature directories."""
    print("test_scan_detects_legacy", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        make_branch_dir(tmp, "platform-identity-auth-login")

        result, code = run(["scan", "--governance-repo", tmp])
        assert_eq("scan status", result["status"], "pass")
        assert_eq("scan exit code", code, 0)
        assert_eq("scan total", result["total"], 1)
        assert_true("has legacy_features", len(result["legacy_features"]) == 1)


def test_scan_derives_components():
    """scan derives domain/service/featureId from old naming."""
    print("test_scan_derives_components", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        make_branch_dir(tmp, "platform-identity-auth-login")
        make_branch_dir(tmp, "platform-identity-auth-login-planning")
        make_branch_dir(tmp, "platform-identity-auth-login-dev")

        result, code = run(["scan", "--governance-repo", tmp])
        assert_eq("scan status", result["status"], "pass")
        assert_eq("total features", result["total"], 1)

        feature = result["legacy_features"][0]
        assert_eq("derived_domain", feature["derived_domain"], "platform")
        assert_eq("derived_service", feature["derived_service"], "identity")
        assert_eq("feature_id", feature["feature_id"], "auth-login")
        assert_eq("old_id", feature["old_id"], "platform-identity-auth-login")
        assert_true("has planning milestone", "planning" in feature["milestones"])
        assert_true("has dev milestone", "dev" in feature["milestones"])
        assert_eq("proposed base_branch", feature["proposed"]["base_branch"], "auth-login")
        assert_eq("proposed plan_branch", feature["proposed"]["plan_branch"], "auth-login-plan")


def test_scan_empty_branches():
    """scan returns empty plan when no branches dir exists."""
    print("test_scan_empty_branches", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        result, code = run(["scan", "--governance-repo", tmp])
        assert_eq("empty scan status", result["status"], "pass")
        assert_eq("empty scan total", result["total"], 0)
        assert_eq("empty scan exit code", code, 0)


def test_scan_multiple_features():
    """scan groups multiple features correctly."""
    print("test_scan_multiple_features", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        make_branch_dir(tmp, "platform-identity-auth-login")
        make_branch_dir(tmp, "core-api-user-mgmt")
        make_branch_dir(tmp, "core-api-user-mgmt-dev")

        result, code = run(["scan", "--governance-repo", tmp])
        assert_eq("scan status", result["status"], "pass")
        assert_eq("total features", result["total"], 2)

        ids = [f["feature_id"] for f in result["legacy_features"]]
        assert_true("has auth-login", "auth-login" in ids)
        assert_true("has user-mgmt", "user-mgmt" in ids)

        user_mgmt = next(f for f in result["legacy_features"] if f["feature_id"] == "user-mgmt")
        assert_true("user-mgmt has dev milestone", "dev" in user_mgmt["milestones"])


def test_migrate_feature_creates_yaml():
    """migrate-feature creates feature.yaml."""
    print("test_migrate_feature_creates_yaml", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        result, code = run([
            "migrate-feature",
            "--governance-repo", tmp,
            "--old-id", "platform-identity-auth-login",
            "--feature-id", "auth-login",
            "--domain", "platform",
            "--service", "identity",
            "--username", "testuser",
        ])
        assert_eq("migrate status", result["status"], "pass")
        assert_eq("migrate exit code", code, 0)
        assert_eq("feature_yaml_created", result["feature_yaml_created"], True)

        feature_path = Path(tmp) / "features" / "platform" / "identity" / "auth-login" / "feature.yaml"
        assert_eq("feature.yaml exists", feature_path.exists(), True)

        with open(feature_path) as f:
            data = yaml.safe_load(f)
        assert_eq("featureId field", data["featureId"], "auth-login")
        assert_eq("domain field", data["domain"], "platform")
        assert_eq("service field", data["service"], "identity")
        assert_eq("migrated_from field", data["migrated_from"], "platform-identity-auth-login")
        assert_eq("phase field", data["phase"], "preplan")
        assert_true("has team lead", data["team"][0]["role"] == "lead")


def test_migrate_feature_creates_index_entry():
    """migrate-feature creates feature-index.yaml entry."""
    print("test_migrate_feature_creates_index_entry", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        result, code = run([
            "migrate-feature",
            "--governance-repo", tmp,
            "--old-id", "core-api-user-mgmt",
            "--feature-id", "user-mgmt",
            "--domain", "core",
            "--service", "api",
            "--username", "testuser",
        ])
        assert_eq("migrate status", result["status"], "pass")
        assert_eq("index_updated", result["index_updated"], True)

        index_path = Path(tmp) / "feature-index.yaml"
        assert_eq("feature-index.yaml exists", index_path.exists(), True)

        with open(index_path) as f:
            index = yaml.safe_load(f)

        feature_ids = [e["featureId"] for e in index.get("features", [])]
        assert_true("feature in index", "user-mgmt" in feature_ids)

        entry = next(e for e in index["features"] if e["featureId"] == "user-mgmt")
        assert_eq("index entry domain", entry["domain"], "core")
        assert_eq("index entry migrated_from", entry["migrated_from"], "core-api-user-mgmt")


def test_migrate_feature_dry_run():
    """migrate-feature dry-run makes no changes."""
    print("test_migrate_feature_dry_run", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        result, code = run([
            "migrate-feature",
            "--governance-repo", tmp,
            "--old-id", "platform-identity-auth-login",
            "--feature-id", "auth-login",
            "--domain", "platform",
            "--service", "identity",
            "--username", "testuser",
            "--dry-run",
        ])
        assert_eq("dry_run status", result["status"], "pass")
        assert_eq("dry_run flag", result["dry_run"], True)
        assert_eq("dry_run no yaml created", result["feature_yaml_created"], False)
        assert_eq("dry_run no index updated", result["index_updated"], False)
        assert_true("planned_actions present", len(result.get("planned_actions", [])) > 0)

        feature_path = Path(tmp) / "features" / "platform" / "identity" / "auth-login" / "feature.yaml"
        assert_eq("feature.yaml not created in dry run", feature_path.exists(), False)

        index_path = Path(tmp) / "feature-index.yaml"
        assert_eq("index not created in dry run", index_path.exists(), False)


def test_check_conflicts_no_conflict():
    """check-conflicts returns pass when target path is free."""
    print("test_check_conflicts_no_conflict", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        result, code = run([
            "check-conflicts",
            "--governance-repo", tmp,
            "--feature-id", "auth-login",
            "--domain", "platform",
            "--service", "identity",
        ])
        assert_eq("no conflict status", result["status"], "pass")
        assert_eq("conflict false", result["conflict"], False)
        assert_eq("exit code 0", code, 0)


def test_check_conflicts_conflict():
    """check-conflicts returns conflict when target path exists."""
    print("test_check_conflicts_conflict", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        feature_dir = Path(tmp) / "features" / "platform" / "identity" / "auth-login"
        feature_dir.mkdir(parents=True)
        with open(feature_dir / "feature.yaml", "w") as f:
            yaml.dump({"featureId": "auth-login"}, f)

        result, code = run([
            "check-conflicts",
            "--governance-repo", tmp,
            "--feature-id", "auth-login",
            "--domain", "platform",
            "--service", "identity",
        ])
        assert_eq("conflict status", result["status"], "conflict")
        assert_eq("conflict true", result["conflict"], True)
        assert_true("existing_path set", bool(result.get("existing_path")))
        assert_eq("conflict exit code", code, 0)


def test_invalid_feature_id():
    """Invalid feature-id slug is rejected with exit code 1."""
    print("test_invalid_feature_id", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        result, code = run([
            "migrate-feature",
            "--governance-repo", tmp,
            "--old-id", "platform-identity-auth-login",
            "--feature-id", "Auth_Login!",
            "--domain", "platform",
            "--service", "identity",
            "--username", "testuser",
        ])
        assert_eq("invalid id status", result["status"], "fail")
        assert_eq("invalid id exit code", code, 1)
        assert_true("error mentions Invalid", "Invalid" in result.get("error", ""))


def test_invalid_domain():
    """Invalid domain (path traversal) is rejected."""
    print("test_invalid_domain", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        result, code = run([
            "migrate-feature",
            "--governance-repo", tmp,
            "--old-id", "platform-identity-auth-login",
            "--feature-id", "auth-login",
            "--domain", "../evil",
            "--service", "identity",
            "--username", "testuser",
        ])
        assert_eq("invalid domain status", result["status"], "fail")
        assert_eq("invalid domain exit code", code, 1)


def test_governance_repo_not_found():
    """Governance repo not found causes exit code 1."""
    print("test_governance_repo_not_found", file=sys.stderr)
    result, code = run([
        "scan",
        "--governance-repo", "/nonexistent/path/to/repo/xyz123",
    ])
    assert_eq("missing repo exit code", code, 1)


def test_governance_repo_not_found_migrate():
    """Governance repo not found on migrate-feature causes exit code 1."""
    print("test_governance_repo_not_found_migrate", file=sys.stderr)
    result, code = run([
        "migrate-feature",
        "--governance-repo", "/nonexistent/path/xyz123",
        "--old-id", "platform-identity-auth",
        "--feature-id", "auth",
        "--domain", "platform",
        "--service", "identity",
        "--username", "testuser",
    ])
    assert_eq("missing repo exit code (migrate)", code, 1)


def test_scan_detects_conflict():
    """scan surfaces conflict when new-model feature.yaml already exists."""
    print("test_scan_detects_conflict", file=sys.stderr)
    with tempfile.TemporaryDirectory() as tmp:
        make_branch_dir(tmp, "platform-identity-auth-login")

        # Pre-create the target feature.yaml
        feature_dir = Path(tmp) / "features" / "platform" / "identity" / "auth-login"
        feature_dir.mkdir(parents=True)
        with open(feature_dir / "feature.yaml", "w") as f:
            yaml.dump({"featureId": "auth-login"}, f)

        result, code = run(["scan", "--governance-repo", tmp])
        assert_eq("scan with conflict status", result["status"], "pass")
        assert_eq("conflicts detected", len(result["conflicts"]), 1)
        assert_eq("conflict old_id", result["conflicts"][0]["old_id"], "platform-identity-auth-login")


if __name__ == "__main__":
    test_scan_detects_legacy()
    test_scan_derives_components()
    test_scan_empty_branches()
    test_scan_multiple_features()
    test_migrate_feature_creates_yaml()
    test_migrate_feature_creates_index_entry()
    test_migrate_feature_dry_run()
    test_check_conflicts_no_conflict()
    test_check_conflicts_conflict()
    test_invalid_feature_id()
    test_invalid_domain()
    test_governance_repo_not_found()
    test_governance_repo_not_found_migrate()
    test_scan_detects_conflict()

    print(f"\n{'='*40}", file=sys.stderr)
    print(f"Results: {PASS} passed, {FAIL} failed", file=sys.stderr)
    print(f"{'='*40}", file=sys.stderr)
    sys.exit(1 if FAIL > 0 else 0)