Minor Fix for rm-dbs command
This commit is contained in:
parent
76f9463910
commit
30e3c17920
@ -2,6 +2,8 @@ import json
|
||||
from pathlib import Path
|
||||
import logging
|
||||
|
||||
from iottb import definitions
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DB_NAME = 'iottb.db'
|
||||
@ -18,10 +20,10 @@ class IottbConfig:
|
||||
def warn():
|
||||
logger.warning(f'DatabaseLocations are DatabaseLocationMap in the class {__name__}')
|
||||
|
||||
def __init__(self, cfg_file):
|
||||
def __init__(self, cfg_file=definitions.CFG_FILE_PATH):
|
||||
logger.info('Initializing Config object')
|
||||
IottbConfig.warn()
|
||||
self.cfg_file = cfg_file
|
||||
self.cfg_file = Path(cfg_file)
|
||||
self.default_database = None
|
||||
self.default_path = None
|
||||
self.DatabaseLocationMap = {}
|
||||
@ -98,6 +100,11 @@ class IottbConfig:
|
||||
|
||||
return self.DatabaseLocationMap.keys()
|
||||
|
||||
def get_know_database_paths(self):
|
||||
"""Get the paths of all known databases"""
|
||||
logger.info(f'Getting known database paths.')
|
||||
return self.DatabaseLocationMap.values()
|
||||
|
||||
|
||||
# TODO: Know issue:
|
||||
class Database:
|
||||
|
||||
@ -20,5 +20,5 @@ LOGFILE_LOG_FORMAT = {
|
||||
MAX_VERBOSITY = len(CONSOLE_LOG_FORMATS) - 1
|
||||
assert len(LOGFILE_LOG_FORMAT) == len(CONSOLE_LOG_FORMATS), 'Log formats must be same size'
|
||||
|
||||
LOGLEVEL = logging.WARNING
|
||||
LOGLEVEL = logging.DEBUG
|
||||
LOGDIR = Path.cwd() / 'logs'
|
||||
|
||||
@ -10,6 +10,7 @@ from iottb.utils.logger_config import setup_logging
|
||||
from iottb import definitions
|
||||
from iottb.contexts import IottbConfig
|
||||
from iottb.commands.initialize_testbed import init_db
|
||||
|
||||
############################################################################
|
||||
# Module shortcuts for global definitions
|
||||
###########################################################################
|
||||
@ -51,7 +52,6 @@ def set_key_in_table_to(ctx, file, table, key, value):
|
||||
"""Edit config or metadata files. TODO: Implement"""
|
||||
click.echo(f'set_key_in_table_to invoked')
|
||||
logger.warning("Unimplemented subcommand invoked.")
|
||||
pass
|
||||
|
||||
|
||||
@click.command()
|
||||
@ -66,27 +66,33 @@ def rm_cfg():
|
||||
click.echo(f'Iottb configuration removed at {CFG_FILE_PATH}')
|
||||
|
||||
|
||||
@click.command
|
||||
@click.option('--db', default=str(Path.home() / 'iottb.db'), help='Delete database')
|
||||
@click.confirmation_option(prompt="Are you certain that you want to delete the database file?")
|
||||
def rm_db(db):
|
||||
""" Removes database from the filesystem.
|
||||
@click.command()
|
||||
@click.confirmation_option(prompt="Are you certain that you want to delete the databases file?")
|
||||
def rm_dbs(dbs):
|
||||
""" Removes ALL(!) databases from the filesystem if they're empty.
|
||||
|
||||
This is mostly a utility during development. Once non-standard database locations are implemented,
|
||||
deleting this would lead to iottb not being able to find them anymore.
|
||||
Development utility currently unfit for use.
|
||||
"""
|
||||
Path(db).unlink()
|
||||
click.echo(f'Iottb configuration removed at {CFG_FILE_PATH}')
|
||||
config = IottbConfig()
|
||||
paths = config.get_know_database_paths()
|
||||
logger.debug(f'Known db paths: {str(paths)}')
|
||||
for dbs in paths:
|
||||
try:
|
||||
Path(dbs).rmdir()
|
||||
click.echo(f'{dbs} deleted')
|
||||
except Exception as e:
|
||||
logger.debug(f'Failed unlinking db {dbs} with error {e}')
|
||||
logger.info(f'All databases deleted')
|
||||
|
||||
|
||||
##################################################################################
|
||||
# Add all subcommands to group here
|
||||
#################################################################################
|
||||
# noinspection PyTypeChecker
|
||||
cli.add_command(init_db)
|
||||
cli.add_command(rm_cfg)
|
||||
cli.add_command(set_key_in_table_to)
|
||||
cli.add_command(rm_db)
|
||||
|
||||
cli.add_command(rm_dbs)
|
||||
|
||||
if __name__ == '__main__':
|
||||
cli(auto_envvar_prefix='IOTTB')
|
||||
|
||||
@ -1,33 +0,0 @@
|
||||
from pathlib import Path
|
||||
from iottb.main import load_config
|
||||
|
||||
|
||||
class TestLoadConfig:
|
||||
|
||||
# Loads configuration from an existing file
|
||||
def test_loads_config_from_existing_file(self, mocker):
|
||||
cfg_file = mocker.Mock()
|
||||
cfg_file.is_file.return_value = True
|
||||
mock_open = mocker.mock_open(
|
||||
read_data=f'{{"DefaultDatabase": "test_db", "DefaultDatabasePath": "{Path.home()}/user"}}')
|
||||
mocker.patch('builtins.open', mock_open)
|
||||
|
||||
result = load_config(cfg_file)
|
||||
|
||||
assert result == {"DefaultDatabase": "test_db", "DefaultDatabasePath": f"{Path.home()}/user"}
|
||||
cfg_file.is_file.assert_called_once()
|
||||
mock_open.assert_called_once_with(cfg_file, 'r')
|
||||
|
||||
# File path is invalid or inaccessible
|
||||
def test_file_path_invalid_or_inaccessible(self, mocker):
|
||||
cfg_file = mocker.Mock()
|
||||
cfg_file.is_file.return_value = False
|
||||
mock_create_default_config = mocker.patch('iottb.main.create_default_config',
|
||||
return_value={"DefaultDatabase": "default_db",
|
||||
"DefaultDatabasePath": f"{Path.home()}/default"})
|
||||
|
||||
result = load_config(cfg_file)
|
||||
|
||||
assert result == {"DefaultDatabase": "default_db", "DefaultDatabasePath": f"{Path.home()}/default"}
|
||||
cfg_file.is_file.assert_called_once()
|
||||
mock_create_default_config.assert_called_once_with(cfg_file)
|
||||
Loading…
x
Reference in New Issue
Block a user