#!/usr/bin/env python3
# Copyright(c) 2007,2013 Progress Software Corporation (PSC). All rights
# Copyright (c) 2013,2015-2016, 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
""" Contains PySys custom runner that adds support for EPL code coverage reporting.
"""
from __future__ import print_function
import sys, stat, os, string, logging, socket, copy, random, types
from pysys import rootLogger
from pysys.constants import *
from pysys.exceptions import *
from pysys.baserunner import BaseRunner
from pysys.process.helper import ProcessWrapper
from apama.correlator import CorrelatorHelper
from pysys.utils.pycompat import *
from pysys.utils.fileutils import deletedir, mkdir
[docs]class ApamaRunner(BaseRunner):
"""
A custom PySys runner for Apama tests.
Supports the ability to generate EPL code coverage reports using ``pysys run -Xeplcoverage``.
Coverage settings can be configured by setting the following
attributes on this class:
``eplcoverageargs, eplcoveragesource, eplcoverageinclude, eplcoverageexclude, eplcoveragetitle``.
"""
eplcoverage = False # can be overridden on the cmd line with -Xeplcoverage
def setup(self):
BaseRunner.setup(self)
self.__eplCoverageFiles = []
# until PySys 1.6.0 we need this workaround to ensure -Xeplcoverage and -Xeplcoverage=true behave the same
if isstring(self.eplcoverage): self.eplcoverage = self.eplcoverage.lower()=='true'
def processCoverageData(self, **kwargs):
super(ApamaRunner, self).processCoverageData(**kwargs)
if not self.eplcoverage: return
runneroutput = self.output
# keep same path as pre-10.5, even though the new PySys has changed the default runer output dir
if os.path.basename(runneroutput) == 'pysys-runner': runneroutput = os.path.dirname(runneroutput)
cov = os.path.normpath(runneroutput+'/eplcoverage')
deletedir(cov)
mkdir(cov)
self.log.info('Writing EPL code coverage output to: %s', cov)
with openfile(cov+'/coverage_files.txt', 'w', encoding='utf-8') as f:
# Writes Generated EPL Coverage file names to coverage_file.txt - always utf-8
for l in self.__eplCoverageFiles:
if isinstance(l, binary_type):
# we have to guess at the encoding; utf-8 is slightly safer than local/default encoding
# since at least 7-bit ascii chars will always work in utf8
l = l.decode('utf-8')
print(l, file=f)
arguments=[
'--output', cov,
'--exclude', '**/Input/**.mon', # test files aren't interesting
# exclude all files from Apama monitor dir.
'--exclude', PROJECT.APAMA_HOME+'/monitors/**',
# not useful
'--exclude', '**/*.qry.mon',
]
def tostringlist(s):
if not s: return []
# accept either a string or a list
if isstring(s):
if ',' in s: return [s.strip() for s in s.split(',')]
return [s]
return s
for s in tostringlist(getattr(self, 'eplcoverageinclude', '')):
arguments.extend(['--include', s])
for s in tostringlist(getattr(self, 'eplcoverageexclude', '')):
arguments.extend(['--exclude', s])
for s in tostringlist(getattr(self, 'eplcoveragesource', '')):
arguments.extend(['--source', s])
for s in tostringlist(getattr(self, 'eplcoverageargs', '')):
arguments.extend([s])
if hasattr(self, 'eplcoveragetitle'):
arguments.extend(['--title', getattr(self, 'eplcoveragetitle', '')])
arguments.append('@%s/coverage_files.txt'%cov)
self.startProcess(PROJECT.APAMA_HOME+'/bin/epl_coverage',
arguments=arguments,
state=FOREGROUND,
environs=dict(os.environ), # need this so we have APAMA_JRE/JAVA_HOME etc available (on Windows, or from Unix core edition)
displayName='epl_coverage (with %d coverage files)'%(len(self.__eplCoverageFiles)),
stdouterr=cov+'/epl_coverage',
workingDir=cov
)
def isPurgableFile(self, path):
# override
return not path.endswith('.eplcoverage') and BaseRunner.isPurgableFile(self, path)
def purgeDirectory(self, dir, delTop=False):
"""Recursively purge a directory removing all files and sub-directories.
:meta private:
Deprecated.
:param dir: The top level directory to be purged
:param delTop: Indicates if the top level directory should also be deleted
"""
try:
for file in os.listdir(dir):
path = os.path.join(dir, file)
if PLATFORM in ['sunos', 'linux']:
mode = os.lstat(path)[stat.ST_MODE]
else:
mode = os.stat(path)[stat.ST_MODE]
if stat.S_ISLNK(mode):
os.unlink(path)
if stat.S_ISREG(mode):
os.remove(path)
elif stat.S_ISDIR(mode):
self.purgeDirectory(path, delTop=True)
if delTop: os.rmdir(dir)
except OSError as ex:
self.log.warning("Caught OSError in purgeDirectory():")
self.log.warning(ex)
self.log.warning("Directory %s may not be completely purged" % dir)
def testComplete(self, testObj, dir):
# As of PySys 1.5.0 we could use collectTestOutput instead of this mechanism,
# but as that would require changes to users' project files stick with the old mechanism for now
if self.eplcoverage and os.path.exists(dir):
for p in os.listdir(dir):
if p.endswith('.eplcoverage'):
self.__eplCoverageFiles.append(dir+'/'+p)
BaseRunner.testComplete(self, testObj, dir)