dcae-cli/dcae_cli/util/tests/test_profiles.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   3 # ================================================================================
   4 # Copyright (c) 2017 AT&T Intellectual Property. All rights reserved.
   5 # ================================================================================
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #      http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17 # ============LICENSE_END=========================================================
  18 #
  19 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20
  21 # -*- coding: utf-8 -*-
  22 """
  23 Tests the profiles module
  24 """
  25 import os, json, copy
  26 from functools import partial
  27
  28 import click
  29 import pytest
  30
  31 from dcae_cli import util
  32 from dcae_cli.util.exc import DcaeException
  33 from dcae_cli.util import profiles
  34 from dcae_cli.util.profiles import (get_active_name, get_profile, get_profiles, get_profiles_path,
  35                                     create_profile, delete_profile, update_profile, ACTIVE,
  36                                     activate_profile, CONSUL_HOST)
  37 from dcae_cli.util import config
  38
  39
  40 def test_profiles(monkeypatch, tmpdir):
  41     '''Tests the creation and initialization of profiles on a clean install'''
  42     # Setup config
  43     config_dict = { "active_profile": "fake-solutioning", "db_url": "some-db" }
  44     config_file = tmpdir.join("config.json")
  45     config_file.write(json.dumps(config_dict))
  46
  47     # Setup profile
  48     profile_dict = { "fake-solutioning": { "cdap_broker": "cdap_broker",
  49         "config_binding_service": "config_binding_service",
  50         "consul_host": "realsolcnsl00.dcae.solutioning.com",
  51         "docker_host": "realsoldokr00.dcae.solutioning.com:2376" }}
  52     profile_file = tmpdir.join("profiles.json")
  53     profile_file.write(json.dumps(profile_dict))
  54
  55     monkeypatch.setattr(click, "get_app_dir", lambda app: str(tmpdir.realpath()))
  56
  57     assert get_active_name() == config_dict["active_profile"]
  58     assert get_profile() == profiles.Profile(**profile_dict["fake-solutioning"])
  59
  60     # Failures looking for unknown profile
  61
  62     with pytest.raises(DcaeException):
  63         get_profile('foo')
  64
  65     with pytest.raises(DcaeException):
  66         delete_profile('foo')
  67
  68     with pytest.raises(DcaeException):
  69         update_profile('foo', **{})  # doesn't exist
  70
  71     # Cannot delete active profile
  72
  73     assert delete_profile(get_active_name()) == False
  74
  75     # Do different get_profiles queries
  76
  77     assert get_profiles(user_only=True) == profile_dict
  78     all_profiles = copy.deepcopy(profile_dict)
  79     all_profiles[ACTIVE] = profile_dict["fake-solutioning"]
  80     assert get_profiles(user_only=False) == all_profiles
  81
  82     # Create and activate new profile
  83
  84     create_profile('foo')
  85     activate_profile('foo')
  86     assert get_active_name() == 'foo'
  87
  88     # Update new profile
  89
  90     update_profile('foo', **{CONSUL_HOST:'bar'})
  91     assert get_profiles()['foo'][CONSUL_HOST] == 'bar'
  92     assert get_profile()._asdict()[CONSUL_HOST] == 'bar'
  93
  94     activate_profile("fake-solutioning")
  95     assert delete_profile('foo') == True
  96
  97
  98 def test_reinit_via_get_profiles(monkeypatch, tmpdir):
  99     monkeypatch.setattr(click, "get_app_dir", lambda app: str(tmpdir.realpath()))
 100
 101     def fake_reinit_failure():
 102         raise profiles.ProfilesInitError("Faked failure")
 103
 104     monkeypatch.setattr(profiles, "reinit_profiles", fake_reinit_failure)
 105
 106     with pytest.raises(DcaeException):
 107         get_profiles()
 108
 109
 110 def test_reinit_profiles(monkeypatch, tmpdir):
 111     monkeypatch.setattr(click, "get_app_dir", lambda app: str(tmpdir.realpath()))
 112
 113     # Setup config (need this because the "active_profile" is needed)
 114     config_dict = { "active_profile": "fake-solutioning", "db_url": "some-db" }
 115     config_file = tmpdir.join("config.json")
 116     config_file.write(json.dumps(config_dict))
 117
 118     # Start with empty profiles
 119
 120     profile_dict = { "fake-solutioning": { "cdap_broker": "cdap_broker",
 121         "config_binding_service": "config_binding_service",
 122         "consul_host": "realsolcnsl00.dcae.solutioning.com",
 123         "docker_host": "realsoldokr00.dcae.solutioning.com:2376" }}
 124
 125     def fetch_profile(target_profile, path):
 126         return target_profile
 127
 128     monkeypatch.setattr(util, "fetch_file_from_nexus", partial(fetch_profile,
 129         profile_dict))
 130     profiles.reinit_profiles()
 131     assert profiles.get_profiles(include_active=False) == profile_dict
 132
 133     # Test update
 134
 135     profile_dict = { "fake-5g": { "cdap_broker": "cdap_broker",
 136         "config_binding_service": "config_binding_service",
 137         "consul_host": "realsolcnsl00.dcae.solutioning.com",
 138         "docker_host": "realsoldokr00.dcae.solutioning.com:2376" }}
 139
 140     monkeypatch.setattr(util, "fetch_file_from_nexus", partial(fetch_profile,
 141         profile_dict))
 142     profiles.reinit_profiles()
 143     all_profiles = profiles.get_profiles(include_active=False)
 144     assert "fake-5g" in all_profiles
 145     assert "fake-solutioning" in all_profiles
 146
 147     # Test fetch failure
 148
 149     def fetch_failure(path):
 150         raise RuntimeError("Mysterious error")
 151
 152     monkeypatch.setattr(util, "fetch_file_from_nexus", fetch_failure)
 153
 154     with pytest.raises(profiles.ProfilesInitError):
 155         profiles.reinit_profiles()
 156
 157
 158 if __name__ == '__main__':
 159     '''Test area'''
 160     pytest.main([__file__, ])