blob: b98b817841755a940041c72f5315a2b285b6ef09 [file] [log] [blame]
import unittest
from mock import patch
import os
from xosconfig import Config
from xosconfig import Config as Config2
basic_conf = os.path.abspath(os.path.dirname(os.path.realpath(__file__)) + "/confs/basic_conf.yaml")
yaml_not_valid = os.path.abspath(os.path.dirname(os.path.realpath(__file__)) + "/confs/yaml_not_valid.yaml")
invalid_format = os.path.abspath(os.path.dirname(os.path.realpath(__file__)) + "/confs/invalid_format.yaml")
sample_conf = os.path.abspath(os.path.dirname(os.path.realpath(__file__)) + "/confs/sample_conf.yaml")
small_schema = os.path.abspath(os.path.dirname(os.path.realpath(__file__)) + "/schemas/small_schema.yaml")
services_list = {
"xos-ws": [],
"xos-db": [],
}
db_service = [
{
"ModifyIndex": 6,
"CreateIndex": 6,
"Node": "0152982c3159",
"Address": "172.19.0.2",
"ServiceID": "0d53ce210785:frontend_xos_db_1:5432",
"ServiceName": "xos-db",
"ServiceTags": [],
"ServiceAddress": "172.18.0.4",
"ServicePort": 5432,
"ServiceEnableTagOverride": "false"
}
]
class XOSConfigTest(unittest.TestCase):
"""
Testing the XOS Config Module
"""
def tearDown(self):
# NOTE clear the config after each test
Config.clear()
def test_initialize_only_once(self):
"""
[XOS-Config] Raise if initialized twice
"""
with self.assertRaises(Exception) as e:
Config.init(sample_conf)
Config2.init(sample_conf)
self.assertEqual(e.exception.message, "[XOS-Config] Module already initialized")
def test_config_not_initialized(self):
"""
[XOS-Config] Raise if accessing properties without initialization
"""
with self.assertRaises(Exception) as e:
Config.get("database")
self.assertEqual(e.exception.message, "[XOS-Config] Module has not been initialized")
def test_missing_file_exception(self):
"""
[XOS-Config] Raise if file not found
"""
with self.assertRaises(Exception) as e:
Config.init("missing_conf")
self.assertEqual(e.exception.message, "[XOS-Config] Config file not found at: missing_conf")
def test_yaml_not_valid(self):
"""
[XOS-Config] Raise if yaml is not valid
"""
with self.assertRaises(Exception) as e:
Config.init(yaml_not_valid)
self.assertEqual(e.exception.message, "[XOS-Config] The config format is wrong: Unable to load any data from source yaml file")
def test_invalid_format(self):
"""
[XOS-Config] Raise if format is not valid (we expect a dictionary)
"""
with self.assertRaises(Exception) as e:
Config.init(invalid_format)
self.assertEqual(e.exception.message, "[XOS-Config] The config format is wrong: Schema validation failed:\n - Value '['I am', 'a yaml', 'but the', 'format is not', 'correct']' is not a dict. Value path: ''.")
def test_env_override(self):
"""
[XOS-Config] the XOS_CONFIG_FILE environment variable should override the config_file
"""
os.environ["XOS_CONFIG_FILE"] = "env.yaml"
with self.assertRaises(Exception) as e:
Config.init("missing_conf")
self.assertEqual(e.exception.message, "[XOS-Config] Config file not found at: env.yaml")
del os.environ["XOS_CONFIG_FILE"]
def test_schema_override(self):
"""
[XOS-Config] the XOS_CONFIG_SCHEMA environment variable should override the config_schema
"""
os.environ["XOS_CONFIG_SCHEMA"] = "env-schema.yaml"
with self.assertRaises(Exception) as e:
Config.init(basic_conf)
self.assertRegexpMatches(e.exception.message, '\[XOS\-Config\] Config schema not found at: (.+)env-schema\.yaml')
# self.assertEqual(e.exception.message, "[XOS-Config] Config schema not found at: env-schema.yaml")
del os.environ["XOS_CONFIG_SCHEMA"]
def test_schema_override_usage(self):
"""
[XOS-Config] the XOS_CONFIG_SCHEMA should be used to validate a config
"""
os.environ["XOS_CONFIG_SCHEMA"] = small_schema
with self.assertRaises(Exception) as e:
Config.init(basic_conf)
self.assertEqual(e.exception.message, "[XOS-Config] The config format is wrong: Schema validation failed:\n - Key 'database' was not defined. Path: ''.")
del os.environ["XOS_CONFIG_SCHEMA"]
def test_get_cli_param(self):
"""
[XOS-Config] Should read CLI -C param
"""
args = ["-A", "Foo", "-c", "Bar", "-C", "config.yaml"]
res = Config.get_cli_param(args)
self.assertEqual(res, "config.yaml")
def test_get_default_val_for_missing_param(self):
"""
[XOS-Config] Should get the default value if nothing is specified
"""
Config.init(basic_conf)
log = Config.get("logging")
self.assertEqual(log, {
"level": "info",
"channels": ["file", "console"],
"logstash_hostport": "cordloghost:5617",
"file": "/var/log/xos.log",
})
def _test_get_missing_param(self):
"""
[XOS-Config] Should raise reading a missing param
"""
Config.init(sample_conf)
res = Config.get("foo")
self.assertEqual(res, None)
def test_get_first_level(self):
"""
[XOS-Config] Should return a first level param
"""
Config.init(sample_conf)
# NOTE we are using Config2 here to be sure that the configuration is readable from any import,
# not only from the one that has been used to initialize it
res = Config2.get("database")
self.assertEqual(res, {
"name": "xos",
"username": "test",
"password": "safe"
})
def _test_get_child_level(self):
"""
[XOS-Config] Should return a child level param
"""
Config.init(sample_conf)
res = Config.get("nested.parameter.for")
self.assertEqual(res, "testing")
def test_get_service_list(self):
"""
[XOS-Config] Should query registrator and return a list of services
"""
with patch("xosconfig.config.requests.get") as mock_get:
mock_get.return_value.json.return_value = services_list
res = Config.get_service_list()
self.assertEqual(res, [
"xos-ws",
"xos-db",
])
def test_get_service_info(self):
"""
[XOS-Config] Should query registrator and return service info
"""
with patch("xosconfig.config.requests.get") as mock_get:
mock_get.return_value.json.return_value = db_service
info = Config.get_service_info("xos-db")
self.assertEqual(info, {
"name": "xos-db",
"url": "172.18.0.4",
"port": 5432
})
def test_fail_get_service_info(self):
"""
[XOS-Config] Should query registrator and return an exception if it"s down
"""
with patch("xosconfig.config.requests.get") as mock_get:
mock_get.return_value.ok = False
with self.assertRaises(Exception) as e:
Config.get_service_info("missing-service")
self.assertEqual(e.exception.message, "[XOS-Config] Registrator is down")
def test_missing_get_service_info(self):
"""
[XOS-Config] Should query registrator and return an exception if service is not there
"""
with patch("xosconfig.config.requests.get") as mock_get:
mock_get.return_value.json.return_value = []
with self.assertRaises(Exception) as e:
Config.get_service_info("missing-service")
self.assertEqual(e.exception.message, "[XOS-Config] The service missing-service looking for does not exist")
def test_get_service_endpoint(self):
"""
[XOS-Config] Should query registrator and return service endpoint
"""
with patch("xosconfig.config.requests.get") as mock_get:
mock_get.return_value.json.return_value = db_service
endpoint = Config.get_service_endpoint("xos-db")
self.assertEqual(endpoint, "http://172.18.0.4:5432")
if __name__ == '__main__':
unittest.main()