Commit d9f208fa authored by Daniel Kastl's avatar Daniel Kastl Committed by Ken Zangelin
Browse files

fix(httpctrl): match POST query stubs on the selector without crashing 

A stub for POST /ngsi-ld/v1/entityOperations/query matches only when the
first entity selector in the request has the same type and id as the
stub's response body. This check had four problems:

- An array response body (what a query returns) caused a TypeError, and
  the mock sent no answer. D011_02_red_02, D011_03_inc_02 and
  D011_04_inc_02 use such stubs.
- The type was compared as plain text. A broker that sends the full type
  IRI (https://ngsi-ld-test-suite/context#Vehicle) never matched a body
  with type Vehicle. D010_03_inc_03 fails this way.
- A selector with only an id, only a type, or no entities caused a
  KeyError.
- Get Stub Count sends no request body, so counting such a stub caused an
  AttributeError.

Move the check into one function and use it in both places. A stub
matches when one entity in its response body (object or array) fits the
type and/or id of the selector. The id may be a list. The type also
matches by its last part after '#' or '/', because the mock does not read
@context. Without a request body (Get Stub Count), the stub matches by
method and URL, like any other stub.
parent 6415e404

libraries/robotframework-httpctrl/src/HttpCtrl/http_stub.py

+37 −22
Changes for libraries/robotframework-httpctrl/src/HttpCtrl/http_stub.py: 37 added lines, 22 removed lines.
Original line number Diff line number Diff line
@@ -14,6 +14,41 @@ from urllib.parse import parse_qs, unquote 
from HttpCtrl.utils.singleton import Singleton





def _same_type(requested, stubbed):

    # The mock does not read @context. So when the broker sends the full

    # type IRI (https://ngsi-ld-test-suite/context#Vehicle), compare only

    # the last part of it (Vehicle).

    if not isinstance(requested, str) or not isinstance(stubbed, str):

        return requested == stubbed

    return requested == stubbed or requested.endswith("#" + stubbed) or requested.endswith("/" + stubbed)





def _query_reply_matches(reply_text, request_bytes):

    # A POST /entityOperations/query stub matches when one entity in its

    # response body (an object or an array) fits the first entity selector

    # of the request.

    if request_bytes is None:

        # count() finds a stub by method and URL only, without a request body.

        return True

    try:

        selector = json.loads(request_bytes.decode('utf-8'))["entities"][0]

    except (ValueError, KeyError, IndexError, TypeError, AttributeError):

        return False

    if not isinstance(selector, dict) or ("type" not in selector and "id" not in selector):

        return False

    reply = json.loads(reply_text)

    ids = selector.get("id") if isinstance(selector.get("id"), list) else [selector.get("id")]

    for entity in (reply if isinstance(reply, list) else [reply]):

        if not isinstance(entity, dict):

            continue

        if "type" in selector and not _same_type(selector["type"], entity.get("type")):

            continue

        if "id" in selector and entity.get("id") not in ids:

            continue

        return True

    return False





class HttpStubCriteria:

    def __init__(self, **kwargs):

        self.method = kwargs.get('method', None)
@@ -167,33 +202,13 @@ class HttpStubContainer(metaclass=Singleton): 
                if stub.criteria.url.rstrip("/") == criteria_url_components[0].rstrip("/"):

                    # if the request is a query via POST, we should have a specific check

                    if stub.criteria.url == "/ngsi-ld/v1/entityoperations/query":

                        response_body = json.loads(stub.response.get_body())

                        body_str = body.decode('utf-8')

                        request_body = json.loads(body_str)

                        # check if the request body contains the same type as the response body

                        if "type" in request_body["entities"][0] or "id" in request_body["entities"][0]:

                            if request_body["entities"][0]["type"] != response_body["type"]:

                                return False

                            if request_body["entities"][0]["id"] != response_body["id"]:

                                return False

                        else:

                            return False

                        return _query_reply_matches(stub.response.get_body(), body)

                    return True

        

        if stub.criteria.url.rstrip("/") == criteria.url.rstrip("/"):

            # if the request is a query via POST, we should have a specific check

            if stub.criteria.url == "/ngsi-ld/v1/entityoperations/query":

                response_body = json.loads(stub.response.get_body())

                body_str = body.decode('utf-8')

                request_body = json.loads(body_str)

                # check if the request body contains the same type as the response body

                if "type" in request_body["entities"][0] or "id" in request_body["entities"][0]:

                    if request_body["entities"][0]["type"] != response_body["type"]:

                        return False

                    if request_body["entities"][0]["id"] != response_body["id"]:

                        return False

                else:

                    return False

                return _query_reply_matches(stub.response.get_body(), body)

            return True



        return False
 
 No newline at end of file