#!/usr/bin/env python3
# Copyright (c) 2020-2021 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 Apama projects, compatible with Software AG Designer.
You can have multiple projects in one test by overriding the parameters::
myProject=self.apama.projectHelper("my-project")
myProject=self.apama.projectHelper("my-project", projectRoot='my-root-path')
myProject=self.apama.projectHelper("my-project", deployedPath='my-deployed-path')
This class provides the ability to create, add and delete bundles, deploy and manipulate projects::
myProject = self.apama.projectHelper("my-project")
myProject.create()
myProject.addBundle("onAppInitialized")
myProject.addBundle("...")
#copy monitor files into project
...
myProject.deploy()
# start the deployed application
myCorrelator = self.apama.startCorrelator('myCorrelator', config=myProject.deployedDir())
Additionally, you can use this command to use an existing project created with Software AG Designer or apama_project::
myProject = self.apama.projectHelper("my-project-from-template")
myProject.create(existingProject='path-to/my-project-template')
#further set-up as above
...
myProject.deploy()
# start the deployed application
myCorrelator = self.apama.startCorrelator('myCorrelator', config=myProject.deployedDir())
:param BaseTest testcase: The ProjectHelper should be initialized with an instance of a test, normally the one in which the helper is created.
:param str name: the project name.
:param str projectRoot: the directory that contains/will contain the project
:param str deployedPath: the directory containing the deployed project, if this exists you must deploy with overwrite=True.
"""
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: Optional L{pysys.process.user.ProcessUser.startProcess} keyword arguments, e.g. timeout, ignoreExitStatus.
"""
result = None
if existingProject is None:
try:
result = self.__testcase.startProcess(self.__projectCommand, ["create",self.__projectName], environs=dict(os.environ), 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,nameList,instanceName=None,**kwargs):
"""Adds a list of bundles or a bundle with instance name to the project::
myProject = ProjectHelper(self, name="my-project")
...
myProject.addBundle(['onAppInitialized'])
myProject.addBundle(['File Adapter', 'Any Extractor', 'Cumulocity IoT > Cumulocity Client'])
myProject.addBundle(['HTTP Server'], instanceName='myHttpServerInstance')
It is most efficient to add all bundles (except those with an instance name) in a single call to addBundle.
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 list[str] nameList: The name of the bundle(s) to add to the project, as a string or a list of strings.
``apama_project list bundles`` can be used to get the current usable list.
:param str instanceName: Certain bundles can specify the instance name (optional). Bundles requiring an
an instance name must be added one at a time.
:param kwargs: Optional L{pysys.process.user.ProcessUser.startProcess} keyword arguments, e.g. timeout, ignoreExitStatus.
"""
# set the __projectCommand and display name
if not isinstance(nameList, list): nameList = [nameList]
nameList = [str(x) for x in nameList] # could in theory be integer indexes, though that's really unlikely in a PySys test
args = ["add","bundle"]+nameList
if instanceName is not None:
args.append("--instance")
args.append(instanceName)
result = None
try:
result = self.__testcase.startProcess(self.__projectCommand, args, environs=dict(os.environ), workingDir=self.__projectDirectory , stdouterr=self.__testcase.allocateUniqueStdOutErr('add_bundle'), displayName='addBundle[%s]'%', '.join(nameList),**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: Optional L{pysys.process.user.ProcessUser.startProcess} keyword arguments, e.g. timeout, ignoreExitStatus.
"""
# 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, environs=dict(os.environ), 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: Optional L{pysys.process.user.ProcessUser.startProcess} keyword arguments, e.g. timeout, ignoreExitStatus.
"""
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:
self.__testcase.deleteDir(self.__deployedPath)
else:
raise Exception("Cannot overwrite existing deployment '"+self.__deployedPath+"' use deploy(overwrite=True) to allow" )
result = self.__testcase.startProcess(self.__deployCommand, args, environs=dict(os.environ), 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