Source code for apama.docker.apamadocker

#!/usr/bin/env python
# Copyright (c) 2015, 2017-2018, 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 

"""
Support for using correlator and IAF Docker images from PySys tests. 
"""

from apama.docker.framework import DockerImage, DockerContainer
from apama.correlator import CorrelatorHelper
from apama.iaf import IAFHelper
import os

[docs]class ApamaDockerContainer(DockerContainer): """Helper class for working with Apama Docker containers. """ @classmethod def fromDefaultImage(cls, parent): ret = DockerContainer.fromImage(DockerImage.fromDockerfile(parent, file=os.path.join(parent.project.DOCKER_PRODUCT_HOME, "image/Dockerfile"), context=parent.project.APAMA_SAG_HOME)) ret.__class__ = ApamaDockerContainer return ret @classmethod def fromImage(cls, image): ret = DockerContainer.fromImage(image) ret.__class__ = ApamaDockerContainer return ret @classmethod def fromExisting(cls, parent, name, own=True): ret = DockerContainer.fromExisting(parent, name, own) ret.__class__ = ApamaDockerContainer return ret
[docs] def getCorrelator(self, port): """ Get an `apama.correlator.CorrelatorHelper` for a correlator running inside this container :param port: The port that the correlator process is running on inside the container :return: The `apama.correlator.CorrelatorHelper` that can be used to interact with the correlator """ return CorrelatorHelper(self.parent, port=self.getExternalPort(port), host=self.getDockerHost())
[docs] def getIAF(self, port): """ Get an `apama.iaf.IAFHelper` for an iaf running inside this container :param port: The port that the IAF process is running on inside the container :return: The `apama.iaf.IAFHelper` that can be used to interact with the IAF """ return IAFHelper(self.parent, port=self.getExternalPort(port), host=self.getDockerHost())
[docs] def getDockerHost(self): """ Get hostname on which this container running :return: docker's hostname """ dockerhost = self.parent.project.DOCKER_HOST or 'localhost' if 'ssh://' in dockerhost: dockerhost = dockerhost[6:] else: dockerhost = dockerhost.split(':')[0] return dockerhost