"""This module contains the

[`ImplementationValidator`][optimade.validator.validator.ImplementationValidator]

class that can be pointed at an OPTIMADE implementation and validated

against the specification via the pydantic models implemented in this package.

"""

import dataclasses

import json

import logging

import random

import re

import sys

import urllib.parse

from typing import Any, Literal

import requests

from optimade.models import DataType, EntryInfoResponse, SupportLevel

from optimade.validator.config import VALIDATOR_CONFIG as CONF

from optimade.validator.utils import (

DEFAULT_CONN_TIMEOUT,

DEFAULT_READ_TIMEOUT,

Client,

ResponseError,

ValidatorEntryResponseMany,

ValidatorEntryResponseOne,

ValidatorResults,

print_failure,

print_notify,

print_success,

print_warning,

test_case,

)

VERSIONS_REGEXP = r".*/v[0-9]+(\.[0-9]+){,2}$"

__all__ = ("ImplementationValidator",)

class ImplementationValidator:

"""Class used to make a series of checks against a particular

OPTIMADE implementation over HTTP.

Uses the pydantic models in [`optimade.models`][optimade.models] to

validate the response from the server and crawl through the

available endpoints.

Attributes:

valid: whether or not the implementation was deemed valid, with

`None` signifying that tests did not run.

Caution:

Only works for current version of the specification as defined

by [`optimade.models`][optimade.models].

"""

valid: bool | None

def __init__(

self,

client: Any | None = None,

base_url: str | None = None,

verbosity: int = 0,

respond_json: bool = False,

page_limit: int = 4,

max_retries: int = 5,

run_optional_tests: bool = True,

fail_fast: bool = False,

as_type: str | None = None,

index: bool = False,

minimal: bool = False,

http_headers: dict[str, str] | None = None,

timeout: float = DEFAULT_CONN_TIMEOUT,

read_timeout: float = DEFAULT_READ_TIMEOUT,

):

"""Set up the tests to run, based on constants in this module

for required endpoints.

Arguments:

client: A client that has a `.get()` method to obtain the

response from the implementation. If `None`, then

[`Client`][optimade.validator.utils.Client] will be used.

base_url: The URL of the implementation to validate. Unless

performing "as_type" validation, this should point to the

base of the OPTIMADE implementation.

verbosity: The verbosity of the output and logging as an integer

(`0`: critical, `1`: warning, `2`: info, `3`: debug).

respond_json: If `True`, print only a JSON representation of the

results of validation to stdout.

page_limit: The default page limit to apply to filters.

max_retries: Argument is passed to the client for how many

attempts to make for a request before failing.

run_optional_tests: Whether to run the tests on optional

OPTIMADE features.

fail_fast: Whether to exit validation after the first failure

of a mandatory test.

as_type: An OPTIMADE entry or endpoint type to coerce the response

from implementation into, e.g. "structures". Requires `base_url`

to be pointed to the corresponding endpoint.

index: Whether to validate the implementation as an index meta-database.

minimal: Whether or not to run only a minimal test set.

http_headers: Dictionary of additional headers to add to every request.

timeout: The connection timeout to use for all requests (in seconds).

read_timeout: The read timeout to use for all requests (in seconds).

"""

self.verbosity = verbosity

self.max_retries = max_retries

self.page_limit = page_limit

self.index = index

self.run_optional_tests = run_optional_tests

self.fail_fast = fail_fast

self.respond_json = respond_json

self.minimal = minimal

if as_type is None:

self.as_type_cls = None

elif self.index:

if as_type not in CONF.response_classes_index:

raise RuntimeError(

f"Provided as_type='{as_type}' not allowed for an Index meta-database."

)

self.as_type_cls = CONF.response_classes_index[as_type]

elif as_type in ("structure", "reference"):

self.as_type_cls = CONF.response_classes[f"{as_type}s/"]

else:

self.as_type_cls = CONF.response_classes[as_type]

if client is None and base_url is None:

raise RuntimeError(

"Need at least a URL or a client to initialize validator."

)

if base_url and client:

raise RuntimeError("Please specify at most one of base_url or client.")

if client:

self.client = client

self.base_url = self.client.base_url

# If a custom client has been provided, try to set custom headers if they have been specified,

# but do not overwrite any existing attributes

if http_headers:

if not hasattr(self.client, "headers"):

self.client.headers = http_headers

else:

print_warning(

f"Not using specified request headers {http_headers} with custom client {self.client}."

)

else:

while base_url.endswith("/"): # type: ignore[union-attr]

base_url = base_url[:-1] # type: ignore[index]

self.base_url = base_url

self.client = Client(

self.base_url, # type: ignore[arg-type]

max_retries=self.max_retries,

headers=http_headers,

timeout=timeout,

read_timeout=read_timeout,

)

self._setup_log()

self._response_classes = (

CONF.response_classes_index if self.index else CONF.response_classes

)

# some simple checks on base_url

self.base_url = str(self.base_url)

self.base_url_parsed = urllib.parse.urlparse(self.base_url)

# only allow filters/endpoints if we are working in "as_type" mode

if self.as_type_cls is None and self.base_url_parsed.query:

raise SystemExit(

f"Base URL {self.base_url} not appropriate: should not contain a filter."

)

self.valid = None

self._test_id_by_type: dict[str, Any] = {}

self._entry_info_by_type: dict[str, Any] = {}

self.results = ValidatorResults(verbosity=self.verbosity)

def _setup_log(self):

"""Define stdout log based on given verbosity."""

self._log = logging.getLogger("optimade").getChild("validator")

self._log.handlers = []

stdout_handler = logging.StreamHandler(sys.stdout)

stdout_handler.setFormatter(

logging.Formatter("%(asctime)s - %(name)s | %(levelname)8s: %(message)s")

)

if not self.respond_json:

self._log.addHandler(stdout_handler)

else:

self.verbosity = -1

if self.verbosity == 0:

self._log.setLevel(logging.CRITICAL)

elif self.verbosity == 1:

self._log.setLevel(logging.WARNING)

elif self.verbosity == 2:

self._log.setLevel(logging.INFO)

elif self.verbosity > 0:

self._log.setLevel(logging.DEBUG)

def print_summary(self):

"""Print a summary of the results of validation."""

if self.respond_json:

print(json.dumps(dataclasses.asdict(self.results), indent=2))

return

if self.results.failure_messages:

print("\n\nFAILURES")

print("========\n")

for message in self.results.failure_messages:

print_failure(message[0])

for line in message[1].split("\n"):

print_warning("\t" + line)

if self.results.optional_failure_messages:

print("\n\nOPTIONAL TEST FAILURES")

print("======================\n")

for message in self.results.optional_failure_messages:

print_notify(message[0])

for line in message[1].split("\n"):

print_warning("\t" + line)

if self.results.internal_failure_messages:

print("\n\nINTERNAL FAILURES")

print("=================\n")

print(

"There were internal validator failures associated with this run.\n"

"If this problem persists, please report it at:\n"

"https://github.com/Materials-Consortia/optimade-python-tools/issues/new\n"

)

for message in self.results.internal_failure_messages:

print_warning(message[0])

for line in message[1].split("\n"):

print_warning("\t" + line)

if self.valid or (not self.valid and not self.fail_fast):

final_message = f"\n\nPassed {self.results.success_count} out of {self.results.success_count + self.results.failure_count + self.results.internal_failure_count} tests."

if not self.valid:

print_failure(final_message)

else:

print_success(final_message)

if self.run_optional_tests and not self.fail_fast:

print(

f"Additionally passed {self.results.optional_success_count} out of "

f"{self.results.optional_success_count + self.results.optional_failure_count} optional tests."

)

def validate_implementation(self):

"""Run all the test cases on the implementation, or the single type test,

depending on what options were provided on initialiation.

Sets the `self.valid` attribute to `True` or `False` depending on the

outcome of the tests.

Raises:

RuntimeError: If it was not possible to start the validation process.

"""

# If a single "as type" has been set, only run that test

if self.as_type_cls is not None:

self._log.debug(

"Validating response of %s with model %s",

self.base_url,

self.as_type_cls,

)

self._test_as_type()

self.valid = not bool(self.results.failure_count)

self.print_summary()

return

# Test entire implementation

if self.verbosity >= 0:

print(f"Testing entire implementation at {self.base_url}")

info_endp = CONF.info_endpoint

self._log.debug("Testing base info endpoint of %s", info_endp)

# Get and validate base info to find endpoints

# If this is not possible, then exit at this stage

base_info = self._test_info_or_links_endpoint(info_endp)

if not base_info:

self._log.critical(

f"Unable to deserialize response from introspective {info_endp!r} endpoint. "

"This is required for all further validation, so the validator will now exit."

)

# Set valid to False to ensure error code 1 is raised at CLI

self.valid = False

self.print_summary()

return

# Grab the provider prefix from base info and use it when looking for provider fields

self.provider_prefix = None

meta = base_info.get("meta", {})

if meta.get("provider") is not None:

self.provider_prefix = meta["provider"].get("prefix")

# Set the response class for all `/info/entry` endpoints based on `/info` response

self.available_json_endpoints, _ = self._get_available_endpoints(

base_info, request=info_endp

)

for endp in self.available_json_endpoints:

self._response_classes[f"{info_endp}/{endp}"] = EntryInfoResponse

# Run some tests on the versions endpoint

self._log.debug("Testing versions endpoint %s", CONF.versions_endpoint)

self._test_versions_endpoint()

self._test_bad_version_returns_553()

# Test that entry info endpoints deserialize correctly

# If they do not, the corresponding entry in _entry_info_by_type

# is set to False, which must be checked for further validation

for endp in self.available_json_endpoints:

entry_info_endpoint = f"{info_endp}/{endp}"

self._log.debug("Testing expected info endpoint %s", entry_info_endpoint)

self._entry_info_by_type[endp] = self._test_info_or_links_endpoint(

entry_info_endpoint

)

# Test that the results from multi-entry-endpoints obey, e.g. page limits,

# and that all entries can be deserialized with the patched models.

# These methods also set the test_ids for each type of entry, which are validated

# in the next loop.

for endp in self.available_json_endpoints:

self._log.debug("Testing multiple entry endpoint of %s", endp)

self._test_multi_entry_endpoint(endp)

# Test that the single IDs scraped earlier work with the single entry endpoint

for endp in self.available_json_endpoints:

self._log.debug("Testing single entry request of type %s", endp)

self._test_single_entry_endpoint(endp)

# Use the _entry_info_by_type to construct filters on the relevant endpoints

if not self.minimal:

for endp in self.available_json_endpoints:

self._log.debug("Testing queries on JSON entry endpoint of %s", endp)

self._recurse_through_endpoint(endp)

# Test that the links endpoint can be serialized correctly

self._log.debug("Testing %s endpoint", CONF.links_endpoint)

self._test_info_or_links_endpoint(CONF.links_endpoint)

self.valid = not (

self.results.failure_count or self.results.internal_failure_count

)

self.print_summary()

@test_case

def _recurse_through_endpoint(self, endp: str) -> tuple[bool | None, str]:

"""For a given endpoint (`endp`), get the entry type

and supported fields, testing that all mandatory fields

are supported, then test queries on every property according

to the reported type, with optionality decided by the

specification-level support level for that field.

Parameters:

endp: Endpoint to be tested.

Returns:

`True` if endpoint passed the tests, and a string summary.

"""

entry_info = self._entry_info_by_type.get(endp)

if not entry_info:

raise ResponseError(

f"Unable to generate filters for endpoint {endp}: 'info/{endp}' response was malformed."

)

_impl_properties = self._check_entry_info(entry_info, endp)

prop_list = list(_impl_properties.keys())

self._check_response_fields(endp, prop_list)

chosen_entry, _ = self._get_archetypal_entry(endp, prop_list)

if not chosen_entry:

return (

None,

f"Unable to generate filters for endpoint {endp}: no valid entries found.",

)

for prop in _impl_properties:

# check support level of property

prop_type = _impl_properties[prop]["type"]

sortable = _impl_properties[prop]["sortable"]

optional = (

CONF.entry_schemas[endp].get(prop, {}).get("queryable")

== SupportLevel.OPTIONAL

)

if optional and not self.run_optional_tests:

continue

self._construct_queries_for_property(

prop,

prop_type,

sortable,

endp,

chosen_entry,

request=f"; testing queries for {endp}->{prop}",

optional=optional,

)

self._test_unknown_provider_property(endp)

self._test_completely_unknown_property(endp)

return True, f"successfully recursed through endpoint {endp}."

@test_case

def _test_completely_unknown_property(self, endp):

request = f"{endp}?filter=crazyfield = 2"

response, _ = self._get_endpoint(

request,

expected_status_code=400,

)

return True, "unknown field returned 400 Bad Request, as expected"

@test_case

def _test_unknown_provider_property(self, endp):

dummy_provider_field = "_crazyprovider_field"

request = f"{endp}?filter={dummy_provider_field}=2"

response, _ = self._get_endpoint(

request,

multistage=True,

request=request,

)

if response is not None:

deserialized, _ = self._deserialize_response(

response, CONF.response_classes[endp], request=request, multistage=True

)

return (

True,

"Unknown provider field was ignored when filtering, as expected",

)

raise ResponseError(

"Failed to handle field from unknown provider; should return without affecting filter results"

)

def _check_entry_info(

self, entry_info: dict[str, Any], endp: str

) -> dict[str, dict[str, Any]]:

"""Checks that `entry_info` contains all the required properties,

and returns the property list for the endpoint.

Parameters:

entry_info: JSON representation of the response from the

entry info endpoint.

endp: The name of the entry endpoint.

Returns:

The list of property names supported by this implementation.

"""

properties = entry_info.get("data", {}).get("properties", [])

self._test_must_properties(

properties, endp, request=f"{CONF.info_endpoint}/{endp}"

)

return properties

@test_case

def _test_must_properties(

self, properties: list[str], endp: str

) -> tuple[bool, str]:

"""Check that the entry info lists all properties with the "MUST"

support level for this endpoint.

Parameters:

properties: The list of property names supported by the endpoint.

endp: The endpoint.

Returns:

`True` if the properties were found, and a string summary.

"""

must_props = {

prop

for prop in CONF.entry_schemas.get(endp, {})

if CONF.entry_schemas[endp].get(prop, {}).get("support")

== SupportLevel.MUST

}

must_props_supported = {prop for prop in properties if prop in must_props}

missing = must_props - must_props_supported

if len(missing) != 0:

raise ResponseError(

f"Some 'MUST' properties were missing from info/{endp}: {missing}"

)

return True, f"Found all required properties in entry info for endpoint {endp}"

@test_case

def _get_archetypal_entry(

self, endp: str, properties: list[str]

) -> tuple[dict[str, Any] | None, str]:

"""Get a random entry from the first page of results for this

endpoint.

Parameters:

endp: The endpoint to query.

Returns:

The JSON representation of the chosen entry and the summary message.

"""

response, message = self._get_endpoint(endp, multistage=True)

if response:

data = response.json().get("data", [])

data_returned = len(data)

if data_returned < 1:

return (

None,

"Endpoint {endp!r} returned no entries, cannot get archetypal entry or test filtering.",

)

archetypal_entry = response.json()["data"][

random.randint(0, data_returned - 1)

]

if "id" not in archetypal_entry:

raise ResponseError(

f"Chosen archetypal entry did not have an ID, cannot proceed: {archetypal_entry!r}"

)

return (

archetypal_entry,

f"set archetypal entry for {endp} with ID {archetypal_entry['id']}.",

)

raise ResponseError(f"Failed to get archetypal entry. Details: {message}")

@test_case

def _check_response_fields(

self, endp: str, fields: list[str]

) -> tuple[bool | None, str]:

"""Check that the response field query parameter is obeyed.

Parameters:

endp: The endpoint to query.

fields: The known fields for this endpoint to test.

Returns:

Bool indicating success and a summary message.

"""

subset_fields = random.sample(fields, min(len(fields) - 1, 3))

test_query = f"{endp}?response_fields={','.join(subset_fields)}&page_limit=1"

response, _ = self._get_endpoint(test_query, multistage=True)

if response and len(response.json()["data"]) > 0:

doc = response.json()["data"][0]

expected_fields = set(subset_fields)

expected_fields -= CONF.top_level_non_attribute_fields

if "attributes" not in doc:

raise ResponseError(

f"Entries are missing `attributes` key.\nReceived: {doc}"

)

returned_fields = set(sorted(list(doc.get("attributes", {}).keys())))

returned_fields -= CONF.top_level_non_attribute_fields

if expected_fields != returned_fields:

raise ResponseError(

f"Response fields not obeyed by {endp!r}:\nExpected: {expected_fields}\nReturned: {returned_fields}"

)

return True, "Successfully limited response fields"

return (

None,

f"Unable to test adherence to response fields as no entries were returned for endpoint {endp!r}.",

)

@test_case

def _construct_queries_for_property(

self,

prop: str,

prop_type: DataType,

sortable: bool,

endp: str,

chosen_entry: dict[str, Any],

) -> tuple[bool | None, str]:

"""For the given property, property type and chose entry, this method

runs a series of queries for each field in the entry, testing that the

initial document is returned where expected.

Parameters:

prop: The property name.

prop_type: The property type.

sortable: Whether the implementation has indicated that the field is sortable.

endp: The corresponding entry endpoint.

chosen_entry: A JSON respresentation of the chosen entry that will be used to

construct the filters.

Returns:

Boolean indicating success (`True`) or failure/irrelevance

(`None`) and the string summary of the test case.

"""

# Explicitly handle top level keys that do not have types in info

if not chosen_entry:

raise ResponseError(

f"Chosen entry of endpoint '/{endp}' failed validation."

)

if prop == "type":

if chosen_entry.get("type") == endp:

return True, f"Successfully validated {prop}"

raise ResponseError(

f"Chosen entry of endpoint '{endp}' had unexpected or missing type: {chosen_entry.get('type')!r}."

)

prop_type = (

CONF.entry_schemas[endp].get(prop, {}).get("type")

if prop_type is None

else prop_type

)

if prop_type is None:

raise ResponseError(

f"Cannot validate queries on {prop!r} as field type was not reported in `/info/{endp}`"

)

# this is the case of a provider field

if prop not in CONF.entry_schemas[endp]:

if self.provider_prefix is None:

raise ResponseError(

f"Found unknown field {prop!r} in `/info/{endp}` and no provider prefix was provided in `/info`"

)

elif not prop.startswith(f"_{self.provider_prefix}_"):

raise ResponseError(

f"Found unknown field {prop!r} that did not start with provider prefix '_{self.provider_prefix}_'"

)

return (

None,

f"Found provider field {prop!r}, will not test queries as they are strictly optional.",

)

query_optional = (

CONF.entry_schemas[endp].get(prop, {}).get("queryable")

== SupportLevel.OPTIONAL

)

return self._construct_single_property_filters(

prop, prop_type, sortable, endp, chosen_entry, query_optional

)

@staticmethod

def _format_test_value(test_value: Any, prop_type: DataType, operator: str) -> str:

"""Formats the test value as a string according to the type of the property.

Parameters:

test_value: The value to format.

prop_type: The OPTIMADE data type of the field.

operator: The operator that will be applied to it.

Returns:

The value formatted as a string to use in an OPTIMADE filter.

"""

if prop_type == DataType.LIST:

if operator in ("HAS ALL", "HAS ANY"):

_vals = sorted(set(test_value))

if isinstance(test_value[0], str):

_vals = [f'"{val}"' for val in _vals]

else:

_vals = [f"{val}" for val in _vals]

_test_value = ",".join(_vals)

elif operator == "LENGTH":

_test_value = f"{len(test_value)}"

else:

if isinstance(test_value[0], str):

_test_value = f'"{test_value[0]}"'

else:

_test_value = test_value[0]

elif prop_type in (DataType.STRING, DataType.TIMESTAMP):

_test_value = f'"{test_value}"'

else:

_test_value = test_value

return _test_value

def _construct_single_property_filters(

self,

prop: str,

prop_type: DataType,

sortable: bool,

endp: str,

chosen_entry: dict[str, Any],

query_optional: bool,

) -> tuple[bool | None, str]:

"""This method constructs appropriate queries using all operators

for a certain field and applies some tests:

- inclusive operators return compatible entries, e.g. `>=` always returns

at least the results of `=`.

- exclusive operators never return contradictory entries, e.g.

`nsites=1` never returns the same entries as `nsites!=1`, modulo

pagination.

Parameters:

prop: The property name.

prop_type: The property type.

sortable: Whether the implementation has indicated that the field is sortable.

endp: The corresponding entry endpoint.

chosen_entry: A JSON respresentation of the chosen entry that will be used to

construct the filters.

query_optional: Whether to treat query success as optional.

Returns:

Boolean indicating success (`True`) or failure/irrelevance

(`None`) and the string summary of the test case.

"""

if prop == "id":

test_value = chosen_entry.get("id")

else:

test_value = chosen_entry.get("attributes", {}).get(prop, "_missing")

if test_value in ("_missing", None):

support = CONF.entry_schemas[endp].get(prop, {}).get("support")

queryable = CONF.entry_schemas[endp].get(prop, {}).get("queryable")

submsg = "had no value" if test_value == "_missing" else "had `None` value"

msg = (

f"Chosen entry {submsg} for {prop!r} with support level {support} and queryability {queryable}, "

f"so cannot construct test queries. This field should potentially be removed from the `/info/{endp}` endpoint response."

)

# None values are allowed for OPTIONAL and SHOULD, so we can just skip

if support in (

SupportLevel.OPTIONAL,

SupportLevel.SHOULD,

):

self._log.info(msg)

return None, msg

# Otherwise, None values are not allowed for MUST's, and entire missing fields are not allowed

raise ResponseError(msg)

using_fallback = False

if prop_type == DataType.LIST:

if not test_value:

test_value = CONF.enum_fallback_values.get(endp, {}).get(prop)

using_fallback = True

if not test_value:

msg = f"Not testing filters on field {prop} of type {prop_type} as no test value was found to use in filter."

self._log.warning(msg)

return None, msg

if isinstance(test_value[0], dict) or isinstance(test_value[0], list):

msg = f"Not testing filters on field {prop} of type {prop_type} with nested dictionary/list test value."

self._log.warning(msg)

return None, msg

# Try to infer if the test value is a float from its string representation

# and decide whether to do inclusive/exclusive query tests

try:

float(test_value[0])

msg = f"Not testing filters on field {prop} of type {prop_type} containing float values."

self._log.warning(msg)

return None, msg

except ValueError:

pass

if prop_type in (DataType.DICTIONARY,):

msg = f"Not testing queries on field {prop} of type {prop_type}."

self._log.warning(msg)

return None, msg

num_data_returned = {}

inclusive_operators = CONF.inclusive_operators[prop_type]

exclusive_operators = CONF.exclusive_operators[prop_type]

field_specific_support_overrides = CONF.field_specific_overrides.get(prop, {})

for operator in inclusive_operators | exclusive_operators:

# Need to pre-format list and string test values for the query

_test_value = self._format_test_value(test_value, prop_type, operator)

query_optional = (

query_optional

or operator

in field_specific_support_overrides.get(SupportLevel.OPTIONAL, [])

)

query = f"{prop} {operator} {_test_value}"

request = f"{endp}?filter={query}"

response, message = self._get_endpoint(

request,

multistage=True,

optional=query_optional,

expected_status_code=(200, 501),

)

if response is None or response.status_code != 200:

if query_optional:

return (

None,

"Optional query {query!r} raised the error: {message}.",

)

raise ResponseError(

f"Unable to perform mandatory query {query!r}, which raised the error: {message}"

)

response = response.json()

if "meta" not in response or "more_data_available" not in response["meta"]:

raise ResponseError(

f"Required field `meta->more_data_available` missing from response for {request}."

)

if not response["meta"]["more_data_available"] and "data" in response:

num_data_returned[operator] = len(response["data"])

else:

num_data_returned[operator] = response["meta"].get("data_returned")

if prop in CONF.unique_properties and operator == "=":

if num_data_returned["="] is not None and num_data_returned["="] == 0:

raise ResponseError(

f"Unable to filter field 'id' for equality, no data was returned for {query}."

)

if num_data_returned["="] is not None and num_data_returned["="] > 1:

raise ResponseError(

f"Filter for an individual 'id' returned {num_data_returned['=']} results, when only 1 was expected."

)

num_response = num_data_returned[operator]

excluded = operator in exclusive_operators

# if we have all results on this page, check that the blessed ID is in the response

if excluded and (

chosen_entry.get("id", "")

in {entry.get("id") for entry in response["data"]}

):

raise ResponseError(

f"Entry {chosen_entry['id']} with value {prop!r}: {test_value} was not excluded by {query!r}"

)

# check that at least the archetypal structure was returned, unless we are using a fallback value

if not excluded and not using_fallback:

if (

num_data_returned[operator] is not None

and num_data_returned[operator] < 1

):

raise ResponseError(

f"Supposedly inclusive query {query!r} did not include original entry ID {chosen_entry['id']!r} "

f"(with field {prop!r} = {test_value}) potentially indicating a problem with filtering on this field."

)

# check that the filter returned no entries that had a null or missing value for the filtered property

if any(

entry.get("attributes", {}).get(prop, entry.get(prop, None)) is None

for entry in response.get("data", [])

):

raise ResponseError(

f"Filter {query!r} on field {prop!r} returned entries that had null or missing values for the field."

)

# Numeric and string comparisons must work both ways...

if prop_type in (

DataType.STRING,

DataType.INTEGER,

DataType.FLOAT,

DataType.TIMESTAMP,

) and operator not in (

"CONTAINS",

"STARTS",

"STARTS WITH",

"ENDS",

"ENDS WITH",

):

reversed_operator = operator.replace("<", ">")

if "<" in operator:

reversed_operator = operator.replace("<", ">")

elif ">" in operator:

reversed_operator = operator.replace(">", "<")

# Don't try to reverse string comparison as it is ill-defined

if prop_type == DataType.STRING and any(

comp in operator for comp in ("<", ">")

):

continue

reversed_query = f"{_test_value} {reversed_operator} {prop}"

reversed_request = f"{endp}?filter={reversed_query}"

reversed_response, message = self._get_endpoint(

reversed_request,

multistage=True,

optional=query_optional,

expected_status_code=(200, 501),

)

if not reversed_response:

if query_optional:

return (

None,

"Optional query {reversed_query!r} raised the error: {message}.",

)

raise ResponseError(

f"Unable to perform mandatory query {reversed_query!r}, which raised the error: {message}"

)

reversed_response = reversed_response.json()

if (

"meta" not in reversed_response

or "more_data_available" not in reversed_response["meta"]

):

raise ResponseError(

f"Required field `meta->more_data_available` missing from response for {request}."

)

if not reversed_response["meta"]["more_data_available"]:

num_reversed_response = len(reversed_response["data"])

else:

num_reversed_response = reversed_response["meta"].get(

"data_returned"

)

if num_response is not None and num_reversed_response is not None:

if reversed_response["meta"].get("data_returned") != response[

"meta"

].get("data_returned"):

raise ResponseError(

f"Query {query} did not work both ways around: {reversed_query}, "

"returning a different number of results each time (as reported by `meta->data_returned`)"

)

# check that the filter returned no entries that had a null or missing value for the filtered property

if any(

entry.get("attributes", {}).get(prop, entry.get(prop, None)) is None

for entry in reversed_response.get("data", [])

):

raise ResponseError(

f"Filter {reversed_query!r} on field {prop!r} returned entries that had null or missing values for the field."

)

return True, f"{prop} passed filter tests"

def _test_info_or_links_endpoint(self, request_str: str) -> Literal[False] | dict:

"""Requests an info or links endpoint and attempts to deserialize

the response.

Parameters:

request_str: The request to make, e.g. "links".

Returns:

`False` if the info response failed deserialization,

otherwise returns the deserialized object.

"""

response, _ = self._get_endpoint(request_str)

if response:

deserialized, _ = self._deserialize_response(

response,

self._response_classes[request_str],

request=request_str,

)

if deserialized:

return deserialized.model_dump()

return False

def _test_single_entry_endpoint(self, endp: str) -> None:

"""Requests and deserializes a single entry endpoint with the

appropriate model.

Parameters:

request_str: The single entry request to make, e.g. "structures/id_1".

"""

response_cls_name = endp + "/"

if response_cls_name in self._response_classes:

response_cls = self._response_classes[response_cls_name]

else:

self._log.warning(

"Deserializing single entry response %s with generic response rather than defined endpoint.",

endp,

)

response_cls = ValidatorEntryResponseOne

response_fields = set()

if endp in CONF.entry_schemas:

response_fields = (

set(CONF.entry_schemas[endp].keys())