TestControlFetch.py #1

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