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 = None
    print(f"Polling mypool/pkgA PR #{package_pr_number} timeline for forwarded PR event...")
    for _ in range(40):
        time.sleep(1)
        timeline_events = gitea_env.get_timeline_events("mypool/pkgA", package_pr_number)
        for event in timeline_events:
            if event.get("type") == "pull_ref":
                if not (ref_issue := event.get("ref_issue")):
                    continue
                url_to_check = ref_issue.get("html_url", "")
                match = re.search(r"myproducts/mySLFO/pulls/(\d+)", url_to_check)
                if match:
                    project_pr_number = int(match.group(1))
                    break
        if project_pr_number:
            break
    
    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
    print("Approving reviews and verifying both PRs are merged...")
    package_merged = False
    project_merged = False
    
    for i in range(20): # Poll for up to 20 seconds
        gitea_env.approve_requested_reviews("mypool/pkgA", package_pr_number)
        gitea_env.approve_requested_reviews("myproducts/mySLFO", project_pr_number)
        
        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.")
        
        # Project PR
        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 automatically."
    assert project_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 = None
    print(f"Polling mypool/pkgA PR #{package_pr_number} timeline for forwarded PR event...")
    for _ in range(40):
        time.sleep(1)
        timeline_events = gitea_env.get_timeline_events("mypool/pkgA", package_pr_number)
        for event in timeline_events:
            if event.get("type") == "pull_ref":
                if not (ref_issue := event.get("ref_issue")):
                    continue
                url_to_check = ref_issue.get("html_url", "")
                match = re.search(r"myproducts/mySLFO/pulls/(\d+)", url_to_check)
                if match:
                    project_pr_number = int(match.group(1))
                    break
        if project_pr_number:
            break
    
    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'.")