import requests
import time
def trigger_commit_and_check_ci(repo_api_url, ci_status_api_url, commit_payload, headers):
# Step 1: Commit code via API (simulate commit)
commit_response = requests.post(repo_api_url, json=commit_payload, headers=headers)
if commit_response.status_code != 201:
raise Exception(f"Commit failed with status {commit_response.status_code}")
commit_id = commit_response.json().get('commit_id')
# Step 2: Poll CI status API until pipeline completes
for _ in range(30): # wait up to ~3 minutes
ci_response = requests.get(f"{ci_status_api_url}/{commit_id}", headers=headers)
if ci_response.status_code == 200:
status = ci_response.json().get('status')
if status in ['passed', 'failed']:
return status
time.sleep(6)
raise TimeoutError("CI pipeline did not complete in expected time")
# Example usage (replace URLs and payload with real values)
# repo_api_url = 'https://api.example.com/repos/myproject/commits'
# ci_status_api_url = 'https://ci.example.com/api/status'
# commit_payload = {"message": "Test commit", "changes": [{"file": "test.txt", "content": "Hello"}]}
# headers = {"Authorization": "Bearer token"}
# result = trigger_commit_and_check_ci(repo_api_url, ci_status_api_url, commit_payload, headers)
# assert result == 'passed', f"CI tests failed with status {result}"This script simulates a code commit by sending a POST request to the repository API. It then polls the CI tool's status API to check if the pipeline triggered by that commit has completed. The polling waits up to about 3 minutes, checking every 6 seconds. Once the pipeline finishes, it returns the status, which should be either 'passed' or 'failed'. The assertion at the end ensures the tests passed.
This approach automates the manual test case steps by programmatically committing code and verifying the CI pipeline result via APIs, which is a common practice in Continuous Integration testing.