6aaff89179
- Add TC-MERGE-008 to 013 for testing MergeMode of workflow-pr - Synchronize integration/test-plan.md with the actual test implementations
332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""
|
|
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)
|