''' Created on Nov 20, 2017 @author: Charlie McLouth ''' import unittest from p4rest.flask.config import P4RestAppConfig import os.path from p4rest.util import osenviron, osremovefile, osmakedirs, rmtree, mkdtemp import logging logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) # debugHandler = logging.StreamHandler() # debugHandler.setFormatter(logging.Formatter("%(levelname)s:%(filename)s:" # "%(lineno)d:%(funcName)s:" # "%(message)s")) # logger.addHandler(debugHandler) class TestP4RestAppConfig(unittest.TestCase): '''Testing P4RestAppConfig''' def setUp(self): self.orgEnviron = {} for k,v in osenviron.items(): if k.startswith("P4"): self.orgEnviron[k] = v self.testdir = os.path.abspath(".") if not os.path.isdir(self.testdir): raise Exception("Test directory ({}) does not exist." "".format(self.testdir)) self.testdir = mkdtemp(suffix=None, prefix=__name__, dir=self.testdir) self.configKey = "TestP4RestAppConfig_key" def tearDown(self): if os.path.isdir(self.testdir): rmtree(self.testdir) else: logger.warn('not a directory?:' + self.testdir) osenviron.update(self.orgEnviron) todelete = [] for k in osenviron.keys(): if k.startswith("P4"): if k not in self.orgEnviron: todelete.append(k) for k in todelete: del(osenviron[k]) if self.configKey in osenviron: del(osenviron[self.configKey]) def test_validation(self): defaultConfig = {} appconfig = P4RestAppConfig(self.testdir, defaultConfig) appconfig.set_program_defaults() # raise exception if key not specified with self.assertRaises(RuntimeError) as cm: appconfig.validate(self.configKey) expected = "The environment variable {} is not set and as such " \ "configuration could not be loaded.".format(self.configKey) actual = cm.exception.args[0] self.assertTrue(actual.startswith(expected)) # specify key as environment variable configfile = os.path.join(self.testdir, "test.json") osenviron[self.configKey] = configfile appconfig.validate(self.configKey) del(osenviron[self.configKey]) # raise exception if key not specified with self.assertRaises(RuntimeError) as cm: appconfig.validate(self.configKey) expected = "The environment variable {} is not set and as such " \ "configuration could not be loaded.".format(self.configKey) actual = cm.exception.args[0] self.assertTrue(actual.startswith(expected)) # specify key as config entry appconfig[self.configKey] = configfile appconfig.validate(self.configKey) # test for missing required values for k in P4RestAppConfig.managed_keys: v = appconfig.pop(k) with self.assertRaises(RuntimeError) as cm: appconfig.validate(self.configKey) expected = "Configuration item not defined: {}".format(k) actual = cm.exception.args[0] self.assertEqual(expected, actual) appconfig[k] = v # testing workdir k = "WORKDIR" v = appconfig[k] appconfig[k] = "relativepath" with self.assertRaises(RuntimeError) as cm: appconfig.validate(self.configKey) expected = "Invalid value for configuration item value '{}': {}" \ "".format(k, appconfig[k]) actual = cm.exception.args[0] self.assertEqual(expected, actual) appconfig[k] = v # fail to create directory rmtree(v, False) with open(v, "w"): pass with self.assertRaises(FileExistsError) as cm: appconfig.validate(self.configKey) osremovefile(v) appconfig.validate(self.configKey) # testing P4TRUST k = "P4TRUST" v = appconfig[k] appconfig[k] = "relativepath/.p4trust" with open(os.path.join(appconfig["WORKDIR"], "relativepath"), "w"): pass with self.assertRaises(FileExistsError) as cm: appconfig.validate(self.configKey) osremovefile(os.path.join(appconfig["WORKDIR"], "relativepath")) appconfig[k] = v appconfig.validate(self.configKey) osremovefile(os.path.join(appconfig["WORKDIR"], v)) osmakedirs(os.path.join(appconfig["WORKDIR"], v), exist_ok=True) with self.assertRaises(PermissionError) as cm: appconfig.validate(self.configKey) expected = appconfig[k] actual = osenviron[k] self.assertEqual(expected, actual) self.assertTrue(os.path.isabs(expected)) rmtree(os.path.join(appconfig["WORKDIR"], v), False) appconfig.validate(self.configKey) # testing P4TRUST k = "P4TICKETS" v = appconfig[k] appconfig[k] = "relativepath/.p4tickets" with open(os.path.join(appconfig["WORKDIR"], "relativepath"), "w"): pass with self.assertRaises(FileExistsError) as cm: appconfig.validate(self.configKey) osremovefile(os.path.join(appconfig["WORKDIR"], "relativepath")) appconfig[k] = v appconfig.validate(self.configKey) osremovefile(os.path.join(appconfig["WORKDIR"], v)) osmakedirs(os.path.join(appconfig["WORKDIR"], v), exist_ok=True) with self.assertRaises(PermissionError) as cm: appconfig.validate(self.configKey) expected = appconfig[k] actual = osenviron[k] self.assertEqual(expected, actual) self.assertTrue(os.path.isabs(expected)) rmtree(os.path.join(appconfig["WORKDIR"], v), False) appconfig.validate(self.configKey) # testing P4CONFIG k = "P4CONFIG" v = appconfig[k] expected = ".p4testconf" appconfig[k] = expected appconfig.validate(self.configKey) actual = osenviron["P4CONFIG"] self.assertEqual(expected, actual) def test_dumpload(self): defaultConfig = {} appconfig = P4RestAppConfig(self.testdir, defaultConfig) appconfig.set_program_defaults() configfile = os.path.join(self.testdir, "test.json") appconfig[self.configKey] = configfile appconfig["P4TRUST"] = os.path.join(appconfig["WORKDIR"], "test.p4trust") appconfig["P4TICKETS"] = os.path.join(appconfig["WORKDIR"], "test.p4tickets") appconfig.validate(self.configKey) with self.assertRaises(FileNotFoundError): appconfig.dump_to_json(os.path.join(self.testdir, "baddir", "badfile.json"), False) appconfig.dump_to_json(configfile, False) appconfig2 = P4RestAppConfig(self.testdir, defaultConfig) appconfig2.from_json(configfile, False) for k in P4RestAppConfig.managed_keys: expected = appconfig[k] actual = appconfig2[k] if k == "P4TRUST": self.assertTrue(os.path.isabs(expected)) self.assertFalse(os.path.isabs(actual)) self.assertEqual(expected, os.path.join(appconfig2["WORKDIR"], actual)) elif k == "P4TICKETS": self.assertTrue(os.path.isabs(expected)) self.assertFalse(os.path.isabs(actual)) self.assertEqual(expected, os.path.join(appconfig2["WORKDIR"], actual)) else: self.assertEqual(expected, actual) # confirm that the config is not reloaded self.assertFalse(appconfig.from_json_if(self.configKey, False)) for k in P4RestAppConfig.managed_keys: expected = appconfig[k] actual = appconfig2[k] if k in ["P4TRUST", "P4TICKETS"]: self.assertTrue(os.path.isabs(expected)) self.assertFalse(os.path.isabs(actual)) self.assertEqual(expected, os.path.join(appconfig2["WORKDIR"], actual)) else: self.assertEqual(expected, actual) # confirm that the config is reloaded by updating the file timestamp with open(configfile, "a") as f: f.write("\n") self.assertTrue(appconfig.from_json_if(self.configKey, False)) for k in P4RestAppConfig.managed_keys: expected = appconfig[k] actual = appconfig2[k] self.assertEqual(expected, actual) if k in ["P4TRUST", "P4TICKETS"]: self.assertFalse(os.path.isabs(actual)) appconfig.validate(self.configKey) for k in ["P4TRUST", "P4TICKETS"]: self.assertTrue(os.path.isabs(appconfig[k])) if __name__ == "__main__": #import sys;sys.argv = ['', 'Test.testName'] unittest.main()