Local Bitbucket Windows Runner Pipeline Test

bitbucket_pipelines_test.py

"""Local Bitbucket Windows Runner Pipeline Test

Run a bitbucket-pipelines.yml locally using Powershell as the interpreter.

This allows you to test a bitbucket-pipelines.yml file locally without pushing it to a repository.

Example:

        $ python bitbucket_pipelines_test.py "Pipeline step 1" --env env.json

Running with no parameters will run all steps in the pipeline.

Environment files are optional. They are useful for defining the environment variables BitBucket defines:

{
    "BITBUCKET_BUILD_NUMBER": "1234",
    "BITBUCKET_REPO_SLUG": "repository-name",
    "BITBUCKET_BRANCH": "branch-name"
}

"""


import argparse
import json
import os
import subprocess
import sys
import tempfile

import yaml

def get_args():
    parser = argparse.ArgumentParser(description='Run a Bitbucket Pipelines file')
    parser.add_argument('--version', action='version', version='%(prog)s {version}'.format(version="0.1.0"))
    parser.add_argument('--filename', type=str, default="bitbucket-pipelines.yml",
                                help='Bitbucket pipelines filename. Default is bitbucket-pipelines.yml.')
    parser.add_argument('--env', type=str, action='append',
                                help='Environment variables (.json). Can have multiple environment variable files: --env env1.json --env env2.json ...')
    parser.add_argument('--keep-script-on-error', type=bool, default=False,
                                help='Keeps step script on error which is useful for debugging powershell script errors.')
    parser.add_argument('steps', nargs='*', type=str,
                                help='Steps to run. All steps will be run if not specified.')

    args = parser.parse_args()

    return args

def execute(cmd):
    popen = subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True)
    for stdout_line in iter(popen.stdout.readline, ""):
        yield stdout_line 
    popen.stdout.close()
    return_code = popen.wait()
    if return_code:
        raise subprocess.CalledProcessError(return_code, cmd)

def run_step(step, keep_script_on_error=False):
    step_script_filename = tempfile.mktemp(suffix=".ps1", prefix=step["name"].replace(" ", "-"))

    try:    
        with open(step_script_filename, "w") as step_script:
            for line in step["script"]:
                print(line, file=step_script)

        for output in execute(["powershell.exe", step_script_filename]):
            print(output, end="")

    finally:
        if not keep_script_on_error:
            os.unlink(step_script_filename)

def main():
    args = get_args()

    if not args.env == None:
        for load_env in args.env:
            for k,v in json.load(open(load_env)).items():
                os.environ[k] = v

    bitbucket_pipelines_filename = args.filename
    step_names = args.steps
    load_env = args.env
   
    pipelines = yaml.safe_load(open(bitbucket_pipelines_filename))

    steps = pipelines["definitions"]["steps"]

    # Restructure steps so it's a dict instead of a list
    steps = {step["step"]["name"]:step["step"] for step in steps}

    if len(step_names) == 0:
        step_names = list(steps.keys())
        print("Running all steps: {}".format(", ".join(step_names)))

    for step_name in step_names:
        print("=" * 70)
        print('Running step "{}"'.format(step_name))
        print("=" * 70)

        if step_name not in steps:
            print('Error: step "{}" does not exist'.format(step_name))
            print("Valid steps: {}".format(", ".join(steps.keys())))
            sys.exit(1)

        run_step(steps[step_name])

if __name__ == '__main__':

    main()