Source code for apama.project

#!/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
from pysys.utils.pycompat import isstring

[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