CheckFolderStructure.py #1

#!/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
# ------------------------------------------------------------------------------

# tag::includeManual[]
"""
NAME:
    CheckFolderStructure.py

DESCRIPTION:
    This trigger is a change-submit trigger for workflow enforcement using Workflow.yaml for streams.

    It will ensure that any attempt to add (or branch) a file does not create
    a new folder name at specific levels.

    As with other Workflow triggers, it looks for a project matching files in the change.
    The value of "new_folder_allowed_level" is an integer and controls at which level within a stream
    new folder names can be created.

    If not set, or set to 0, then new folders can be created in the root of the stream.
    If set to 1, then for stream
        //streams/main/new_folder/a.txt would not be allowed, but
        //streams/main/existing_folder/new_folder/a.txt would be allowed.
    Note that files can be added just not folders.

    The field "new_folder_exceptions" - an array of users/groups who can override the trigger.

    To install, add a line to your Perforce triggers table like the following:

        check-folder-structure change-submit //... "python3 /p4/common/bin/triggers/CheckFolderStructure.py -p %serverport% -u perforce %change%"

    or (if server is standard SDP and has appropriate environment defaults for P4PORT and P4USER):

        check-folder-structure change-submit //... "python3 /p4/common/bin/triggers/CheckFolderStructure.py %change%"

    You may need to provide the full path to python executable, or edit the path to the trigger.

    Also, don't forget to make the file executable.
"""
# end::includeManual[]

# Python 2.7/3.3 compatibility.
from __future__ import print_function

import sys
import P4
import WorkflowTriggers
from os.path import basename, splitext
from collections import defaultdict

trigger_name = basename(splitext(__file__)[0])

class NewDirFinder:
    "Finds new directories in paths - seperate class for ease of testing"

    def __init__(self, p4):
        self.p4 = p4
        self.dirCache = {}

    # This method returns None if no new directory found, or the directory name
    def findNewDir(self, path, level):
        path = path[2:]  # cut the // at the beginning of the depot off
        dirs = path.split('/')
        depot = dirs.pop(0)
        stream = dirs.pop(0)
        file = dirs.pop()  # Remove file
        return self.findNewDirs('//%s/%s' % (depot, stream), dirs, level)

    # Look for new directories
    # Recursive algorithm, descending down the directory paths
    # Caching directories as we find them
    def findNewDirs(self, top, dirs, level):
        if len(dirs) == 0 or level == 0:
            return None
        aDir = dirs.pop(0)
        path = top + '/' + aDir

        newdir = None
        if top in self.dirCache:
            existingDirs = self.dirCache[top]
        else:
            # result is in tagged mode, single entry "dir"=>directory name
            existingDirs = [d["dir"] for d in self.p4.run_dirs(top + "/*")]
            self.dirCache[top] = existingDirs
        for d in existingDirs:
            if d == path:
                return self.findNewDirs(path, dirs, level - 1)
        return path  # It is a new directory

class CheckFolderStructure(WorkflowTriggers.WorkflowTrigger):
    """See module doc string for details"""

    def __init__(self, *args, **kwargs):
        WorkflowTriggers.WorkflowTrigger.__init__(self, **kwargs)
        self.parse_args(__doc__, args)

    def add_parse_args(self, parser):
        """Specific args for this trigger - also calls super class to add common trigger args"""
        parser.add_argument('-c', '--config-file', default=None,
                            help="Configuration file for trigger. Default: Workflow.yaml")
        parser.add_argument('change', help="Change to validate - %%change%% argument from triggers entry.")
        super(CheckFolderStructure, self).add_parse_args(parser)

    def run(self):
        """Runs trigger"""
        try:
            self.logger.debug("%s firing" % trigger_name)
            config = self.load_config(self.options.config_file)
            errors = []
            for k in "msg_new_folder_not_allowed".split():
                if k not in config:
                    errors.append("Config file %s missing definition for %s" %
                                  (self.options.config_file, k))
            if errors:
                msg = "%s: Invalid config file for trigger %s\n" % (trigger_name, str(errors))
                self.message(msg)
                return 1

            self.setupP4()
            self.p4.connect()

            change = self.getChange(self.options.change)
            files = [x.depotFile for x in change.files]

            # Only process if adding or branching files
            actions = [x.revisions[0].action for x in change.files if x.revisions[0].action in ['add', 'branch', 'move/add']]
            if not actions:
                self.logger.debug("%s: Ignoring change as no add actions" % trigger_name)
                return 0

            # If no project then don't check further
            prj = self.get_project_by_files(config, files)
            self.logger.debug("prj: %s" % str(prj))
            if not prj:
                self.logger.debug("%s: Allowing folder as no project affected" % trigger_name)
                return 0

            if not 'new_folder_allowed_level' in prj:
                self.logger.debug("%s: Allowing changes as 'new_folder_allowed_level' not specified" % trigger_name)
                return 0

            level = 0
            try:
                level = int(prj['new_folder_allowed_level'])
            except ValueError:
                level = 0
            if level <= 0:
                self.logger.debug("%s: Allowing changes as level not specified" % trigger_name)
                return 0

            # Now check for users and groups who are exceptions
            if 'new_folder_exceptions' in prj:
                gchk = GroupMemberChecker(self.p4)
                if gchk.IsMember(change.user, prj['new_folder_exceptions']):
                    self.logger.debug("%s: User allowed to bypass trigger: %s" % (trigger_name, user))
                    return 0

            # Finally we check the individual files
            errors = []
            # Todo - calculate stream depth properly for streams depots referenced.

            new_files = [x.depotFile for x in change.files if
                         x.revisions[0].action in ['add', 'branch', 'move/add']]

            df = NewDirFinder(self.p4)
            for f in new_files:
                newdir = df.findNewDir(f, level)
                if newdir:
                    errors.append(newdir)

            if errors:
                err_msg = "\n".join(config['msg_new_folder_not_allowed'])
                msg = err_msg + "\nNew folders being created:\n%s\n\n" % ", ".join(errors)
                self.message(msg)
                return 1
        except Exception:
            return self.reportException()

        return 0


if __name__ == '__main__':
    """ Main Program"""
    trigger = CheckFolderStructure(*sys.argv[1:])
    sys.exit(trigger.run())