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