Jenkins is a very familiar tool today. We use it for CI/CD workflow everyday at work.
I usually just often need to configure Jenkins for triggering build in cron or when there is new changes happended in source code in gitlab or github.
But there was a time, I was assigned a task to validate the response from an API whenever we detect the response changes.
We do not know when the response will change. Setting up the jenkins job to check, let's say every 1 hour is very spamming the builds.
That's when I needed to make some research and found out about Jenkins API.
Below is the illustration and we will be implemented in Python.
The Jenkins API wrapper I'm using is Python-Jenkins
You can refer to python-jenkins for details
I.Remote access API in Jenkins
For more details , you can visit this page in jenkins-wiki
II.Some simple Jenkins API
In our current project, the jenkins server is like: http://01.0.00.206:8080
If you want to get the general of jenkins information, you can call to http://01.0.00.206:8080/api/json by using get method
Or if you want to get the job information: http://01.0.00.206:8080/job/API_Test/job/CocCoc_Music
III.Real world problem using Jenkins API
We actually had a task to check the api for getting playlists item.
We need to validate the object model of the response, whether it contains certain fields, or whether the field value is in correct format (regex).
This validation need to trigger whenever there is new playlists from the api.
Unfortunately, we don't know when there are new playlists, so we had to check based on the difference of the response text compared to the previous text.
→ If we just setup the jobs in Jenkins server → there will be a lot of spam builds (for example we trigger the job every 2 hours).
→ This time we can use the remote api of Jenkins.
We setup to trigger the job via the API whenever the json response is changed.
What we would need to do is.
1.Call the getting playlists api to retrieve the response text
class Playlist:
def get_all_playlist(self):
print(f'URL is : {CocCocMusicApiConfigs.COCCOC_MUSIC_API_DOMAIN}{CocCocMusicApiConfigs.COCCOC_MUSIC_API_PATH_LISTING}')
return requests.get(CocCocMusicApiConfigs.COCCOC_MUSIC_API_DOMAIN
+ CocCocMusicApiConfigs.COCCOC_MUSIC_API_PATH_LISTING,
headers={'Content-Type': 'application/json'})
2.Compare the texts before and after
Read the text_file:
def read_text_file(self, file):
import codecs
with codecs.open(file, 'r', 'utf-8') as f:
return f.read()
Compare the text:
response_txt != read_text_stored_file()
3.If the text is changed, trigger the job and override the text in the file
def do_the_job_check():
response_txt = get_music_playlist_lists()
if response_txt != read_text_stored_file():
print("Response API listing is CHANGED")
trigger_api_coccoc_music_job()
write_text_response_to_txt(response_txt)
else:
print("Nothing changes")
4.Validate the response model
class PlaylistSchema(Schema):
song_schema = SongSchema()
id = fields.Str(required=True, allow_none=False, validate=validate.Regexp('.+-playlist-+.*'))
type = fields.Str(required=True, validate=validate.OneOf(['1', '2']), allow_none=False)
genre = fields.Str(required=True, allow_none=True)
name = fields.Str(required=True, allow_none=False)
artists = fields.Str(required=True, allow_none=True)
image = fields.Str(required=True, allow_none=True)
items = fields.List(fields.Nested(song_schema), required=True, allow_none=False)
@validates('name')
def validate_name(self, value):
if value == '':
raise ValidationError('Name should not be blank')
elif value == 'null':
raise ValidationError('Name: Why are you doing like this, convert to string "null" =))))')
@validates('genre')
def validate_genre(self, value):
if value == '':
raise ValidationError('Genre should not be blank')
elif value == 'null':
raise ValidationError('Genre: Why are you doing like this, convert to string "null" =))))')
@validates('artists')
def validate_artists(self, value):
if value == 'null':
raise ValidationError('Artists: Why are you doing like this, convert to string "null" =))))')
@validates('image')
def validate_image(self, value):
if value == 'null':
raise ValidationError('Image: Why are you doing like this, convert to string "null" =))))')
5.After validation, send comments to Jira, send message to Skype
@pytest.mark.hookwrapper
def pytest_runtest_makereport():
global total_test_failed
outcome = yield
report = outcome.get_result()
if report.when == 'call' and report.failed:
total_test_failed += 1
@pytest.fixture(scope='session', autouse=True)
def update_result_jira_skype():
yield
jira_utils.add_comment(jira_issue, comment_for_jira_skype(total_test_failed))
skype_utils.send_message_group_skype(SkypeGroupIds.TEST_GROUP_ID, comment_for_jira_skype(total_test_failed))
def comment_for_jira_skype(total_failed):
general_comment = "\nThis is result for checking CocCoc Music API to return list of playlists"
if total_failed > 0:
failed_comment = f"\nTest result is FAILED\nTotal test failed is :{total_failed}" \
f"\nPlease check test run with ID=2358 in test rail" \
f"and jenkins job for api test for details"
return general_comment + failed_comment
elif total_failed == 0:
success_comment = "\nTest result is SUCCESS"
return general_comment + success_comment
Notes: If you feel this blog help you and want to show the appreciation, feel free to drop by :
This will help me to contributing more valued contents.
Top comments (0)