lifted: Add delete_profile function and tests
Also adds a helper to providers, _get_profile_path, so that the code doesn't need to be repeated in all the functions.
This commit is contained in:
parent
5d7aa95f2f
commit
b2fc391677
@ -23,6 +23,35 @@ import stat
|
|||||||
import pylorax.api.toml as toml
|
import pylorax.api.toml as toml
|
||||||
|
|
||||||
|
|
||||||
|
def _get_profile_path(ucfg, provider_name, profile, exists=True):
|
||||||
|
"""Helper to return the directory and path for a provider's profile file
|
||||||
|
|
||||||
|
:param ucfg: upload config
|
||||||
|
:type ucfg: object
|
||||||
|
:param provider_name: the name of the cloud provider, e.g. "azure"
|
||||||
|
:type provider_name: str
|
||||||
|
:param profile: the name of the profile to save
|
||||||
|
:type profile: str != ""
|
||||||
|
:returns: Full path of the profile .toml file
|
||||||
|
:rtype: str
|
||||||
|
:raises: ValueError when passed invalid settings or an invalid profile name
|
||||||
|
:raises: RuntimeError when the provider or profile couldn't be found
|
||||||
|
"""
|
||||||
|
if not profile:
|
||||||
|
raise ValueError("Profile name cannot be empty!")
|
||||||
|
if not provider_name:
|
||||||
|
raise ValueError("Provider name cannot be empty!")
|
||||||
|
|
||||||
|
directory = os.path.join(ucfg["settings_dir"], provider_name)
|
||||||
|
# create the settings directory if it doesn't exist
|
||||||
|
os.makedirs(directory, exist_ok=True)
|
||||||
|
|
||||||
|
path = os.path.join(directory, f"{profile}.toml")
|
||||||
|
if exists and not os.path.isfile(path):
|
||||||
|
raise RuntimeError(f'Couldn\'t find profile "{profile}"!')
|
||||||
|
|
||||||
|
return path
|
||||||
|
|
||||||
def resolve_provider(ucfg, provider_name):
|
def resolve_provider(ucfg, provider_name):
|
||||||
"""Get information about the specified provider as defined in that
|
"""Get information about the specified provider as defined in that
|
||||||
provider's `provider.toml`, including the provider's display name and expected
|
provider's `provider.toml`, including the provider's display name and expected
|
||||||
@ -149,16 +178,9 @@ def save_settings(ucfg, provider_name, profile, settings):
|
|||||||
:type settings: dict
|
:type settings: dict
|
||||||
:raises: ValueError when passed invalid settings or an invalid profile name
|
:raises: ValueError when passed invalid settings or an invalid profile name
|
||||||
"""
|
"""
|
||||||
if not profile:
|
path = _get_profile_path(ucfg, provider_name, profile, exists=False)
|
||||||
raise ValueError("Profile name cannot be empty!")
|
|
||||||
validate_settings(ucfg, provider_name, settings, image_name=None)
|
validate_settings(ucfg, provider_name, settings, image_name=None)
|
||||||
|
|
||||||
directory = os.path.join(ucfg["settings_dir"], provider_name)
|
|
||||||
|
|
||||||
# create the settings directory if it doesn't exist
|
|
||||||
os.makedirs(directory, exist_ok=True)
|
|
||||||
|
|
||||||
path = os.path.join(directory, f"{profile}.toml")
|
|
||||||
# touch the TOML file if it doesn't exist
|
# touch the TOML file if it doesn't exist
|
||||||
if not os.path.isfile(path):
|
if not os.path.isfile(path):
|
||||||
open(path, "a").close()
|
open(path, "a").close()
|
||||||
@ -189,19 +211,26 @@ def load_settings(ucfg, provider_name, profile):
|
|||||||
This also calls validate_settings on the loaded settings, potentially
|
This also calls validate_settings on the loaded settings, potentially
|
||||||
raising an error if the saved settings are invalid.
|
raising an error if the saved settings are invalid.
|
||||||
"""
|
"""
|
||||||
if not profile:
|
path = _get_profile_path(ucfg, provider_name, profile)
|
||||||
raise ValueError("Profile name cannot be empty!")
|
|
||||||
if not provider_name:
|
|
||||||
raise ValueError("Provider name cannot be empty!")
|
|
||||||
directory = os.path.join(ucfg["settings_dir"], provider_name)
|
|
||||||
if not os.path.isdir(directory):
|
|
||||||
raise RuntimeError(f'Couldn\'t find provider "{provider_name}"!')
|
|
||||||
|
|
||||||
path = os.path.join(directory, f"{profile}.toml")
|
|
||||||
if not os.path.isfile(path):
|
|
||||||
raise RuntimeError(f'Couldn\'t find provider "{provider_name}"!')
|
|
||||||
|
|
||||||
with open(path) as file:
|
with open(path) as file:
|
||||||
settings = toml.load(file)
|
settings = toml.load(file)
|
||||||
validate_settings(ucfg, provider_name, settings)
|
validate_settings(ucfg, provider_name, settings)
|
||||||
return settings
|
return settings
|
||||||
|
|
||||||
|
def delete_profile(ucfg, provider_name, profile):
|
||||||
|
"""Delete a provider's profile settings file
|
||||||
|
|
||||||
|
:param ucfg: upload config
|
||||||
|
:type ucfg: object
|
||||||
|
:param provider_name: the name of the cloud provider, e.g. "azure"
|
||||||
|
:type provider_name: str
|
||||||
|
:param profile: the name of the profile to save
|
||||||
|
:type profile: str != ""
|
||||||
|
:raises: ValueError when passed invalid settings or an invalid profile name
|
||||||
|
:raises: RuntimeError when the provider or profile couldn't be found
|
||||||
|
"""
|
||||||
|
path = _get_profile_path(ucfg, provider_name, profile)
|
||||||
|
|
||||||
|
if os.path.exists(path):
|
||||||
|
os.unlink(path)
|
||||||
|
@ -21,7 +21,7 @@ import unittest
|
|||||||
|
|
||||||
import lifted.config
|
import lifted.config
|
||||||
from lifted.providers import list_providers, resolve_provider, resolve_playbook_path, save_settings
|
from lifted.providers import list_providers, resolve_provider, resolve_playbook_path, save_settings
|
||||||
from lifted.providers import load_profiles, validate_settings, load_settings
|
from lifted.providers import load_profiles, validate_settings, load_settings, delete_profile
|
||||||
import pylorax.api.config
|
import pylorax.api.config
|
||||||
from pylorax.sysutils import joinpaths
|
from pylorax.sysutils import joinpaths
|
||||||
|
|
||||||
@ -118,3 +118,31 @@ class ProvidersTestCase(unittest.TestCase):
|
|||||||
for p in list_providers(self.config["upload"]):
|
for p in list_providers(self.config["upload"]):
|
||||||
settings = load_settings(self.config["upload"], p, test_profiles[p][0])
|
settings = load_settings(self.config["upload"], p, test_profiles[p][0])
|
||||||
self.assertEqual(settings, test_profiles[p][1])
|
self.assertEqual(settings, test_profiles[p][1])
|
||||||
|
|
||||||
|
# This *must* run after all the save and load tests, but *before* the actual delete test
|
||||||
|
# _zz_ ensures this happens
|
||||||
|
def test_zz_delete_settings_errors(self):
|
||||||
|
"""Test raising the correct errors when deleting"""
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
delete_profile(self.config["upload"], "", "")
|
||||||
|
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
delete_profile(self.config["upload"], "", "default")
|
||||||
|
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
delete_profile(self.config["upload"], "azure", "")
|
||||||
|
|
||||||
|
with self.assertRaises(RuntimeError):
|
||||||
|
delete_profile(self.config["upload"], "azure", "missing-test")
|
||||||
|
|
||||||
|
# This *must* run after all the save and load tests, _zzz_ ensures this happens
|
||||||
|
def test_zzz_delete_settings(self):
|
||||||
|
"""Test raising the correct errors when deleting"""
|
||||||
|
# Ensure the profile is really there
|
||||||
|
settings = load_settings(self.config["upload"], "azure", test_profiles["azure"][0])
|
||||||
|
self.assertEqual(settings, test_profiles["azure"][1])
|
||||||
|
|
||||||
|
delete_profile(self.config["upload"], "azure", test_profiles["azure"][0])
|
||||||
|
|
||||||
|
with self.assertRaises(RuntimeError):
|
||||||
|
load_settings(self.config["upload"], "azure", test_profiles["azure"][0])
|
||||||
|
Loading…
Reference in New Issue
Block a user