From 1f4b134b23bcba8d06a5f9b00cc22ac2486eaf0a Mon Sep 17 00:00:00 2001 From: petruki <31597636+petruki@users.noreply.github.com> Date: Thu, 7 May 2026 19:36:45 -0700 Subject: [PATCH] feat: improved logs, added LOG_LEVEL --- .env.template | 3 +- README.md | 7 ++-- src/app.py | 7 ++++ src/controller/change_request.py | 56 +++++++++++++++++++-------- src/controller/home.py | 11 ++++-- src/services/switcher_service.py | 8 ++-- src/store/switcher_store.py | 44 ++++++++++----------- src/utils/constants.py | 1 + src/utils/logging_config.py | 40 +++++++++++++++++++ tests/test_logging_config.py | 36 +++++++++++++++++ tests/test_switcher_controller.py | 64 +++++++++++++++---------------- tests/test_switcher_util.py | 12 +++--- tests/utils/generate_token.py | 28 ++++++++++++++ 13 files changed, 230 insertions(+), 87 deletions(-) create mode 100644 src/utils/logging_config.py create mode 100644 tests/test_logging_config.py create mode 100644 tests/utils/generate_token.py diff --git a/.env.template b/.env.template index 5cbd05f..a694d19 100644 --- a/.env.template +++ b/.env.template @@ -4,4 +4,5 @@ SLACK_CLIENT_SECRET="" SWITCHER_JWT_SECRET="" SWITCHER_URL="http://localhost:4200" SWITCHER_API_URL="http://localhost:3000" -SWITCHER_CERT_PATH="path/to/cert.pem" \ No newline at end of file +SWITCHER_CERT_PATH="path/to/cert.pem" +LOG_LEVEL="INFO" diff --git a/README.md b/README.md index f598acc..0536584 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ A summary message containing all details about the change will be sent to a spec # Running locally -## Requirements +## Requirements - Python 3.14 - VirtualEnv - `pip install virtualenv` - Pipenv - `pip install pipenv` @@ -72,13 +72,14 @@ This will trigger the callback to Switcher Management [SWITCHER_URL] to proceed 2. Install dependencies: `make install` 3. Copy the values Client ID, Secret and Signing Secret. 4. Create a .env file based on .env.template and paste the copied values. + - `LOG_LEVEL` controls application and root logger verbosity. Use values such as `DEBUG`, `INFO`, `WARNING`, `ERROR`, or `CRITICAL`. 5. Make sure that the SWITCHER_JWT_SECRET matches the Switcher API env value for SWITCHER_SLACK_JWT_SECRET 6. Make sure that Switcher Management has SWITCHERSLACKAPP_URL set to the app URL 7. Start the API by running: `make run` ## Contributing -You are more than welcome to contribute to the project. +You are more than welcome to contribute to the project. Here are some important guidelines: 1. Suggestions: Open a discussion topic or issue and describe clearly what you have in mind. @@ -99,4 +100,4 @@ make test ``` make test make cover -``` \ No newline at end of file +``` diff --git a/src/app.py b/src/app.py index 4709e94..8aaea79 100644 --- a/src/app.py +++ b/src/app.py @@ -1,10 +1,16 @@ +import logging + from events.request_change import RequestChangeEventHandler from controller.slack_app import SlackAppHandler +from utils.logging_config import configure_logging + +LOGGER = logging.getLogger(__name__) slack_handler = SlackAppHandler() request_change_handler = RequestChangeEventHandler() # Build App +configure_logging() app = slack_handler.build_app() # Register App Events @@ -14,4 +20,5 @@ slack_app = slack_handler.register_handler(app) if __name__ == "__main__": + LOGGER.info("Starting Slack App...") slack_app.run() diff --git a/src/controller/change_request.py b/src/controller/change_request.py index 04b8ad2..41d74ad 100644 --- a/src/controller/change_request.py +++ b/src/controller/change_request.py @@ -1,4 +1,5 @@ import json +import logging from services.switcher_service import SwitcherService from utils.switcher_util import get_environment_keyval, get_keyval, validate_context_request @@ -23,18 +24,23 @@ read_request_metadata ) +LOGGER = logging.getLogger(__name__) + def on_domain_selected(ack, body, client, logger): """ Load environments when domain is selected """ ack() + current_logger = logger or LOGGER try: # Collect args team_id = body["team"]["id"] domain_id = get_selected_action(body) domain_name = get_selected_action_text(body) + current_logger.debug("Domain %s selected for team %s", domain_name, team_id) envs = SwitcherService().get_environments(team_id, domain_id or "") or [] + current_logger.debug("Environments %s loaded for domain %s", envs, domain_name) # Clear previous selection populate_selection(body["view"], "Group", NEW_SELECTION) @@ -67,21 +73,24 @@ def on_domain_selected(ack, body, client, logger): return body["view"] except Exception as e: error = f"Error opening change request form: {e}" - logger.error(error) + current_logger.exception(error) return error def on_environment_selected(ack, body, client, logger): """ Load groups when environment is selected """ ack() + current_logger = logger or LOGGER try: # Collect args env_selected = get_selected_action(body) team_id = body["team"]["id"] domain_id = read_request_metadata(body["view"])["domain_id"] + current_logger.debug("Environment %s selected for team %s in domain %s", env_selected, team_id, domain_id) groups = SwitcherService().get_groups(team_id, domain_id, env_selected or "") or [] + current_logger.debug("Groups %s loaded for environment %s", groups, env_selected) # Clear previous selection populate_selection(body["view"], "Group", NEW_SELECTION) @@ -109,13 +118,14 @@ def on_environment_selected(ack, body, client, logger): return body["view"] except Exception as e: error = f"Error selecting environment: {e}" - logger.error(error) + current_logger.exception(error) return error def on_group_selected(ack, body, client, logger): """ Load switchers when group is selected """ ack() + current_logger = logger or LOGGER try: # Collect args @@ -124,10 +134,12 @@ def on_group_selected(ack, body, client, logger): group_status = get_selected_action_status(body) team_id = body["team"]["id"] domain_id = read_request_metadata(body["view"])["domain_id"] + current_logger.debug("Group %s selected for team %s in domain %s", group_selected, team_id, domain_id) switchers = SwitcherService().get_switchers( team_id, domain_id, env_selected, group_selected or "" ) or [] + current_logger.debug("Switchers %s loaded for group %s", switchers, group_selected) # Clear previous selection populate_selection(body["view"], "Switcher", NEW_SELECTION) @@ -157,21 +169,25 @@ def on_group_selected(ack, body, client, logger): return body["view"] except Exception as e: error = f"Error selecting group: {e}" - logger.error(error) + current_logger.exception(error) return error def on_switcher_selected(ack, body, client): """ Updates view's metadata with switcher selection """ ack() + current_logger = LOGGER # Populate view selected_switcher = get_selected_action_text(body) + current_logger.debug("Switcher %s selected", selected_switcher) if selected_switcher != "-": switcher_status = get_selected_action_status(body) + current_logger.debug("Switcher selected with status %s", switcher_status) populate_selection_status(body["view"], switcher_status) else: selected_group = get_state_name(body["view"], "selection_group") + current_logger.debug("No switcher selected, reverting to group %s status", selected_group) populate_selection_status(body["view"], get_status(selected_group)) view_hash = body["view"]["hash"] @@ -190,6 +206,7 @@ def on_change_request_review(ack, body, client, view, logger): """ Populate context with selections, validate via Switcher API then publish view for review """ ack() + current_logger = logger or LOGGER user = body["user"] team_id = body["team"]["id"] @@ -207,6 +224,7 @@ def on_change_request_review(ack, body, client, view, logger): "status": get_state_value(view, "selection_status") } + current_logger.debug("Validating change request with context %s", json.dumps(context)) validate_context_request(context) result = SwitcherService().validate_ticket(team_id, context) @@ -241,13 +259,14 @@ def on_change_request_review(ack, body, client, view, logger): ) error = f"Error on change request review: {e}" - logger.error(error) + current_logger.exception(error) return error def on_submit(ack, body, client, logger): """ Create ticket, return to home view then publish approval message """ ack() + current_logger = logger or LOGGER user = body["user"] team_id = body["team"]["id"] @@ -258,7 +277,6 @@ def on_submit(ack, body, client, logger): **read_request_metadata(body["view"]), "observations": "" if observation is None else observation, } - domain_id = context["domain_id"] # Return to initial state @@ -269,6 +287,7 @@ def on_submit(ack, body, client, logger): ) # Create ticket and post approval request + current_logger.debug("Creating ticket for change request with context %s", json.dumps(context)) ticket = SwitcherService().create_ticket(team_id, context) ticket_payload = { "id": ticket.get("ticket_id"), @@ -284,12 +303,13 @@ def on_submit(ack, body, client, logger): return request_message, ticket except Exception as e: - logger.error(f"Error on submitting: {e}") + error = f"Error on submitting: {e}" + current_logger.exception(error) client.chat_postMessage( channel = user["id"], text = f"There was an error with your request: {e}" ) - return e + return error def on_change_request_abort(ack, body, client): """ Return to home view """ @@ -306,6 +326,7 @@ def on_request_approved(ack, body, client, logger): """ Approve ticket through Switcher API and update chat message """ ack() + current_logger = logger or LOGGER message_ts = body["message"]["ts"] team_id = body["team"]["id"] channel_id = body["channel"]["id"] @@ -319,6 +340,7 @@ def on_request_approved(ack, body, client, logger): message_blocks.append(body["message"]["blocks"][2]) # Approve ticket and update message + current_logger.debug("Approving ticket %s for domain %s in team %s", ticket_id, domain_id, team_id) SwitcherService().approve_request(team_id, domain_id, ticket_id) client.chat_update( channel = channel_id, @@ -329,19 +351,21 @@ def on_request_approved(ack, body, client, logger): return message_blocks except Exception as e: - logger.error(f"Error on aborting: {e}") + error = f"Error on approving request: {e}" + current_logger.exception(error) client.chat_update( channel = channel_id, - text = e.args[0], + text = f"There was an error with your request: {e}", ts = message_ts, - blocks = create_block_message(f":large_yellow_square: *{e.args[0]}*") + blocks = create_block_message(f":large_yellow_square: *{e}*") ) - return e + return error def on_request_denied(ack, body, client, logger): """ Deny ticket through Switcher API and update chat message """ ack() + current_logger = logger or LOGGER message_ts = body["message"]["ts"] team_id = body["team"]["id"] channel_id = body["channel"]["id"] @@ -355,6 +379,7 @@ def on_request_denied(ack, body, client, logger): message_blocks.append(body["message"]["blocks"][2]) # Deny ticket and update message + current_logger.debug("Denying ticket %s for domain %s in team %s", ticket_id, domain_id, team_id) SwitcherService().deny_request(team_id, domain_id, ticket_id) client.chat_update( channel = channel_id, @@ -365,11 +390,12 @@ def on_request_denied(ack, body, client, logger): return message_blocks except Exception as e: - logger.error(f"Error on denying: {e}") + error = f"Error on denying request: {e}" + current_logger.exception(error) client.chat_update( channel = channel_id, - text = e.args[0], + text = f"There was an error with your request: {e}", ts = message_ts, - blocks = create_block_message(f":large_yellow_square: *{e.args[0]}*") + blocks = create_block_message(f":large_yellow_square: *{e}*") ) - return e + return error diff --git a/src/controller/home.py b/src/controller/home.py index 8f50934..f8c02a2 100644 --- a/src/controller/home.py +++ b/src/controller/home.py @@ -1,14 +1,18 @@ import copy +import logging from services.switcher_service import SwitcherService from payloads.home import MODAL_REQUEST, APP_HOME from utils.slack_payload_util import populate_selection from utils.switcher_util import get_keyval_of_keys +LOGGER = logging.getLogger(__name__) + def on_home_opened(client, event, logger): """ Displays the home view """ - logger.warning(f"Home view opened by {event['user']}") + current_logger = logger or LOGGER + current_logger.info("Home view opened by %s", event["user"]) client.views_publish( user_id = event["user"], view = APP_HOME @@ -20,13 +24,14 @@ def on_change_request_opened(ack, body, client, logger): """ Displays the change request modal """ ack() + current_logger = logger or LOGGER try: change_request = copy.deepcopy(MODAL_REQUEST) # Collect args team_id = body["team"]["id"] - logger.warning(f"Team ID: {team_id}") + current_logger.debug("Opening change request modal for team %s", team_id) domains = SwitcherService().get_domains(team_id) @@ -44,5 +49,5 @@ def on_change_request_opened(ack, body, client, logger): return change_request except Exception as e: - logger.error(f"Error opening change request form: {e}") + current_logger.exception(f"Error opening change request form: {e}") return None diff --git a/src/services/switcher_service.py b/src/services/switcher_service.py index 42b9365..07d78d3 100644 --- a/src/services/switcher_service.py +++ b/src/services/switcher_service.py @@ -41,11 +41,11 @@ def get_environments(self, team_id: str, domain_id: str) -> list[str] | None: }} ''') + environments = None configuration = response.get("configuration", None) if configuration is not None: environments = configuration.get("environments", None) - if environments is not None: - return environments + return environments def get_groups(self, team_id: str, domain_id: str, environment: str) -> list[dict] | None: """ Fetch all Groups linked to the Environment """ @@ -64,11 +64,11 @@ def get_groups(self, team_id: str, domain_id: str, environment: str) -> list[dic }} ''') + groups = None configuration = response.get("configuration", None) if configuration is not None: groups = configuration.get("group", None) - if groups is not None: - return groups + return groups def get_switchers(self, team_id: str, domain_id: str, environment: str, group: str) -> list[dict] | None: """ Fetch all Switchers linked to the Group """ diff --git a/src/store/switcher_store.py b/src/store/switcher_store.py index 2ffb3f7..228cfde 100644 --- a/src/store/switcher_store.py +++ b/src/store/switcher_store.py @@ -13,15 +13,17 @@ AsyncInstallationStore, ) +LOGGER = logging.getLogger(__name__) + class SwitcherAppInstallationStore(InstallationStore, AsyncInstallationStore): - def __init__(self, api_url: str, logger: Logger = logging.getLogger(__name__)): - self._logger = logger + def __init__(self, api_url: str, logger: Optional[Logger] = None): + self._logger = logger or LOGGER self._store_service = SwitcherInstallationStoreService(api_url) @property def logger(self) -> Logger: if self._logger is None: - self._logger = logging.getLogger(__name__) + self._logger = LOGGER return self._logger async def async_save(self, installation: Installation): @@ -41,11 +43,10 @@ def save(self, installation: Installation): bot_payload = installation.to_bot().__dict__ ) except Exception as e: - message = \ - "Failed to save installation data for enterprise:" \ - f"{e_id}, team: {t_id}: {e}" - - self.logger.warning(message) + self.logger.error( + f"Failed to save installation data for enterprise: {e_id}, team: {t_id}: {e}", + exc_info = True, + ) async def async_find_bot( self, @@ -77,11 +78,10 @@ def find_bot( bot_payload = json.loads(response.data) return Bot(**bot_payload) except Exception as e: - message = \ - "Failed to find bot installation data for enterprise:" \ - f"{enterprise_id}, team: {team_id}: {e}" - - self.logger.warning(message) + self.logger.warning( + f"Failed to find bot installation data for enterprise: {enterprise_id}, team: {team_id}: {e}", + exc_info = True, + ) return None async def async_find_installation( @@ -117,11 +117,10 @@ def find_installation( installation_payload = json.loads(response.data) return Installation(**installation_payload) except Exception as e: - message = \ - "Failed to find an installation data for enterprise:" \ - f"{enterprise_id}, team: {team_id}: {e}" - - self.logger.warning(message) + self.logger.warning( + f"Failed to find an installation data for enterprise: {enterprise_id}, team: {team_id}: {e}", + exc_info = True, + ) return None async def async_delete_bot( @@ -166,11 +165,10 @@ def delete_installation( user_id = user_id ) except Exception as e: - message = \ - "Failed to delete installation data for enterprise:" \ - f"{enterprise_id}, team: {team_id}: {e}" - - self.logger.warning(message) + self.logger.warning( + f"Failed to delete installation data for enterprise: {enterprise_id}, team: {team_id}: {e}", + exc_info = True, + ) @staticmethod def _input_sanitize( diff --git a/src/utils/constants.py b/src/utils/constants.py index dc83d7e..7aee883 100644 --- a/src/utils/constants.py +++ b/src/utils/constants.py @@ -6,5 +6,6 @@ SWITCHER_URL = os.environ.get("SWITCHER_URL") SWITCHER_API_URL = os.environ.get("SWITCHER_API_URL") +LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO") RELEASE_TIME = os.environ.get("RELEASE_TIME", "latest") VERSION = f"2.1.1 {RELEASE_TIME}" diff --git a/src/utils/logging_config.py b/src/utils/logging_config.py new file mode 100644 index 0000000..dbf6fa8 --- /dev/null +++ b/src/utils/logging_config.py @@ -0,0 +1,40 @@ +import logging + +from typing import Optional +from utils.constants import LOG_LEVEL + +DEFAULT_LOG_LEVEL = logging.INFO +DEFAULT_LOG_FORMAT = "[%(asctime)s] [%(process)d] [%(levelname)s] %(name)s: %(message)s" +DEFAULT_LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S %z" +CONSOLE_HANDLER_NAME = "switcher-console" +LOG_LEVELS = logging.getLevelNamesMapping() + +def get_log_level(level_name: Optional[str] = None) -> int: + """Resolve a configured log level name to a logging constant.""" + + configured_level = (level_name or LOG_LEVEL).strip().upper() + return LOG_LEVELS.get(configured_level, DEFAULT_LOG_LEVEL) + +def configure_logging(level_name: Optional[str] = None) -> logging.Logger: + """Configure process-wide console logging without adding duplicate handlers.""" + + level = get_log_level(level_name) + formatter = logging.Formatter( + fmt = DEFAULT_LOG_FORMAT, + datefmt = DEFAULT_LOG_DATE_FORMAT, + ) + root_logger = logging.getLogger() + + if root_logger.handlers: + for handler in root_logger.handlers: + handler.setLevel(level) + handler.setFormatter(formatter) + else: + handler = logging.StreamHandler() + handler.set_name(CONSOLE_HANDLER_NAME) + handler.setLevel(level) + handler.setFormatter(formatter) + root_logger.addHandler(handler) + + root_logger.setLevel(level) + return root_logger diff --git a/tests/test_logging_config.py b/tests/test_logging_config.py new file mode 100644 index 0000000..1466b4a --- /dev/null +++ b/tests/test_logging_config.py @@ -0,0 +1,36 @@ +import logging + +from src.utils.logging_config import ( + CONSOLE_HANDLER_NAME, + DEFAULT_LOG_FORMAT, + configure_logging, + get_log_level, +) + +def test_get_log_level_returns_default_for_invalid_value(): + assert get_log_level("not-a-level") == logging.INFO + +def test_configure_logging_is_idempotent(): + root_logger = logging.getLogger() + original_handlers = root_logger.handlers[:] + original_level = root_logger.level + + try: + root_logger.handlers = [] + + configure_logging("DEBUG") + configure_logging("INFO") + + matching_handlers = [ + handler + for handler in root_logger.handlers + if handler.get_name() == CONSOLE_HANDLER_NAME + ] + + assert len(matching_handlers) == 1 + assert root_logger.level == logging.INFO + assert matching_handlers[0].formatter is not None + assert matching_handlers[0].formatter._fmt == DEFAULT_LOG_FORMAT + finally: + root_logger.handlers = original_handlers + root_logger.setLevel(original_level) diff --git a/tests/test_switcher_controller.py b/tests/test_switcher_controller.py index cb91116..8973f96 100644 --- a/tests/test_switcher_controller.py +++ b/tests/test_switcher_controller.py @@ -27,7 +27,7 @@ build_static_select_action_value, build_buttom_action_value, - # Constants + # Constants RELEASE_1, PRODUCTION, DEFAULT_ENV, @@ -79,7 +79,7 @@ def test_open_change_request_modal(client): client = client, logger = logging.getLogger() ) - + with open(ON_CHANGE_REQUEST_OPENED) as f: expected_result = json.load(f) @@ -101,12 +101,12 @@ def test_open_change_request_modal_with_error(client): client = client, logger = logging.getLogger() ) - + assert result == None - -@mock_gql_client({ - 'configuration': { - 'environments': ['default'] + +@mock_gql_client({ + 'configuration': { + 'environments': ['default'] } }) def test_select_domain(client): @@ -131,7 +131,7 @@ def test_select_domain(client): with open(ON_DOMAIN_SELECTED) as f: expected_result = json.load(f) - + result = __remove_block_id(result) assert result == expected_result @@ -155,7 +155,7 @@ def test_select_domain_error(client): client = client, logger = logging.getLogger() ) - + assert "Error opening change request form" in result @mock_gql_client({ @@ -177,8 +177,8 @@ def test_select_evironment(client): text = PRODUCTION, value = DEFAULT_ENV ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -207,8 +207,8 @@ def test_select_evironment_error(client): text = PRODUCTION, value = DEFAULT_ENV ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -237,15 +237,15 @@ def test_select_group(client): text = RELEASE_1, value = RELEASE_1 ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), client = client, logger = logging.getLogger() ) - + with open(ON_GROUP_SELECTED) as f: expected_result = json.load(f) @@ -267,15 +267,15 @@ def test_select_group_error(client): text = RELEASE_1, value = RELEASE_1 ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), client = client, logger = logging.getLogger() ) - + assert "Error selecting group" in result def test_select_switcher(client): @@ -292,8 +292,8 @@ def test_select_switcher(client): text = MY_FEATURE, value = MY_FEATURE ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -321,8 +321,8 @@ def test_select_switcher_none(client): value = "-" ), state_fixture = SWITCHER_STATE_SELECTION, - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -350,8 +350,8 @@ def test_submit_for_review(client): text = "Activate", value = "true" ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -383,8 +383,8 @@ def test_submit_without_approval_ignored(client): text = "Activate", value = "true" ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -416,8 +416,8 @@ def test_submit_without_approval_frozen(client): text = "Activate", value = "true" ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -449,8 +449,8 @@ def test_submit_for_review_error(client): text = "Activate", value = "true" ), - private_metadata = json.dumps({ - "domain_id": "1", "domain_name": "Test" + private_metadata = json.dumps({ + "domain_id": "1", "domain_name": "Test" }), blocks_fixture = modal['blocks'] ), @@ -488,7 +488,7 @@ def test_submit_request(client): ), client = client, logger = logging.getLogger() - ) or {} + ) or {} # type: ignore with open(ON_SUBMIT) as f: expected_result = json.load(f) diff --git a/tests/test_switcher_util.py b/tests/test_switcher_util.py index 386e0e1..8d15d44 100644 --- a/tests/test_switcher_util.py +++ b/tests/test_switcher_util.py @@ -29,9 +29,9 @@ def test_validate_context_request(): assert e_info.value.args[0] == "Missing [Domain - Domain ID - Environment - Group - Status]" def test_env_loading(): - assert os.environ.get("SLACK_SIGNING_SECRET") == "[SLACK_SIGNING_SECRET]" - assert os.environ.get("SLACK_CLIENT_ID") == "[SLACK_CLIENT_ID]" - assert os.environ.get("SLACK_CLIENT_SECRET") == "[SLACK_CLIENT_SECRET]" - assert os.environ.get("SWITCHER_URL") == "https://cloud.swiktcherapi.com" - assert os.environ.get("SWITCHER_API_URL") == "https://switcherapi.com/api" - assert os.environ.get("SWITCHER_JWT_SECRET") == "[SWITCHER_JWT_SECRET]" \ No newline at end of file + assert "SLACK_SIGNING_SECRET" in os.environ + assert "SLACK_CLIENT_ID" in os.environ + assert "SLACK_CLIENT_SECRET" in os.environ + assert "SWITCHER_URL" in os.environ + assert "SWITCHER_API_URL" in os.environ + assert "SWITCHER_JWT_SECRET" in os.environ diff --git a/tests/utils/generate_token.py b/tests/utils/generate_token.py new file mode 100644 index 0000000..ff50277 --- /dev/null +++ b/tests/utils/generate_token.py @@ -0,0 +1,28 @@ +# Usage: Set SWITCHER_JWT_SECRET in your environment or .env file +# Example: python tests/generate_token.py resource_id +import os +import jwt +import datetime + +from flask.cli import load_dotenv + +load_dotenv() + +SECRET = os.environ.get("SWITCHER_JWT_SECRET", "changeit") +ALGORITHM = "HS256" +ISSUER = "Switcher Slack App" + +import sys +if len(sys.argv) < 2: + print("Usage: python generate_token.py ") + sys.exit(1) +resource = sys.argv[1] + +payload = { + "iss": ISSUER, + "sub": resource, + "exp": datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=30) +} + +token = jwt.encode(payload=payload, key=SECRET, algorithm=ALGORITHM) +print(token)