Make content-type response checks on '/versions` endpoint optional

optimade/validator/validator.py

@@ -926,8 +926,10 @@ def _test_versions_endpoint(self):
         # First, check that there is a versions endpoint in the appropriate place:
         # If passed a versioned URL, then strip that version from
         # the URL before looking for `/versions`.
+        _old_base_url = self.base_url
         if re.match(VERSIONS_REGEXP, self.base_url_parsed.path) is not None:
             self.client.base_url = "/".join(self.client.base_url.split("/")[:-1])
+            self.base_url = self.client.base_url
 
         response, _ = self._get_endpoint(
             CONF.versions_endpoint, expected_status_code=200
@@ -936,28 +938,31 @@ def _test_versions_endpoint(self):
         self._test_versions_endpoint_content(response, request=CONF.versions_endpoint)
 
         # If passed a versioned URL, first reset the URL of the client to the
-        # versioned one, then that this versioned URL does NOT host a versions endpoint
+        # versioned one, then test that this versioned URL does NOT host a versions endpoint
         if re.match(VERSIONS_REGEXP, self.base_url_parsed.path) is not None:
-            self.client.base_url = self.base_url
+            self.client.base_url = _old_base_url
+            self.base_url = _old_base_url
             self._get_endpoint(CONF.versions_endpoint, expected_status_code=404)
 
     @test_case
     def _test_versions_endpoint_content(
         self, response: requests.Response
-    ) -> Tuple[Optional[requests.Response], str]:
+    ) -> Tuple[requests.Response, str]:
         """Checks that the response from the versions endpoint complies
-        with the specification.
+        with the specification and that its 'Content-Type' header complies with
+        [RFC 4180](https://tools.ietf.org/html/rfc4180.html).
 
         Parameters:
             response: The HTTP response from the versions endpoint.
 
+        Raises:
+            ResponseError: If any content checks fail.
+
         Returns:
             The successful HTTP response or `None`, and a summary string.
 
         """
         text_content = response.text.strip().split("\n")
-        headers = response.headers
-
         if text_content[0] != "version":
             raise ResponseError(
                 f"First line of `/{CONF.versions_endpoint}` response must be 'version', not {text_content[0]}"
@@ -976,23 +981,57 @@ def _test_versions_endpoint_content(
                     f"Version numbers reported by `/{CONF.versions_endpoint}` must be integers specifying the major version, not {text_content}."
                 )
 
-        content_type = headers.get("content-type")
-        if content_type is not None:
-            content_type = [_.replace(" ", "") for _ in content_type.split(";")]
-        if not content_type or content_type[0].strip() != "text/csv":
+        content_type = response.headers.get("content-type")
+        if not content_type:
             raise ResponseError(
-                f"Incorrect content-type header {content_type} instead of 'text/csv'"
+                "Missing 'Content-Type' in response header from `/versions`."
             )
 
-        for type_parameter in content_type:
-            if type_parameter == "header=present":
-                break
-        else:
+        content_type = [_.replace(" ", "") for _ in content_type.split(";")]
+
+        self._test_versions_headers(
+            content_type, ("text/csv", "text/plain"), optional=True
+        )
+        self._test_versions_headers(content_type, "header=present", optional=True)
+
+        return response, "`/versions` endpoint responded correctly."
+
+    @test_case
+    def _test_versions_headers(
+        self,
+        content_type: Dict[str, Any],
+        expected_parameter: Union[str, List[str]],
+    ) -> Tuple[Dict[str, Any], str]:
+        """Tests that the `Content-Type` field of the `/versions` header contains
+        the passed parameter.
+
+        Arguments:
+            content_type: The 'Content-Type' field from the response of the `/versions` endpoint.
+            expected_paramter: A substring or list of substrings that are expected in
+                the Content-Type of the response. If multiple strings are passed, they will
+                be treated as possible alternatives to one another.
+
+        Raises:
+            ResponseError: If the expected 'Content-Type' parameter is missing.
+
+        Returns:
+            The HTTP response headers and a summary string.
+
+        """
+
+        if isinstance(expected_parameter, str):
+            expected_parameter = [expected_parameter]
+
+        if not any(param in content_type for param in expected_parameter):
             raise ResponseError(
-                f"Missing 'header=present' parameter in content-type {content_type}"
+                f"Incorrect 'Content-Type' header {';'.join(content_type)!r}.\n"
+                f"Missing at least one expected parameter(s): {expected_parameter!r}"
             )
 
-        return response, "`/versions` endpoint responded correctly."
+        return (
+            content_type,
+            f"`/versions` response had one of the expected Content-Type parameters {expected_parameter}.",
+        )
 
     def _test_as_type(self) -> None:
         """Tests that the base URL of the validator (i.e. with no

tests/validator/test_utils.py

@@ -367,3 +367,92 @@ def test_that_system_exit_is_fatal_in_test_case():
     assert validator.results.failure_count == 0
     assert validator.results.optional_failure_count == 0
     assert validator.results.internal_failure_count == 0
+
+
+def test_versions_test_cases():
+    """Check the `/versions` test cases."""
+    from requests import Response
+    from functools import partial
+
+    unversioned_base_url = "https://example.org"
+    versioned_base_url = unversioned_base_url + "/v1"
+
+    def monkey_patched_get_endpoint(
+        url,
+        expected_status_code=200,
+        content=b"version\n1",
+        content_type="text/csv;header=present",
+        **kwargs
+    ):
+        r = Response()
+        if expected_status_code == 200:
+            r._content = content
+            r.headers["content-type"] = content_type
+            r.status_code = 200
+        else:
+            r.status_code = 404
+
+        gets.append(url)
+
+        return r, None
+
+    # Test that the versioned base URL correctly tests the unversioned and then resets
+    validator = ImplementationValidator(base_url=versioned_base_url, verbosity=5)
+    gets = []
+    validator._get_endpoint = monkey_patched_get_endpoint
+    validator._test_versions_endpoint()
+    assert len(gets) == 2
+    assert validator.client.base_url == versioned_base_url
+    assert validator.results.success_count == 1
+    assert validator.results.optional_success_count == 2
+    assert validator.results.failure_count == 0
+
+    # Test that the unversioned base URL correctly tests just the unversioned URL
+    validator = ImplementationValidator(base_url=unversioned_base_url, verbosity=0)
+    gets = []
+    validator._get_endpoint = monkey_patched_get_endpoint
+    validator._test_versions_endpoint()
+    assert len(gets) == 1
+    assert validator.client.base_url == unversioned_base_url
+    assert validator.results.success_count == 1
+    assert validator.results.optional_success_count == 2
+    assert validator.results.failure_count == 0
+
+    # Test that bad content successfully triggers failures
+    bad_content = (b"versions\n1", b"version\n", b"version\n1.1")
+    for content in bad_content:
+        validator = ImplementationValidator(base_url=versioned_base_url, verbosity=0)
+        validator._get_endpoint = partial(monkey_patched_get_endpoint, content=content)
+        validator._test_versions_endpoint()
+        assert validator.client.base_url == versioned_base_url
+        assert validator.results.success_count == 0
+        assert validator.results.failure_count == 1
+        assert validator.results.optional_failure_count == 0
+
+    # Test that missing content-type header triggers a hard failure
+    validator = ImplementationValidator(base_url=versioned_base_url, verbosity=0)
+    validator._get_endpoint = partial(monkey_patched_get_endpoint, content_type="")
+    validator._test_versions_endpoint()
+    assert validator.client.base_url == versioned_base_url
+    assert validator.results.success_count == 0
+    assert validator.results.failure_count == 1
+    assert validator.results.optional_failure_count == 0
+
+    # Test that bad content-type headers successfully trigger *optional* failures
+    bad_content_type = (
+        ("text/csv", 1),
+        ("text/plain", 1),
+        ("header=present", 1),
+        ("x;y", 2),
+        ("application/json;charset=utf-8", 2),
+    )
+    for content_type, num_errors in bad_content_type:
+        validator = ImplementationValidator(base_url=versioned_base_url, verbosity=0)
+        validator._get_endpoint = partial(
+            monkey_patched_get_endpoint, content_type=content_type
+        )
+        validator._test_versions_endpoint()
+        assert validator.client.base_url == versioned_base_url
+        assert validator.results.success_count == 1
+        assert validator.results.failure_count == 0
+        assert validator.results.optional_failure_count == num_errors