From 22cbc52bf2bdb214520057049c89159ef1c2bac2 Mon Sep 17 00:00:00 2001 From: Kate Sugden <107400614+ksugden@users.noreply.github.com> Date: Mon, 13 Jan 2025 12:01:50 +0000 Subject: [PATCH] refactor: DBTP-1678 maintenance page domain refactor (#717) Co-authored-by: chiaramapellimt --- .../domain/maintenance_page.py | 64 +++++++++++++------ 1 file changed, 44 insertions(+), 20 deletions(-) diff --git a/dbt_platform_helper/domain/maintenance_page.py b/dbt_platform_helper/domain/maintenance_page.py index 6b7a74b49..b40f76b5b 100644 --- a/dbt_platform_helper/domain/maintenance_page.py +++ b/dbt_platform_helper/domain/maintenance_page.py @@ -3,38 +3,62 @@ import re import string from pathlib import Path +from typing import Callable from typing import List from typing import Union import boto3 import click +from dbt_platform_helper.platform_exception import PlatformException from dbt_platform_helper.providers.load_balancers import ListenerNotFoundException from dbt_platform_helper.providers.load_balancers import ListenerRuleNotFoundException from dbt_platform_helper.providers.load_balancers import LoadBalancerNotFoundException from dbt_platform_helper.providers.load_balancers import find_https_listener +from dbt_platform_helper.utils.application import Application from dbt_platform_helper.utils.application import Environment from dbt_platform_helper.utils.application import Service from dbt_platform_helper.utils.application import load_application -class MaintenancePage: - def activate(self, app, env, svc, template, vpc): - application = load_application(app) - application_environment = get_app_environment(app, env) +class MaintenancePageException(PlatformException): + pass + + +class LoadBalancedWebServiceNotFoundException(MaintenancePageException): + pass + +class MaintenancePage: + def __init__( + self, + user_prompt_callback: Callable[[str], bool] = click.confirm, + echo: Callable[[str], str] = click.secho, + ): + self.user_prompt_callback = user_prompt_callback + self.echo = echo + + def _get_deployed_load_balanced_web_services(self, app: Application, svc: str): if "*" in svc: - services = [ - s for s in application.services.values() if s.kind == "Load Balanced Web Service" - ] + services = [s for s in app.services.values() if s.kind == "Load Balanced Web Service"] else: - all_services = [get_app_service(app, s) for s in list(svc)] + all_services = [get_app_service(app.name, s) for s in list(svc)] services = [s for s in all_services if s.kind == "Load Balanced Web Service"] - if not services: - click.secho(f"No services deployed yet to {app} environment {env}", fg="red") + raise LoadBalancedWebServiceNotFoundException + return services + + def activate(self, app, env, svc, template, vpc): + try: + services = self._get_deployed_load_balanced_web_services(load_application(app), svc) + except LoadBalancedWebServiceNotFoundException: + # TODO DBTP-1643 - this bit of logic does not depend on env, so env shouldn't really be in the exception + # message + # Exception should be propagated to command and caught there. + self.echo(f"No services deployed yet to {app} environment {env}", fg="red") raise click.Abort + application_environment = get_app_environment(app, env) try: https_listener = find_https_listener(application_environment.session, app, env) current_maintenance_page = get_maintenance_page( @@ -42,7 +66,7 @@ def activate(self, app, env, svc, template, vpc): ) remove_current_maintenance_page = False if current_maintenance_page: - remove_current_maintenance_page = click.confirm( + remove_current_maintenance_page = self.user_prompt_callback( f"There is currently a '{current_maintenance_page}' maintenance page for the {env} " f"environment in {app}.\nWould you like to replace it with a '{template}' " f"maintenance page?" @@ -50,7 +74,7 @@ def activate(self, app, env, svc, template, vpc): if not remove_current_maintenance_page: raise click.Abort - if remove_current_maintenance_page or click.confirm( + if remove_current_maintenance_page or self.user_prompt_callback( f"You are about to enable the '{template}' maintenance page for the {env} " f"environment in {app}.\nWould you like to continue?" ): @@ -68,7 +92,7 @@ def activate(self, app, env, svc, template, vpc): allowed_ips, template, ) - click.secho( + self.echo( f"Maintenance page '{template}' added for environment {env} in application {app}", fg="green", ) @@ -76,13 +100,13 @@ def activate(self, app, env, svc, template, vpc): raise click.Abort except LoadBalancerNotFoundException: - click.secho( + self.echo( f"No load balancer found for environment {env} in the application {app}.", fg="red" ) raise click.Abort except ListenerNotFoundException: - click.secho( + self.echo( f"No HTTPS listener found for environment {env} in the application {app}.", fg="red" ) raise click.Abort @@ -96,28 +120,28 @@ def deactivate(self, app, env): application_environment.session, https_listener ) if not current_maintenance_page: - click.secho("There is no current maintenance page to remove", fg="red") + self.echo("There is no current maintenance page to remove", fg="red") raise click.Abort - if not click.confirm( + if not self.user_prompt_callback( f"There is currently a '{current_maintenance_page}' maintenance page, " f"would you like to remove it?" ): raise click.Abort remove_maintenance_page(application_environment.session, https_listener) - click.secho( + self.echo( f"Maintenance page removed from environment {env} in application {app}", fg="green" ) except LoadBalancerNotFoundException: - click.secho( + self.echo( f"No load balancer found for environment {env} in the application {app}.", fg="red" ) raise click.Abort except ListenerNotFoundException: - click.secho( + self.echo( f"No HTTPS listener found for environment {env} in the application {app}.", fg="red" ) raise click.Abort