move to util

Author: bdraco

onvif/client.py

@@ -8,8 +8,6 @@
 import os.path
 from typing import Any
 from collections.abc import Callable
-from yarl import URL
-from multidict import CIMultiDict
 import httpx
 from httpx import AsyncClient, BasicAuth, DigestAuth
 from zeep.cache import SqliteCache
@@ -28,7 +26,14 @@
 from .settings import DEFAULT_SETTINGS
 from .transport import ASYNC_TRANSPORT
 from .types import FastDateTime, ForgivingTime
-from .util import create_no_verify_ssl_context, normalize_url, path_isfile, utcnow
+from .util import (
+    create_no_verify_ssl_context,
+    normalize_url,
+    path_isfile,
+    utcnow,
+    strip_user_pass_url,
+    obscure_user_pass_url,
+)
 from .wrappers import retry_connection_error  # noqa: F401
 from .wsa import WsAddressingIfMissingPlugin
 
@@ -45,38 +50,6 @@
 _WRITE_TIMEOUT = 90
 _HTTPX_LIMITS = httpx.Limits(keepalive_expiry=KEEPALIVE_EXPIRY)
 _NO_VERIFY_SSL_CONTEXT = create_no_verify_ssl_context()
-_CREDENTIAL_KEYS = ("username", "user", "pass", "password")
-
-
-def strip_user_pass_url(url: str) -> str:
-    """Strip password from URL."""
-    parsed_url = URL(url)
-    query = parsed_url.query
-    new_query: CIMultiDict | None = None
-    for key in _CREDENTIAL_KEYS:
-        if key in query:
-            if new_query is None:
-                new_query = CIMultiDict(parsed_url.query)
-            new_query.popall(key)
-    if new_query is not None:
-        return str(parsed_url.with_query(new_query))
-    return url
-
-
-def obscure_user_pass_url(url: str) -> str:
-    """Obscure user and password from URL."""
-    parsed_url = URL(url)
-    query = parsed_url.query
-    new_query: CIMultiDict | None = None
-    for key in _CREDENTIAL_KEYS:
-        if key in query:
-            if new_query is None:
-                new_query = CIMultiDict(parsed_url.query)
-            new_query.popall(key)
-            new_query[key] = "********"
-    if new_query is not None:
-        return str(parsed_url.with_query(new_query))
-    return url
 
 
 def safe_func(func):

onvif/util.py

@@ -9,7 +9,8 @@
 import ssl
 from typing import Any
 from urllib.parse import ParseResultBytes, urlparse, urlunparse
-
+from yarl import URL
+from multidict import CIMultiDict
 from zeep.exceptions import Fault
 
 utcnow: partial[dt.datetime] = partial(dt.datetime.now, dt.timezone.utc)
@@ -18,6 +19,8 @@
 # to minimize the impact of the blocking I/O.
 path_isfile = lru_cache(maxsize=128)(os.path.isfile)
 
+_CREDENTIAL_KEYS = ("username", "password", "user", "pass")
+
 
 def normalize_url(url: bytes | str | None) -> str | None:
     """Normalize URL.
@@ -105,3 +108,34 @@ def create_no_verify_ssl_context() -> ssl.SSLContext:
     # ssl.OP_LEGACY_SERVER_CONNECT is only available in Python 3.12a4+
     sslcontext.options |= getattr(ssl, "OP_LEGACY_SERVER_CONNECT", 0x4)
     return sslcontext
+
+
+def strip_user_pass_url(url: str) -> str:
+    """Strip password from URL."""
+    parsed_url = URL(url)
+    query = parsed_url.query
+    new_query: CIMultiDict | None = None
+    for key in _CREDENTIAL_KEYS:
+        if key in query:
+            if new_query is None:
+                new_query = CIMultiDict(parsed_url.query)
+            new_query.popall(key)
+    if new_query is not None:
+        return str(parsed_url.with_query(new_query))
+    return url
+
+
+def obscure_user_pass_url(url: str) -> str:
+    """Obscure user and password from URL."""
+    parsed_url = URL(url)
+    query = parsed_url.query
+    new_query: CIMultiDict | None = None
+    for key in _CREDENTIAL_KEYS:
+        if key in query:
+            if new_query is None:
+                new_query = CIMultiDict(parsed_url.query)
+            new_query.popall(key)
+            new_query[key] = "********"
+    if new_query is not None:
+        return str(parsed_url.with_query(new_query))
+    return url

tests/test_client.py

@@ -4,6 +4,7 @@
 
 import pytest
 from zeep.loader import parse_xml
+from onvif.util import strip_user_pass_url, obscure_user_pass_url
 
 from onvif.client import ONVIFCamera
 from onvif.settings import DEFAULT_SETTINGS
@@ -40,3 +41,16 @@ async def test_normalize_url_with_missing_url():
     )
     result = operation.process_reply(envelope)
     assert normalize_url(result.SubscriptionReference.Address._value_1) is None
+
+
+def test_strip_user_pass_url():
+    assert strip_user_pass_url("http://1.2.3.4/?user=foo&pass=bar") == "http://1.2.3.4/"
+    assert strip_user_pass_url("http://1.2.3.4/") == "http://1.2.3.4/"
+
+
+def test_obscure_user_pass_url():
+    assert (
+        obscure_user_pass_url("http://1.2.3.4/?user=foo&pass=bar")
+        == "http://1.2.3.4/?user=********&pass=********"
+    )
+    assert obscure_user_pass_url("http://1.2.3.4/") == "http://1.2.3.4/"