'''
Created on Dec 21, 2017
@author: Charlie McLouth
'''
from p4rest.p4 import P4
from P4Server import P4TestcaseShared
import os.path
from p4rest.command import Command
from p4rest.exceptions import P4RestCommandError
import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
debugHandler = logging.StreamHandler()
debugHandler.setFormatter(logging.Formatter("%(levelname)s:%(filename)s:"
"%(lineno)d:%(funcName)s:"
"%(message)s"))
logger.addHandler(debugHandler)
class TestP4RestSessionP4(P4TestcaseShared):
'''Test P4 environment within the P4RestSessionP4'''
def initTestData(self, _p4api):
# create 10 of every object
p4api = P4()
p4api.port = _p4api.port
p4api.user = _p4api.user
p4api.exception_level = P4.RAISE_ERROR
p4api.connect()
for plural in ["users", "depots", "streams", "clients", "repos",
"labels", "branches", "changes", "jobs", "groups",
"ldaps", "remotes", "servers"]:
single, unused = p4api.specfields.get(plural)
for i in range(0,10):
args = ["{}_{!s}".format(single, i)]
p4api.user = "user_{!s}".format(i)
if single == "depot":
if i == 0:
args = ["-t" "stream", "{}_{!s}".format(single, i)]
elif i == 1:
args = ["-t" "graph", "{}_{!s}".format(single, i)]
elif single == "change":
args = []
p4api.client = "client_{!s}".format(i)
elif single == "user":
args.clear()
elif single == "stream":
args = ["-t" "mainline", "//depot_0/{}{}".format(single, i)]
elif single == "repo":
args = ["//depot_1/{}{}".format(single, i)]
func = getattr(p4api, "fetch_{}".format(single))
spec = func(args)
args = [spec]
if single == "branch":
spec["View"][0] = "//depot_2/a/... //depot_2/b/..."
elif single == "client":
spec["Root"] = self.p4server.clientroot
options = spec["Options"].split()
options[-1] = "rmdir"
spec["Options"] = " ".join(options)
elif single in ["change", "job"]:
spec["Description"] = "description {!s}".format(i)
elif single == "ldap":
spec["SimplePattern"] = "%user%"
elif single == "group":
spec["Owners"] = [p4api.user]
spec["Users"] = [p4api.user]
func = getattr(p4api, "save_{}".format(single))
spec = func(args)
p4api.user = _p4api.user
p4api.client = _p4api.client
for i in range(0,10):
p4api.user = "user_{!s}".format(i)
p4api.client = "client_{!s}".format(i)
spec = p4api.fetch_change()
spec["Description"] = "submitted change"
changeno = p4api.save_change(spec)[0].split(maxsplit=2)[1]
for ii in range(0,10):
filename = p4api.run_where("//depot_2/{!s}/file{!s}.txt".format(i, ii))[0]["path"]
if not os.path.isdir(os.path.dirname(filename)):
os.makedirs(os.path.dirname(filename))
with open(filename, mode="a") as f:
f.write("a line {!s}".format(ii))
p4api.run_add("-c", changeno, filename)
for line in p4api.run_submit("-c", changeno):
if isinstance(line, dict) and "submittedChange" in line:
changeno = line["submittedChange"]
break
spec = p4api.fetch_change()
spec["Description"] = "shelved change"
changeno = p4api.save_change(spec)[0].split(maxsplit=2)[1]
p4api.run_edit("-c", changeno, "//depot_2/{!s}/...".format(i))
p4api.run_shelve("-r", "-c", changeno)
spec = p4api.fetch_change()
spec["Description"] = "pending change"
changeno = p4api.save_change(spec)[0].split(maxsplit=2)[1]
# p4api.run_reopen("-c", changeno, "//depot_2/{!s}/...".format(i))
p4api.run_revert("//...")
p4api.disconnect()
def testValidateCommand(self):
p4api = P4()
p4api.user="user_0"
p4api.port=self.p4server.p4api.port
p4api.exception_level = P4.RAISE_ERRORS
p4api.connect()
p4api.run_login(password="Perf0rce")
# null p4api
with self.assertRaises(P4RestCommandError) as cm:
c = Command(None, None)
c.validateCommand(None)
expected = "p4api not set"
actual = cm.exception.value
self.assertEqual(expected, actual)
# null command
c = Command(p4api, None)
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand(None)
expected = (404, "command not set")
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
# empty command
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand("")
expected = (404, "command not set")
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
# unknown command
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand("unknown")
expected = (404, "unknown command (unknown)")
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
# unsupported command
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand("parse_spec")
expected = (404, "unsupported command (parse_spec)")
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand("format_spec")
expected = (404, "unsupported command (format_spec)")
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand("run_client")
expected = (404, "unsupported command (run_client)")
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
# with self.assertRaises(P4RestCommandError) as cm:
# c.validateCommand("run_clients")
# expected = (404, "unsupported command (run_clients)")
# actual = (cm.exception.httpstatuscode, cm.exception.value)
# self.assertEqual(expected, actual)
# unknown commands
for command, unused in Command.CT_NAME.items():
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand(command)
expected = (404, "unknown command ({})".format(command))
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
for command, unused in Command.CT_NAME.items():
command = command + "_"
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand(command)
expected = (404, "unknown command ({})".format(command))
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
for command, commandtype in Command.CT_NAME.items():
if commandtype in Command.UNSUPPORTED_COMMANDTYPES \
or commandtype == Command.CT_RUN:
continue
command = command + "_foo"
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand(command)
expected = (404, "unknown command ({})".format(command))
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual)
p4api.disconnect()
def testValidateCommandArgs(self):
p4api = P4()
p4api.user="user_0"
p4api.port=self.p4server.p4api.port
p4api.exception_level = P4.RAISE_ERRORS
p4api.connect()
p4api.run_login(password="Perf0rce")
c = Command(p4api, None)
# missing arguments
for i in range(0, 4):
for commandtype in [Command.CT_FETCH, Command.CT_DELETE,
Command.CT_UPDATE, Command.CT_SAVE]:
command = "{}_client".format(Command.CT_LOOKUP[commandtype])
logger.debug(command)
if commandtype == Command.CT_SAVE:
expected = (400, "command ({}) expects at least one "
"argument of type dict".format(command))
if i == 3:
expected = (400, "command ({}) expects first argument "
"to be of dict type".format(command))
elif commandtype == Command.CT_UPDATE:
expected = (404, "command ({}) expects at least two "
"arguments".format(command))
elif i == 3:
continue
else:
expected = (404, "command ({}) expects at least one "
"argument of simple type".format(command))
with self.assertRaises(P4RestCommandError) as cm:
if i == 0:
c.validateCommand(command)
elif i == 1:
c.validateCommand(command, )
elif i == 2:
cmdargs = []
c.validateCommand(command, *cmdargs)
elif i == 3:
cmdargs = ["a"]
c.validateCommand(command, *cmdargs)
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual, command)
# empty arguments
for commandtype in [Command.CT_FETCH, Command.CT_DELETE,
Command.CT_UPDATE, Command.CT_SAVE]:
command = "{}_client".format(Command.CT_LOOKUP[commandtype])
cmdargs = [""]
if commandtype == Command.CT_SAVE:
cmdargs = [{}]
expected = (400, "command ({}) expects first argument to be of "
"dict type".format(command))
elif commandtype == Command.CT_UPDATE:
cmdargs = ["", {}]
expected = (404, "command ({}) expects second to last argument"
" to be of simple type".format(command))
else:
expected = (404, "command ({}) expects last argument to be"\
" of simple type".format(command))
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand(command, *cmdargs)
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual, command)
commandtype = Command.CT_UPDATE
command = "{}_client".format(Command.CT_LOOKUP[commandtype])
cmdargs = ["a", {}]
expected = (400, "command ({}) expects last argument to be of "
"dict type".format(command))
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand(command, *cmdargs)
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual, command)
# dict in arguments
for commandtype in [Command.CT_RUN, Command.CT_FETCH,
Command.CT_DELETE, Command.CT_ITERATE]:
command = "{}_client".format(Command.CT_LOOKUP[commandtype])
if commandtype == Command.CT_RUN:
command = "run_info"
elif commandtype == Command.CT_ITERATE:
command = command + "s"
cmdargs = [{"a":"a"}, "a"]
expected = (400, "command ({}) expects all arguments to be"\
" of simple type".format(command))
with self.assertRaises(P4RestCommandError) as cm:
c.validateCommand(command, *cmdargs)
actual = (cm.exception.httpstatuscode, cm.exception.value)
self.assertEqual(expected, actual, command)
p4api.disconnect()
def testExecute_Run(self):
p4api = P4()
p4api.user="user_0"
p4api.port=self.p4server.p4api.port
p4api.exception_level = P4.RAISE_ERRORS
p4api.connect()
p4api.run_login(password="Perf0rce")
command = "run_info"
c = Command(p4api, command)
result = c.executeCommand(command)
self.assertIsInstance(result, tuple, command)
(data, code) = result
self.assertEqual(200, code, command)
self.assertIsInstance(data, list, command)
self.assertEqual(1, len(data), command)
self.assertGreater(len(data[0]), 0, command)
resultlimit = 99
command = "run_files"
c = Command(p4api, command)
result = c.executeCommand(command, "//...", limit=resultlimit)
self.assertIsInstance(result, tuple, command)
(data, code) = result
self.assertEqual(200, code, command)
self.assertIsInstance(data, dict, command)
self.assertEqual(2, len(data), command)
self.assertEqual(99, len(data["results"]), command)
command = "run_filelog"
c = Command(p4api, command)
result = c.executeCommand(command, "//...", limit=resultlimit)
self.assertIsInstance(result, tuple, command)
(data, code) = result
self.assertEqual(200, code, command)
self.assertIsInstance(data, dict, command)
self.assertEqual(2, len(data), command)
self.assertEqual(99, len(data["results"]), command)
command = "run_files"
c = Command(p4api, command)
result = c.executeCommand(command, "//depot_2/baddir/...", "//depot_3/baddir/...",limit=resultlimit)
self.assertIsInstance(result, tuple, command)
(data, code) = result
self.assertEqual(200, code, command)
self.assertIsInstance(data, dict, command)
self.assertEqual(2, len(data), command)
self.assertEqual(0, len(data["results"]), command)
self.assertEqual(2, len(data["warnings"]), command)
p4api.disconnect()