autogits/integration/tests/conftest.py

"""
This module contains pytest fixtures for setting up the test environment.
"""

import pytest
import requests
import time
import os
import json
import base64
from tests.lib.common_test_utils import GiteaAPIClient

BRANCH_CONFIG_COMMON = {
    "workflow.config": {
        "Workflows": ["pr"],
        "Organization": "mypool",
        "Reviewers": ["-autogits_obs_staging_bot"],
        "GitProjectName": "myproducts/mySLFO#{branch}"
    },
    "_maintainership.json": {
        "": ["ownerX", "ownerY"],
        "pkgA": ["ownerA"],
        "pkgB": ["ownerB", "ownerBB"]
    }
}

BRANCH_CONFIG_CUSTOM = {
    "main": {},
    "staging-main": {
        "workflow.config": {
            "ManualMergeProject": True
        },
        "staging.config": {
            "ObsProject": "openSUSE:Leap:16.0",
            "StagingProject": "openSUSE:Leap:16.0:PullRequest"
        }
    },
    "merge": {
        "workflow.config": {
            "Reviewers": ["+usera", "+userb", "-autogits_obs_staging_bot"]
        }
    },
    "maintainer-merge": {
        "workflow.config": {
        }
    },
    "review-required": {
        "workflow.config": {
            "ReviewRequired": True
        }
    },
    "dev": {
        "workflow.config": {
            "ManualMergeProject": True,
            "NoProjectGitPR": True
        }
    },
    "manual-merge": {
        "workflow.config": {
            "ManualMergeOnly": True,
            "Reviewers": ["+usera", "+userb", "-autogits_obs_staging_bot"]
        }
    },
    "label-test": {
        "workflow.config": {
            "ManualMergeProject": True,
            "Reviewers": ["*usera"],
            "ReviewRequired": True,
            "Labels": { 
                "StagingAuto": "staging/Backlog",
                "ReviewPending": "review/Pending"
            }
        }
    }
}

# Global state to track created Gitea objects during a pytest run
_CREATED_ORGS = set()
_CREATED_REPOS = set()
_CREATED_USERS = set()
_CREATED_LABELS = set()
_ADDED_COLLABORATORS = set() # format: (org_repo, username)

def setup_users_from_config(client: GiteaAPIClient, wf: dict, mt: dict):
    """
    Parses workflow.config and _maintainership.json, creates users, and adds them as collaborators.
    """
    all_users = set()
    
    # Extract from workflow.config Reviewers
    reviewers = wf.get("Reviewers", [])
    for r in reviewers:
        username = r.lstrip("+-*")
        if username and username not in ["autogits_obs_staging_bot", "workflow-pr"]:
            all_users.add(username)
            
    # Extract from maintainership
    for pkg, users in mt.items():
        for username in users:
            all_users.add(username)
            
    # Create all users
    for username in all_users:
        if username not in _CREATED_USERS:
            client.create_user(username, "password123", f"{username}@example.com")
            _CREATED_USERS.add(username)
        
        if ("myproducts/mySLFO", username) not in _ADDED_COLLABORATORS:
            client.add_collaborator("myproducts", "mySLFO", username, "write")
            _ADDED_COLLABORATORS.add(("myproducts/mySLFO", username))
        
    # Set specific repository permissions based on maintainership
    for pkg, users in mt.items():
        repo_name = pkg if pkg else None
        for username in users:
            if not repo_name:
                for r in ["pkgA", "pkgB"]:
                    if (f"mypool/{r}", username) not in _ADDED_COLLABORATORS:
                        client.add_collaborator("mypool", r, username, "write")
                        _ADDED_COLLABORATORS.add((f"mypool/{r}", username))
            else:
                if (f"mypool/{repo_name}", username) not in _ADDED_COLLABORATORS:
                    client.add_collaborator("mypool", repo_name, username, "write")
                    _ADDED_COLLABORATORS.add((f"mypool/{repo_name}", username))

def ensure_config_file(client: GiteaAPIClient, owner: str, repo: str, branch: str, file_name: str, expected_content_dict: dict):
    """
    Checks if a config file exists and has the correct content.
    Returns True if a change was made, False otherwise.
    """
    file_info = client.get_file_info(owner, repo, file_name, branch=branch)
    expected_content = json.dumps(expected_content_dict, indent=4)
    
    if file_info:
        current_content_raw = base64.b64decode(file_info["content"]).decode("utf-8")
        try:
            current_content_dict = json.loads(current_content_raw)
            if current_content_dict == expected_content_dict:
                return False
        except json.JSONDecodeError:
            pass # Overwrite invalid JSON
            
    client.create_file(owner, repo, file_name, expected_content, branch=branch)
    return True

@pytest.fixture(scope="session")
def gitea_env():
    """
    Global fixture to set up the Gitea environment for all tests.
    """
    gitea_url = "http://127.0.0.1:3000"
    admin_token_path = "./gitea-data/admin.token"
    
    admin_token = None
    try:
        with open(admin_token_path, "r") as f:
            admin_token = f.read().strip()
    except FileNotFoundError:
        raise Exception(f"Admin token file not found at {admin_token_path}.")

    client = GiteaAPIClient(base_url=gitea_url, token=admin_token)

    # Wait for Gitea
    for i in range(10):
        try:
            if client._request("GET", "version").status_code == 200:
                break
        except:
            pass
        time.sleep(1)
    else:
        raise Exception("Gitea not available.")

    print("--- Starting Gitea Global Setup ---")
    for org in ["myproducts", "mypool"]:
        if org not in _CREATED_ORGS:
            client.create_org(org)
            _CREATED_ORGS.add(org)

    for org, repo in [("myproducts", "mySLFO"), ("mypool", "pkgA"), ("mypool", "pkgB")]:
        if f"{org}/{repo}" not in _CREATED_REPOS:
            client.create_repo(org, repo)
            client.update_repo_settings(org, repo)
            _CREATED_REPOS.add(f"{org}/{repo}")
    
    # Create labels
    for name, color in [("staging/Backlog", "#0000ff"), ("review/Pending", "#ffff00")]:
        if ("myproducts/mySLFO", name) not in _CREATED_LABELS:
            client.create_label("myproducts", "mySLFO", name, color=color)
            _CREATED_LABELS.add(("myproducts/mySLFO", name))
    
    # Submodules in mySLFO
    client.add_submodules("myproducts", "mySLFO")

    for repo_full, bot in [("myproducts/mySLFO", "autogits_obs_staging_bot"), 
                          ("myproducts/mySLFO", "workflow-pr"),
                          ("mypool/pkgA", "workflow-pr"),
                          ("mypool/pkgB", "workflow-pr")]:
        if (repo_full, bot) not in _ADDED_COLLABORATORS:
            org_part, repo_part = repo_full.split("/")
            client.add_collaborator(org_part, repo_part, bot, "write")
            _ADDED_COLLABORATORS.add((repo_full, bot))

    restart_needed = False
    
    # Setup all branches and configs
    for branch_name, custom_configs in BRANCH_CONFIG_CUSTOM.items():
        # Ensure branch exists in all 3 repos
        for owner, repo in [("myproducts", "mySLFO"), ("mypool", "pkgA"), ("mypool", "pkgB")]:
            if branch_name != "main":
                try:
                    main_sha = client._request("GET", f"repos/{owner}/{repo}/branches/main").json()["commit"]["id"]
                    client.create_branch(owner, repo, branch_name, main_sha)
                except Exception as e:
                    if "already exists" not in str(e).lower():
                        raise
        
        # Merge configs
        merged_configs = {}
        for file_name, common_content in BRANCH_CONFIG_COMMON.items():
            merged_configs[file_name] = common_content.copy()
            # Dynamically format values containing {branch}
            if file_name == "workflow.config":
                if "GitProjectName" in merged_configs[file_name]:
                    merged_configs[file_name]["GitProjectName"] = merged_configs[file_name]["GitProjectName"].format(branch=branch_name)
                # Inject branch name dynamically
                merged_configs[file_name]["Branch"] = branch_name
            
        for file_name, custom_content in custom_configs.items():
            if file_name in merged_configs:
                merged_configs[file_name].update(custom_content)
            else:
                merged_configs[file_name] = custom_content

        # Ensure config files in myproducts/mySLFO
        for file_name, content_dict in merged_configs.items():
            if ensure_config_file(client, "myproducts", "mySLFO", branch_name, file_name, content_dict):
                restart_needed = True
                
        # Setup users (using configs from this branch)
        setup_users_from_config(client, merged_configs.get("workflow.config", {}), merged_configs.get("_maintainership.json", {}))

    if restart_needed:
        client.restart_service("workflow-pr")
        time.sleep(2) # Give it time to pick up changes

    print("--- Gitea Global Setup Complete ---")
    yield client

@pytest.fixture(scope="session")
def automerge_env(gitea_env):
    return gitea_env, "myproducts/mySLFO", "merge"

@pytest.fixture(scope="session")
def staging_main_env(gitea_env):
    return gitea_env, "myproducts/mySLFO", "staging-main"

@pytest.fixture(scope="session")
def manual_merge_env(gitea_env):
    return gitea_env, "myproducts/mySLFO", "manual-merge"

@pytest.fixture(scope="session")
def maintainer_env(gitea_env):
    return gitea_env, "myproducts/mySLFO", "maintainer-merge"

@pytest.fixture(scope="session")
def review_required_env(gitea_env):
    return gitea_env, "myproducts/mySLFO", "review-required"

@pytest.fixture(scope="session")
def no_project_git_pr_env(gitea_env):
    return gitea_env, "myproducts/mySLFO", "dev"

@pytest.fixture(scope="session")
def label_env(gitea_env):
    return gitea_env, "myproducts/mySLFO", "label-test"

@pytest.fixture(scope="session")
def usera_client(gitea_env):
    return GiteaAPIClient(base_url=gitea_env.base_url, token=gitea_env.headers["Authorization"].split(" ")[1], sudo="usera")

@pytest.fixture(scope="session")
def ownerA_client(gitea_env):
    return GiteaAPIClient(base_url=gitea_env.base_url, token=gitea_env.headers["Authorization"].split(" ")[1], sudo="ownerA")

@pytest.fixture(scope="session")
def ownerB_client(gitea_env):
    return GiteaAPIClient(base_url=gitea_env.base_url, token=gitea_env.headers["Authorization"].split(" ")[1], sudo="ownerB")

@pytest.fixture(scope="session")
def ownerBB_client(gitea_env):
    return GiteaAPIClient(base_url=gitea_env.base_url, token=gitea_env.headers["Authorization"].split(" ")[1], sudo="ownerBB")

@pytest.fixture(scope="session")
def staging_bot_client(gitea_env):
    return GiteaAPIClient(base_url=gitea_env.base_url, token=gitea_env.headers["Authorization"].split(" ")[1], sudo="autogits_obs_staging_bot")

@pytest.fixture(scope="session")
def test_user_client(gitea_env):
    username = f"test-user-{int(time.time())}"
    gitea_env.create_user(username, "password123", f"{username}@example.com")
    gitea_env.add_collaborator("mypool", "pkgA", username, "write")
    gitea_env.add_collaborator("myproducts", "mySLFO", username, "write")
    return GiteaAPIClient(base_url=gitea_env.base_url, token=gitea_env.headers["Authorization"].split(" ")[1], sudo=username)