diff --git a/lean/commands/cloud/live/deploy.py b/lean/commands/cloud/live/deploy.py index c667ae40..2be51450 100644 --- a/lean/commands/cloud/live/deploy.py +++ b/lean/commands/cloud/live/deploy.py @@ -11,7 +11,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import List, Tuple, Optional +from typing import Dict, List, Tuple, Optional from click import prompt, option, argument, confirm from lean.click import LeanCommand, ensure_options, CaseInsensitiveChoice from lean.components.api.api_client import APIClient @@ -152,6 +152,24 @@ def _configure_notifications(logger: Logger) -> Tuple[bool, bool, List[QCNotific return notify_order_events, notify_insights, notify_methods +def _parse_webhook_config(config: str) -> Tuple[str, Dict[str, str]]: + """Splits a 'url:HEADER_1=VALUE_1:HEADER_2=VALUE_2' webhook configuration into its address and headers. + + The address can contain colons itself (scheme and port), so only the trailing 'NAME=VALUE' parts are headers. + + :param config: the webhook configuration given by the user + :return: the address and the headers of the webhook + """ + from re import match + + parts = config.split(":") + headers = [] + while len(parts) > 1 and match(r"^[\w-]+=", parts[-1]): + headers.insert(0, parts.pop().split("=", 1)) + + return ":".join(parts), dict(headers) + + def _configure_auto_restart(logger: Logger) -> bool: """Interactively configures whether automatic algorithm restarting must be enabled. @@ -260,8 +278,7 @@ def deploy(project: str, if notify_webhooks is not None: for config in notify_webhooks.split(","): - address, *headers = config.split(":") - headers = {header.split("=")[0]: header.split("=")[1] for header in headers} + address, headers = _parse_webhook_config(config) notify_methods.append(QCWebhookNotificationMethod(address=address, headers=headers)) if notify_sms is not None: diff --git a/tests/commands/cloud/live/test_cloud_live_commands.py b/tests/commands/cloud/live/test_cloud_live_commands.py index c6740e27..fd25fe8e 100644 --- a/tests/commands/cloud/live/test_cloud_live_commands.py +++ b/tests/commands/cloud/live/test_cloud_live_commands.py @@ -231,6 +231,9 @@ def test_cloud_live_deploy_with_tradier_using_tradier_datafeed() -> None: ("webhooks", "customAddress:header1=value1"), ("webhooks", "customAddress:header1=value1:header2=value2"), ("webhooks", "customAddress1:header1=value1:header2=value2,customAddress2:header3=value3"), + ("webhooks", "https://example.com/hook"), + ("webhooks", "https://example.com:8080/hook?id=1:header1=value1"), + ("webhooks", "https://example.com/hook:Authorization=Basic dXNlcjpwYXNz=="), ("sms", "customNumber"), ("sms", "customNumber1,customNumber2,customNumber3"), ("telegram", "customId"), @@ -241,6 +244,18 @@ def test_cloud_live_deploy_with_tradier_using_tradier_datafeed() -> None: ("telegram", "customId1:custom:token1,customId2"), ("telegram", "customId1:custom:token1,customId2:custom:token2")]) def test_cloud_live_deploy_with_notifications(notice_method: str, configs: str) -> None: + expected_webhooks = { + "customAddress:header1=value1": ("customAddress", {"header1": "value1"}), + "customAddress:header1=value1:header2=value2": ("customAddress", {"header1": "value1", "header2": "value2"}), + "customAddress1:header1=value1:header2=value2": ("customAddress1", {"header1": "value1", "header2": "value2"}), + "customAddress2:header3=value3": ("customAddress2", {"header3": "value3"}), + "https://example.com/hook": ("https://example.com/hook", {}), + "https://example.com:8080/hook?id=1:header1=value1": ("https://example.com:8080/hook?id=1", + {"header1": "value1"}), + "https://example.com/hook:Authorization=Basic dXNlcjpwYXNz==": ("https://example.com/hook", + {"Authorization": "Basic dXNlcjpwYXNz=="}) + } + create_fake_lean_cli_directory() api_client = mock.Mock() @@ -272,12 +287,7 @@ def test_cloud_live_deploy_with_notifications(notice_method: str, configs: str) notification.append(QCEmailNotificationMethod(address=address, subject=subject)) elif notice_method == "webhooks": - address, headers = config.split(":", 1) - headers_dict = {} - - for header in headers.split(":"): - key, value = header.split("=") - headers_dict[key] = value + address, headers_dict = expected_webhooks[config] notification.append(QCWebhookNotificationMethod(address=address, headers=headers_dict))