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