Allow VespaSearchAdaptor instantiation with no certs (#98)

* Adding a failing test.

* Bumping the cpr_sdk version.

* Adding fix.

* Update src/cpr_sdk/vespa.py

Co-authored-by: Jesse Claven <[email protected]>

* Adding import for Optional.

* Return no cert path if we can't find application name.

---------

Co-authored-by: Mark <[email protected]>
Co-authored-by: Jesse Claven <[email protected]>

src/cpr_sdk/search_adaptors.py

@@ -1,24 +1,25 @@
 """Adaptors for searching CPR data"""
 
+import logging
 import time
 from abc import ABC
 from pathlib import Path
 from typing import Optional
-import logging
+
+from requests.exceptions import HTTPError
+from vespa.application import Vespa
+from vespa.exceptions import VespaError
 
 from cpr_sdk.embedding import Embedder
 from cpr_sdk.exceptions import DocumentNotFoundError, FetchError, QueryError
 from cpr_sdk.models.search import Hit, SearchParameters, SearchResponse
 from cpr_sdk.vespa import (
+    VespaErrorDetails,
     build_vespa_request_body,
     find_vespa_cert_paths,
     parse_vespa_response,
     split_document_id,
-    VespaErrorDetails,
 )
-from requests.exceptions import HTTPError
-from vespa.application import Vespa
-from vespa.exceptions import VespaError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -67,10 +68,10 @@ def __init__(
         if cert_directory is None:
             cert_path, key_path = find_vespa_cert_paths()
         else:
-            cert_path = Path(cert_directory) / "cert.pem"
-            key_path = Path(cert_directory) / "key.pem"
+            cert_path = (Path(cert_directory) / "cert.pem").__str__()
+            key_path = (Path(cert_directory) / "key.pem").__str__()
 
-        self.client = Vespa(url=instance_url, cert=str(cert_path), key=str(key_path))
+        self.client = Vespa(url=instance_url, cert=cert_path, key=key_path)
         self.embedder = embedder or Embedder()
 
     def search(self, parameters: SearchParameters) -> SearchResponse:

src/cpr_sdk/version.py

@@ -1,6 +1,6 @@
 _MAJOR = "1"
 _MINOR = "4"
-_PATCH = "2"
+_PATCH = "3"
 _SUFFIX = ""
 
 VERSION_SHORT = "{0}.{1}".format(_MAJOR, _MINOR)

src/cpr_sdk/vespa.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Any, List
+from typing import Any, List, Optional
 
 import yaml
 from vespa.exceptions import VespaError
@@ -36,7 +36,7 @@ def split_document_id(document_id: str) -> tuple[str, str, str]:
     return namespace, schema, data_id
 
 
-def find_vespa_cert_paths() -> tuple[Path, Path]:
+def find_vespa_cert_paths() -> tuple[Optional[str], Optional[str]]:
     """
     Automatically find the certificate and key files for the vespa instance
 
@@ -54,11 +54,21 @@ def find_vespa_cert_paths() -> tuple[Path, Path]:
     # read the config.yaml file to find the application name
     with open(vespa_directory / "config.yaml", "r", encoding="utf-8") as yaml_file:
         data = yaml.safe_load(yaml_file)
+        if "application" not in data:
+            return None, None
         application_name = data["application"]
 
     cert_directory = vespa_directory / application_name
-    cert_path = list(cert_directory.glob("*cert.pem"))[0]
-    key_path = list(cert_directory.glob("*key.pem"))[0]
+
+    cert_directory_certs = list(cert_directory.glob("*cert.pem"))
+    cert_path = (
+        str(list(cert_directory.glob("*cert.pem"))[0]) if cert_directory_certs else None
+    )
+    cert_directory_keys = list(cert_directory.glob("*key.pem"))
+    key_path = (
+        str(list(cert_directory.glob("*key.pem"))[0]) if cert_directory_keys else None
+    )
+
     return cert_path, key_path
 
 

tests/test_search_adaptors.py

@@ -37,6 +37,18 @@ def profile_search(
     return avg_ms
 
 
+@pytest.mark.parametrize(
+    "search_adaptor_params",
+    [
+        {"instance_url": "http://localhost:8080", "cert_directory": "/tmp"},
+        {"instance_url": "http://localhost:8080"},
+    ],
+)
+@pytest.mark.vespa
+def test_vespa_search_adaptor_instantiation(search_adaptor_params: dict) -> None:
+    VespaSearchAdapter(**search_adaptor_params)
+
+
 @pytest.mark.vespa
 def test_vespa_search_adaptor__works(test_vespa):
     request = SearchParameters(query_string="the")