Commit cbf2982e authored by Trishna Saeharaseelan's avatar Trishna Saeharaseelan
Browse files

fix: tests

...@@ -9,7 +9,7 @@ ...@@ -9,7 +9,7 @@
"encoded": false "encoded": false
}, },
"payload":{ "payload":{
"message_type": "platform_status", "message_type": "planning_configuration",
"planning_config_ID": 3, "planning_config_ID": 3,
"exclusion_zones": [ "exclusion_zones": [
{ {
......
...@@ -15,7 +15,6 @@ ...@@ -15,7 +15,6 @@
"status_source": "onboard_platform", "status_source": "onboard_platform",
"autonomy_engine_plan_ID": 1, "autonomy_engine_plan_ID": 1,
"battery_remaining_capacity": 80.2, "battery_remaining_capacity": 80.2,
"active": true,
"platform_state": "ABORT", "platform_state": "ABORT",
"mission_plan_ID": 1, "mission_plan_ID": 1,
"mission_track_ID": 4, "mission_track_ID": 4,
...@@ -30,7 +29,6 @@ ...@@ -30,7 +29,6 @@
"usbl_fix_seconds_ago": 0, "usbl_fix_seconds_ago": 0,
"range_to_go": 124.3, "range_to_go": 124.3,
"sensor_config": { "sensor_config": {
"sensor_ID": 22,
"serial": "sidescan-2x", "serial": "sidescan-2x",
"sensor_on": true, "sensor_on": true,
"additional_data": { "additional_data": {
......
...@@ -15,7 +15,6 @@ ...@@ -15,7 +15,6 @@
"status_source": "onboard_platform", "status_source": "onboard_platform",
"autonomy_engine_plan_ID": 1, "autonomy_engine_plan_ID": 1,
"battery_remaining_capacity": 80.2, "battery_remaining_capacity": 80.2,
"active": true,
"platform_state": "ABORT", "platform_state": "ABORT",
"mission_plan_ID": 1, "mission_plan_ID": 1,
"mission_track_ID": 4, "mission_track_ID": 4,
...@@ -30,8 +29,7 @@ ...@@ -30,8 +29,7 @@
"usbl_fix_seconds_ago": 0, "usbl_fix_seconds_ago": 0,
"range_to_go": 124.3, "range_to_go": 124.3,
"sensor_config": { "sensor_config": {
"sensor_ID": 22, "sensor_serial": "sidescan-2x",
"serial": "sidescan-2x",
"sensor_on": true, "sensor_on": true,
"additional_data": { "additional_data": {
"whiskers_on": true "whiskers_on": true
......
...@@ -9,7 +9,7 @@ ...@@ -9,7 +9,7 @@
"encoded": false "encoded": false
}, },
"payload":{ "payload":{
"message_type": "platform_status", "message_type": "planning_configuration",
"planning_config_ID": 3, "planning_config_ID": 3,
"exclusion_zones": [ "exclusion_zones": [
{ {
......
...@@ -18,6 +18,7 @@ sensor_schema = { ...@@ -18,6 +18,7 @@ sensor_schema = {
"example": True, "example": True,
}, },
"additional_data": { "additional_data": {
"type": "object", # TODO: Check 000
"description": "Any addition fields/data to be added here", "description": "Any addition fields/data to be added here",
"example": {"payload": [1.2, 434]}, "example": {"payload": [1.2, 434]},
}, },
......
from openapi_spec_validator import openapi_v30_spec_validator from openapi_spec_validator import openapi_v30_spec_validator
from openapi_spec_validator.readers import read_from_filename from openapi_spec_validator.readers import read_from_filename
from openapi_schema_validator import validate from openapi_schema_validator import validate
from jsonschema import FormatChecker from jsonschema.validators import RefResolver
from tests.fixtures.schemas import (
acknowledgement_schema,
message_header,
observation_schema,
planning_configuration_schema,
platform_status_schema,
mission_plan_schema,
)
import unittest import unittest
import json import json
import os import os
from jsonschema.validators import RefResolver
from openapi_schema_validator import validate
# class ValidateResponse:
# def __init__(self, valid, error):
# self.valid = valid
# self.error = error
SCHEMA_DIR = "projects/soar/swagger.json" SCHEMA_DIR = "projects/soar/swagger.json"
MOCK_DATA_DIR = "examples/" MOCK_DATA_DIR = "examples/"
class SchemaError(Exception): class SchemaError(Exception):
""" """
Test config specs of swagger.json for projects
""" """
def __init__(self, exception_type, message): def __init__(self, exception_type, message):
self.type = exception_type self.type = exception_type
self.message = message self.message = message
...@@ -42,6 +30,7 @@ class TestSpecs(unittest.TestCase): ...@@ -42,6 +30,7 @@ class TestSpecs(unittest.TestCase):
schema, spec_url = read_from_filename("tests/fixtures/swagger.json") schema, spec_url = read_from_filename("tests/fixtures/swagger.json")
self.assertIsNone(openapi_v30_spec_validator.validate(schema)) self.assertIsNone(openapi_v30_spec_validator.validate(schema))
class TestAllMessageExamples(unittest.TestCase): class TestAllMessageExamples(unittest.TestCase):
def test_schema_specs(self): def test_schema_specs(self):
""" """
...@@ -51,253 +40,37 @@ class TestAllMessageExamples(unittest.TestCase): ...@@ -51,253 +40,37 @@ class TestAllMessageExamples(unittest.TestCase):
schema = json.load(schema_ref) schema = json.load(schema_ref)
schema_ref.close() schema_ref.close()
partner_dir_list = os.listdir(MOCK_DATA_DIR) partner_dir_list = os.listdir(MOCK_DATA_DIR)
for partner in partner_dir_list: for partner in partner_dir_list:
message_list = os.listdir(os.path.join(MOCK_DATA_DIR, partner)) if not partner.endswith(".py"):
for message_type in message_list: print("TEST: Partner is %s" % partner.upper())
print("Testing message %s by %s now..." % (message_type, partner)) message_list = os.listdir(os.path.join(MOCK_DATA_DIR, partner))
for message_type in message_list:
f = open(os.path.join(MOCK_DATA_DIR, partner, message_type)) if message_type.endswith(".json"):
mock_message = json.load(f) print("Testing %s now..." % message_type)
ref_resolver = RefResolver.from_schema(schema)
f = open(
os.path.join(
MOCK_DATA_DIR,
partner,
message_type,
)
)
mock_message = json.load(f)
ref_resolver = RefResolver.from_schema(schema)
self.assertIsNone(
validate(
mock_message,
schema["components"]["schemas"]["MESSAGE"],
resolver=ref_resolver,
)
)
f.close()
print("Done.")
self.assertIsNone(
validate(
mock_message,
schema["components"]["schemas"]["MESSAGE"],
resolver=ref_resolver,
)
)
f.close()
schema_ref.close() schema_ref.close()
class TestSchema(unittest.TestCase):
def test_sample_hydrosurv_messages(self):
"""
Test schema - Hydrosurv Messages
"""
message_types = ["acknowledgement", "mission_plan", "platform_status"]
for item in message_types:
f = open("examples/hydrosurv_adapter/%s.json" % item)
mock_data = json.load(f)
header_data = mock_data["header"]
payload_data = mock_data["payload"]
self.assertIsNone(
validate(
header_data,
message_header,
format_checker=FormatChecker(),
)
)
if item == "acknowledgement":
self.assertIsNone(
validate(
payload_data,
acknowledgement_schema,
format_checker=FormatChecker(),
)
)
elif item == "mission_plan":
self.assertIsNone(
validate(
payload_data,
mission_plan_schema,
format_checker=FormatChecker(),
)
)
elif item == "platform_status":
self.assertIsNone(
validate(
payload_data,
platform_status_schema,
format_checker=FormatChecker(),
)
)
f.close()
def test_sample_gui_messages(self):
"""
Test schema - GUI Messages
"""
message_types = ["planning_configuration"]
for item in message_types:
f = open("examples/gui_adapter/%s.json" % item)
mock_data = json.load(f)
header_data = mock_data["header"]
payload_data = mock_data["payload"]
self.assertIsNone(
validate(
header_data,
message_header,
format_checker=FormatChecker(),
)
)
if item == "planning_configuration":
self.assertIsNone(
validate(
payload_data,
planning_configuration_schema,
format_checker=FormatChecker(),
)
)
f.close()
def test_sample_ecosub_messages(self):
"""
Test schema - Ecosub Messages
"""
message_types = [
"observation",
"mission_plan",
"platform_status",
"platform_status-from_usbl_example",
]
for item in message_types:
f = open("examples/ecosub_adapter/%s.json" % item)
mock_data = json.load(f)
header_data = mock_data["header"]
payload_data = mock_data["payload"]
self.assertIsNone(
validate(
header_data,
message_header,
format_checker=FormatChecker(),
)
)
if item == "platform_status-from_usbl_example.json":
self.assertIsNone(
validate(
payload_data,
platform_status_schema,
format_checker=FormatChecker(),
)
)
elif item == "mission_plan":
self.assertIsNone(
validate(
payload_data,
mission_plan_schema,
format_checker=FormatChecker(),
)
)
elif item == "platform_status":
self.assertIsNone(
validate(
payload_data,
platform_status_schema,
format_checker=FormatChecker(),
)
)
elif item == "observation":
self.assertIsNone(
validate(
payload_data,
observation_schema,
format_checker=FormatChecker(),
)
)
f.close()
def test_sample_autonomy_engine_messages(self):
"""
Test schema - Autonomy Engine Messages
"""
message_types = [
"planning_configuration",
"mission_plan_AH1",
"mission_plan_ECOSUB",
"mission_plan_HYDROSURV",
"platform_status",
"platform_status-from_usbl_example",
"acknowledgement",
]
for item in message_types:
f = open("examples/autonomy_engine_adapter/%s.json" % item)
mock_data = json.load(f)
header_data = mock_data["header"]
payload_data = mock_data["payload"]
self.assertIsNone(
validate(
header_data,
message_header,
format_checker=FormatChecker(),
)
)
if item == "platform_status-from_usbl_example.json":
self.assertIsNone(
validate(
payload_data,
platform_status_schema,
format_checker=FormatChecker(),
)
)
elif item == "mission_plan_AH1":
self.assertIsNone(
validate(
payload_data,
mission_plan_schema,
format_checker=FormatChecker(),
)
)
elif item == "mission_plan_ECOSUB":
self.assertIsNone(
validate(
payload_data,
mission_plan_schema,
format_checker=FormatChecker(),
)
)
elif item == "mission_plan_HYDROSURV":
self.assertIsNone(
validate(
payload_data,
mission_plan_schema,
format_checker=FormatChecker(),
)
)
elif item == "platform_status":
self.assertIsNone(
validate(
payload_data,
platform_status_schema,
format_checker=FormatChecker(),
)
)
elif item == "observation":
self.assertIsNone(
validate(
payload_data,
observation_schema,
format_checker=FormatChecker(),
)
)
elif item == "planning_configuration":
self.assertIsNone(
validate(
payload_data,
planning_configuration_schema,
format_checker=FormatChecker(),
)
)
elif item == "acknowledgement":
self.assertIsNone(
validate(
payload_data,
acknowledgement_schema,
format_checker=FormatChecker(),
)
)
f.close()
if __name__ == "__main__": if __name__ == "__main__":
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment