# -*- encoding: UTF8 -*- # Test harness for ControlFetch.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 parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, parent_dir) from ControlFetch import ControlFetch os.environ["LOGS"] = "." LOGGER_NAME = "TestControlFetch" LOG_FILE = "log-TestControlFetch.log" class TestControlFetch(TestCase): def __init__(self, methodName='runTest'): super(TestControlFetch, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName) def setUp(self): self.serverSrc = P4Server(suffix="Src") self.serverTarg = P4Server(suffix="Targ", cleanup=False) trigpath = os.path.join(parent_dir, "ControlFetch.py") self.config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "~test_config.yaml") p4 = self.serverSrc.p4 self.p4 = p4 p4.logger = self.logger # This works if no spaces in server root pathname! port = p4.port.replace('"', '') self.logger.debug("port: |%s|" % port) triggers = p4.fetch_triggers() python = "python" if python3: python = "python3" triggers['Triggers'] = ['control-fetch command pre-rmt-Fetch " {} {} -p %quote%{}%quote% ' '-u {} -c {} %user% "'.format(python, trigpath, port, p4.user, self.config_path), ] self.logger.debug(triggers) p4.save_triggers(triggers) p4.run("configure", "set", "server.allowfetch=3") self.serverTarg.p4.run("configure", "set", "server.allowfetch=3") self.serverTarg.p4.run("configure", "set", "server.allowpush=3") p4 = self.serverTarg.p4 remote = p4.fetch_remote("origin") remote["Address"] = self.serverSrc.port.replace('"', '') remote["DepotMap"] = ["//depot/... //depot/..."] p4.save_remote(remote) # Reconnect to pick up changes p4.disconnect() p4.connect() self.serverTarg.p4.disconnect() self.serverTarg.p4.connect() self.p4 = self.serverTarg.p4 def tearDown(self): pass def testControlFetch(self): """check that it works when called as a trigger""" p4 = self.serverSrc.p4 inside = localDirectory(self.serverSrc.client_root, "inside") inside_file1 = os.path.join(inside, "inside_file1.yaml") create_file(inside_file1, "some content") p4.run('add', inside_file1) p4.run('submit', '-d', 'inside_file1 added') file2 = os.path.join(inside, "file2.txt") create_file(file2, "some content") p4.run('add', file2) p4.run('submit', '-d', 'file2 added') # Target p4 = self.p4 create_file(self.config_path, """ msg_fetch_not_allowed: - "" - "You are not allowed to do a p4 fetch." fetch_allowed_users_or_groups: - user1 - group1 - group2 """) try: p4.run_fetch() self.assertTrue(False, "Expected exception not found") except P4.P4Exception as e: self.assertRegex(str(e), r"You are not allowed to do a p4 fetch") create_file(self.config_path, """ msg_fetch_not_allowed: - "" - "You are not allowed to do p4 fetch." fetch_allowed_users_or_groups: - testuser """) # Fetches default origin remote spec - should work fine p4.run_fetch("-v", "@1,1") # Test group is OK - first create a change to fetch group = self.serverSrc.p4.fetch_group("allowed_users") group["Users"] = ["testuser"] self.serverSrc.p4.save_group(group) create_file(self.config_path, """ msg_fetch_not_allowed: - "" - "You are not allowed to do p4 fetch." fetch_allowed_users_or_groups: - allowed_users """) p4.user = "fred" try: p4.run_fetch() self.assertTrue(False, "Expected exception not found") except P4.P4Exception as e: self.assertRegex(str(e), r"You are not allowed to do p4 fetch") p4.user = "testuser" p4.run_fetch("-v") if __name__ == '__main__': unittest.main()
# | Change | User | Description | Committed | |
---|---|---|---|---|---|
#1 | 28240 | C. Thomas Tyler |
Released SDP 2021.1.28238 (2021/11/12). Copy Up using 'p4 copy -r -b perforce_software-sdp-dev'. |
||
//guest/perforce_software/sdp/dev/Unsupported/Samples/triggers/tests/TestControlFetch.py | |||||
#1 | 27999 | Robert Cowham | Sample trigger to control the ability to run a fetch from a remote server. |