diff --git a/tests/unit_tests/test_tethys_cli/test_proxyapps_commands.py b/tests/unit_tests/test_tethys_cli/test_proxyapps_commands.py index fc484ecd0..57f220c59 100644 --- a/tests/unit_tests/test_tethys_cli/test_proxyapps_commands.py +++ b/tests/unit_tests/test_tethys_cli/test_proxyapps_commands.py @@ -1,16 +1,20 @@ from tethys_apps.models import ProxyApp from unittest import mock -from io import StringIO -from tethys_cli.proxyapps_commands import add_proxyapp, update_proxyapp, list_apps +from tethys_cli.proxyapps_commands import ( + add_proxyapp, + update_proxyapp, + list_proxyapps, + get_engine, +) from django.test import TestCase import unittest -class TestProxyAppsCommand(TestCase): +class TestProxyAppsCommand(unittest.TestCase): def setUp(self): - self.app_name = "My Proxy App for Testing" + self.app_name = "My_Proxy_App_for_Testing" self.endpoint = "http://foo.example.com/my-proxy-app" self.back_url = "http://bar.example.com/apps/" self.logo = "http://foo.example.com/my-proxy-app/logo.png" @@ -28,16 +32,147 @@ def setUp(self): open_in_new_tab=self.open_in_new_tab, ) self.proxy_app.save() - def tearDown(self): - print("hellos") self.proxy_app.delete() - - @mock.patch('sys.stdout', new_callable=StringIO) - def test_list_proxy_apps(self,mock_stdout): - expected_output = "Proxy Apps:\n My Proxy App for Testing" - list_apps() - self.assertEqual(mock_stdout.getvalue().strip(), expected_output) - - \ No newline at end of file + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + @mock.patch("tethys_cli.proxyapps_commands.read_settings") + def test_get_engine_no_db_settings_error(self, mock_settings, mock_write_error): + mock_settings.return_value = {} + get_engine() + mock_write_error.assert_called_with( + "No database settings defined in the portal_config.yml file" + ) + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + @mock.patch("tethys_cli.proxyapps_commands.read_settings") + def test_get_engine_no_default_db_settings_error( + self, mock_settings, mock_write_error + ): + mock_settings.return_value = {"DATABASES": {}} + get_engine() + mock_write_error.assert_called_with( + "No default database defined in the portal_config.yml file" + ) + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + @mock.patch("tethys_cli.proxyapps_commands.read_settings") + def test_get_engine_connection_error(self, mock_settings, mock_write_error): + # database settings is empty here, so it will cause an error + mock_settings.return_value = { + "DATABASES": {"default": {"ENGINE": "django.db.backends.postgresql"}} + } + get_engine() + mock_write_error.assert_called_with("Error when connecting to the database") + + @mock.patch("tethys_cli.proxyapps_commands.create_engine") + @mock.patch("tethys_cli.proxyapps_commands.read_settings") + def test_get_engine_sqlite(self, mock_settings, mock_create_engine): + mock_settings.return_value = { + "DATABASES": { + "default": {"ENGINE": "django.db.backends.sqlite3", "NAME": "db_name"} + } + } + mock_engine = mock_create_engine.return_value + mock_engine.connect.return_value = "Mocked SQLite Engine" + + # Call the function that uses create_engine + engine = get_engine() + + # Assertions + mock_create_engine.assert_called_once() + mock_create_engine.assert_called_with("sqlite:///db_name", pool_pre_ping=True) + + @mock.patch("tethys_cli.proxyapps_commands.write_info") + @mock.patch("tethys_cli.proxyapps_commands.print") + def test_list_proxy_apps(self, mock_print, mock_write_info): + list_proxyapps() + rts_call_args = mock_print.call_args_list + check_list = [] + + for i in range(len(rts_call_args)): + check_list.append(rts_call_args[i][0][0]) + + mock_write_info.assert_called_with("Proxy Apps:") + self.assertIn(" My_Proxy_App_for_Testing", check_list) + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_update_proxy_apps_no_app_name(self, mock_write_error): + mock_args = [] + update_proxyapp(mock_args) + mock_write_error.assert_called_with("proxy_app_name cannot be empty") + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_update_proxy_apps_no_app_key_name(self, mock_write_error): + mock_args = [self.app_name] + update_proxyapp(mock_args) + mock_write_error.assert_called_with("proxy_app_key cannot be empty") + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_update_proxy_apps_no_app_value_name(self, mock_write_error): + mock_args = [self.app_name, "logo_url"] + update_proxyapp(mock_args) + mock_write_error.assert_called_with("proxy_app_value cannot be empty") + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_update_proxy_apps_no_app(self, mock_write_error): + mock_args = ["My_Proxy_App_for_Testing2", "logo_url", "https://fake.com"] + update_proxyapp(mock_args) + mock_write_error.assert_called_with( + f"Proxy app My_Proxy_App_for_Testing2 does not exits" + ) + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_update_proxy_apps_no_correct_key(self, mock_write_error): + mock_args = [ + self.app_name, + "non_existing_key", + "https://fake.com", + ] + update_proxyapp(mock_args) + mock_write_error.assert_called_with( + f"Attribute non_existing_key does not exists in Proxy app {self.app_name}" + ) + + @mock.patch("tethys_cli.proxyapps_commands.write_success") + def test_update_proxy_apps(self, mock_write_success): + mock_args = [ + self.app_name, + "logo_url", + "https://fake.com", + ] + update_proxyapp(mock_args) + mock_write_success.assert_called_with(f"Proxy app {self.app_name} was updated") + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_add_proxy_apps_no_app_name(self, mock_write_error): + mock_args = [] + add_proxyapp(mock_args) + mock_write_error.assert_called_with(f"proxy_app_name argument cannot be empty") + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_add_proxy_apps_no_endpoint(self, mock_write_error): + mock_args = ["new_proxy_app"] + add_proxyapp(mock_args) + mock_write_error.assert_called_with( + f"proxy_app_endpoint argument cannot be empty" + ) + + @mock.patch("tethys_cli.proxyapps_commands.write_error") + def test_add_proxy_apps_with_existing_proxy_app(self, mock_write_error): + mock_args = ["My_Proxy_App_for_Testing", "http://foo.example.com/my-proxy-app"] + add_proxyapp(mock_args) + mock_write_error.assert_called_with( + f"There is already a proxy app with that name: {self.app_name}" + ) + + @mock.patch("tethys_cli.proxyapps_commands.write_success") + def test_add_proxyapp_only_two_arguments(self, mock_write_success): + app_name_mock = "My_Proxy_App_for_Testing_2" + endpoint_mock = "http://foo.example.com/my-proxy-app" + mock_args = [app_name_mock, endpoint_mock] + add_proxyapp(mock_args) + mock_write_success.assert_called_with(f"Proxy app {app_name_mock} added") + new_proxy_app = ProxyApp.objects.get(name=app_name_mock) + new_proxy_app.delete() diff --git a/tethys_cli/proxyapps_commands.py b/tethys_cli/proxyapps_commands.py index 1aefac678..68c9eb55f 100644 --- a/tethys_cli/proxyapps_commands.py +++ b/tethys_cli/proxyapps_commands.py @@ -3,14 +3,21 @@ write_success, write_info, ) -from django.db import IntegrityError +from pathlib import Path -def add_proxyapps_parser(subparsers): - import django +from sqlalchemy import create_engine, exc +from sqlalchemy.ext.automap import automap_base +from sqlalchemy.orm import sessionmaker +from tethys_apps.utilities import ( + get_tethys_home_dir, +) +from .settings_commands import read_settings - django.setup() +TETHYS_HOME = Path(get_tethys_home_dir()) + +def add_proxyapps_parser(subparsers): # Setup list command proxyapps_parser = subparsers.add_parser( "proxyapp", help="Add proxy apps and list proxy apps into the Tethys Platform" @@ -24,7 +31,7 @@ def add_proxyapps_parser(subparsers): proxyapps_parser.add_argument( "-a", "--add", - help="Add a new proxy app. Arguments: proxy_app_name endpoint [description] [logo_url] [tags] [enabled] [show_in_apps_library] [back_url] [open_new_tab] [display_external_icon]", + help="Add a new proxy app. Arguments: proxy_app_name endpoint [description] [logo_url] [tags] [enabled] [show_in_apps_library] [back_url] [open_new_tab] [display_external_icon] [app_order]", nargs="+", ) proxyapps_parser.add_argument( @@ -37,19 +44,63 @@ def add_proxyapps_parser(subparsers): proxyapps_parser.set_defaults(func=proxyapp_command, urls=False) -def list_apps(): - from tethys_apps.models import ProxyApp +def list_proxyapps(): + engine = get_engine() + if engine: + Session = sessionmaker(bind=engine) + session = Session() + + Base = automap_base() + Base.prepare(engine, reflect=True) + + ProxyAppsClass = Base.classes.tethys_apps_proxyapp + proxy_apps = session.query(ProxyAppsClass).all() + + write_info("Proxy Apps:") + for proxy_app in proxy_apps: + print(f" {proxy_app.name}") + + session.close() + + +def get_engine(): + engine = None + + tethys_settings = read_settings() + database_settings = tethys_settings.get("DATABASES", "") + + if database_settings == "": + write_error(f"No database settings defined in the portal_config.yml file") + return engine + + database_default_settings = database_settings.get("default", "") + + if database_default_settings == "": + write_error(f"No default database defined in the portal_config.yml file") + return engine + + database_engine = database_default_settings.get("ENGINE", "") + database_name = database_default_settings.get("NAME", "") + database_host = database_default_settings.get("HOST", "") + database_password = database_default_settings.get("PASSWORD", "") + database_port = database_default_settings.get("PORT", "") + database_user = database_default_settings.get("USER", "") + try: + if database_engine == "django.db.backends.postgresql": + postgres_uri = f"postgresql://{database_user}:{database_password}@{database_host}:{database_port}/{database_name}" + else: + postgres_uri = f"sqlite:///{database_name}" + engine = create_engine(postgres_uri, pool_pre_ping=True) - # Function to list proxy apps - proxy_apps = ProxyApp.objects.all() - write_info("Proxy Apps:") - for proxy_app in proxy_apps: - print(f" {proxy_app.name}") + except Exception as e: + write_error(f"Error when connecting to the database") + + return engine def proxyapp_command(args): if args.list: - list_apps() + list_proxyapps() elif args.add: add_proxyapp(args.add) elif args.update: @@ -57,8 +108,6 @@ def proxyapp_command(args): def update_proxyapp(args): - from tethys_apps.models import ProxyApp - app_name = args[0] if len(args) > 0 else None app_key = args[1] if len(args) > 1 else None app_value = args[2] if len(args) > 2 else None @@ -72,16 +121,35 @@ def update_proxyapp(args): if app_value is None: write_error(f"proxy_app_value cannot be empty") return - try: - proxy_app = ProxyApp.objects.get(name=app_name) - if not hasattr(proxy_app, app_key): - write_error(f"Attribute {app_key} does not exists in Proxy app {app_name}") - return - setattr(proxy_app, app_key, app_value) - proxy_app.save() - write_success(f"Proxy app {app_name} was updated") - except ProxyApp.DoesNotExist: - write_error(f"Proxy app {app_name} does not exits") + + engine = get_engine() + + if engine: + Session = sessionmaker(bind=engine) + session = Session() + + Base = automap_base() + Base.prepare(engine, reflect=True) + + ProxyAppsClass = Base.classes.tethys_apps_proxyapp + + proxy_app = ( + session.query(ProxyAppsClass) + .filter(ProxyAppsClass.name == app_name) + .first() + ) + if proxy_app: + if not hasattr(proxy_app, app_key): + write_error( + f"Attribute {app_key} does not exists in Proxy app {app_name}" + ) + return + setattr(proxy_app, app_key, app_value) + session.commit() + write_success(f"Proxy app {app_name} was updated") + else: + write_error(f"Proxy app {app_name} does not exits") + session.close() def add_proxyapp(args): @@ -89,8 +157,6 @@ def add_proxyapp(args): Add Proxy app """ - from tethys_apps.models import ProxyApp - app_name = args[0] if len(args) > 0 else "" app_endpoint = args[1] if len(args) > 1 else "" app_description = args[2] if len(args) > 2 else "" @@ -101,30 +167,42 @@ def add_proxyapp(args): app_back_url = args[7] if len(args) > 7 else "" app_open_new_tab = args[8] if len(args) > 8 else True app_display_external_icon = args[9] if len(args) > 9 else False + app_order = args[10] if len(args) > 10 else 0 if app_name == "": - write_error(f"proxy_app_name cannot be empty") + write_error(f"proxy_app_name argument cannot be empty") return if app_endpoint == "": - write_error(f"proxy_app_endpoint cannot be empty") - return - try: - proxy_app = ProxyApp.objects.create( - name=app_name, - endpoint=app_endpoint, - logo_url=app_logo_url, - back_url=app_back_url, - description=app_description, - tags=app_tags, - show_in_apps_library=app_show_in_app_library, - enabled=app_enabled, - open_in_new_tab=app_open_new_tab, - display_external_icon=app_display_external_icon, - ) - proxy_app.save() - - write_success(f"Proxy app {app_name} added") - except IntegrityError as e: - # Handle the IntegrityError here - write_error(f"there is already a proxy app with that name: {app_name}") + write_error(f"proxy_app_endpoint argument cannot be empty") return + engine = get_engine() + if engine: + Session = sessionmaker(bind=engine) + session = Session() + + Base = automap_base() + Base.prepare(engine, reflect=True) + + ProxyAppsClass = Base.classes.tethys_apps_proxyapp + try: + proxy_app = ProxyAppsClass( + name=app_name, + endpoint=app_endpoint, + logo_url=app_logo_url, + back_url=app_back_url, + description=app_description, + tags=app_tags, + show_in_apps_library=app_show_in_app_library, + enabled=app_enabled, + open_in_new_tab=app_open_new_tab, + display_external_icon=app_display_external_icon, + order=app_order, + ) + session.add(proxy_app) + session.commit() + write_success(f"Proxy app {app_name} added") + + except exc.IntegrityError: + # Handle the IntegrityError here + write_error(f"There is already a proxy app with that name: {app_name}") + session.close()