''' 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()