TestCheckFolderStructure.py #2

# -*- 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()