Commit 95d9a3da authored by Victor Nogueira's avatar Victor Nogueira Committed by Jakub Kicinski
Browse files

selftests: tc-testing: Add matchJSON to tdc 



This allows the use of a matchJSON field in tests to match
against JSON output from the command under test, if that
command outputs JSON.

You specify what you want to match against as a JSON array
or object in the test's matchJSON field. You can leave out
any fields you don't want to match against that are present
in the output and they will be skipped.

An example matchJSON value would look like this:

"matchJSON": [
  {
    "Value": {
      "neighIP": {
        "family": 4,
        "addr": "AQIDBA==",
        "width": 32
      },
      "nsflags": 142,
      "ncflags": 0,
      "LLADDR": "ESIzRFVm"
    }
  }
]

The real output from the command under test might have some
extra fields that we don't care about for matching, and
since we didn't include them in our matchJSON value, those
fields will not be attempted to be matched. If everything
we included above has the same values as the real command
output, the test will pass.

The matchJSON field's type must be the same as the command
output's type, otherwise the test will fail. So if the
command outputs an array, then the value of matchJSON must
also be an array.

If matchJSON is an array, it must not contain more elements
than the command output's array, otherwise the test will
fail.

Signed-off-by: default avatarJeremy Carter <jeremy@mojatatu.com>
Signed-off-by: default avatarVictor Nogueira <victor@mojatatu.com>
Acked-by: default avatarJamal Hadi Salim <jhs@mojatatu.com>
Link: https://lore.kernel.org/r/20221024111603.2185410-1-victor@mojatatu.com


Signed-off-by: default avatarJakub Kicinski <kuba@kernel.org>
parent 3a6573b7

tools/testing/selftests/tc-testing/tdc.py

+118 −7
Original line number Diff line number Diff line
@@ -246,6 +246,110 @@ def prepare_env(args, pm, stage, prefix, cmdlist, output = None): 
                stage, output,

                '"{}" did not complete successfully'.format(prefix))



def verify_by_json(procout, res, tidx, args, pm):

    try:

        outputJSON = json.loads(procout)

    except json.JSONDecodeError:

        res.set_result(ResultState.fail)

        res.set_failmsg('Cannot decode verify command\'s output. Is it JSON?')

        return res



    matchJSON = json.loads(json.dumps(tidx['matchJSON']))



    if type(outputJSON) != type(matchJSON):

        failmsg = 'Original output and matchJSON value are not the same type: output: {} != matchJSON: {} '

        failmsg = failmsg.format(type(outputJSON).__name__, type(matchJSON).__name__)

        res.set_result(ResultState.fail)

        res.set_failmsg(failmsg)

        return res



    if len(matchJSON) > len(outputJSON):

        failmsg = "Your matchJSON value is an array, and it contains more elements than the command under test\'s output:\ncommand output (length: {}):\n{}\nmatchJSON value (length: {}):\n{}"

        failmsg = failmsg.format(len(outputJSON), outputJSON, len(matchJSON), matchJSON)

        res.set_result(ResultState.fail)

        res.set_failmsg(failmsg)

        return res

    res = find_in_json(res, outputJSON, matchJSON, 0)



    return res



def find_in_json(res, outputJSONVal, matchJSONVal, matchJSONKey=None):

    if res.get_result() == ResultState.fail:

        return res



    if type(matchJSONVal) == list:

        res = find_in_json_list(res, outputJSONVal, matchJSONVal, matchJSONKey)



    elif type(matchJSONVal) == dict:

        res = find_in_json_dict(res, outputJSONVal, matchJSONVal)

    else:

        res = find_in_json_other(res, outputJSONVal, matchJSONVal, matchJSONKey)



    if res.get_result() != ResultState.fail:

        res.set_result(ResultState.success)

        return res



    return res



def find_in_json_list(res, outputJSONVal, matchJSONVal, matchJSONKey=None):

    if (type(matchJSONVal) != type(outputJSONVal)):

        failmsg = 'Original output and matchJSON value are not the same type: output: {} != matchJSON: {}'

        failmsg = failmsg.format(outputJSONVal, matchJSONVal)

        res.set_result(ResultState.fail)

        res.set_failmsg(failmsg)

        return res



    if len(matchJSONVal) > len(outputJSONVal):

        failmsg = "Your matchJSON value is an array, and it contains more elements than the command under test\'s output:\ncommand output (length: {}):\n{}\nmatchJSON value (length: {}):\n{}"

        failmsg = failmsg.format(len(outputJSONVal), outputJSONVal, len(matchJSONVal), matchJSONVal)

        res.set_result(ResultState.fail)

        res.set_failmsg(failmsg)

        return res



    for matchJSONIdx, matchJSONVal in enumerate(matchJSONVal):

        res = find_in_json(res, outputJSONVal[matchJSONIdx], matchJSONVal,

                           matchJSONKey)

    return res



def find_in_json_dict(res, outputJSONVal, matchJSONVal):

    for matchJSONKey, matchJSONVal in matchJSONVal.items():

        if type(outputJSONVal) == dict:

            if matchJSONKey not in outputJSONVal:

                failmsg = 'Key not found in json output: {}: {}\nMatching against output: {}'

                failmsg = failmsg.format(matchJSONKey, matchJSONVal, outputJSONVal)

                res.set_result(ResultState.fail)

                res.set_failmsg(failmsg)

                return res



        else:

            failmsg = 'Original output and matchJSON value are not the same type: output: {} != matchJSON: {}'

            failmsg = failmsg.format(type(outputJSON).__name__, type(matchJSON).__name__)

            res.set_result(ResultState.fail)

            res.set_failmsg(failmsg)

            return rest



        if type(outputJSONVal) == dict and (type(outputJSONVal[matchJSONKey]) == dict or

                type(outputJSONVal[matchJSONKey]) == list):

            if len(matchJSONVal) > 0:

                res = find_in_json(res, outputJSONVal[matchJSONKey], matchJSONVal, matchJSONKey)

            # handling corner case where matchJSONVal == [] or matchJSONVal == {}

            else:

                res = find_in_json_other(res, outputJSONVal, matchJSONVal, matchJSONKey)

        else:

            res = find_in_json(res, outputJSONVal, matchJSONVal, matchJSONKey)

    return res



def find_in_json_other(res, outputJSONVal, matchJSONVal, matchJSONKey=None):

    if matchJSONKey in outputJSONVal:

        if matchJSONVal != outputJSONVal[matchJSONKey]:

            failmsg = 'Value doesn\'t match: {}: {} != {}\nMatching against output: {}'

            failmsg = failmsg.format(matchJSONKey, matchJSONVal, outputJSONVal[matchJSONKey], outputJSONVal)

            res.set_result(ResultState.fail)

            res.set_failmsg(failmsg)

            return res



    return res



def run_one_test(pm, args, index, tidx):

    global NAMES

    result = True
@@ -292,16 +396,22 @@ def run_one_test(pm, args, index, tidx): 
    else:

        if args.verbose > 0:

            print('-----> verify stage')

        match_pattern = re.compile(

            str(tidx["matchPattern"]), re.DOTALL | re.MULTILINE)

        (p, procout) = exec_cmd(args, pm, 'verify', tidx["verifyCmd"])

        if procout:

            if 'matchJSON' in tidx:

                verify_by_json(procout, res, tidx, args, pm)

            elif 'matchPattern' in tidx:

                match_pattern = re.compile(

                    str(tidx["matchPattern"]), re.DOTALL | re.MULTILINE)

                match_index = re.findall(match_pattern, procout)

                if len(match_index) != int(tidx["matchCount"]):

                    res.set_result(ResultState.fail)

                    res.set_failmsg('Could not match regex pattern. Verify command output:\n{}'.format(procout))

                else:

                    res.set_result(ResultState.success)

            else:

                res.set_result(ResultState.fail)

                res.set_failmsg('Must specify a match option: matchJSON or matchPattern\n{}'.format(procout))

        elif int(tidx["matchCount"]) != 0:

            res.set_result(ResultState.fail)

            res.set_failmsg('No output generated by verify command.')
@@ -365,6 +475,7 @@ def test_runner(pm, args, filtered_tests): 
            res.set_result(ResultState.skip)

            res.set_errormsg(errmsg)

            tsr.add_resultdata(res)

            index += 1

            continue

        try:

            badtest = tidx  # in case it goes bad