# -*- encoding: UTF8 -*- # Test harness for CheckFolderStructure.py from __future__ import print_function import sys import unittest import os import re import P4 from p4testutils import TestCase, P4Server, localDirectory, create_file, append_to_file python3 = sys.version_info[0] >= 3 if python3: from unittest.mock import Mock else: from mock import Mock parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, parent_dir) from CheckFolderStructure import CheckFolderStructure, NewDirFinder from WorkflowTriggers import GroupMemberChecker os.environ["LOGS"] = "." LOGGER_NAME = "TestCheckFolderStructure" LOG_FILE = "log-TestCheckFolderStructure.log" class TestCheckFolderStructure(TestCase): def __init__(self, methodName='runTest'): super(TestCheckFolderStructure, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName) def setUp(self): self.server = P4Server() trigpath = os.path.join(parent_dir, "CheckFolderStructure.py") self.config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "~test_config.yaml") p4 = self.server.p4 self.p4 = p4 p4.logger = self.logger depot = p4.fetch_depot("streams") depot["Type"] = "stream" p4.save_depot(depot) # This works if no spaces in server root pathname! port = p4.port.replace('"', '') self.logger.debug("port: |%s|" % port) triggers = p4.fetch_triggers() triggers['Triggers'] = ['check-folder-structure change-submit //streams/... " python {} -p %quote%{}%quote% ' '-u {} -c {} %change% "'.format(trigpath, port, p4.user, self.config_path), ] self.logger.debug(triggers) p4.save_triggers(triggers) # Reconnect to pick up changes p4.disconnect() p4.connect() stream = p4.fetch_stream("-tmainline", "//streams/main") p4.save_stream(stream) client = p4.fetch_client() client['Stream'] = "//streams/main" p4.save_client(client) def tearDown(self): pass def testCheckNewDirs(self): """unit testing with mock framework""" mock_p4 = Mock() mock_p4.run_dirs.return_value = [] df = NewDirFinder(mock_p4) level = 0 self.assertFalse(df.findNewDir("//depot/stream/fred.txt", level)) self.assertFalse(df.findNewDir("//depot/stream/dir/fred.txt", level)) self.assertFalse(df.findNewDir("//depot/stream/dir/subdir/fred.txt", level)) mock_p4.run_dirs.return_value = [] df = NewDirFinder(mock_p4) level = 1 self.assertFalse(df.findNewDir("//depot/stream/fred.txt", level)) self.assertEqual("//depot/stream/dir", df.findNewDir("//depot/stream/dir/fred.txt", level)) mock_p4.run_dirs.return_value = [{'dir': "//depot/stream/dir"}] call_count = mock_p4.run_dirs.call_count df = NewDirFinder(mock_p4) newDir = df.findNewDir("//depot/stream/dir/fred.txt", level) self.assertFalse(newDir) call_count += 1 self.assertEqual(call_count, mock_p4.run_dirs.call_count) # Test caching - so don't expect call count to be incremented even if we call it # multiple times mock_p4.run_dirs.return_value = [{'dir': "//depot/stream/dir"}] call_count = mock_p4.run_dirs.call_count df = NewDirFinder(mock_p4) self.assertFalse(df.findNewDir("//depot/stream/dir/fred.txt", level)) self.assertFalse(df.findNewDir("//depot/stream/dir/fred2.txt", level)) self.assertFalse(df.findNewDir("//depot/stream/dir/fred3.txt", level)) call_count += 1 self.assertEqual(call_count, mock_p4.run_dirs.call_count) # Test multiple dirs under that level mock_p4.run_dirs.return_value = [{'dir': "//depot/stream/dir1"}, {'dir': "//depot/stream/dir2"}] call_count = mock_p4.run_dirs.call_count df = NewDirFinder(mock_p4) self.assertFalse(df.findNewDir("//depot/stream/dir1/fred.txt", level)) self.assertFalse(df.findNewDir("//depot/stream/dir2/fred2.txt", level)) self.assertFalse(df.findNewDir("//depot/stream/dir2/fred3.txt", level)) self.assertEqual("//depot/stream/dir3", df.findNewDir("//depot/stream/dir3/fred4.txt", level)) call_count += 1 self.assertEqual(call_count, mock_p4.run_dirs.call_count) # Test multiple depots - so multiple calls to dirs mock_p4.run_dirs.side_effect = [[{'dir': "//depotA/streamX/dir1"}], [{'dir': "//depotB/streamY/dir2"}]] call_count = mock_p4.run_dirs.call_count df = NewDirFinder(mock_p4) self.assertFalse(df.findNewDir("//depotA/streamX/dir1/fred.txt", level)) self.assertFalse(df.findNewDir("//depotB/streamY/dir2/fred2.txt", level)) self.assertEqual("//depotB/streamY/dir3", df.findNewDir("//depotB/streamY/dir3/fred4.txt", level)) call_count += 2 self.assertEqual(call_count, mock_p4.run_dirs.call_count) mock_p4.run_dirs.side_effect = None # Now test level = 2 # Test multiple dirs under that level level = 2 call_count = mock_p4.run_dirs.call_count mock_p4.run_dirs.side_effect = [[{'dir': "//depot/stream/dir1"}], [{'dir': "//depot/stream/dir1/dir2"}]] df = NewDirFinder(mock_p4) self.assertFalse(df.findNewDir("//depot/stream/dir1/fred.txt", level)) self.assertFalse(df.findNewDir("//depot/stream/dir1/dir2/fred2.txt", level)) self.assertEqual("//depot/stream/dir3", df.findNewDir("//depot/stream/dir3/fred4.txt", level)) self.assertEqual("//depot/stream/dir1/dir3", df.findNewDir("//depot/stream/dir1/dir3/fred4.txt", level)) call_count += 2 self.assertEqual(call_count, mock_p4.run_dirs.call_count) def testGroupMemberChecker(self): """test group membership""" mock_p4 = Mock() mock_p4.run_groups.return_value = [ {'user': 'user1', 'group': 'G1', 'isSubGroup': '0', 'isOwner': '0', 'isUser': '1'}, {'user': 'user2', 'group': 'G1', 'isSubGroup': '0', 'isOwner': '0', 'isUser': '1'}, {'user': 'user3', 'group': 'G3', 'isSubGroup': '0', 'isOwner': '1', 'isUser': '0'}, # Next record means G1 is a subgroup of G2 {'user': 'G1', 'group': 'G2', 'isSubGroup': '1', 'isOwner': '0', 'isUser': '0'}, {'user': 'G2', 'group': 'G3', 'isSubGroup': '1', 'isOwner': '0', 'isUser': '0'}] gchk = GroupMemberChecker(mock_p4) exceptions_list = ['user1'] self.assertTrue(gchk.IsMember('user1', exceptions_list)) self.assertFalse(gchk.IsMember('user2', exceptions_list)) exceptions_list = ['user1', 'user2'] self.assertTrue(gchk.IsMember('user2', exceptions_list)) exceptions_list = ['G1'] self.assertTrue(gchk.IsMember('user1', exceptions_list)) self.assertTrue(gchk.IsMember('user2', exceptions_list)) exceptions_list = ['G2'] self.assertTrue(gchk.IsMember('user1', exceptions_list)) self.assertTrue(gchk.IsMember('user2', exceptions_list)) self.assertFalse(gchk.IsMember('user3', exceptions_list)) exceptions_list = ['G3'] # G3 has subgroup G2 which has subgroup G1 self.assertTrue(gchk.IsMember('user1', exceptions_list)) self.assertTrue(gchk.IsMember('user2', exceptions_list)) self.assertTrue(gchk.IsMember('user3', exceptions_list)) self.assertFalse(gchk.IsMember('user4', exceptions_list)) def testCheckFolderStructure(self): """check that it works when called as a trigger""" p4 = self.p4 # Create a directory dir1 = localDirectory(self.server.client_root, "dir1") file1 = os.path.join(dir1, "file1") create_file(file1, "Some content") # Allow new dirs at all levels with open(self.config_path, "w") as f: f.write(""" msg_new_folder_not_allowed: - "" - "You are not allowed to create new folders at this level in the stream." - "Please add your files/folders to an existing folder at this level." projects: - name: ProjectA new_folder_allowed_level: 0 depot_paths: - //streams/... """) p4.run('add', file1) result = p4.run('submit', '-d', 'file1 added') self.assertTrue('submittedChange' in result[-1]) # Now block level 1 with open(self.config_path, "w") as f: f.write(""" msg_new_folder_not_allowed: - "" - "You are not allowed to create new folders at this level in the stream." - "Please add your files/folders to an existing folder at this level." projects: - name: ProjectA new_folder_allowed_level: 1 depot_paths: - //streams/... """) dir2 = localDirectory(self.server.client_root, "dir2") file2 = os.path.join(dir2, "file2") create_file(file2, "Some content") p4.run('add', file2) try: p4.run('submit', '-d', 'file2 added') self.assertTrue(False, "Expected exception not found") except P4.P4Exception as e: self.assertRegex(str(e), r"You are not allowed to create new folders") p4.run('revert', file2) # But creating file in existing dir is OK file3 = os.path.join(dir1, "file3") create_file(file3, "Some content") p4.run('add', file3) result = p4.run('submit', '-d', 'file3 added') self.assertTrue('submittedChange' in result[-1]) # Moving a file also OK in existing dir but not a new dir file4 = os.path.join(dir1, "file4") p4.run('edit', file1) p4.run('move', file1, file4) result = p4.run('submit', '-d', 'file1 moved') self.assertTrue('submittedChange' in result[-1]) file5 = os.path.join(dir2, "file5") p4.run('edit', file3) p4.run('move', file3, file5) try: p4.run('submit', '-d', 'file3 renamed') self.assertTrue(False, "Expected exception not found") except P4.P4Exception as e: self.assertRegex(str(e), r"You are not allowed to create new folders") if __name__ == '__main__': unittest.main()
# | Change | User | Description | Committed | |
---|---|---|---|---|---|
#1 | 27331 | C. Thomas Tyler |
Released SDP 2020.1.27325 (2021/01/29). Copy Up using 'p4 copy -r -b perforce_software-sdp-dev'. |
||
//guest/perforce_software/sdp/dev/Unsupported/Samples/triggers/tests/TestCheckFolderStructure.py | |||||
#1 | 26652 | Robert Cowham |
This is Tom's change: Introduced new 'Unsupported' directory to clarify that some files in the SDP are not officially supported. These files are samples for illustration, to provide examples, or are deprecated but not yet ready for removal from the package. The Maintenance and many SDP triggers have been moved under here, along with other SDP scripts and triggers. Added comments to p4_vars indicating that it should not be edited directly. Added reference to an optional site_global_vars file that, if it exists, will be sourced to provide global user settings without needing to edit p4_vars. As an exception to the refactoring, the totalusers.py Maintenance script will be moved to indicate that it is supported. Removed settings to support long-sunset P4Web from supported structure. Structure under new .../Unsupported folder is: Samples/bin Sample scripts. Samples/triggers Sample trigger scripts. Samples/triggers/tests Sample trigger script tests. Samples/broker Sample broker filter scripts. Deprecated/triggers Deprecated triggers. To Do in a subsequent change: Make corresponding doc changes. |
||
//guest/perforce_software/sdp/dev/Server/Unix/p4/common/bin/triggers/tests/TestCheckFolderStructure.py | |||||
#3 | 26433 | Robert Cowham | Refactored - group checker moved to WorkflowTriggers.py | ||
#2 | 25626 | Robert Cowham | Add group membership checking for exceptions list (to bypass trigger). | ||
#1 | 25625 | Robert Cowham |
First version of trigger to enforce stream folder naming structure - control at which level in a stream users are allowed to create new folders. |