"""Send Webhooks
This module is used to send a JSON payload to the URL specified in the config.
Run the following for help:
python send_webhooks.py --help
"""
import argparse
import json
import os
import sys
from collections import namedtuple
import requests
from P4 import P4, P4Exception
ARG_PARSER = argparse.ArgumentParser(description='Send Webhooks.')
ARG_PARSER.add_argument('-cf', '--config-file',
help='Configuration file for the script.',
required=True, action='store')
ARG_PARSER.add_argument('-cl', '--changelist',
help='Number of the changelist submitted.',
required=True, action='store')
ARG_PARSER.add_argument('-p', '--project',
help='Project name and key for config file section.',
required=True, action='store')
ARG_PARSER.add_argument('-tp', '--trigger-path',
help='For a change-commit trigger this is the third field in trigger definition. The path for which the trigger is expected to match.',
required=False, action='store')
ARG_PARSER.add_argument('-u', '--user',
help='The user who submitted the change.',
required=True, action='store')
ARG_PARSER.add_argument('-d', '--debug',
help='Turns on debug output.',
required=False, action='store_true')
def load_config(config_file: str) -> dict:
"""Loads the json configuration file.
Args:
config_file: Path to the configuration file.
Returns:
A dictionary object with the config file information, or empty if there
was an error.
"""
if os.path.exists(config_file):
with open(config_file, 'r') as config:
return json.load(config)
return {}
# load_config -----------------------------------------------------------------
def get_p4_result_by_key(p4_result: list, key: str) -> namedtuple:
"""Searches through the p4 command result for a value based on the key provided.
Args:
p4_result: The result from running a P4 command.
key: The key of the key value pair you want to find.
"""
P4Result = namedtuple('P4Result', 'is_valid value')
# Change specs are returned as a plain dict not a dict(s) in a list
# Hooray for consistency! ... :(
if isinstance(p4_result, dict):
if key in p4_result:
return P4Result(True, p4_result[key])
for result_set in p4_result:
if key in result_set:
return P4Result(True, result_set[key])
return P4Result(False, None)
# get_p4_result_by_key --------------------------------------------------------
def generate_payload(target: str, p4_params: namedtuple) -> dict:
"""Creates the payload for the given target.
Args:
target: The target of the payload.
p4_params: Params given from Helix.
Returns:
A dictionary object with the payload, or empty if there is no
implementation for the given target.
"""
if target == 'jenkins':
# Jenkins is weird and we are not actually returning a dict object...
# We are returning a string and
payload = "payload={'change':%s,'p4port':'helix.etcconnect.com:1666'}" % p4_params.change_num
elif target == 'teams':
payload = {
"@type": "MessageCard",
"@context": "http://schema.org/extensions",
"summary": f"{p4_params.username} submitted a change",
"sections": [
{
"activityTitle": f"{p4_params.username} submitted change {p4_params.change_num}",
"activitySubtitle": f"{p4_params.trigger_path}",
"activityImage": "https://swarm.workshop.perforce.com/view/guest/perforce_software/slack/main/images/60x60-Helix-Bee.png",
"text": p4_params.desc,
"markdown": False
} ],
"potentialAction": [
{
"@type": "OpenUri",
"name": "View Changelist in Swarm",
"targets": [
{
"os": "default",
"uri": f"http://swarm.etcconnect.com/changes/{p4_params.change_num}"
} ]
} ]
}
else:
payload = {}
return payload
# generate_payload ------------------------------------------------------------
def main(args: namedtuple):
"""Sends the post request to the provided url.
Args:
args: Args given via CLI
"""
config_file = load_config(args.config_file)
if not config_file:
if args.debug:
print(f'SEND_WEBHOOKS [ERROR]: Could not read config ({args.config_file}) Aborting...')
sys.exit(0) # Don't think we want Helix to think there was an error.
if args.project not in config_file:
if args.debug:
print(f'SEND_WEBHOOKS [ERROR]: Project ({args.project}) not in config file. Aborting...')
sys.exit(0) # Don't think we want Helix to think there was an error.
# Grab these now so we don't need to mess with accessing...
config = config_file['config']
excludes = config_file[args.project]['excludes']
targets = config_file[args.project]['targets']
if args.debug:
print(f'SEND_WEBHOOKS [DEBUG]: Excludes: {excludes}')
print(f'SEND_WEBHOOKS [DEBUG]: Targets: {targets}')
# Flags for later use
is_excluded = False
is_merge_or_copy = False
p4 = P4(user=config['user'], port=config['port'])
secret_file = config['pass_file']
if os.path.exists(secret_file):
with open(secret_file) as f:
secret_key = f.read().strip()
else:
if args.debug:
print(f'SEND_WEBHOOKS [ERROR]: Could not find key for login. Aborting...')
sys.exit(0)
p4.password = secret_key
try: # All p4 commands are run in this try block
p4.connect()
p4.run_login()
# Get changelist information
submitted_change_spec = p4.run('describe', '-s', args.changelist)
p4_desc_result = get_p4_result_by_key(submitted_change_spec, 'desc')
p4_action_result = get_p4_result_by_key(submitted_change_spec, 'action')
description = ''
if p4_desc_result.is_valid:
description = p4_desc_result.value
# Remove any quotes that may mess up the JSON and replace with backticks
description = description.replace('\"', '"')
description = description.replace('\'', ''')
description = description.replace('\n', '<br>')
# Check to see if the change is a merge/integration
if p4_action_result.is_valid:
if 'integrate' in p4_action_result.value:
is_merge_or_copy = True
if args.debug:
print(f'SEND_WEBHOOKS [DEBUG]: Change Spec: {submitted_change_spec}')
print(f'SEND_WEBHOOKS [DEBUG]: Description: {description}')
# Make a list of the last five changes on each of the exclude streams
# and then check if the given changelist is in the list.
if not is_merge_or_copy:
changes_in_excluded_paths = []
for path in excludes:
changes = p4.run('changes', '-m', config['change_limit'], path)
for change in changes:
p4_result = get_p4_result_by_key(change, 'change')
if p4_result.is_valid:
changes_in_excluded_paths.append(p4_result.value)
if args.debug:
print(f'SEND_WEBHOOKS [DEBUG]: Changes from excluded paths: {changes_in_excluded_paths}')
if args.changelist in changes_in_excluded_paths:
is_excluded = True
# Get full name from the username.
p4_user_name = args.user # default to username if the other part fails
p4_user_spec = p4.run('user', '-o', args.user)
p4_user_result = get_p4_result_by_key(p4_user_spec, 'FullName')
if p4_user_result.is_valid:
p4_user_name = p4_user_result.value
except P4Exception as exception: # Don't want to throw ANY p4exceptions
if args.debug:
print(f'SEND_WEBHOOKS [WARN]: P4Exception occurred {exception}')
sys.exit(0)
# Creating a named tuple to pass in the args given from Helix so as we
# add more (if we do...) the generate_payload function doesn't have to
# have more and more params added to it. Unless we want to do that?
P4Params = namedtuple('P4Params', 'change_num username trigger_path desc')
p4_params = P4Params(change_num=args.changelist,
username=p4_user_name,
trigger_path=args.trigger_path,
desc=description)
for target in targets:
# Bail if the url is blank
if not targets[target]:
if args.debug:
print(f'SEND_WEBHOOKS [DEBUG]: Skipping {target} due to blank URL...')
continue
# skip this target if it is in an excluded path or if it is a merge/copy
if target == 'teams' and (is_excluded or is_merge_or_copy):
if args.debug:
print(f'SEND_WEBHOOKS [DEBUG]: Skipping {target}...')
print(f'SEND_WEBHOOKS [DEBUG]: ... Is Excluded? {is_excluded} -- Is Merge/Copy? {is_merge_or_copy}')
continue
payload = generate_payload(target, p4_params)
if args.debug:
print(f'SEND_WEBHOOKS [DEBUG]: Generated Payload: {payload}')
# Doing this check is a safety net. If a new type of target is added to
# the config file but the implementation for generating its payload has
# not been completed, then we will get an empty dict which resolves as
# False.
if not payload:
if args.debug:
print(f'SEND_WEBHOOKS [ERROR]: Payload not generated for {target}')
continue
# Now is time to do the actual sending of the payload to the target.
try:
headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
if target == 'jenkins':
request = requests.post(targets[target], data=payload)
else:
request = requests.post(targets[target], data=json.dumps(payload), headers=headers)
if args.debug:
print(f'SEND_WEBHOOKS [DEBUG]: Webhook returned status code: {request.status_code}')
print(f'SEND_WEBHOOKS [DEBUG]: Reason for code: {request.reason}')
except Exception as exception: # Don't want to throw ANY exceptions
if args.debug:
print(f'SEND_WEBHOOKS [WARN]: Exception occurred while sending webhook ({target} -> {targets[target]}).')
print(f'SEND_WEBHOOKS [WARN]: {exception}')
if p4.connected():
p4.disconnect()
if __name__ == '__main__':
GIVEN_ARGS = ARG_PARSER.parse_args()
main(GIVEN_ARGS)