Last active
December 31, 2015 12:59
-
-
Save kostajh/7989792 to your computer and use it in GitHub Desktop.
Call with something like `python3 setup.py --username ACQUIA_CLOUD_USERNAME --password ACQUIA_CLOUD_PASSWORD --branch GIT_BRANCH`
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
import subprocess | |
import json | |
import time | |
import sys | |
import argparse | |
baseurl = 'https://cloudapi.acquia.com/v1/sites/devcloud:MYSITE' | |
parser = argparse.ArgumentParser(description='Get username and password.') | |
parser.add_argument('--username', type=str, help='Acquia Cloud username') | |
parser.add_argument('--password', type=str, help='Acquia Cloud password') | |
parser.add_argument('--branch', type=str, help='The git branch to test') | |
args = vars(parser.parse_args()) | |
username = args['username'] | |
password = args['password'] | |
git_branch = args['branch'].replace('origin/', '') | |
def acquiaCloudApi(url, requestType): | |
""" Send a GET or POST request to Acquia Cloud API. """ | |
print('API call: %s' % url) | |
if (requestType == 'POST'): | |
subp = subprocess.Popen(['curl', '-s', '-u', '%s:%s' % (username, | |
password), '-X', 'POST', url], | |
stdout=subprocess.PIPE, stderr=subprocess.PIPE) | |
else: | |
subp = subprocess.Popen(['curl', '-s', '-u', '%s:%s' % (username, | |
password), url], stdout=subprocess.PIPE, | |
stderr=subprocess.PIPE) | |
curlstdout, curlstderr = subp.communicate() | |
output = curlstdout.decode('utf-8') | |
error = curlstderr.decode('utf-8') | |
try: | |
output_check = json.loads(output) | |
if 'message' in output_check: | |
if output_check['message'] == 'Resource not found': | |
print('API call %s failed' % url) | |
sys.exit(1) | |
except Exception: | |
print('Failed to decode JSON') | |
sys.exit(1) | |
return output, error | |
def waitForActiveTasks(): | |
""" Waits for active tasks to complete. """ | |
def checkTaskCompleted(task_id): | |
""" Checks if a task is completed, returns True or False. """ | |
curlstdout, curlstderr = acquiaCloudApi('%s/tasks/%d.json' | |
% (baseurl, task_id), 'GET') | |
print('Checking to see if task %d is done' % task_id) | |
try: | |
data = json.loads(curlstdout) | |
if data['state'] == 'done': | |
return True | |
elif data['state'] == 'error': | |
print('Task %d failed!' % task_id) | |
sys.exit(1) | |
else: | |
return False | |
except Exception: | |
print('Failed to parse JSON string, exiting.') | |
sys.exit(1) | |
def getActiveTasks(): | |
""" Returns a list of active task IDs. """ | |
curlstdout, curlstderr = acquiaCloudApi('%s/tasks.json' | |
% baseurl, 'GET') | |
try: | |
tasks = json.loads(curlstdout) | |
active_tasks = list() | |
# Build a list of active task IDs. | |
for task in tasks: | |
if task['state'] == 'started': | |
active_tasks.append(int(task['id'])) | |
return active_tasks | |
except Exception: | |
print(Exception) | |
print('Error!') | |
sys.exit(1) | |
# Start of waitForActiveTasks() | |
active_tasks = getActiveTasks() | |
if len(active_tasks) == 0: | |
print('No active tasks, continuing.') | |
return | |
print('Found the following active tasks: ', active_tasks) | |
i = 0 | |
for active_task_id in active_tasks: | |
while checkTaskCompleted(active_task_id) is False: | |
time.sleep(5) | |
i += 1 | |
if (i > 48): | |
print('Task %d did not complete within 240 seconds!' | |
% active_task_id) | |
sys.exit(1) | |
break | |
# Script start. | |
# If we have active tasks, wait until all of them are completed. | |
print('Setting up Acquia Dev Cloud for the tests.') | |
waitForActiveTasks() | |
# Initiate code deploy of branch | |
acquiaCloudApi('%s/envs/test/code-deploy.json?path=%s' | |
% (baseurl, git_branch), 'POST') | |
print('Initiated code deploy of branch %s to @test' % git_branch) | |
# Copy files from Dev to Test | |
acquiaCloudApi('%s/files-copy/dev/test.json' % baseurl, 'POST') | |
print('Initiated copy of files from @dev to @test') | |
# Copy the DB from Dev to Test | |
acquiaCloudApi('%s/dbs/MYDATABASE/db-copy/dev/test.json' % baseurl, 'POST') | |
print('Initiated copy of DB from @dev to @test') | |
print('Waiting for tasks to finish before proceeding') | |
time.sleep(5) | |
waitForActiveTasks() | |
print('All active tasks have been completed') | |
sys.exit(0) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment