# Version 2.2.0
# tag::includeManual[]
"""
CaseCheckTrigger.py
This trigger ensures users are not adding new files (directly or by branching) which only
differ in case (either filename or a directory element of their path) from existing depot paths.
It is useful for both case-sensitive and case-insensitive servers, although most used for the former.
Example 1: Typical usage from the Helix Core server triggers table:
Triggers:
CheckCaseTrigger change-submit //... "/usr/bin/python3 /p4/common/bin/triggers/CheckCaseTrigger.py %changelist% myuser=%user%"
SAMPLE OUTPUT:
Submit validation failed -- fix problems then use 'p4 submit -c 1234'.
'CheckCaseTrigger' validation failed:
Your submission has been rejected because the following files
are inconsistent in their use of case with respect to existing
directories:
Your file: '//depot/dir/test'
existing file/dir: '//depot/DIR'
BYPASS LOGIC:
By default, this trigger can be bypassed by any user by adding the
token BYPASS_CASE_CHECK to the changelist description. Specify
'allowbypass=no' on the command line to disable the ability to
bypass this trigger
As an exception, if the user is 'git-fusion-user', the case check
is always bypassed if 'myuser' is defined.
DEPENDENCIES:
This trigger requires P4Triggers.py and the P4Python API.
SEE ALSO:
See 'p4 help triggers'.
"""
# end::includeManual[]
from __future__ import print_function
import logging
import os
import platform
import re
import subprocess
import sys
import P4
import P4Triggers
# Method canonical
# IN: string, utf8 compatible
# OUT: unicode string, all lower case
def canonical(aString):
return aString.lower()
class CheckCaseTrigger(P4Triggers.P4Trigger):
# tag::includeManual[]
"""CaseCheckTrigger is a subclass of P4Trigger. Use this trigger to ensure
that your depot does not contain two filenames or directories that only
differ in case.
Having files with different case spelling will cause problems in mixed
environments where both case-insensitive clients like Windows and case-
sensitive clients like UNIX access the same server.
"""
# end::includeManual[]
def __init__(self, *args, **kwargs):
kwargs['charset'] = 'none'
kwargs['api_level'] = 71
self.allowBypass=AllowBypass
fileFilter = None
if 'filefilter' in kwargs:
fileFilter = kwargs['filefilter']
del kwargs['filefilter']
P4Triggers.P4Trigger.__init__(self, **kwargs)
self.parse_args(__doc__, args)
# Ensure that -ztag global option is used.
self.p4.tagged = True
# need to reset the args in case a p4config file overwrote them
for (k, v) in kwargs.items():
if k != "log":
try:
setattr(self.p4, k, v)
except:
self.logger.error("error setting p4 property: '%s' to '%s'" % (k, v))
self.map = None
if fileFilter:
try:
with open(fileFilter) as f:
self.map = P4.Map()
for line in f:
self.map.insert(line.strip())
except IOError:
self.logger.error("Could not open filter file %s" % fileFilter)
self.depotCache = {}
self.masterCache = {}
self.maxWildcards = 9
self.loggingEnabled = self.logger.isEnabledFor(logging.DEBUG)
self.caseSensitive = platform.system() == "Linux" # Default - will be checked later
def add_parse_args(self, parser):
"""Specific args for this trigger - also calls super class to add common trigger args"""
parser.add_argument('change', help="Change to validate - %%change%% argument from triggers entry.")
parser.add_argument('-m', '--max-errors', default=10, help="Max no of errors before aborting submit. Default 10.")
super(CheckCaseTrigger, self).add_parse_args(parser)
def setUp(self):
info = self.p4.run_info()[0]
if "unicode" in info and info["unicode"] == "enabled":
self.p4.charset = "utf8"
self.p4.exception_level = 1 # ignore WARNINGS like "no such file"
self.p4.prog = "CheckCaseTrigger"
if self.allowBypass:
self.USER_MESSAGE="""
Your changelist submit attempt has been rejected because
one or more file paths opened for add vary only by case
from existing files/directory paths. Creating file/folders
that vary only in case from existing paths causes inconsistent
behavior across platforms with different case handling behaviors
(e.g. Windows, Linux/UNIX, Mac OSX). Thus, adding case-only
variations of existing paths is strongly discouraged.
If you are certain the files to be added will only be accessed
from workspaces on case-sensitive platforms (like UNIX/Linux),
then this trigger can be bypassed by adding the token
BYPASS_CASE_CHECK to the changelist description and attempting
the submit again.
Alternately, you can revert any files opened for add in your
changelists that vary only in case from existing files, or
move them to new names that don't conflict with existing files.
Offending files:
"""
else:
self.USER_MESSAGE="""
Your changelist submit attempt has been rejected because
one or more file paths opened for add vary only by case
from existing files/directory paths. Creating file/folders
that vary only in case from existing paths causes inconsistent
behavior across platforms with different case handling behaviors
(e.g. Windows, Linux/UNIX, Mac OSX). Thus, adding case-only
variations of existing paths is disallowed.
To move forward, you can revert any files opened for add in your
changelists that vary only in case from existing files, or
move them to new names that don't conflict with existing files.
Offending files:
"""
self.BADFILE_FORMAT="""
Your file: '%s'
existing file/dir: '%s'
"""
def validate(self):
"""Here the fun begins. This method overrides P4Trigger.validate()"""
badlist = {}
info = self.p4.run_info()
if "caseHandling" in info[0]:
self.caseSensitive = "insensitive" != info[0]["caseHandling"]
self.logger.debug("validate: p4d caseSensitive %s", self.caseSensitive)
files = self.change.files
if self.loggingEnabled:
self.logger.debug("validate: Files to submit: %s", files)
self.filterRenames(files)
# Determine valid file list and depots to cache
validFiles = []
uniqueDepots = {}
for file in files:
action = file.revisions[0].action
if self.map and self.map.includes(file.depotFile):
continue
if not action in ("add", "branch", "move/add"):
continue
path = file.depotFile[2:]
if self.loggingEnabled:
self.logger.debug("validate: path = %s", path)
validFiles.append(file.depotFile)
if self.loggingEnabled:
self.logger.debug("validate: file.depotFile = %s", file.depotFile)
if self.loggingEnabled:
self.logger.debug("validate: uniqueDepots = %s", uniqueDepots)
self.logger.debug("validate: validFiles = %s", validFiles)
# Build cache for each unique depot.
self.buildCache(validFiles)
if self.loggingEnabled:
self.logger.debug("validate: masterCache_1 = %s", self.masterCache)
# Look for files in cache. This includes looking for directories in file path along the way.
self.searchCache(validFiles, badlist)
if self.loggingEnabled:
self.logger.debug("validate: badlist = %s", badlist)
self.logger.debug("validate: masterCache_2 = %s", self.masterCache)
if len(badlist) > 0:
self.report(badlist)
return len(badlist) == 0
# This method returns a list of all dirs between root and lowest level in the filelist
# Can then run "p4 dirs -i a/* a/b/*" against this list to find any other potential conflicts at each level
# IN: filelist
# OUT: dirlist for dirs command
def getDirList(self, fileList):
# Files:
# //D/a/f.txt
# //D/a/b/c/f.txt
# Output:
# //D
# //D/a
# //D/a/b
# //D/a/b/c
# We don't need to go any deeper than max path of files in list
dirList = {}
for f in fileList:
cf = canonical(f)
parts = f[2:].split('/') # Original case
cparts = cf[2:].split('/') # Lower case
for i in range(1, len(cparts)): # Process up to the parent dir of the file
if self.caseSensitive:
p = "/".join(parts[:i])
else:
p = "/".join(cparts[:i])
if not p in dirList:
dirList[p] = "/".join(parts[:i])
return dirList
# This method returns a list of dirs containing files in the filelist (includes intermediat dirs to allow for
# dir and filename collision
# Can then run "p4 files -i a/b/c/* a/b/d/*" against this list to find any other potential conflicts at each level
# IN: filelist
# OUT: dirlist for files command
def getFileDirList(self, fileList):
# Files:
# //D/a/c.txt
# //D/a/b/c/d.txt
# Output:
# //D/a
# //D/a/b
# //D/a/b/C - sensitive
# //D/a/b/c - insensitive
# We don't need to go any deeper than max path of files in list
dirList = {}
for f in fileList:
parts = f[2:].split('/')
if self.caseSensitive:
for i in range(1, len(parts)): # Process up to the parent dir of the file
p = "/".join(parts[:i])
if not p in dirList:
dirList[p] = "/".join(parts[:-1])
else:
cf = canonical(f)
cparts = cf[2:].split('/')
for i in range(1, len(cparts)): # Process up to the parent dir of the file
p = "/".join(cparts[:i])
if not p in dirList:
dirList[p] = "/".join(parts[:-1])
return dirList
# Builds a global cache to use for mismatch searches.
# IN: depots to use
# fileList to parse
# OUT: None
def buildCache(self, fileList):
if self.loggingEnabled:
self.logger.debug("buildCache: fileList = %s", fileList)
depots = self.p4.run_depots()
for d in depots:
dname = d["name"]
cd = canonical(dname)
if self.caseSensitive:
self.depotCache[dname] = dname
else:
self.depotCache[cd] = dname
dirList = self.getDirList(fileList)
if self.loggingEnabled:
self.logger.debug("buildCache: dirList = %s", dirList)
# Note depots will exist in the list but we need to ensure correct case is used
if not self.caseSensitive:
for d in self.depotCache:
if not d in dirList:
dirList[d] = d
if self.caseSensitive:
dirParams = ["//" + d + "/*" for d in dirList]
else:
dirParams = ["//" + dirList[d] + "/*" for d in dirList]
cdirs = {}
if dirParams:
for d in self.p4.run_dirs(*dirParams):
d = d["dir"] # result is in tagged mode, single entry "dir"=>directory name
cd = d.lower()
self.masterCache[cd] = d
if self.caseSensitive:
if not d in dirList:
cdirs[cd] = d
else:
if cd != d and not d in dirList:
cdirs[cd] = d
# If necessary, repeat the dirs command on case sensitive systems with any extra dirs found from
# previous call
if self.caseSensitive and len(cdirs) > 0:
dirParams = [d + "/*" for d in cdirs.keys()]
for d in self.p4.run_dirs(*dirParams):
d = d["dir"] # result is in tagged mode, single entry "dir"=>directory name
cd = d.lower()
self.masterCache[cd] = d
fileDirList = self.getFileDirList(fileList)
if not fileDirList:
return
fileParams = ["//" + d + "/*" for d in fileDirList]
for f in self.p4.run_files(*fileParams):
if not "delete" in f["action"]:
f = f["depotFile"]
cf = f.lower()
if not cf in self.masterCache:
self.masterCache[cf] = f
# Method filterRenames:
# Removes pairs of files that only differ in case
# where one action is branch, and the other delete
def filterRenames(self, files):
branches = [x for x in files if x.revisions[0].action == 'branch']
deletes = [x.depotFile.lower() for x in files if x.revisions[0].action == 'delete']
for f in branches:
if f.depotFile.lower() in deletes:
files.remove(f)
def report(self, badfiles):
msg = self.USER_MESSAGE
for (n, (file, mismatch)) in enumerate(badfiles.items()):
if n >= self.options.max_errors:
break
msg += self.BADFILE_FORMAT % (file, mismatch)
self.message(msg)
def run(self):
"""Runs trigger"""
try:
self.logger.debug("CheckCaseTrigger firing")
self.setupP4()
return self.parseChange(self.options.change)
except Exception:
return self.reportException()
# This method searches a global cache to find case mismatches for changelist files.
# File subdirectories and file itself are added to cache if no mismatches are found.
# IN: List of changelist files for which we want verify there are no case mismatches.
# Mismatch dictionary that records mismatches.
# OUT: May modify mismatches parameter
def searchCache(self, cfiles, mismatches):
if self.loggingEnabled:
self.logger.debug("searchCache: changelist files = %s, mismatches = %s", cfiles, mismatches)
self.logger.debug("searchCache: masterCache = %s", self.masterCache)
for f in cfiles:
if self.loggingEnabled:
self.logger.debug("searchCache: f = %s", f)
mismatch = ""
# Check depot
parts = f[2:].split("/")
depot = parts[0]
cdepot = canonical(depot)
# Search on depots - require depot to be in cache
if self.caseSensitive:
if not depot in self.depotCache:
mismatch = self.depotCache[cdepot]
if self.loggingEnabled:
self.logger.debug("depot not found: %s", depot)
mismatches[f] = mismatch
continue
else:
if depot != self.depotCache[cdepot]:
mismatch = self.depotCache[cdepot]
if self.loggingEnabled:
self.logger.debug("mismatch2: sd = %s, f = %s, m = %s", cdepot, depot, mismatch)
mismatches[f] = mismatch
continue
# Look for file and continue if it's in the cache. It's already been added.
cf = canonical(f)
if cf in self.masterCache:
if self.loggingEnabled:
self.logger.debug("searchCache: found %s: %s", f, self.masterCache[cf])
if f != self.masterCache[cf]:
mismatch = self.masterCache[cf]
if self.loggingEnabled:
self.logger.debug("mismatch3: cf = %s, f = %s, m = %s", cf, f, mismatch)
mismatches[f] = mismatch
continue
# Need to check for mismatch of file path components.
# If none, add components to cache.
cparts = cf[2:].split('/')
for i in range(1, len(cparts) + 1):
cp = "//" + "/".join(cparts[:i])
if not cp in self.masterCache: # Save in cache
self.masterCache[cp] = "//" + "/".join(parts[:i])
if self.loggingEnabled:
self.logger.debug("adding to cache: %s", self.masterCache[cp])
else:
p = "//" + "/".join(parts[:i])
m = self.masterCache[cp]
if m != p:
mismatch = m
if self.loggingEnabled:
self.logger.debug("mismatch4: cp = %s, p = %s, mismatch = %s", cp, p, mismatch)
break
if mismatch:
mismatches[f] = mismatch
else:
self.masterCache[cf] = f
if __name__ == "__main__":
# Generate new args - parsing out port=123 style way of specifying
# parameters intended for p4 properties
kwargs = {}
args = []
for arg in sys.argv[1:]:
p = arg.split("=", 1)
if len(p) == 1:
args.append(arg)
else:
kwargs[p[0]] = p[1]
# Example of how to exclude the 'git-fusion-user'
# Note: Need to remove 'myuser' after test as it's not a valid P4 argument.
if 'myuser' in kwargs:
if kwargs['myuser'] == 'git-fusion-user':
sys.exit(0)
else:
del kwargs['myuser']
AllowBypass = 1
if 'allowbypass' in kwargs:
if kwargs['allowbypass'] == 'no':
AllowBypass = 0
# Remove 'allowbypass' after test as it's not a valid P4 argument.
del kwargs['allowbypass']
if AllowBypass:
# Grab the changelist description, and scan for the bypass token string.
# If the token is detected, silently and immediately exit with a happy 0
# exit code.
changelist = sys.argv[1]
cmd = "%s -ztag -F %%desc%% describe -f -s %s" % (os.getenv('P4BIN','p4'), changelist)
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
(changeDesc, err) = p.communicate()
p_status = p.wait()
# If the changelist description contains the text BYPASS_CASE_CHECK,
# bypass the case check logic.
if (re.search (b'BYPASS_CASE_CHECK', changeDesc, re.MULTILINE)):
sys.exit(0)
ct = CheckCaseTrigger(*args, **kwargs)
sys.exit(ct.run())