From 029cbdcf5d183b344fe572b21157293cef1c62e0 Mon Sep 17 00:00:00 2001 From: Anton Kiselev Date: Tue, 7 Mar 2017 01:35:28 +0700 Subject: scapy_server: fix template tests Signed-off-by: Anton Kiselev --- .../stl/services/scapy_server/scapy_service.py | 53 +++++++++------------- .../services/scapy_server/unit_tests/basetest.py | 8 ++-- .../scapy_server/unit_tests/test_scapy_service.py | 52 ++++++++++----------- 3 files changed, 50 insertions(+), 63 deletions(-) diff --git a/scripts/automation/trex_control_plane/stl/services/scapy_server/scapy_service.py b/scripts/automation/trex_control_plane/stl/services/scapy_server/scapy_service.py index 7b19896b..98eeb749 100755 --- a/scripts/automation/trex_control_plane/stl/services/scapy_server/scapy_service.py +++ b/scripts/automation/trex_control_plane/stl/services/scapy_server/scapy_service.py @@ -981,41 +981,30 @@ class Scapy_service(Scapy_service_api): def _get_templates(self): - # Make sure you understand the three return values of os.walk: - # - # for root, subdirs, files in os.walk(rootdir): - # has the following meaning: - # - # root: Current path which is "walked through" - # subdirs: Files in root of type directory - # files: Files in root (not in subdirs) of type other than directory - # And please use os.path.join instead of concatenating with a slash! - # Your problem is filePath = rootdir + '/' + file - you must concatenate the currently "walked" folder instead of the topmost folder. - # So that must be filePath = os.path.join(root, file). BTW "file" is a builtin, so you don't normally use it as variable name. - templates = [] for root, subdirs, files in os.walk("templates"): for file in files: - if file.endswith('.trp'): - try: - f = os.path.join(root, file) - o = open(f) - c = json.loads(o.read()) - o.close() - id = f.replace("templates" + os.path.sep, "", 1) - id = id.split(os.path.sep) - id[-1] = id[-1].replace(".trp", "", 1) - id = "/".join(id) - t = { - "id": id, - "meta": { - "name": c["metadata"]["caption"], - "description": "" - } - } - templates.append(t) - except: - pass + if not file.endswith('.trp'): + next + try: + f = os.path.join(root, file) + c = None + with open(f, 'r') as templatefile: + c = json.loads(templatefile.read()) + id = f.replace("templates" + os.path.sep, "", 1) + id = id.split(os.path.sep) + id[-1] = id[-1].replace(".trp", "", 1) + id = "/".join(id) + t = { + "id": id, + "meta": { + "name": c["metadata"]["caption"], + "description": "" + } + } + templates.append(t) + except: + pass return templates def _get_template(self,template): diff --git a/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/basetest.py b/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/basetest.py index 9836c794..0e95fa32 100644 --- a/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/basetest.py +++ b/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/basetest.py @@ -92,8 +92,8 @@ def adapt_json_protocol_fields(protocols_array): def get_templates(): return pass_result(service.get_templates(v_handler)) - - -def get_template(t): - return pass_result(service.get_template(v_handler, t)) +def get_template_by_id(templateId): + params = {"id": templateId} + template_b64 = service.get_template(v_handler, params) + return pass_result(base64.b64decode(template_b64)) diff --git a/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/test_scapy_service.py b/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/test_scapy_service.py index 1ece5d1e..7d70931b 100644 --- a/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/test_scapy_service.py +++ b/scripts/automation/trex_control_plane/stl/services/scapy_server/unit_tests/test_scapy_service.py @@ -124,6 +124,9 @@ def test_get_definitions_all(): for instruction in fe_instructions: print(instruction['help']) assert("help" in instruction) + assert(len(defs['feInstructionParameters']) > 0) + assert(len(defs['feParameters']) > 0) + assert(len(defs['feTemplates']) > 0) def test_get_definitions_ether(): res = get_definitions(["Ether"]) @@ -291,41 +294,36 @@ def test_generate_vm_instructions(): assert(ttl_instruction['max_value'] == 64) -def test_get_templates(): - tt = get_templates() - assert(tt[0]['id']) - assert(tt[7]["meta"]['name']) - try: - assert(tt[9]['id']) - except: - pass +def test_list_templates_hierarchy(): + ids = [] + for template_info in get_templates(): + assert(template_info["meta"]["name"]) + assert("description" in template_info["meta"]) + ids.append(template_info['id']) + assert('IPv4/TCP' in ids) + assert('IPv4/UDP' in ids) + assert('TCP-SYN' in ids) + assert('ICMP echo request' in ids) +def test_get_template_root(): + obj = json.loads(get_template_by_id('TCP-SYN')) + assert(obj['packet'][0]['id'] == 'Ether') + assert(obj['packet'][1]['id'] == 'IP') + assert(obj['packet'][2]['id'] == 'TCP') -def test_get_template(): - tt = get_templates() - t = tt[0] - res = get_template(t) - res2 = base64.b64decode(res) - obj = json.loads(res2) +def test_get_template_IP_ICMP(): + obj = json.loads(get_template_by_id('IPv4/ICMP')) assert(obj['packet'][0]['id'] == 'Ether') assert(obj['packet'][1]['id'] == 'IP') assert(obj['packet'][2]['id'] == 'ICMP') - -def test_get_template2(): - tt = get_templates() - t = tt[7] - res = get_template(t) - res2 = base64.b64decode(res) - obj = json.loads(res2) +def test_get_template_IPv6_UDP(): + obj = json.loads(get_template_by_id('IPv6/UDP')) assert(obj['packet'][0]['id'] == 'Ether') assert(obj['packet'][1]['id'] == 'IPv6') assert(obj['packet'][2]['id'] == 'UDP') +def test_templates_no_relative_path(): + res = get_template_by_id("../templates/IPv6/UDP") + assert(res == "") -def test_get_template3(): - tt = get_templates() - t = tt[7] - t["id"] = "../../" + t["id"] - res = get_template(t) - assert(res == '') -- cgit 1.2.3-korg