Skip to content

Instantly share code, notes, and snippets.

@automaticgiant
Created May 14, 2024 16:28
Show Gist options
  • Save automaticgiant/e5ec2d492c226d9b772a4a7e88ae82d0 to your computer and use it in GitHub Desktop.
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 😆.
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="workflow-job@2.40">
<description></description>
<keepDependencies>false</keepDependencies>
<properties></properties>
<definition class="org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition" plugin="workflow-cps@2.92">
<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