added error handling for bad schema urls

apps/schematic/api/schematic_api/controllers/utils.py

@@ -3,6 +3,7 @@
 import urllib.request
 import shutil
 import tempfile
+from urllib.error import HTTPError
 
 from flask import request  # type: ignore
 from synapseclient.core.exceptions import (  # type: ignore
@@ -58,6 +59,11 @@ def func(*args: Any, **kwargs: Any) -> tuple[Union[Any, BasicError], int]:
             status = 403
             res = BasicError("Synapse entity access error", status, str(error))
             return res, status
+
+        except InvalidSchemaURL as error:
+            status = 404
+            res = BasicError("Invalid URL", status, str(error))
+            return res, status
 
         except Exception as error:  # pylint: disable=broad-exception-caught
             status = 500
@@ -67,18 +73,49 @@ def func(*args: Any, **kwargs: Any) -> tuple[Union[Any, BasicError], int]:
     return func
 
 
+class InvalidSchemaURL(Exception):
+    """Raised when a provided url for a schema is incorrect"""
+
+    def __init__(self, message: str, url: str):
+        """
+        Args:
+            message (str): The error message
+            url (str): The provided incorrect URL
+        """
+        self.message = message
+        self.url = url
+        super().__init__(self.message)
+
+    def __str__(self) -> str:
+        return f"{self.message}: {self.url}"
+
+
 def download_schema_file_as_jsonld(schema_url: str) -> str:
     """Downloads a schema and saves it as temp file
 
     Args:
         schema_url (str): The URL of the schema
 
+    Raises:
+        InvalidSchemaURL: When the schema url doesn't exist or is badly formatted
+
     Returns:
         str: The path fo the schema jsonld file
     """
-    with urllib.request.urlopen(schema_url) as response:
-        with tempfile.NamedTemporaryFile(
-            delete=False, suffix=".model.jsonld"
-        ) as tmp_file:
-            shutil.copyfileobj(response, tmp_file)
-            return tmp_file.name
+    try:
+        with urllib.request.urlopen(schema_url) as response:
+            with tempfile.NamedTemporaryFile(
+                delete=False, suffix=".model.jsonld"
+            ) as tmp_file:
+                shutil.copyfileobj(response, tmp_file)
+                return tmp_file.name
+    except ValueError as error:
+        # checks for specific ValueError where the url isn't correctly formatted
+        if str(error).startswith("unknown url type"):
+            raise InvalidSchemaURL("The provided URL is incorrect", schema_url) from error
+        # reraises the ValueError if it isn't the specific type above
+        else:
+            raise
+    except HTTPError as error:
+        raise InvalidSchemaURL("The provided URL could not be found", schema_url) from error
+

apps/schematic/api/schematic_api/test/test_utils.py

@@ -1,13 +1,14 @@
 """Tests for utils"""
 
+import pytest
 
 from synapseclient.core.exceptions import (  # type: ignore
     SynapseNoCredentialsError,
     SynapseAuthenticationError,
 )
 from schematic.exceptions import AccessCredentialsError  # type: ignore
 
-from schematic_api.controllers.utils import handle_exceptions
+from schematic_api.controllers.utils import handle_exceptions, download_schema_file_as_jsonld, InvalidSchemaURL
 from schematic_api.models.basic_error import BasicError
 
 
@@ -38,8 +39,35 @@ def test_403(self) -> None:
         assert status == 403
         assert isinstance(res, BasicError)
 
+    def test_404(self) -> None:
+        "Tests for 404 status"
+        res, status = func(InvalidSchemaURL("message", "url"))
+        assert status == 404
+        assert isinstance(res, BasicError)
+
     def test_500(self) -> None:
         "Tests for 500 status"
         res, status = func(TypeError)
         assert status == 500
         assert isinstance(res, BasicError)
+
+class TestDownloadSchemaFileAsJsonLD:
+    "tests download_schema_file_as_jsonld"
+    def test_success(self, test_schema_url: str) -> None:
+        "tests for successful download"
+        file_path = download_schema_file_as_jsonld(test_schema_url)
+        assert file_path
+
+    def test_failure(self) -> None:
+        "tests for exception"
+        with pytest.raises(
+            InvalidSchemaURL,
+            match="The provided URL is incorrect: xxx"
+        ):
+            download_schema_file_as_jsonld("xxx")
+
+        with pytest.raises(
+            InvalidSchemaURL,
+            match="The provided URL could not be found: https://raw.github.com/model.jsonld"
+        ):
+            download_schema_file_as_jsonld("https://raw.github.com/model.jsonld")