#!/usr/bin/env python3
# Copyright (c) 2020 Software AG, Darmstadt, Germany and/or Software AG USA Inc., Reston, VA, USA, and/or its subsidiaries and/or its affiliates and/or their licensors.
# Use, reproduction, transfer, publication or disclosure is prohibited except as specifically provided for in your License Agreement with Software AG
""" Contains PySys extensions for projects: creating, manipulating and deploying.
"""
import os, shutil
[docs]class ProjectHelper:
"""Helper class for creating, manipulating and deploying Software AG Designer-compatible projects.
Creates an instance of the ProjectHelper class. You can have multiple projects in one test by overriding the parameters::
myProject=ProjectHelper(self,"my-project")
myProject=ProjectHelper(self,"my-project",projectRoot='my-root-path')
myProject=ProjectHelper(self,name="my-project",deployedPath='my-deployed-path')
:param testcase: The ProjectHelper should be initialized with an instance of a test, normally the one in which the helper is created.
:param name: the project name.
:param projectRoot: the directory that contains/will contain the project
:param deployedPath: the directory containing the deployed project, if this exists you must deploy with overwrite=True.
This class provides the ability to create, add and delete bundles, deploy and manipulate projects::
mycorrelator = CorrelatorHelper(self, name='mycorrelator')
myProject = ProjectHelper(self,name="my-project")
myProject.create()
myProject.addBundle("onAppInitialized")
myProject.addBundle("...")
#copy monitor files into project
...
myProject.deploy()
#start the deployed application
mycorrelator.start(config=myProject.deployedDir())
Additionally, you can use this command to use an existing project created with Software AG Designer or apama_project::
mycorrelator = CorrelatorHelper(self, name='mycorrelator')
myProject = ProjectHelper(self,name="my-project-from-template")
myProject.create(existingProject='path-to/my-project-template')
#further set-up as above
...
myProject.deploy()
#start the deployed application
mycorrelator.start(config=myProject.deployedDir())
"""
def __init__(self, testcase, name, projectRoot=None, deployedPath=None):
self.__projectName = name
self.__testcase = testcase
if projectRoot is None:
self.__projectRoot = self.__testcase.output
else:
self.__projectRoot = projectRoot
if deployedPath is None:
self.__deployedPath = os.path.join(self.__projectRoot, self.__projectName+'-deployed')
else:
self.__deployedPath = deployedPath
self.__projectCommand = self.__testcase.project.APAMA_HOME + '/bin/apama_project'
self.__deployCommand = self.__testcase.project.APAMA_HOME + '/bin/engine_deploy'
self.__projectDirectory = os.path.join(self.__projectRoot,self.__projectName)
self.__monitorsDirectory = os.path.join(self.__projectDirectory,"monitors")
self.__configDirectory = os.path.join(self.__projectDirectory,"config")
[docs] def create(self,existingProject=None,**kwargs):
"""Creates the base project in the output directory by default or projectRoot if set in the constructor::
myProject = ProjectHelper(self,name="my-project-from-template")
#create new
myProject.create()
#use existing, copies project into correct location name changed to the project name
myProject.create(existingProject='path-to/my-project-template')
:param existingProject: The path of an existing Designer or apama_project project.
:param kwargs: Allow named parameters, for example, timeout=XXX.
"""
result = None
if existingProject is None:
try:
result = self.__testcase.startProcess(self.__projectCommand, ["create",self.__projectName], workingDir=self.__projectRoot, stdouterr=self.__testcase.allocateUniqueStdOutErr('create_project'), displayName="create_"+self.__projectName,**kwargs)
except:
if result is not None:
self.__testcase.logFileContents(result.stdout)
raise
else:
shutil.copytree(existingProject, self.__projectDirectory)
return result
[docs] def addBundle(self,name,instanceName=None,**kwargs):
"""Adds a specific bundle to the project::
myProject = ProjectHelper(self,name="my-project")
...
myProject.addBundle('onAppInitialized')
myProject.addBundle('HTTP Server',instanceName='myHttpServerInstance')
Note that this command does not set any of the configuration. This must be done by the user in the test environment in the same way as a Designer or apama_project user needs to do this.
:param name: The bundlename to add to the project. apama_project list bundles can be used to get the current usable list.
:param instanceName: certain bundles can specify the instance name (optional).
:param kwargs: Allow named parameters, for example,timeout=XXX.
"""
# set the __projectCommand and display name
dispName = "addBundle_"+name
args = ["add","bundle",name]
if instanceName is not None:
args.append("--instance")
args.append("\""+instanceName+"\"")
dispName = dispName + "_" + instanceName
result = None
try:
result = self.__testcase.startProcess(self.__projectCommand, args, workingDir=self.__projectDirectory , stdouterr=self.__testcase.allocateUniqueStdOutErr('add_bundle'), displayName=dispName,**kwargs)
except:
if result is not None:
self.__testcase.logFileContents(result.stdout)
raise
return result
[docs] def removeBundle(self,name=None,instanceName=None,**kwargs):
"""Removes a specific bundle from the project::
myProject = ProjectHelper(self,name="my-project")
...
myProject.removeBundle('onAppInitialized')
#will remove ALL instances as well as the bundle
myProject.removeBundle(name ='HTTP Server')
#removes the instance but not the bundle unless it is the only instance
myProject.removeBundle(instanceName='myHttpServerInstance')
Only one of the parameters should be specified.
:param name: If this is specified, all instances of that bundle will be removed.
:param instanceName: If this is specified, the specific instance will be removed.
:param kwargs: Allow named parameters, for example,timeout=XXX.
"""
# set the __projectCommand and display name
dispName = "remove_bundle_"
args = ["remove","bundle"]
if instanceName is not None:
args.append(instanceName)
dispName += instanceName
else:
args.append(name)
dispName += name
result = None
try:
result = self.__testcase.startProcess(self.__projectCommand, args, workingDir=self.__projectDirectory , stdouterr=self.__testcase.allocateUniqueStdOutErr('remove_bundle'), displayName=dispName,**kwargs)
except:
if result is not None:
self.__testcase.logFileContents(result.stdout)
raise
return
[docs] def deploy(self,overwrite=False,dest=None,**kwargs):
"""Deploys a runnable project.
Creates a <project name>-deployed directory in the output directory by default or projectRoot if set in the constructor. If you try to deploy to an existing directory, it will fail by default. Use overwrite=True to enable it::
myProject = ProjectHelper(self,name="my-project")
...
#create runnable application, add 120s timeout to deployment via kwargs
myProject.deploy(timeout=120)
#create runnable application, allow redeployment to the same name
myProject.deploy(overwrite=true)
#create runnable application, set/override the deployment directory
myProject.deploy(dest='path-to/my-override-dir',timeout=120)
:param dest: Override the destination directory for the deployed project.
:param overwrite: Remove an existing deployment with the same path (default is False).
:param kwargs: Allow named parameters, for example, timeout=XXX.
"""
if dest is not None:
self.__deployedPath = dest
# set the deploy_command and display name
args = ["--outputDeployDir",self.__deployedPath, self.__projectDirectory]
dispName = "deploy_" + self.__projectName
result = None
try:
if os.path.exists(self.__deployedPath):
if overwrite:
shutil.rmtree(self.__deployedPath)
else:
raise Exception("Cannot overwrite existing deployment '"+self.__deployedPath+"' use deploy(overwrite=True) to allow" )
result = self.__testcase.startProcess(self.__deployCommand, args, workingDir=self.__testcase.output , stdouterr=self.__testcase.allocateUniqueStdOutErr('deploy'), displayName=dispName,**kwargs)
except:
if result is not None:
self.__testcase.logFileContents(result.stderr)
raise
return
[docs] def deployedDir(self):
"""Retrieves the deployed directory.
If the project has not been deployed then the command will return the path to <projectname>-deployed.
If the deployed project was overridden at deployment then this will return that path.
"""
return self.__deployedPath
[docs] def projectDir(self):
"""Returns the path to the project.
"""
return self.__projectDirectory
[docs] def monitorsDir(self):
"""Returns the directory path to the project monitors.
"""
return self.__monitorsDirectory
[docs] def configDir(self):
"""Returns the directory path of the project configuration.
"""
return self.__configDirectory