Lazy caching of WSDL documents (martyanov#6)

Author: tamirOK

aonvif/client.py

@@ -1,4 +1,5 @@
 import datetime
+import functools
 import logging
 import os
 import typing
@@ -9,6 +10,7 @@
 import zeep.helpers
 import zeep.proxy
 import zeep.transports
+import zeep.wsdl
 import zeep.wsse.username
 
 from . import exceptions
@@ -32,6 +34,18 @@ def wrapped(*args, **kwargs):
     return wrapped
 
 
+@functools.cache
+def _get_wsdl_document(path: str):
+    return zeep.wsdl.Document(
+        location=path,
+        transport=zeep.transports.AsyncTransport(cache=MemoryCache()),
+        settings=zeep.client.Settings(
+            strict=False,
+            xml_huge_tree=True,
+        ),
+    )
+
+
 class UsernameToken(zeep.wsse.username.UsernameToken):
     """UsernameDigestToken class, with a time drift parameter that can be adjusted.
 
@@ -140,6 +154,9 @@ def __init__(
         use_token_digest: bool = True,
         device_time_drift: typing.Optional[datetime.timedelta] = None,
     ):
+        if not isinstance(url, str) or not os.path.exists(url):
+            raise RuntimeError('ONVIFService url must be valid path')
+
         self.url = url
         self.xaddr = xaddr
 
@@ -156,9 +173,12 @@ def __init__(
         settings.strict = False
         settings.xml_huge_tree = True
 
+        # Parse wsdl document
+        wsdl = _get_wsdl_document(url)
+
         # Create client
         self._client = zeep.client.AsyncClient(
-            wsdl=url,
+            wsdl=wsdl,
             wsse=wsse,
             settings=settings,
             transport=zeep.transports.AsyncTransport(cache=MemoryCache()),

tests/test_client.py

@@ -1,4 +1,6 @@
 import pytest
+import zeep
+import zeep.wsdl
 
 import aonvif
 import aonvif.client
@@ -205,3 +207,42 @@ async def test_client_update_xaddrs_with_custom_capabilities(
         'http://www.onvif.org/ver20/imaging/wsdl': 'http://testhost/onvif/imaging_service',
     }
     mocked_device_mgmt_service.GetCapabilities.assert_not_awaited()
+
+
+def test_onvif_service_with_invalid_url_type():
+    with pytest.raises(
+        aonvif.exceptions.ONVIFError,
+        match='ONVIFService url must be valid path',
+    ):
+        aonvif.client.ONVIFService(
+            xaddr='http://testhost/onvif',
+            username='test',
+            password='password',
+            url=object(),
+            binding_name='ptz',
+        )
+
+
+def test_onvif_service_with_nonexistent_path():
+    with pytest.raises(
+        aonvif.exceptions.ONVIFError,
+        match='ONVIFService url must be valid path',
+    ):
+        aonvif.client.ONVIFService(
+            xaddr='http://testhost/onvif',
+            username='test',
+            password='password',
+            url='/tmp/nonexistent/path',
+            binding_name='ptz',
+        )
+
+
+def test_onvif_service_ensure_wsdl_documents_cached():
+    client1 = aonvif.client.ONVIFCamera('testhost1', 80, 'testuser1', 'password')
+    service1 = client1.create_devicemgmt_service()
+
+    client2 = aonvif.client.ONVIFCamera('testhost2', 80, 'testuser2', 'password')
+    service2 = client2.create_devicemgmt_service()
+
+    assert isinstance(service1._client.wsdl, zeep.wsdl.Document)
+    assert service1._client.wsdl is service2._client.wsdl