git-workflow/autogits
SHA256
21
2
Fork 16
Code Issues 44 Pull Requests 16 Actions Projects 2 Activity

t: Optimize setups #132

Merged
andriinikitin merged 2 commits from andriinikitin/autogits:t-optimize-setup into main 2026-02-22 20:50:29 +01:00
Conversation 0 Commits 2 Files Changed 5 +217 -644
5 changed files with 217 additions and 644 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
integration/gitea/Dockerfile
View File

@@ -11,7 +11,7 @@ RUN zypper -n install \
openssh \
jq \
devel_Factory_git-workflow:gitea \
&& rm -rf /var/cache/zypp/*
&& rm -rf /var/cache/zypp/* || ( tail -n 1000 /var/log/zypper.log ; exit 1 )
# Copy the minimal set of required files from the local 'container-files' directory
COPY container-files/ /

818
integration/tests/conftest.py
View File

@@ -6,231 +6,100 @@ import pytest
import requests
import time
import os
# Assuming GiteaAPIClient is in tests/lib/common_test_utils.py
import json
import base64
from tests.lib.common_test_utils import GiteaAPIClient
@pytest.fixture(scope="session")
def gitea_env():
"""
Sets up the Gitea environment with dummy data and provides a GiteaAPIClient instance.
"""
gitea_url = "http://127.0.0.1:3000"
# Read admin token
admin_token_path = "./gitea-data/admin.token" # Corrected path
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}. Ensure it's generated and accessible.")
BRANCH_CONFIGS = {
"main": {
"workflow.config": {
"Workflows": ["pr"],
"GitProjectName": "products/SLFO#main",
"Organization": "pool",
"Branch": "main",
"ManualMergeProject": True,
"Reviewers": ["-autogits_obs_staging_bot"]
},
"staging.config": {
"ObsProject": "openSUSE:Leap:16.0",
"StagingProject": "openSUSE:Leap:16.0:PullRequest"
},
"_maintainership.json": {
"": ["ownerX", "ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB", "ownerBB"]
}
},
"merge": {
"workflow.config": {
"Workflows": ["pr"],
"GitProjectName": "products/SLFO#merge",
"Organization": "pool",
"Branch": "merge",
"Reviewers": ["+usera", "+userb", "-autogits_obs_staging_bot"]
},
"_maintainership.json": {
"": ["ownerX", "ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB", "ownerBB"]
}
},
"maintainer-merge": {
"workflow.config": {
"Workflows": ["pr"],
"GitProjectName": "products/SLFO#maintainer-merge",
"Organization": "pool",
"Branch": "maintainer-merge",
"Reviewers": ["-autogits_obs_staging_bot"]
},
"_maintainership.json": {
"": ["ownerX", "ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB", "ownerBB"]
}
},
"review-required": {
"workflow.config": {
"Workflows": ["pr"],
"GitProjectName": "products/SLFO#review-required",
"Organization": "pool",
"Branch": "review-required",
"Reviewers": ["-autogits_obs_staging_bot"],
"ReviewRequired": True
},
"_maintainership.json": {
"": ["ownerX", "ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB", "ownerBB"]
}
},
"dev": {
"workflow.config": {
"Workflows": ["pr"],
"GitProjectName": "products/SLFO#dev",
"Organization": "pool",
"Branch": "dev",
"ManualMergeProject": True,
"Reviewers": ["-autogits_obs_staging_bot"],
"NoProjectGitPR": True
},
"_maintainership.json": {
"": ["ownerX", "ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB", "ownerBB"]
}
}
}
# Headers for authenticated requests
auth_headers = {"Authorization": f"token {admin_token}", "Content-Type": "application/json"}
# Wait for Gitea to be available
print(f"Waiting for Gitea at {gitea_url}...")
max_retries = 5
for i in range(max_retries):
try:
# Check a specific API endpoint that indicates readiness
response = requests.get(f"{gitea_url}/api/v1/version", headers=auth_headers, timeout=5)
if response.status_code == 200:
print("Gitea API is available.")
break
except requests.exceptions.ConnectionError:
pass
print(f"Gitea not ready ({response.status_code if 'response' in locals() else 'ConnectionError'}), retrying in 1 seconds... ({i+1}/{max_retries})")
time.sleep(1)
else:
raise Exception("Gitea did not become available within the expected time.")
client = GiteaAPIClient(base_url=gitea_url, token=admin_token)
# Setup dummy data
print("--- Starting Gitea Dummy Data Setup from Pytest Fixture ---")
client.create_org("products")
client.create_org("pool")
client.create_repo("products", "SLFO")
client.create_repo("pool", "pkgA")
client.create_repo("pool", "pkgB")
# The add_submodules method also creates workflow.config and staging.config
client.add_submodules("products", "SLFO")
time.sleep(1)
workflow_config_content = """{
"Workflows": ["pr"],
"GitProjectName": "products/SLFO#main",
"Organization": "pool",
"Branch": "main",
"ManualMergeProject": true,
"Reviewers": [ "-autogits_obs_staging_bot" ]
}"""
client.create_file("products", "SLFO", "workflow.config", workflow_config_content)
staging_config_content = """{
"ObsProject": "openSUSE:Leap:16.0",
"StagingProject": "openSUSE:Leap:16.0:PullRequest"
}"""
client.create_file("products", "SLFO", "staging.config", staging_config_content)
client.add_collaborator("products", "SLFO", "autogits_obs_staging_bot", "write")
client.add_collaborator("products", "SLFO", "workflow-pr", "write")
client.add_collaborator("pool", "pkgA", "workflow-pr", "write")
client.add_collaborator("pool", "pkgB", "workflow-pr", "write")
client.update_repo_settings("products", "SLFO")
client.update_repo_settings("pool", "pkgA")
client.update_repo_settings("pool", "pkgB")
print("--- Gitea Dummy Data Setup Complete ---")
time.sleep(1) # Give workflow-pr bot time to become fully active
yield client
@pytest.fixture(scope="session")
def configured_dev_branch_env(gitea_env: GiteaAPIClient, request):
"""
Fixture to set up a 'dev' branch in products/SLFO and pool/pkgA,
and configure workflow.config in products/SLFO#dev with specific content.
Yields (gitea_env, test_full_repo_name, dev_branch_name).
"""
test_org_name = "products"
test_repo_name = "SLFO"
test_full_repo_name = f"{test_org_name}/{test_repo_name}"
dev_branch_name = "dev"
workflow_config_content = request.param # Get config content from parametrization
print(f"--- Setting up 'dev' branch and workflow.config in {test_full_repo_name}#{dev_branch_name} ---")
# Get the latest commit SHA of the main branch
main_branch_sha = gitea_env._request("GET", f"repos/{test_org_name}/{test_repo_name}/branches/main").json()["commit"]["id"]
# Create 'dev' branch from 'main' in products/SLFO
gitea_env.create_branch(test_org_name, test_repo_name, dev_branch_name, main_branch_sha)
# Create 'dev' branch in pool/pkgA as well
pool_pkga_main_sha = gitea_env._request("GET", "repos/pool/pkgA/branches/main").json()["commit"]["id"]
gitea_env.create_branch("pool", "pkgA", dev_branch_name, pool_pkga_main_sha)
# Create 'dev' branch in pool/pkgB as well
pool_pkgb_main_sha = gitea_env._request("GET", "repos/pool/pkgB/branches/main").json()["commit"]["id"]
gitea_env.create_branch("pool", "pkgB", dev_branch_name, pool_pkgb_main_sha)
# Create/update workflow.config with the provided content
gitea_env.create_file(test_org_name, test_repo_name, "workflow.config", workflow_config_content, branch=dev_branch_name)
print(f"Created workflow.config with specific content in {test_full_repo_name}#{dev_branch_name}")
# Restart workflow-pr service to pick up new project config
gitea_env.restart_service("workflow-pr")
time.sleep(1) # Give the service time to restart and re-initialize
yield gitea_env, test_full_repo_name, dev_branch_name
# Teardown (optional, depending on test strategy)
# For now, we'll leave resources for inspection. If a clean slate is needed for each test,
# this fixture's scope would be 'function' and teardown logic would be added here.
@pytest.fixture(scope="session")
def no_project_git_pr_env(gitea_env: GiteaAPIClient):
"""
Sets up 'dev' branch in products/SLFO and pool/pkgA,
and configures workflow.config in products/SLFO#dev with NoProjectGitPR: true.
"""
test_org_name = "products"
test_repo_name = "SLFO"
test_full_repo_name = f"{test_org_name}/{test_repo_name}"
dev_branch_name = "dev"
print(f"--- Setting up workflow.config in {test_full_repo_name}#{dev_branch_name} for No Project PR ---")
# Get the latest commit SHA of the main branch
main_branch_sha = gitea_env._request("GET", f"repos/{test_org_name}/{test_repo_name}/branches/main").json()["commit"]["id"]
# Create 'dev' branch from 'main' in products/SLFO
try:
gitea_env.create_branch(test_org_name, test_repo_name, dev_branch_name, main_branch_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create 'dev' branch in pool/pkgA as well
pool_pkga_main_sha = gitea_env._request("GET", "repos/pool/pkgA/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgA", dev_branch_name, pool_pkga_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create 'dev' branch in pool/pkgB as well
pool_pkgb_main_sha = gitea_env._request("GET", "repos/pool/pkgB/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgB", dev_branch_name, pool_pkgb_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Setup workflow.config to have "NoProjectGitPR": true
workflow_config_content_no_project_pr = f"""{{
"Workflows": ["pr"],
"GitProjectName": "{test_full_repo_name}#{dev_branch_name}",
"Organization": "pool",
"Branch": "dev",
"ManualMergeProject": true,
"Reviewers": [ "-autogits_obs_staging_bot" ],
"NoProjectGitPR": true
}}"""
gitea_env.create_file(test_org_name, test_repo_name, "workflow.config", workflow_config_content_no_project_pr, branch=dev_branch_name)
print(f"Created workflow.config with NoProjectGitPR: true in {test_full_repo_name}#{dev_branch_name}")
# Restart workflow-pr service
gitea_env.restart_service("workflow-pr")
time.sleep(1) # Give the service time to restart and re-initialize
return gitea_env, test_full_repo_name, dev_branch_name
@pytest.fixture(scope="session")
def test_user_client(gitea_env: GiteaAPIClient):
"""
Creates a new unique user and returns a GiteaAPIClient instance for them using sudo.
This user should not have write permissions to the test repositories by default.
"""
username = f"user-{int(time.time())}"
password = "password123"
email = f"{username}@example.com"
gitea_env.create_user(username, password, email)
# Grant write access to pool/pkgA
gitea_env.add_collaborator("pool", "pkgA", username, "write")
# Use admin token with Sudo header
admin_token = gitea_env.headers["Authorization"].split(" ")[1]
return GiteaAPIClient(base_url=gitea_env.base_url, token=admin_token, sudo=username)
def setup_users_from_config(client: GiteaAPIClient, workflow_config: str, maintainership_config: str):
def setup_users_from_config(client: GiteaAPIClient, wf: dict, mt: dict):
"""
Parses workflow.config and _maintainership.json, creates users, and adds them as collaborators.
"""
import json
wf = json.loads(workflow_config)
mt = json.loads(maintainership_config)
all_users = set()
# Extract from workflow.config Reviewers
reviewers = wf.get("Reviewers", [])
for r in reviewers:
# Strip +, - prefixes
username = r.lstrip("+-")
if username and username not in ["autogits_obs_staging_bot", "workflow-pr"]:
all_users.add(username)
@@ -243,8 +112,6 @@ def setup_users_from_config(client: GiteaAPIClient, workflow_config: str, mainta
# Create all users
for username in all_users:
client.create_user(username, "password123", f"{username}@example.com")
# Global maintainers (empty key) get write access to everything
# Actually, let's just make them collaborators on SLFO, pkgA, pkgB for simplicity in tests
client.add_collaborator("products", "SLFO", username, "write")
# Set specific repository permissions based on maintainership
@@ -252,469 +119,138 @@ def setup_users_from_config(client: GiteaAPIClient, workflow_config: str, mainta
repo_name = pkg if pkg else None
for username in users:
if not repo_name:
# Global maintainer - already added to SLFO, add to pkgA/pkgB
client.add_collaborator("pool", "pkgA", username, "write")
client.add_collaborator("pool", "pkgB", username, "write")
else:
client.add_collaborator("pool", repo_name, username, "write")
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():
"""
Sets up the Gitea environment with dummy data and provides a GiteaAPIClient instance.
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"
# Read admin token
admin_token_path = "./gitea-data/admin.token" # Corrected path
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}. Ensure it's generated and accessible.")
# Headers for authenticated requests
auth_headers = {"Authorization": f"token {admin_token}", "Content-Type": "application/json"}
# Wait for Gitea to be available
print(f"Waiting for Gitea at {gitea_url}...")
max_retries = 5
for i in range(max_retries):
try:
# Check a specific API endpoint that indicates readiness
response = requests.get(f"{gitea_url}/api/v1/version", headers=auth_headers, timeout=5)
if response.status_code == 200:
print("Gitea API is available.")
break
except requests.exceptions.ConnectionError:
pass
print(f"Gitea not ready ({response.status_code if 'response' in locals() else 'ConnectionError'}), retrying in 1 seconds... ({i+1}/{max_retries})")
time.sleep(1)
else:
raise Exception("Gitea did not become available within the expected time.")
raise Exception(f"Admin token file not found at {admin_token_path}.")
client = GiteaAPIClient(base_url=gitea_url, token=admin_token)
# Setup dummy data
print("--- Starting Gitea Dummy Data Setup from Pytest Fixture ---")
# 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 ---")
client.create_org("products")
client.create_org("pool")
client.create_repo("products", "SLFO")
client.create_repo("pool", "pkgA")
client.create_repo("pool", "pkgB")
# The add_submodules method also creates workflow.config and staging.config
client.update_repo_settings("products", "SLFO")
client.update_repo_settings("pool", "pkgA")
client.update_repo_settings("pool", "pkgB")
# Submodules in SLFO
client.add_submodules("products", "SLFO")
time.sleep(1)
workflow_config_content = """{
"Workflows": ["pr"],
"GitProjectName": "products/SLFO#main",
"Organization": "pool",
"Branch": "main",
"ManualMergeProject": true,
"Reviewers": [ "-autogits_obs_staging_bot" ]
}"""
client.create_file("products", "SLFO", "workflow.config", workflow_config_content)
staging_config_content = """{
"ObsProject": "openSUSE:Leap:16.0",
"StagingProject": "openSUSE:Leap:16.0:PullRequest"
}"""
client.create_file("products", "SLFO", "staging.config", staging_config_content)
maintainership_content = """{
"": ["ownerX","ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB","ownerBB"]
}"""
# Create users from default main config
setup_users_from_config(client, workflow_config_content, maintainership_content)
client.add_collaborator("products", "SLFO", "autogits_obs_staging_bot", "write")
client.add_collaborator("products", "SLFO", "workflow-pr", "write")
client.add_collaborator("pool", "pkgA", "workflow-pr", "write")
client.add_collaborator("pool", "pkgB", "workflow-pr", "write")
client.update_repo_settings("products", "SLFO")
client.update_repo_settings("pool", "pkgA")
client.update_repo_settings("pool", "pkgB")
print("--- Gitea Dummy Data Setup Complete ---")
time.sleep(1) # Give workflow-pr bot time to become fully active
restart_needed = False
# Setup all branches and configs
for branch_name, configs in BRANCH_CONFIGS.items():
# Ensure branch exists in all 3 repos
for owner, repo in [("products", "SLFO"), ("pool", "pkgA"), ("pool", "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
# Ensure config files in products/SLFO
for file_name, content_dict in configs.items():
if ensure_config_file(client, "products", "SLFO", branch_name, file_name, content_dict):
restart_needed = True
# Setup users (using configs from this branch)
setup_users_from_config(client, configs.get("workflow.config", {}), 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, "products/SLFO", "merge"
@pytest.fixture(scope="session")
def configured_dev_branch_env(gitea_env: GiteaAPIClient, request):
"""
Fixture to set up a 'dev' branch in products/SLFO and pool/pkgA,
and configure workflow.config in products/SLFO#dev with specific content.
Yields (gitea_env, test_full_repo_name, dev_branch_name).
"""
test_org_name = "products"
test_repo_name = "SLFO"
test_full_repo_name = f"{test_org_name}/{test_repo_name}"
dev_branch_name = "dev"
workflow_config_content = request.param # Get config content from parametrization
print(f"--- Setting up 'dev' branch and workflow.config in {test_full_repo_name}#{dev_branch_name} ---")
# Get the latest commit SHA of the main branch
gitea_env.ensure_branch_exists(test_org_name, test_repo_name, "main")
main_branch_sha = gitea_env._request("GET", f"repos/{test_org_name}/{test_repo_name}/branches/main").json()["commit"]["id"]
# Create 'dev' branch from 'main' in products/SLFO
gitea_env.create_branch(test_org_name, test_repo_name, dev_branch_name, main_branch_sha)
# Create 'dev' branch in pool/pkgA as well
gitea_env.ensure_branch_exists("pool", "pkgA", "main")
pool_pkga_main_sha = gitea_env._request("GET", "repos/pool/pkgA/branches/main").json()["commit"]["id"]
gitea_env.create_branch("pool", "pkgA", dev_branch_name, pool_pkga_main_sha)
# Create 'dev' branch in pool/pkgB as well
gitea_env.ensure_branch_exists("pool", "pkgB", "main")
pool_pkgb_main_sha = gitea_env._request("GET", "repos/pool/pkgB/branches/main").json()["commit"]["id"]
gitea_env.create_branch("pool", "pkgB", dev_branch_name, pool_pkgb_main_sha)
# Create/update workflow.config with the provided content
gitea_env.create_file(test_org_name, test_repo_name, "workflow.config", workflow_config_content, branch=dev_branch_name)
# For this fixture, we use default maintainership as we don't receive it in request.param
maintainership_content = """{
"": ["ownerX","ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB","ownerBB"]
}"""
setup_users_from_config(gitea_env, workflow_config_content, maintainership_content)
print(f"Created workflow.config with specific content in {test_full_repo_name}#{dev_branch_name}")
# Restart workflow-pr service to pick up new project config
gitea_env.restart_service("workflow-pr")
time.sleep(1) # Give the service time to restart and re-initialize
yield gitea_env, test_full_repo_name, dev_branch_name
def maintainer_env(gitea_env):
return gitea_env, "products/SLFO", "maintainer-merge"
@pytest.fixture(scope="session")
def no_project_git_pr_env(gitea_env: GiteaAPIClient):
"""
Sets up 'dev' branch in products/SLFO and pool/pkgA,
and configures workflow.config in products/SLFO#dev with NoProjectGitPR: true.
"""
test_org_name = "products"
test_repo_name = "SLFO"
test_full_repo_name = f"{test_org_name}/{test_repo_name}"
dev_branch_name = "dev"
print(f"--- Setting up workflow.config in {test_full_repo_name}#{dev_branch_name} for No Project PR ---")
# Get the latest commit SHA of the main branch
gitea_env.ensure_branch_exists(test_org_name, test_repo_name, "main")
main_branch_sha = gitea_env._request("GET", f"repos/{test_org_name}/{test_repo_name}/branches/main").json()["commit"]["id"]
# Create 'dev' branch from 'main' in products/SLFO
try:
gitea_env.create_branch(test_org_name, test_repo_name, dev_branch_name, main_branch_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create 'dev' branch in pool/pkgA as well
gitea_env.ensure_branch_exists("pool", "pkgA", "main")
pool_pkga_main_sha = gitea_env._request("GET", "repos/pool/pkgA/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgA", dev_branch_name, pool_pkga_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create 'dev' branch in pool/pkgB as well
gitea_env.ensure_branch_exists("pool", "pkgB", "main")
pool_pkgb_main_sha = gitea_env._request("GET", "repos/pool/pkgB/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgB", dev_branch_name, pool_pkgb_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Setup workflow.config to have "NoProjectGitPR": true
workflow_config_content = f"""{{
"Workflows": ["pr"],
"GitProjectName": "{test_full_repo_name}#{dev_branch_name}",
"Organization": "pool",
"Branch": "dev",
"ManualMergeProject": true,
"Reviewers": [ "-autogits_obs_staging_bot" ],
"NoProjectGitPR": true
}}"""
gitea_env.create_file(test_org_name, test_repo_name, "workflow.config", workflow_config_content, branch=dev_branch_name)
maintainership_content = """{
"": ["ownerX","ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB","ownerBB"]
}"""
setup_users_from_config(gitea_env, workflow_config_content, maintainership_content)
print(f"Created workflow.config with NoProjectGitPR: true in {test_full_repo_name}#{dev_branch_name}")
# Restart workflow-pr service
gitea_env.restart_service("workflow-pr")
time.sleep(1) # Give the service time to restart and re-initialize
return gitea_env, test_full_repo_name, dev_branch_name
def review_required_env(gitea_env):
return gitea_env, "products/SLFO", "review-required"
@pytest.fixture(scope="session")
def test_user_client(gitea_env: GiteaAPIClient):
"""
Creates a new unique user and returns a GiteaAPIClient instance for them using sudo.
This user should not have write permissions to the test repositories by default.
"""
username = f"user-{int(time.time())}"
password = "password123"
email = f"{username}@example.com"
gitea_env.create_user(username, password, email)
# Grant write access to pool/pkgA
def no_project_git_pr_env(gitea_env):
return gitea_env, "products/SLFO", "dev"
@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 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("pool", "pkgA", username, "write")
# Use admin token with Sudo header
admin_token = gitea_env.headers["Authorization"].split(" ")[1]
return GiteaAPIClient(base_url=gitea_env.base_url, token=admin_token, sudo=username)
@pytest.fixture(scope="session")
def automerge_env(gitea_env: GiteaAPIClient):
"""
Sets up 'merge' branch and custom workflow.config for automerge tests.
"""
test_org_name = "products"
test_repo_name = "SLFO"
test_full_repo_name = f"{test_org_name}/{test_repo_name}"
merge_branch_name = "merge"
print(f"--- Setting up '{merge_branch_name}' branch and workflow.config in {test_full_repo_name} ---")
# Get the latest commit SHA of the main branch
gitea_env.ensure_branch_exists(test_org_name, test_repo_name, "main")
main_branch_sha = gitea_env._request("GET", f"repos/{test_org_name}/{test_repo_name}/branches/main").json()["commit"]["id"]
# Create 'merge' branch from 'main' in products/SLFO
try:
gitea_env.create_branch(test_org_name, test_repo_name, merge_branch_name, main_branch_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create 'merge' branch in pool/pkgA as well
gitea_env.ensure_branch_exists("pool", "pkgA", "main")
pool_pkga_main_sha = gitea_env._request("GET", "repos/pool/pkgA/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgA", merge_branch_name, pool_pkga_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create 'merge' branch in pool/pkgB as well
gitea_env.ensure_branch_exists("pool", "pkgB", "main")
pool_pkgb_main_sha = gitea_env._request("GET", "repos/pool/pkgB/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgB", merge_branch_name, pool_pkgb_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
custom_workflow_config = f"""{{
"Workflows": ["pr"],
"GitProjectName": "{test_full_repo_name}#{merge_branch_name}",
"Organization": "pool",
"Branch": "{merge_branch_name}",
"Reviewers": [ "+usera", "+userb", "-autogits_obs_staging_bot" ]
}}"""
gitea_env.create_file(test_org_name, test_repo_name, "workflow.config", custom_workflow_config, branch=merge_branch_name)
maintainership_content = """{
"": ["ownerX","ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB","ownerBB"]
}"""
gitea_env.create_file(test_org_name, test_repo_name, "_maintainership.json", maintainership_content, branch=merge_branch_name)
setup_users_from_config(gitea_env, custom_workflow_config, maintainership_content)
# Restart workflow-pr service
gitea_env.restart_service("workflow-pr")
time.sleep(1)
return gitea_env, test_full_repo_name, merge_branch_name
@pytest.fixture(scope="session")
def maintainer_env(gitea_env: GiteaAPIClient):
"""
Sets up 'maintainer-merge' branch and workflow.config without mandatory reviewers.
"""
test_org_name = "products"
test_repo_name = "SLFO"
test_full_repo_name = f"{test_org_name}/{test_repo_name}"
branch_name = "maintainer-merge"
print(f"--- Setting up '{branch_name}' branch and workflow.config in {test_full_repo_name} ---")
# Get the latest commit SHA of the main branch
gitea_env.ensure_branch_exists(test_org_name, test_repo_name, "main")
main_branch_sha = gitea_env._request("GET", f"repos/{test_org_name}/{test_repo_name}/branches/main").json()["commit"]["id"]
# Create branch in products/SLFO
try:
gitea_env.create_branch(test_org_name, test_repo_name, branch_name, main_branch_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create branch in pool/pkgA
gitea_env.ensure_branch_exists("pool", "pkgA", "main")
pool_pkga_main_sha = gitea_env._request("GET", "repos/pool/pkgA/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgA", branch_name, pool_pkga_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create branch in pool/pkgB
gitea_env.ensure_branch_exists("pool", "pkgB", "main")
pool_pkgb_main_sha = gitea_env._request("GET", "repos/pool/pkgB/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgB", branch_name, pool_pkgb_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
custom_workflow_config = f"""{{
"Workflows": ["pr"],
"GitProjectName": "{test_full_repo_name}#{branch_name}",
"Organization": "pool",
"Branch": "{branch_name}",
"Reviewers": [ "-autogits_obs_staging_bot" ]
}}"""
gitea_env.create_file(test_org_name, test_repo_name, "workflow.config", custom_workflow_config, branch=branch_name)
maintainership_content = """{
"": ["ownerX","ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB","ownerBB"]
}"""
gitea_env.create_file(test_org_name, test_repo_name, "_maintainership.json", maintainership_content, branch=branch_name)
setup_users_from_config(gitea_env, custom_workflow_config, maintainership_content)
gitea_env.add_collaborator(test_org_name, test_repo_name, "autogits_obs_staging_bot", "write")
# Restart workflow-pr service
gitea_env.restart_service("workflow-pr")
time.sleep(1)
return gitea_env, test_full_repo_name, branch_name
@pytest.fixture(scope="session")
def review_required_env(gitea_env: GiteaAPIClient):
"""
Sets up 'review-required' branch and workflow.config with ReviewRequired: true.
"""
test_org_name = "products"
test_repo_name = "SLFO"
test_full_repo_name = f"{test_org_name}/{test_repo_name}"
branch_name = "review-required"
print(f"--- Setting up '{branch_name}' branch and workflow.config in {test_full_repo_name} ---")
# Get the latest commit SHA of the main branch
gitea_env.ensure_branch_exists(test_org_name, test_repo_name, "main")
main_branch_sha = gitea_env._request("GET", f"repos/{test_org_name}/{test_repo_name}/branches/main").json()["commit"]["id"]
# Create branch in products/SLFO
try:
gitea_env.create_branch(test_org_name, test_repo_name, branch_name, main_branch_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create branch in pool/pkgA
gitea_env.ensure_branch_exists("pool", "pkgA", "main")
pool_pkga_main_sha = gitea_env._request("GET", "repos/pool/pkgA/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgA", branch_name, pool_pkga_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Create branch in pool/pkgB
gitea_env.ensure_branch_exists("pool", "pkgB", "main")
pool_pkgb_main_sha = gitea_env._request("GET", "repos/pool/pkgB/branches/main").json()["commit"]["id"]
try:
gitea_env.create_branch("pool", "pkgB", branch_name, pool_pkgb_main_sha)
except Exception as e:
if "already exists" not in str(e).lower():
raise
custom_workflow_config = f"""{{
"Workflows": ["pr"],
"GitProjectName": "{test_full_repo_name}#{branch_name}",
"Organization": "pool",
"Branch": "{branch_name}",
"Reviewers": [ "-autogits_obs_staging_bot" ],
"ReviewRequired": true
}}"""
gitea_env.create_file(test_org_name, test_repo_name, "workflow.config", custom_workflow_config, branch=branch_name)
maintainership_content = """{
"": ["ownerX","ownerY"],
"pkgA": ["ownerA"],
"pkgB": ["ownerB","ownerBB"]
}"""
gitea_env.create_file(test_org_name, test_repo_name, "_maintainership.json", maintainership_content, branch=branch_name)
setup_users_from_config(gitea_env, custom_workflow_config, maintainership_content)
gitea_env.add_collaborator(test_org_name, test_repo_name, "autogits_obs_staging_bot", "write")
# Restart workflow-pr service
gitea_env.restart_service("workflow-pr")
time.sleep(1)
return gitea_env, test_full_repo_name, branch_name
@pytest.fixture(scope="session")
def ownerA_client(gitea_env: GiteaAPIClient):
"""
Returns a GiteaAPIClient instance for ownerA.
"""
admin_token = gitea_env.headers["Authorization"].split(" ")[1]
return GiteaAPIClient(base_url=gitea_env.base_url, token=admin_token, sudo="ownerA")
@pytest.fixture(scope="session")
def ownerB_client(gitea_env: GiteaAPIClient):
"""
Returns a GiteaAPIClient instance for ownerB.
"""
admin_token = gitea_env.headers["Authorization"].split(" ")[1]
return GiteaAPIClient(base_url=gitea_env.base_url, token=admin_token, sudo="ownerB")
@pytest.fixture(scope="session")
def ownerBB_client(gitea_env: GiteaAPIClient):
"""
Returns a GiteaAPIClient instance for ownerBB.
"""
admin_token = gitea_env.headers["Authorization"].split(" ")[1]
return GiteaAPIClient(base_url=gitea_env.base_url, token=admin_token, sudo="ownerBB")
gitea_env.add_collaborator("products", "SLFO", username, "write")
return GiteaAPIClient(base_url=gitea_env.base_url, token=gitea_env.headers["Authorization"].split(" ")[1], sudo=username)

37
integration/tests/workflow_pr_review_test.py
View File

@@ -5,6 +5,43 @@ import base64
from pathlib import Path
from tests.lib.common_test_utils import GiteaAPIClient
@pytest.mark.t001
def test_001_review_requests_matching_config(automerge_env, ownerA_client):
"""
Test scenario:
1. The package PR for pkgB is opened by ownerA (who is not a maintainer of pkgB).
2. Check that review request comes to ownerB and ownerBB (package maintainers)
AND usera and userb (from workflow.config).
"""
gitea_env, test_full_repo_name, branch_name = automerge_env
# 1. Create a package PR for pool/pkgB as ownerA
diff = """diff --git a/pkgB_test_001.txt b/pkgB_test_001.txt
new file mode 100644
index 0000000..e69de29
"""
print(f"--- Creating package PR in pool/pkgB on branch {branch_name} as ownerA ---")
package_pr = ownerA_client.create_gitea_pr("pool/pkgB", diff, "Test Review Requests Config", True, base_branch=branch_name)
package_pr_number = package_pr["number"]
print(f"Created package PR pool/pkgB#{package_pr_number}")
# 2. Check that review requests came to ownerB, ownerBB, usera, and userb
print("Checking for review requests from maintainers and workflow.config...")
reviewers_requested = set()
expected_reviewers = {"ownerB", "ownerBB", "usera", "userb"}
for _ in range(30):
reviews = gitea_env.list_reviews("pool/pkgB", package_pr_number)
reviewers_requested = {r["user"]["login"] for r in reviews if r["state"] == "REQUEST_REVIEW"}
if expected_reviewers.issubset(reviewers_requested):
break
time.sleep(1)
for reviewer in expected_reviewers:
assert reviewer in reviewers_requested, f"{reviewer} was not requested for review. Requested: {reviewers_requested}"
print(f"Confirmed: {expected_reviewers} were requested for review.")
@pytest.mark.t004
@pytest.mark.xfail(reason="the bot sometimes re-requests review from autogits_obs_staging_bot despite having the approval")
def test_004_maintainer(maintainer_env, ownerA_client):

2
integration/workflow-pr/Dockerfile.local
View File

@@ -6,7 +6,7 @@ COPY integration/rabbitmq-config/certs/cert.pem /usr/share/pki/trust/anchors/git
RUN update-ca-certificates
# Install git and ssh
RUN zypper -n in git-core openssh-clients binutils git-lfs
RUN zypper -n in git-core openssh-clients binutils git-lfs || (tail -n 1000 /var/log/zypper.log; exit 1)
# Copy the pre-built binary into the container
COPY workflow-pr/workflow-pr /usr/local/bin/workflow-pr

2
integration/workflow-pr/Dockerfile.package
View File

@@ -9,7 +9,7 @@ RUN zypper ar -f http://download.opensuse.org/repositories/devel:/Factory:/git-w
RUN zypper --gpg-auto-import-keys ref
# Install git and ssh
RUN zypper -n in git-core openssh-clients autogits-workflow-pr binutils git-lfs
RUN zypper -n in git-core openssh-clients autogits-workflow-pr binutils git-lfs || ( tail -n 1000 /var/log/zypper.log; exit 1 )
COPY integration/workflow-pr/entrypoint.sh /usr/local/bin/entrypoint.sh
RUN chmod +4755 /usr/local/bin/entrypoint.sh