autogits/integration/tests/workflow_pr_merge_test.py

import pytest
import re
import time
from pathlib import Path
from tests.lib.common_test_utils import GiteaAPIClient

@pytest.mark.t001
def test_001_automerge(automerge_env, test_user_client):
    """
    Test scenario TC-MERGE-001:
    1. Create a PackageGit PR.
    2. Ensure all mandatory reviews are completed on both project and package PRs.
    3. Verify the PR is merged automatically.
    """
    gitea_env, test_full_repo_name, merge_branch_name = automerge_env

    # 1. Create a package PR
    diff = """diff --git a/automerge_test.txt b/automerge_test.txt
new file mode 100644
index 0000000..e69de29
"""
    print(f"--- Creating package PR in mypool/pkgA on branch {merge_branch_name} ---")
    package_pr = test_user_client.create_gitea_pr("mypool/pkgA", diff, "Test Automerge Fixture", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    print(f"Created package PR mypool/pkgA#{package_pr_number}")

    # 2. Make sure the workflow-pr service created related project PR
    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None, "Workflow bot did not create a project PR."
    print(f"Found project PR: myproducts/mySLFO#{project_pr_number}")

    # 3. Approve reviews and verify merged
    pkg_merged, prj_merged = gitea_env.approve_and_wait_merge("mypool/pkgA", package_pr_number, project_pr_number)
    assert pkg_merged, f"Package PR mypool/pkgA#{package_pr_number} was not merged automatically."
    assert prj_merged, f"Project PR myproducts/mySLFO#{project_pr_number} was not merged automatically."
    print("Both PRs merged successfully.")

@pytest.mark.t002
def test_002_manual_merge(manual_merge_env, test_user_client, usera_client, staging_bot_client):
    """
    Test scenario TC-MERGE-002:
    1. Create a PackageGit PR with ManualMergeOnly set to true.
    2. Ensure all mandatory reviews are completed on both project and package PRs.
    3. Comment "merge ok" on the package PR from the account of a requested reviewer.
    4. Verify the PR is merged.
    """
    gitea_env, test_full_repo_name, merge_branch_name = manual_merge_env

    # 1. Create a package PR
    diff = """diff --git a/manual_merge_test.txt b/manual_merge_test.txt
new file mode 100644
index 0000000..e69de29
"""
    print(f"--- Creating package PR in mypool/pkgA on branch {merge_branch_name} ---")
    package_pr = test_user_client.create_gitea_pr("mypool/pkgA", diff, "Test Manual Merge Fixture", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    print(f"Created package PR mypool/pkgA#{package_pr_number}")

    # 2. Make sure the workflow-pr service created related project PR
    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None, "Workflow bot did not create a project PR."
    print(f"Found project PR: myproducts/mySLFO#{project_pr_number}")

    # 3. Approve reviews and verify NOT merged
    print("Waiting for all expected review requests and approving them...")
    # Expected reviewers based on manual-merge branch config and pkgA maintainership
    expected_reviewers = {"usera", "userb", "ownerA", "ownerX", "ownerY"}
    
    # ManualMergeOnly still requires regular reviews to be satisfied.
    # We poll until all expected reviewers are requested, then approve them.
    all_requested = False
    for _ in range(30):
        # Trigger approvals for whatever is already requested
        gitea_env.approve_requested_reviews("mypool/pkgA", package_pr_number)
        gitea_env.approve_requested_reviews("myproducts/mySLFO", project_pr_number)
        
        # Explicitly handle staging bot if it is requested or pending
        prj_reviews = gitea_env.list_reviews("myproducts/mySLFO", project_pr_number)
        if any(r["user"]["login"] == "autogits_obs_staging_bot" and r["state"] in ["REQUEST_REVIEW", "PENDING"] for r in prj_reviews):
            print("Staging bot has a pending/requested review. Approving...")
            staging_bot_client.create_review("myproducts/mySLFO", project_pr_number, event="APPROVED", body="Staging bot approves")

        # Check if all expected reviewers have at least one review record (any state)
        pkg_reviews = gitea_env.list_reviews("mypool/pkgA", package_pr_number)
        current_reviewers = {r["user"]["login"] for r in pkg_reviews}
        
        if expected_reviewers.issubset(current_reviewers):
            # Also ensure they are all approved (not just requested)
            approved_reviewers = {r["user"]["login"] for r in pkg_reviews if r["state"] == "APPROVED"}
            if expected_reviewers.issubset(approved_reviewers):
                # And check project PR for bot approval
                prj_approved = any(r["user"]["login"] == "autogits_obs_staging_bot" and r["state"] == "APPROVED" for r in prj_reviews)
                if prj_approved:
                    all_requested = True
                    print(f"All expected reviewers {expected_reviewers} and staging bot have approved.")
                    break
        
        pkg_details = gitea_env.get_pr_details("mypool/pkgA", package_pr_number)
        prj_details = gitea_env.get_pr_details("myproducts/mySLFO", project_pr_number)
        
        assert not pkg_details.get("merged"), "Package PR merged prematurely (ManualMergeOnly ignored?)"
        assert not prj_details.get("merged"), "Project PR merged prematurely (ManualMergeOnly ignored?)"
        
        time.sleep(2)

    assert all_requested, f"Timed out waiting for all expected reviewers {expected_reviewers} to approve. Current: {current_reviewers}"
    print("Both PRs have all required approvals but are not merged (as expected with ManualMergeOnly).")

    # 4. Comment "merge ok" from a requested reviewer (usera)
    print("Commenting 'merge ok' on package PR...")
    usera_client.create_issue_comment("mypool/pkgA", package_pr_number, "merge ok")

    # 5. Verify both PRs are merged
    print("Polling for PR merge status...")
    package_merged = False
    project_merged = False
    
    for i in range(20): # Poll for up to 20 seconds
        if not package_merged:
            pkg_details = gitea_env.get_pr_details("mypool/pkgA", package_pr_number)
            if pkg_details.get("merged"):
                package_merged = True
                print(f"Package PR mypool/pkgA#{package_pr_number} merged.")
        
        if not project_merged:
            prj_details = gitea_env.get_pr_details("myproducts/mySLFO", project_pr_number)
            if prj_details.get("merged"):
                project_merged = True
                print(f"Project PR myproducts/mySLFO#{project_pr_number} merged.")
        
        if package_merged and project_merged:
            break
            
        time.sleep(1)
            
    assert package_merged, f"Package PR mypool/pkgA#{package_pr_number} was not merged after 'merge ok'."
    assert project_merged, f"Project PR myproducts/mySLFO#{project_pr_number} was not merged after 'merge ok'."
    print("Both PRs merged successfully after 'merge ok'.")

@pytest.mark.t008
def test_008_merge_mode_ff_only_success(merge_ff_env, test_user_client):
    """
    Test MergeMode "ff-only" - Success case (FF-mergeable)
    """
    gitea_env, test_full_repo_name, merge_branch_name = merge_ff_env

    # 1. Create a package PR (this will be FF-mergeable by default)
    diff = """diff --git a/ff_test.txt b/ff_test.txt
new file mode 100644
index 0000000..e69de29
"""
    package_pr = test_user_client.create_gitea_pr("mypool/pkgA", diff, "Test FF Merge", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    
    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None

    pkg_merged, prj_merged = gitea_env.approve_and_wait_merge("mypool/pkgA", package_pr_number, project_pr_number)
    assert pkg_merged and prj_merged

@pytest.mark.t009
def test_009_merge_mode_ff_only_failure(merge_ff_env, ownerA_client):
    """
    Test MergeMode "ff-only" - Failure case (Content Conflict, should NOT merge)
    """
    gitea_env, test_full_repo_name, merge_branch_name = merge_ff_env

    ts = time.strftime("%H%M%S")
    filename = f"ff_fail_test_{ts}.txt"

    # 1. Create a package PR that adds a file
    diff = f"""diff --git a/{filename} b/{filename}
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/{filename}
@@ -0,0 +1 @@
+PR content
"""
    package_pr = ownerA_client.create_gitea_pr("mypool/pkgA", diff, "Test FF Merge Failure (Conflict)", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    
    # 2. Wait for project PR to be created
    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None

    print("Making PR non-FF by creating a content conflict in the base branch...")
    gitea_env.create_file("mypool", "pkgA", filename, "Conflicting base content\n", branch=merge_branch_name)

    print("Approving reviews initially...")
    gitea_env.approve_requested_reviews("mypool/pkgA", package_pr_number)
    gitea_env.approve_requested_reviews("myproducts/mySLFO", project_pr_number)

    print("Pushing another change to PR branch to trigger sync...")
    gitea_env.modify_gitea_pr("mypool/pkgA", package_pr_number, 
                             "diff --git a/sync_test.txt b/sync_test.txt\nnew file mode 100644\nindex 0000000..e69de29\n",
                             "Trigger Sync")

    # The bot should detect it's not FF and NOT merge, and re-request reviews because of the new commit
    print("Waiting for reviews to be re-requested and approving again...")
    time.sleep(10) # Wait for bot to process sync
    
    # Approve again and verify it is NOT merged
    print("Approving again and verifying PR is NOT merged (because it's not FF)...")
    for i in range(15):
        gitea_env.approve_requested_reviews("mypool/pkgA", package_pr_number)
        gitea_env.approve_requested_reviews("myproducts/mySLFO", project_pr_number)
        time.sleep(1)
        
    pkg_details = gitea_env.get_pr_details("mypool/pkgA", package_pr_number)
    assert not pkg_details.get("merged"), "Package PR merged despite NOT being FF-mergeable!"
    
    print("FF-only failure (not merged after sync) verified.")

@pytest.mark.t010
def test_010_merge_mode_devel_success(merge_devel_env, ownerA_client):
    """
    Test MergeMode "devel" - Success case (Content Conflict, should still merge via force-push)
    """
    gitea_env, test_full_repo_name, merge_branch_name = merge_devel_env

    ts = time.strftime("%H%M%S")
    filename = f"devel_test_{ts}.txt"

    # 1. Create a package PR that adds a file
    diff = f"""diff --git a/{filename} b/{filename}
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/{filename}
@@ -0,0 +1 @@
+PR content
"""
    package_pr = ownerA_client.create_gitea_pr("mypool/pkgA", diff, "Test Devel Merge (Conflict)", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    
    # 2. Create a content conflict by committing the same file to the base branch
    gitea_env.create_file("mypool", "pkgA", filename, "Conflicting base content\n", branch=merge_branch_name)

    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None

    # Before merge, get the head sha of the package pr and project pr
    pkg_details = gitea_env.get_pr_details("mypool/pkgA", package_pr_number)
    pkg_head_sha = pkg_details["head"]["sha"]
    prj_details = gitea_env.get_pr_details("myproducts/mySLFO", project_pr_number)
    prj_head_sha = prj_details["head"]["sha"]

    pkg_merged, prj_merged = gitea_env.approve_and_wait_merge("mypool/pkgA", package_pr_number, project_pr_number)
    assert pkg_merged and prj_merged
    print("Devel merge (force-push) successful.")

    # Verify that pkgA submodule points to the correct SHA
    pkgA_submodule_info = gitea_env.get_file_info("myproducts", "mySLFO", "pkgA", branch=merge_branch_name)
    assert pkgA_submodule_info["sha"] == pkg_head_sha, f"Submodule pkgA should point to {pkg_head_sha} but points to {pkgA_submodule_info['sha']}"

@pytest.mark.t011
def test_011_merge_mode_replace_success(merge_replace_env, ownerA_client):
    """
    Test MergeMode "replace" - Success case (Content Conflict, bot should add merge commit)
    """
    gitea_env, test_full_repo_name, merge_branch_name = merge_replace_env

    ts = time.strftime("%H%M%S")
    filename = f"replace_test_{ts}.txt"

    # 1. Create a package PR that adds a file
    diff = f"""diff --git a/{filename} b/{filename}
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/{filename}
@@ -0,0 +1 @@
+PR content
"""
    package_pr = ownerA_client.create_gitea_pr("mypool/pkgA", diff, "Test Replace Merge (Conflict)", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    
    # Enable "Allow edits from maintainers"
    ownerA_client.update_gitea_pr_properties("mypool/pkgA", package_pr_number, allow_maintainer_edit=True)
    
    # 2. Create a content conflict by committing the same file to the base branch
    gitea_env.create_file("mypool", "pkgA", filename, "Conflicting base content\n", branch=merge_branch_name)

    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None

    # Before merge, get the head sha of the package pr and project pr
    pkg_details = gitea_env.get_pr_details("mypool/pkgA", package_pr_number)
    pkg_head_sha = pkg_details["head"]["sha"]
    prj_details = gitea_env.get_pr_details("myproducts/mySLFO", project_pr_number)
    prj_head_sha = prj_details["head"]["sha"]

    pkg_merged, prj_merged = gitea_env.approve_and_wait_merge("mypool/pkgA", package_pr_number, project_pr_number, timeout=60)
    assert pkg_merged and prj_merged
    print("Replace merge successful.")

    # Verify that the project branch HEAD is a merge commit
    branch_info = gitea_env._request("GET", f"repos/myproducts/mySLFO/branches/{merge_branch_name}").json()
    new_head_sha = branch_info["commit"]["id"]
    
    commit_details = gitea_env._request("GET", f"repos/myproducts/mySLFO/git/commits/{new_head_sha}").json()
    assert len(commit_details["parents"]) > 1, f"Project branch {merge_branch_name} HEAD should be a merge commit but has {len(commit_details['parents'])} parents"

    # Verify that pkgA submodule points to the correct SHA
    pkgA_submodule_info = gitea_env.get_file_info("myproducts", "mySLFO", "pkgA", branch=merge_branch_name)
    assert pkgA_submodule_info["sha"] == pkg_head_sha, f"Submodule pkgA should point to {pkg_head_sha} but points to {pkgA_submodule_info['sha']}"

@pytest.mark.t012
def test_012_merge_mode_devel_ff_success(merge_devel_env, ownerA_client):
    """
    Test MergeMode "devel" - Success case (No Conflict, should fast-forward)
    """
    gitea_env, test_full_repo_name, merge_branch_name = merge_devel_env

    ts = time.strftime("%H%M%S")
    filename = f"devel_ff_test_{ts}.txt"

    # 1. Create a package PR (this will be FF-mergeable by default)
    diff = f"""diff --git a/{filename} b/{filename}
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/{filename}
@@ -0,0 +1 @@
+PR content
"""
    package_pr = ownerA_client.create_gitea_pr("mypool/pkgA", diff, "Test Devel FF Merge", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    
    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None

    pkg_details = gitea_env.get_pr_details("mypool/pkgA", package_pr_number)
    pkg_head_sha = pkg_details["head"]["sha"]

    pkg_merged, prj_merged = gitea_env.approve_and_wait_merge("mypool/pkgA", package_pr_number, project_pr_number)
    assert pkg_merged and prj_merged
    print("Devel FF merge successful.")

    # Verify that the package base branch HEAD is the same as the PR head (FF)
    branch_info = gitea_env._request("GET", f"repos/mypool/pkgA/branches/{merge_branch_name}").json()
    new_head_sha = branch_info["commit"]["id"]
    assert new_head_sha == pkg_head_sha, f"Package branch {merge_branch_name} HEAD should be {pkg_head_sha} but is {new_head_sha}"
    
    commit_details = gitea_env._request("GET", f"repos/mypool/pkgA/git/commits/{new_head_sha}").json()
    assert len(commit_details["parents"]) == 1, f"Package branch {merge_branch_name} HEAD should have 1 parent but has {len(commit_details['parents'])}"

@pytest.mark.t013
def test_013_merge_mode_replace_ff_success(merge_replace_env, ownerA_client):
    """
    Test MergeMode "replace" - Success case (No Conflict, should fast-forward)
    """
    gitea_env, test_full_repo_name, merge_branch_name = merge_replace_env

    ts = time.strftime("%H%M%S")
    filename = f"replace_ff_test_{ts}.txt"

    # 1. Create a package PR (this will be FF-mergeable by default)
    diff = f"""diff --git a/{filename} b/{filename}
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/{filename}
@@ -0,0 +1 @@
+PR content
"""
    package_pr = ownerA_client.create_gitea_pr("mypool/pkgA", diff, "Test Replace FF Merge", False, base_branch=merge_branch_name)
    package_pr_number = package_pr["number"]
    
    project_pr_number = gitea_env.wait_for_project_pr("mypool/pkgA", package_pr_number)
    assert project_pr_number is not None

    pkg_details = gitea_env.get_pr_details("mypool/pkgA", package_pr_number)
    pkg_head_sha = pkg_details["head"]["sha"]

    pkg_merged, prj_merged = gitea_env.approve_and_wait_merge("mypool/pkgA", package_pr_number, project_pr_number)
    assert pkg_merged and prj_merged
    print("Replace FF merge successful.")

    # Verify that the package base branch HEAD is the same as the PR head (FF)
    branch_info = gitea_env._request("GET", f"repos/mypool/pkgA/branches/{merge_branch_name}").json()
    new_head_sha = branch_info["commit"]["id"]
    assert new_head_sha == pkg_head_sha, f"Package branch {merge_branch_name} HEAD should be {pkg_head_sha} but is {new_head_sha}"
    
    commit_details = gitea_env._request("GET", f"repos/mypool/pkgA/git/commits/{new_head_sha}").json()
    assert len(commit_details["parents"]) == 1, f"Package branch {merge_branch_name} HEAD should have 1 parent but has {len(commit_details['parents'])}"