wc-test

#!/usr/bin/env python3

import argparse
import json
import os
import random
import re
import shlex
import subprocess
import textwrap
import time
import urllib.parse

CI_CONFIG = ".ciconfig"

def read_config(path):
  if not os.path.isfile(path):
    path = os.path.expanduser(os.path.join('~', path))
    if not os.path.isfile(path):
      return {}
  with open(path) as file:
    config = json.load(file)
    return config

def run_command(command):
  status, output = subprocess.getstatusoutput(command)
  if status:
    raise SystemExit(command + '\n' + output)
  return output

def get_queued_build(build_url, params):
  while True:
    response = run_command("curl --http1.1 -fs '{0}api/json' -u $(whoami):{apiToken}"
      .format(build_url, **params))
    build = json.loads(response)
    if ("executable" in build) or build["blocked"] or build["buildable"]:
      return build
    time.sleep(1)

class HelpFormatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter):
  pass

if __name__ == "__main__":
  parser = argparse.ArgumentParser(
    formatter_class=HelpFormatter,
    argument_default=argparse.SUPPRESS,
    description=textwrap.dedent(
      """
      Test current working copy on Jenkins.

      Please follow instructions from the link below to generate an API token:
      https://jenkins.io/doc/book/system-administration/authenticating-scripted-clients

      Example ~/{config}:
      {{
        \"url\": \"https://venice-ci.stg.linkedin.com:8443\",
        \"apiToken\": \"**********************************\"
      }}
      """.format(config=CI_CONFIG)))

  parser.add_argument("-j", "--job", help="jenkins job to run")
  parser.add_argument("-b", "--baseline", help="baseline branch/tag to use for testing")
  parser.add_argument("-m", "--message", help="optional description of the changes")
  parser.add_argument("patch", nargs='?', help="diff-formatted patch to submit for testing")

  params = dict(job="Venice-WcTest", baseline="upstream/main")
  params.update(read_config(CI_CONFIG))
  parser.set_defaults(**params)
  params.update(vars(parser.parse_args()))

  if isinstance(params["url"], list):
    params["url"] = random.choice(params["url"])

  if "message" not in params:
    params["message"] = "" if "patch" in params else run_command("git log --format=%s -1 HEAD")

  command = "curl --http1.1 -fv '{url}/job/{job}/buildWithParameters?{0}' -u $(whoami):{apiToken} -F patch.diff=@" \
    .format(urllib.parse.urlencode(dict(baseline=params["baseline"], message=params["message"])), **params)

  if "patch" in params:
    response = run_command(command + shlex.quote(params["patch"]))
  else:
    run_command("git fetch upstream")
    changes = run_command("git diff --find-renames --compact-summary " + params["baseline"])
    if not changes:
      raise SystemExit("No changes")

    revision = '#' + run_command("git rev-parse --short HEAD")
    response = run_command("git diff --binary --find-renames --compact-summary {0} | {1}"
      .format(params["baseline"], command + shlex.quote("-;filename=" + revision)))

    print("Submitted working copy {0} for testing:".format(revision))
    print(changes)
    print()

  build_url = re.findall("< Location: (https?://.+)", response)[0]
  build = get_queued_build(build_url, params)
  if build.get("cancelled"):
    print("Build cancelled")
  elif build["blocked"] or build["buildable"]:
    print(build["task"]["url"])
  else:
    print(build["executable"]["url"])