Code Samples : Python Code
  
Python Code
The Zerto Orchestrator uses Python code to enable automatically scheduled failover tests. The following code is available in full by downloading and deploying the Zerto Orchestrator in the /Zerto directory.
The following code uses the use the Requests library for making the REST calls: http://docs.python-requests.org/en/latest/.
import logging
import os.path
import sys
import subprocess
import os
import datetime
import time
import requests
import json
import base64
def GenerateZertoRestSession_WindowsAuthentication(zvmIp, user, password):
  credStr = user + ":" + password
  encodedCredStr = "Basic " + base64.b64encode(credStr)
  print encodedCredStr
 
  headers = {'Authorization': encodedCredStr}
  url = "https://" + zvmIp + ":9669/v1/session/add"
  r = requests.post(url, headers=headers, verify=False)
  print r.status_code
  if(r.status_code != requests.codes.ok):
    raise Exception("Failed authenticating to the Zerto Virtual Manager")
  res = r.headers.get('x-zerto-session')
  print "session = " + res
  return res
def GenerateZertoRestSession_VcenterAuthentication(zvmIp, user, password):
  credStr = user + ":" + password
  encodedCredStr = "Basic " + base64.b64encode(credStr)
  print encodedCredStr
 
  payload = {"AuthenticationMethod": 1}
  dataval = json.dumps(payload)
  print (dataval)
 
  headers = {'Authorization': encodedCredStr, 'content-type': 'application/json'}
  url = "https://" + zvmIp + ":9669/v1/session/add"
  r = requests.post(url, data=dataval, headers=headers, verify=False)
  print r.status_code
  if(r.status_code != requests.codes.ok):
  raise Exception("Failed authenticating to the Zerto Virtual Manager")
  res = r.headers.get('x-zerto-session')
  print "session = " + res
  return res
def GetVpgIdFromName(zvmIp, sessionId, vpgName):
  url = "https://" + zvmIp + ":9669/v1/vpgs?name=" + vpgName
  headers = {'x-zerto-session': sessionId, 'content-type': 'application/json'}
  r = requests.get(url, headers=headers, verify=False)
  if(r.status_code != requests.codes.ok):
    raise Exception("Call to ZVM failed")
  print r.status_code
  print r.text
  vpgData = json.loads(r.text)
  if(len(vpgData) == 0):
    raise Exception("VPG by name " + vpgName + " not found")
  if(len(vpgData) > 1):
  raise Exception("multiple vpgs named " + vpgName + " found")
  thisVpg = vpgData[0]
  res = thisVpg["VpgIdentifier"]
  print thisVpg
  print res
  return res
def GetLatestCheckpointForVpg(zvmIp, sessionId, vpgId):
  url = "https://" + zvmIp + ":9669/v1/vpgs/" + vpgId + "/checkpoints"
  headers = {'x-zerto-session': sessionId, 'content-type': 'application/json'}
  r = requests.get(url, headers=headers, verify=False)
  print r.status_code
  if(r.status_code != requests.codes.ok):
    raise Exception("Failed GetLatestCheckpointForVpg")
  allcpData = json.loads(r.text)
  print str(len(allcpData))
  if(len(allcpData) == 0):
    raise Exception("no checkpoints found for vpg " + vpgId)
  sortedCps = sorted(allcpData, key=lambda x: x["TimeStamp"], reverse=True)
  latestCp = sortedCps[0]
  print latestCp
  return latestCp["CheckpointIdentifier"]
def StartFailoverTest(zvmIp, sessionId, vpgId, cpId):
  url = "https://" + zvmIp + ":9669/v1/vpgs/" + vpgId + "/failovertest"
  payload = {"CheckpointId": cpId}
  dataval = json.dumps(payload)
  headers = {'x-zerto-session': sessionId, 'content-type': 'application/json'}
  r = requests.post(url, data=dataval, headers=headers, verify=False)
  print r.status_code
  if(r.status_code != requests.codes.ok):
    raise Exception("Failed StartFailoverTest")
  taskId = json.loads(r.text)
  print "task id = " + taskId
  return taskId
def StopFailoverTest(zvmIp, sessionId, vpgId, status):
  url = "https://" + zvmIp + ":9669/v1/vpgs/" + vpgId + "/failoverteststop"
  payload = {"FailoverTestSuccess": status, "FailoverTestSummary":"Automatically"}
  dataval = json.dumps(payload)
  headers = {'x-zerto-session': sessionId, 'content-type': 'application/json'}
  r = requests.post(url, data=dataval, headers=headers, verify=False)
  print r.status_code
  if(r.status_code != requests.codes.ok):
    raise Exception("Failed StopFailoverTest")
  taskId = json.loads(r.text)
  print "task id = " + taskId
  return taskId
def GetVmNamesForVpg(zvmIp, sessionId, vpgName):
  url = "https://" + zvmIp + ":9669/v1/vms?vpgName=" + vpgName
  headers = {'x-zerto-session': sessionId, 'content-type': 'application/json'}
  r = requests.get(url, headers=headers, verify=False)
  print r.status_code
  if(r.status_code != requests.codes.ok):
    raise Exception("Failed GetVmNamesForVpg")
  allvms = json.loads(r.text)
  print allvms
  res = []
  for vm in allvms:
    print "cur"
    print vm
    res.append(vm["VmName"])
  print res
  return res
def IsTaskComplete(zvmIp, sessionId, stopFotTaskId):
  url = "https://" + zvmIp + ":9669/v1/tasks/" + stopFotTaskId
  headers = {'x-zerto-session': sessionId, 'content-type': 'application/json'}
  r = requests.get(url, headers=headers, verify=False)
  print r.status_code
  if(r.status_code != requests.codes.ok):
    raise Exception("Failed getting tasks")
  tasksRes = json.loads(r.text)
  print tasksRes
  taskStatus = tasksRes["Status"]
  state = taskStatus["State"]
  print state
  if(state == 4 or state == 5):
    raise Exception("task failed")
  res = state == 6
  print res
  return res