#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ==============================================================================
# Copyright and license info is available in the LICENSE file included with
# the Server Deployment Package (SDP), and also available online:
# https://swarm.workshop.perforce.com/projects/perforce-software-sdp/view/main/LICENSE
# ------------------------------------------------------------------------------
"""
NAME:
WorkflowTriggers.py
DESCRIPTION:
Base class with utility functions for interfacing with Swarm and parsing workflow yaml file.
"""
# Python 2.7/3.3 compatibility.
from __future__ import print_function
import sys
import os
import P4Triggers
import P4
import requests
import yaml
from string import Template
from collections import OrderedDict
def ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
class OrderedLoader(Loader):
pass
def construct_mapping(loader, node):
loader.flatten_mapping(node)
return object_pairs_hook(loader.construct_pairs(node))
OrderedLoader.add_constructor(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
construct_mapping)
return yaml.load(stream, OrderedLoader)
class WorkflowTrigger(P4Triggers.P4Trigger):
"""See module doc string for details"""
def load_config(self, config_file=None):
if not config_file:
config_file = "/p4/common/config/Workflow.yaml"
config = {}
if not os.path.exists(config_file):
return config
try:
with open(config_file, 'r') as f:
config = ordered_load(f, yaml.SafeLoader)
except Exception as e:
self.logger.error(str(e))
return config
def get_swarm_url(self):
"""Read from Perforce server propery value"""
if self.options.test_mode:
return "http://swarm.dev/"
prop = self.p4.run('property', '-l' , '-n', 'P4.Swarm.URL')
self.logger.debug("Property: %s" % str(prop))
url = prop[0]['value']
if url[-1] != '/':
url += '/'
return url
def get_project(self, config, change):
"""Return first matching project for a file in the change"""
files = []
if change.shelved and not change.files:
try:
files = [x['depotFile'] for x in self.p4.run_files("@=%s" % change.change)]
except:
self.logger.warning("No shelved files found for change @=%s" % change.change)
else:
files = [x.depotFile for x in change.files]
return self.get_project_by_files(config, files)
def get_project_by_files(self, config, files):
"""Return first matching project for a list of files"""
if not 'projects' in config:
return {}
for prj in config['projects']:
if not 'name' in prj or not 'depot_paths' in prj:
return False
map = P4.Map()
for p in prj['depot_paths']:
map.insert(p)
for df in files:
if map.includes(df):
return prj
return {}
def project_flag_true(self, config, change, key):
"""Returns True if specified field (key) has a value of 'y' for a matching project"""
if not 'projects' in config:
return False
prj = self.get_project(config, change)
self.logger.debug("prj: %s" % prj)
if not key in prj or not prj[key] == 'y':
return False
return True
def formatReviewDescription(self, review_description, **kwargs):
"""Format using specified format options - see call below
Assumes review_description is an array of lines in config file"""
desc = "\n".join(review_description)
desc = desc.replace("\\n", "\n")
t = Template(desc)
result = t.safe_substitute(**kwargs)
return result