bug: accept urls in --notify-webhooks

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:

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))