Created
May 14, 2024 16:28
-
-
Save automaticgiant/e5ec2d492c226d9b772a4a7e88ae82d0 to your computer and use it in GitHub Desktop.
as close to a jenkinsfile repl as you can get. it's just a chunky repl 😆.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
import time | |
import json | |
import os | |
# Jenkins details | |
jenkins_url = "REDACTED" | |
job_name = "test" | |
user_name = os.getenv('JENKINS_USER_NAME') | |
api_token = os.getenv('JENKINS_API_TOKEN') | |
crumb_issuer_url = f"{jenkins_url}/crumbIssuer/api/xml?xpath=concat(//crumbRequestField,\":\",//crumb)" | |
# Get Jenkins crumb | |
response = requests.get(crumb_issuer_url, auth=(user_name, api_token)) | |
crumb_field, crumb = response.text.split(':') | |
# Read Jenkinsfile contents | |
with open('Jenkinsfile2', 'r') as file: | |
jenkinsfile_contents = file.read() | |
# XML job configuration template. Insert Jenkinsfile contents. | |
config_xml = f'''<?xml version='1.1' encoding='UTF-8'?> | |
<flow-definition plugin="[email protected]"> | |
<description></description> | |
<keepDependencies>false</keepDependencies> | |
<properties></properties> | |
<definition class="org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition" plugin="[email protected]"> | |
<script>{jenkinsfile_contents}</script> | |
<sandbox>true</sandbox> | |
</definition> | |
<triggers></triggers> | |
</flow-definition>''' | |
# Create job (try silently delete conflicting first) | |
delete_job_url = f"{jenkins_url}/job/{job_name}/doDelete" | |
create_job_url = f"{jenkins_url}/createItem?name={job_name}" | |
headers = { | |
'Content-Type': 'application/xml', | |
crumb_field: crumb | |
} | |
response = requests.post(delete_job_url, headers={crumb_field: crumb}, auth=(user_name, api_token)) | |
response = requests.post(create_job_url, headers=headers, data=config_xml, auth=(user_name, api_token)) | |
if response.status_code == 200: | |
print(f"Job '{job_name}' created successfully.") | |
else: | |
print(f"Failed to create job. Status code: {response.status_code}") | |
exit() | |
# Build job | |
build_job_url = f"{jenkins_url}/job/{job_name}/build" | |
response = requests.post(build_job_url, headers={crumb_field: crumb}, auth=(user_name, api_token)) | |
if response.status_code == 201: | |
print(f"Job '{job_name}' build triggered.") | |
else: | |
print(f"Failed to trigger job build. Status code: {response.status_code}") | |
exit() | |
# Poll for job start | |
while True: | |
build_info_url = f"{jenkins_url}/job/{job_name}/lastBuild/api/json" | |
response = requests.get(build_info_url, auth=(user_name, api_token)) | |
if response.text: # Check if the response is not empty | |
try: | |
build_info = response.json() | |
# Break if the build is building or has finished | |
if build_info['building'] or build_info['result'] is not None: | |
break | |
except json.decoder.JSONDecodeError: | |
pass # Ignore JSON decoding errors | |
time.sleep(1) | |
# Stream console output | |
console_output_url = f"{jenkins_url}/job/{job_name}/lastBuild/logText/progressiveText" | |
build_info_url = f"{jenkins_url}/job/{job_name}/lastBuild/api/json" | |
start = 0 | |
while True: | |
headers = {'start': str(start)} | |
response = requests.post(console_output_url, headers=headers, auth=(user_name, api_token)) | |
if response.status_code == 200: | |
print(response.text, end='') | |
start += len(response.text) | |
elif response.status_code == 404: | |
print("Build not started yet. Waiting...") | |
else: | |
print("Failed to get build console output.") | |
break | |
# Check if the build has finished | |
response = requests.get(build_info_url, auth=(user_name, api_token)) | |
build_info = response.json() | |
if build_info['result'] is not None: | |
break # Stop streaming the console output | |
time.sleep(1) | |
# Get build info | |
build_info_url = f"{jenkins_url}/job/{job_name}/lastBuild/api/json" | |
response = requests.get(build_info_url, auth=(user_name, api_token)) | |
build_info = response.json() | |
# Delete job if it failed | |
if build_info['result'] == 'SUCCESS': | |
delete_job_url = f"{jenkins_url}/job/{job_name}/doDelete" | |
response = requests.post(delete_job_url, headers={crumb_field: crumb}, auth=(user_name, api_token)) | |
if response.status_code == 200: | |
print(f"Job '{job_name}' deleted successfully.") | |
else: | |
print(f"Failed to delete job. Status code: {response.status_code}") | |
else: | |
print(f"Job '{job_name}' not completed successfully. You can view the job at {jenkins_url}/job/{job_name}") | |
print(f"To delete the job, visit {jenkins_url}/job/{job_name}/doDelete") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment