from jsonschema import validate, ValidationError
import requests
-from dcae_cli.util import reraise_with_msg
+from dcae_cli.util import reraise_with_msg, fetch_file_from_web
from dcae_cli.util import config as cli_config
from dcae_cli.util.exc import DcaeException
from dcae_cli.util.logger import get_logger
# c) Both
#
-# TODO: Source this from app's configuration [ONAP URL TBD]
-_nexus_uri = "http://make-me-valid"
class FetchSchemaError(RuntimeError):
pass
-def _fetch_schema_from_web(server_uri, schema_path):
+def _fetch_schema(schema_path):
try:
- schema_url = "{0}/{1}".format(server_uri, schema_path)
- r = requests.get(schema_url)
- r.raise_for_status()
- return json.loads(r.text)
+ server_url = cli_config.get_server_url()
+ return fetch_file_from_web(server_url, schema_path)
except requests.HTTPError as e:
raise FetchSchemaError("HTTP error from fetching schema", e)
except Exception as e:
raise FetchSchemaError("Unexpected error from fetching schema", e)
-_fetch_schema_from_nexus = partial(_fetch_schema_from_web, _nexus_uri)
-
def _safe_dict(obj):
'''Returns a dict from a dict or json string'''
except FetchSchemaError as e:
reraise_with_msg(e, as_dcae=True)
-_validate_using_nexus = partial(_validate, _fetch_schema_from_nexus)
+_validate_using_nexus = partial(_validate, _fetch_schema)
_path_component_spec = cli_config.get_path_component_spec()
"""
# Apply health check defaults
healthcheck_type = config["healthcheck"]["type"]
- component_spec = _fetch_schema_from_nexus(_path_component_spec)
+ component_spec = _fetch_schema(_path_component_spec)
if healthcheck_type in ["http", "https"]:
apply_defaults_func = partial(apply_defaults,
import json
from click.testing import CliRunner
import time
+import pytest
from dcae_cli.cli import cli
from dcae_cli.catalog import MockCatalog
assert comp_model_spec == json.loads(spec_str)
+@pytest.mark.skip(reason="This is not a pure unit test. Need a way to setup dependencies and trigger in the appropriate stages of testing.")
def test_comp_cdap(obj=None):
"""
This is not a unit test. It is bigger than that. It Does a full "workflow" test:
generate_dir = os.path.join(TEST_DIR, 'mocked_components', 'model', 'generatedir')
cmd = "data_format generate --keywords name:1.0.2 {:} ".format(generate_dir).split()
- out_str = runner.invoke(cli, cmd, obj=obj).output
- assert '{\n "dataformatversion": "1.0.0", \n "jsonschema": {\n "$schema": "http://json-schema.org/draft-04/schema#", \n "description": "", \n "properties": {\n "foobar": {\n "description": "", \n "maxLength": 0, \n "minLength": 0, \n "pattern": "", \n "type": "string"\n }, \n "foobar2": {\n "description": "", \n "maxLength": 0, \n "minLength": 0, \n "pattern": "", \n "type": "string"\n }\n }, \n "type": "object"\n }, \n "self": {\n "description": "", \n "name": "name", \n "version": "1.0.2"\n }\n}\n' == out_str
+ actual = json.loads(runner.invoke(cli, cmd, obj=obj).output)
+ expected = json.loads('{\n "dataformatversion": "1.0.0", \n "jsonschema": {\n "$schema": "http://json-schema.org/draft-04/schema#", \n "description": "", \n "properties": {\n "foobar": {\n "description": "", \n "maxLength": 0, \n "minLength": 0, \n "pattern": "", \n "type": "string"\n }, \n "foobar2": {\n "description": "", \n "maxLength": 0, \n "minLength": 0, \n "pattern": "", \n "type": "string"\n }\n }, \n "type": "object"\n }, \n "self": {\n "description": "", \n "name": "name", \n "version": "1.0.2"\n }\n}\n')
+ assert actual == expected
generate_dir = os.path.join(TEST_DIR, 'mocked_components', 'model', 'generatedir')
cmd = "data_format generate name:1.0.2 {:} ".format(generate_dir).split()
- out_str = runner.invoke(cli, cmd, obj=obj).output
- assert '{\n "dataformatversion": "1.0.0", \n "jsonschema": {\n "$schema": "http://json-schema.org/draft-04/schema#", \n "description": "", \n "properties": {\n "foobar": {\n "description": "", \n "type": "string"\n }, \n "foobar2": {\n "description": "", \n "type": "string"\n }\n }, \n "type": "object"\n }, \n "self": {\n "description": "", \n "name": "name", \n "version": "1.0.2"\n }\n}\n' == out_str
+ actual = json.loads(runner.invoke(cli, cmd, obj=obj).output)
+ expected = json.loads('{\n "dataformatversion": "1.0.0", \n "jsonschema": {\n "$schema": "http://json-schema.org/draft-04/schema#", \n "description": "", \n "properties": {\n "foobar": {\n "description": "", \n "type": "string"\n }, \n "foobar2": {\n "description": "", \n "type": "string"\n }\n }, \n "type": "object"\n }, \n "self": {\n "description": "", \n "name": "name", \n "version": "1.0.2"\n }\n}\n'
+ )
+ assert actual == expected
generate_dir = os.path.join(TEST_DIR, 'mocked_components', 'model', 'generatedir', 'ex1.json')
cmd = "data_format generate name:1.0.2 {:} ".format(generate_dir).split()
- out_str = runner.invoke(cli, cmd, obj=obj).output
- assert '{\n "dataformatversion": "1.0.0", \n "jsonschema": {\n "$schema": "http://json-schema.org/draft-04/schema#", \n "additionalproperties": true, \n "description": "", \n "properties": {\n "foobar": {\n "description": "", \n "type": "string"\n }\n }, \n "required": [\n "foobar"\n ], \n "type": "object"\n }, \n "self": {\n "description": "", \n "name": "name", \n "version": "1.0.2"\n }\n}\n' == out_str
+ actual = json.loads(runner.invoke(cli, cmd, obj=obj).output)
+ expected = json.loads('{\n "dataformatversion": "1.0.0", \n "jsonschema": {\n "$schema": "http://json-schema.org/draft-04/schema#", \n "additionalproperties": true, \n "description": "", \n "properties": {\n "foobar": {\n "description": "", \n "type": "string"\n }\n }, \n "required": [\n "foobar"\n ], \n "type": "object"\n }, \n "self": {\n "description": "", \n "name": "name", \n "version": "1.0.2"\n }\n}\n')
+ assert actual == expected
if __name__ == '__main__':
raise DcaeException("File '{}' appears to be a malformed JSON.".format(path))
-def fetch_file_from_nexus(path, transform_func=json.loads):
- """Fetch file from nexus
+def fetch_file_from_web(server_url, path, transform_func=json.loads):
+ """Fetch file from a web server
The default behavior is to transform the response to a json.
"""
- # TODO: Source this from app's configuration [ONAP URL TBD]
- server_uri = "https://make-me-valid"
- nexus_url = "{0}/{1}".format(server_uri, path)
- r = requests.get(nexus_url)
+ artifact_url = "{0}/{1}".format(server_url, path)
+ r = requests.get(artifact_url)
r.raise_for_status()
if transform_func:
return transform_func(r.text)
else:
click.echo("Invalid user id. Please try again.")
+def _init_config_server_url():
+ return click.prompt('Please enter the remote server url', type=str).strip()
def _init_config():
'''Returns an initial dict for populating the config'''
# Grab the remote config and merge it in
try:
- new_config = util.fetch_file_from_nexus("/dcae-cli/config.json")
+ server_url = _init_config_server_url()
+ new_config = util.fetch_file_from_web(server_url, "/dcae-cli/config.json")
+ new_config["server_url"] = server_url
except:
# REVIEW: Should we allow users to manually setup their config if not
# able to pull from remote server?
'''Returns the configuration dictionary'''
return get_pref(get_config_path(), _init_config)
+def get_server_url():
+ """Returns the remote server url
+
+ The remote server holds the artifacts that the dcae-cli requires like the
+ seed config json and seed profiles json, and json schemas.
+ """
+ return get_config().get("server_url")
# These functions are used to fetch the configurable path to the various json
# schema files used in validation.
"""
mapping = defaultdict(set)
for instance in instances:
- match = _inst_re.match(instance_target)
+ match = _inst_re.match(instance)
if match is None:
continue
"""
# Grab the remote profiles and merge it in
try:
- new_profiles = util.fetch_file_from_nexus("/dcae-cli/profiles.json")
+ server_url = config.get_server_url()
+ new_profiles = util.fetch_file_from_web(server_url, "/dcae-cli/profiles.json")
except:
# REVIEW: Should we allow users to manually setup their config if not
# able to pull from remote server?
def test_init_config(monkeypatch):
monkeypatch.setattr(config, '_init_config_user', lambda: "bigmama")
- monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_nexus',
- lambda path: { "db_url": "conn" })
+ monkeypatch.setattr(config, '_init_config_server_url',
+ lambda: "http://some-nexus-in-the-sky.com")
+ monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_web',
+ lambda server_url, path: { "db_url": "conn" })
monkeypatch.setattr("dcae_cli._version.__version__", "2.X.X")
- expected = {'cli_version': '2.X.X', 'user': 'bigmama', 'db_url': 'conn'}
+ expected = {'cli_version': '2.X.X', 'user': 'bigmama', 'db_url': 'conn',
+ 'server_url': 'http://some-nexus-in-the-sky.com'}
assert expected == config._init_config()
# Test using of db fallback
- monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_nexus',
- lambda path: { "db_url": "" })
+ monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_web',
+ lambda server_url, path: { "db_url": "" })
assert "sqlite" in config._init_config()["db_url"]
- monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_nexus',
- lambda path: {})
+ monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_web',
+ lambda server_url, path: {})
assert "sqlite" in config._init_config()["db_url"]
# Simulate error trying to fetch
- def fetch_simulate_error(path):
+ def fetch_simulate_error(server_url, path):
raise RuntimeError("Simulated error")
- monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_nexus',
+ monkeypatch.setattr(dcae_cli.util, 'fetch_file_from_web',
fetch_simulate_error)
with pytest.raises(config.ConfigurationInitError):
instances_map = dis._make_instances_map(instances_latest_format)
assert instances_map.get(("sandbox.platform.laika", "0.5.0")) == set(instances_latest_format)
- instances_other_format = ["mike.39f809be-7d13-4799-b8d0-ba0d5358282b.0-5-0.sandbox-platform-laika.rework-central.dcae.com"]
-
- instances_map = dis._make_instances_map(instances_other_format)
- assert instances_map.get(("sandbox.platform.laika", "0.5.0")) == set(instances_other_format)
-
- instances_other_format = ["mike.b5435a8c-ef56-4371-94d8-560a61f7352f.0-5-0.sandbox-platform-laika.solutioning-central.dcae.com"]
-
- instances_map = dis._make_instances_map(instances_other_format)
- assert instances_map.get(("sandbox.platform.laika", "0.5.0")) == set(instances_other_format)
-
- instances_other_format = ["mike.b5435a8c-ef56-4371-94d8-560a61f7352f.0-5-0.sandbox-platform-laika.dcae.com"]
-
- instances_map = dis._make_instances_map(instances_other_format)
- assert instances_map.get(("sandbox.platform.laika", "0.5.0")) == set(instances_other_format)
-
def test_get_component_instances(monkeypatch):
instances = [
"consul_host": "realsolcnsl00.dcae.solutioning.com",
"docker_host": "realsoldokr00.dcae.solutioning.com:2376" }}
- def fetch_profile(target_profile, path):
+ def fetch_profile(target_profile, server_url, path):
return target_profile
- monkeypatch.setattr(util, "fetch_file_from_nexus", partial(fetch_profile,
+ monkeypatch.setattr(util, "fetch_file_from_web", partial(fetch_profile,
profile_dict))
profiles.reinit_profiles()
assert profiles.get_profiles(include_active=False) == profile_dict
"consul_host": "realsolcnsl00.dcae.solutioning.com",
"docker_host": "realsoldokr00.dcae.solutioning.com:2376" }}
- monkeypatch.setattr(util, "fetch_file_from_nexus", partial(fetch_profile,
+ monkeypatch.setattr(util, "fetch_file_from_web", partial(fetch_profile,
profile_dict))
profiles.reinit_profiles()
all_profiles = profiles.get_profiles(include_active=False)
# Test fetch failure
- def fetch_failure(path):
+ def fetch_failure(server_url, path):
raise RuntimeError("Mysterious error")
- monkeypatch.setattr(util, "fetch_file_from_nexus", fetch_failure)
+ monkeypatch.setattr(util, "fetch_file_from_web", fetch_failure)
with pytest.raises(profiles.ProfilesInitError):
profiles.reinit_profiles()
Code Review / dcaegen2 / platform / cli.git / commitdiff