diff --git a/examples/autonomy_engine_adapter/planning_configuration.json b/examples/autonomy_engine_adapter/planning_configuration.json
index f5a82819acb029aee08d71ae63366d41b3beb950..476e56e7cb0b4c174691c13355c8de44413b283a 100644
--- a/examples/autonomy_engine_adapter/planning_configuration.json
+++ b/examples/autonomy_engine_adapter/planning_configuration.json
@@ -9,7 +9,7 @@
"encoded": false
},
"payload":{
- "message_type": "platform_status",
+ "message_type": "planning_configuration",
"planning_config_ID": 3,
"exclusion_zones": [
{
diff --git a/examples/autonomy_engine_adapter/platform_status.json b/examples/autonomy_engine_adapter/platform_status.json
index 59a24f1f48a47b3a3aeb1b37d05212c7e7cde0a7..7c1c5c2daaec8fe8ee09ae4001bfb55fd76007c9 100644
--- a/examples/autonomy_engine_adapter/platform_status.json
+++ b/examples/autonomy_engine_adapter/platform_status.json
@@ -15,7 +15,6 @@
"status_source": "onboard_platform",
"autonomy_engine_plan_ID": 1,
"battery_remaining_capacity": 80.2,
- "active": true,
"platform_state": "ABORT",
"mission_plan_ID": 1,
"mission_track_ID": 4,
@@ -30,7 +29,6 @@
"usbl_fix_seconds_ago": 0,
"range_to_go": 124.3,
"sensor_config": {
- "sensor_ID": 22,
"serial": "sidescan-2x",
"sensor_on": true,
"additional_data": {
diff --git a/examples/ecosub_adapter/platform_status.json b/examples/ecosub_adapter/platform_status.json
index 59a24f1f48a47b3a3aeb1b37d05212c7e7cde0a7..84bc4160ec3dd4f2a16dc81e9f7dc2b8b6af4822 100644
--- a/examples/ecosub_adapter/platform_status.json
+++ b/examples/ecosub_adapter/platform_status.json
@@ -15,7 +15,6 @@
"status_source": "onboard_platform",
"autonomy_engine_plan_ID": 1,
"battery_remaining_capacity": 80.2,
- "active": true,
"platform_state": "ABORT",
"mission_plan_ID": 1,
"mission_track_ID": 4,
@@ -30,8 +29,7 @@
"usbl_fix_seconds_ago": 0,
"range_to_go": 124.3,
"sensor_config": {
- "sensor_ID": 22,
- "serial": "sidescan-2x",
+ "sensor_serial": "sidescan-2x",
"sensor_on": true,
"additional_data": {
"whiskers_on": true
diff --git a/examples/gui_adapter/planning_configuration.json b/examples/gui_adapter/planning_configuration.json
index 106bac90c64cbf065605acb4dbd2d168b18c7406..e5d0f128c70075e615c4af964ff1408ccd9a9b72 100644
--- a/examples/gui_adapter/planning_configuration.json
+++ b/examples/gui_adapter/planning_configuration.json
@@ -9,7 +9,7 @@
"encoded": false
},
"payload":{
- "message_type": "platform_status",
+ "message_type": "planning_configuration",
"planning_config_ID": 3,
"exclusion_zones": [
{
diff --git a/formats/platform_status.py b/formats/platform_status.py
index 3593a5239e132c5ccec208cddfaf29bf1a6fb27a..88a6759b765963f73dcf30ebe8b3afc4305811c4 100644
--- a/formats/platform_status.py
+++ b/formats/platform_status.py
@@ -18,6 +18,7 @@ sensor_schema = {
"example": True,
},
"additional_data": {
+ "type": "object", # TODO: Check 000
"description": "Any addition fields/data to be added here",
"example": {"payload": [1.2, 434]},
},
diff --git a/tests/test_schemas.py b/tests/test_schemas.py
index c893ac1e8cd1113544ca709c922a646f38179815..830525c214a62e80a7861e1a6f872e0b5992313f 100644
--- a/tests/test_schemas.py
+++ b/tests/test_schemas.py
@@ -1,33 +1,21 @@
from openapi_spec_validator import openapi_v30_spec_validator
from openapi_spec_validator.readers import read_from_filename
from openapi_schema_validator import validate
-from jsonschema import FormatChecker
-from tests.fixtures.schemas import (
- acknowledgement_schema,
- message_header,
- observation_schema,
- planning_configuration_schema,
- platform_status_schema,
- mission_plan_schema,
-)
-
+from jsonschema.validators import RefResolver
import unittest
import json
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"
MOCK_DATA_DIR = "examples/"
+
class SchemaError(Exception):
"""
+ Test config specs of swagger.json for projects
"""
+
def __init__(self, exception_type, message):
self.type = exception_type
self.message = message
@@ -42,6 +30,7 @@ class TestSpecs(unittest.TestCase):
schema, spec_url = read_from_filename("tests/fixtures/swagger.json")
self.assertIsNone(openapi_v30_spec_validator.validate(schema))
+
class TestAllMessageExamples(unittest.TestCase):
def test_schema_specs(self):
"""
@@ -51,253 +40,37 @@ class TestAllMessageExamples(unittest.TestCase):
schema = json.load(schema_ref)
schema_ref.close()
-
partner_dir_list = os.listdir(MOCK_DATA_DIR)
-
for partner in partner_dir_list:
- message_list = os.listdir(os.path.join(MOCK_DATA_DIR, partner))
- for message_type in message_list:
- print("Testing message %s by %s now..." % (message_type, partner))
-
- f = open(os.path.join(MOCK_DATA_DIR, partner, message_type))
- mock_message = json.load(f)
- ref_resolver = RefResolver.from_schema(schema)
-
+ if not partner.endswith(".py"):
+ print("TEST: Partner is %s" % partner.upper())
+ message_list = os.listdir(os.path.join(MOCK_DATA_DIR, partner))
+ for message_type in message_list:
+ if message_type.endswith(".json"):
+ print("Testing %s now..." % message_type)
+
+ 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()
-
-
-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__":