""" 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" } } }, "merge-ff": { "workflow.config": { "MergeMode": "ff-only" } }, "merge-replace": { "workflow.config": { "MergeMode": "replace" } }, "merge-devel": { "workflow.config": { "MergeMode": "devel" } } } # 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 merge_ff_env(gitea_env): return gitea_env, "myproducts/mySLFO", "merge-ff" @pytest.fixture(scope="session") def merge_replace_env(gitea_env): return gitea_env, "myproducts/mySLFO", "merge-replace" @pytest.fixture(scope="session") def merge_devel_env(gitea_env): return gitea_env, "myproducts/mySLFO", "merge-devel" @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)