#
# P4Triggers.py
#
# Version 2.0.6
#
# Copyright (c) 2008-2021, Perforce Software, Inc. All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL PERFORCE SOFTWARE, INC. BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# # Base class for all Python based P4 triggers
#
from __future__ import print_function
import P4
from datetime import datetime
import sys
import os
import logging
import traceback
import textwrap
import argparse
# If working on a server with the SDP, the 'LOGS' environment variable contains
# the path the standard logging directory. The '-L <logfile>' argument should
# be specified in non-SDP environments.
LOGDIR = os.getenv('LOGS', '/tmp')
DEFAULT_LOG_FILE = "p4triggers.log"
if os.path.exists(LOGDIR):
DEFAULT_LOG_FILE = "%s/p4triggers.log" % LOGDIR
DEFAULT_VERBOSITY = 'INFO'
LOGGER_NAME = 'P4Triggers'
class P4Change:
"""Encapsulates a Perforce change. Basically a pretty wrapping around p4.run_describe()"""
def __init__(self, desc):
self.change = desc["change"]
self.user = desc["user"]
self.client = desc["client"]
self.desc = desc["desc"]
self.time = datetime.utcfromtimestamp(int(desc["time"]))
self.status = desc["status"]
self.shelved = "shelved" in desc
self.files = []
if "depotFile" in desc:
for n, d in enumerate(desc["depotFile"]):
df = P4.DepotFile(d)
dr = df.new_revision()
dr.type = desc["type"][n]
dr.rev = desc["rev"][n]
dr.action = desc["action"][n]
self.files.append(df)
self.jobs = {}
if "job" in desc:
for n, j in enumerate(desc["job"]):
self.jobs[j] = desc["jobstat"][n]
class P4Trigger(object):
"""Base class for Perforce Triggers"""
def __init__(self, *args, **kwargs):
"""Constructor for P4Trigger.
Keyword arguments are passed to the P4.P4() instance used"""
kwargs['charset'] = 'none'
# API Levels are defined here: http://answers.perforce.com/articles/KB/3197
# Ensure this does not exceed the value for the P4Python version used.
# API Level 79 is for p4d 2015.2.
kwargs['api_level'] = 79
self.p4 = P4.P4(**kwargs)
self.options = None
self.logger = None
self.change = None
def parse_args(self, doc, args):
"""Common parsing and setting up of args"""
desc = textwrap.dedent(doc)
parser = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description=desc,
epilog="Copyright (c) 2008-2017 Perforce Software, Inc."
)
self.add_parse_args(parser) # Should be implemented by subclass
self.options = parser.parse_args(args=args)
self.init_logger()
self.logger.debug("Command Line Options: %s\n" % self.options)
def add_parse_args(self, parser, default_log_file=None, default_verbosity=None):
"""Default trigger arguments - common to all triggers
:param default_verbosity:
:param default_log_file:
:param parser:
"""
if not default_log_file:
default_log_file = DEFAULT_LOG_FILE
if not default_verbosity:
default_verbosity = DEFAULT_VERBOSITY
parser.add_argument('-p', '--port', default=None,
help="Perforce server port - set using %%serverport%%. Default: $P4PORT")
parser.add_argument('-u', '--p4user', default=None, help="Perforce user. Default: $P4USER")
parser.add_argument('-L', '--log', default=default_log_file, help="Default: " + default_log_file)
parser.add_argument('-T', '--tickets', help="P4TICKETS file full path")
parser.add_argument('-v', '--verbosity',
nargs='?',
const="INFO",
default=default_verbosity,
choices=('DEBUG', 'WARNING', 'INFO', 'ERROR', 'FATAL'),
help="Output verbosity level. Default is: " + default_verbosity)
def init_logger(self, logger_name=None):
if not logger_name:
logger_name = LOGGER_NAME
self.logger = logging.getLogger(logger_name)
self.logger.setLevel(self.options.verbosity)
logformat = '%(levelname)s %(asctime)s %(filename)s %(lineno)d: %(message)s'
logging.basicConfig(format=logformat, filename=self.options.log, level=self.options.verbosity)
def setupP4(self):
if self.options.port:
self.p4.port = self.options.port
if self.options.p4user:
self.p4.user = self.options.p4user
if self.options.tickets:
self.p4.ticket_file = self.options.tickets
self.p4.logger = self.logger
self.logger.debug("P4 port: '%s', user: '%s'" % (self.p4.port, self.p4.user))
def parseChange(self, changeNo):
try:
self.p4.connect()
self.setUp()
# Always call describe with -s flag.
self.change = self.getChange(changeNo,"-s")
return 0 if self.validate() else 1
except Exception:
return self.reportException()
def getChange(self, changeNo, flag=None):
if flag:
result = self.p4.run_describe(flag, changeNo)
else:
result = self.p4.run_describe(changeNo)
chg = P4Change(result[0])
if chg.shelved: # Files not listed by describe -s
try:
result = self.p4.run_files("@=%s" % changeNo)
except:
result = []
for f in result:
df = P4.DepotFile(f['depotFile'])
dr = df.new_revision()
dr.type = f["type"]
dr.rev = f["rev"]
dr.action = f["action"]
chg.files.append(df)
return chg
def validate(self):
"""Intended to be implemented in sub-class"""
return True
# method that subclasses can overwrite in order to complete the setup of P4 connection
def setUp(self):
pass
def message(self, msg):
"""Method to send a message to the user. Just writes to stdout, but it's
nice to encapsulate that here.
:param msg: """
print(msg)
def errorMessage(self):
return """
An error was encountered during trigger execution. Please
contact your Perforce administrator and ask them to
investigate the cause of this error in %s
""" % self.options.log
def reportException(self):
"""Method to encapsulate error reporting to make sure
all errors are reported in a consistent way"""
exc_type, exc_value, exc_tb = sys.exc_info()
self.message("Exception during trigger execution: %s %s %s" % (exc_type, exc_value, exc_tb))
self.reportP4Errors()
self.logger.error("called from:\n%s", "".join(traceback.format_exception(exc_type, exc_value, exc_tb)))
self.logger.error("port %s user %s tickets %s" % (self.p4.port, self.p4.user, self.p4.ticket_file))
# return message to the user
self.message(self.errorMessage())
return 1
def reportP4Errors(self):
lines = []
for e in self.p4.errors:
lines.append("P4 ERROR: %s" % e)
for w in self.p4.warnings:
lines.append("P4 WARNING: %s" % w)
if lines:
self.message("\n".join(lines))