CDD API and feature flag support

tests/default_value.py

@@ -24,6 +24,11 @@
 
 IMPLEMENTED_API_VERSION = 0
 
+TEST_CDC1 = [
+    { "endpoint": 1,       "payload": bytes.fromhex("AABBCCDDEEFF") },
+    { "endpoint": 0x20000, "payload": bytes.fromhex("334455") },
+]
+
 # Todo add more fields in config
 NODE_CONFIG_1 = dict([("sink_id", SINK_ID), ("node_address", 123)])
 
@@ -55,5 +60,6 @@
      ("stored_type", SCRATCHPAD_TYPE),
      ("stored_status", SCRATCHPAD_STATUS),
      ("processed_scratchpad", SCRATCHPAD_INFO),
-     ("stored_scratchpad", SCRATCHPAD_INFO)
+     ("stored_scratchpad", SCRATCHPAD_INFO),
+     ("configuration_data_content", TEST_CDC1)
     ])

tests/test_feature_flags.py

@@ -0,0 +1,20 @@
+# flake8: noqa
+
+import pytest
+import wirepas_mesh_messaging
+from wirepas_mesh_messaging.proto import GatewayFeature as GatewayFeature_pb
+
+@pytest.mark.parametrize("protobuf_feature_flag", GatewayFeature_pb.items())
+def test_decoding_errors(protobuf_feature_flag):
+    """
+    Make sure all the feature flags in the internal
+    (wirepas_mesh_messaging.OptionalGatewayFeature) enum are aligned with the
+    protobuf side if the protobuf definition is updated in this library.
+    """
+
+    name, value = protobuf_feature_flag
+    try:
+        wirepas_mesh_messaging.GatewayFeature(value)
+    except ValueError as e:
+        raise ValueError(f"{name} is not defined in GatewayFeature") from e
+

tests/test_get_config_data_item.py

@@ -0,0 +1,122 @@
+# flake8: noqa
+
+import pytest
+import base64
+import wirepas_mesh_messaging
+from wirepas_mesh_messaging.proto import GenericMessage
+from google.protobuf.json_format import ParseDict
+from default_value import *
+
+
+def test_parse_request_message():
+    TEST_REQ_ID = 200
+    TEST_ENDPOINT = 3000
+
+    message_description = {
+        "wirepas": {
+            "get_configuration_data_item_req": {
+                "header": {"req_id": TEST_REQ_ID},
+                "endpoint": TEST_ENDPOINT
+            }
+        }
+    }
+
+    message = ParseDict(message_description, GenericMessage())
+    assert message.IsInitialized()
+
+    request = wirepas_mesh_messaging.GetConfigurationDataItemRequest.from_payload(
+        message.SerializeToString()
+    )
+
+    assert TEST_REQ_ID == request.req_id
+    assert TEST_ENDPOINT == request.cdc_endpoint
+
+def test_encode_request_message():
+    TEST_ENDPOINT = 2000
+
+    request = wirepas_mesh_messaging.GetConfigurationDataItemRequest(SINK_ID, TEST_ENDPOINT)
+
+    parsed_request = wirepas_mesh_messaging.GetConfigurationDataItemRequest.from_payload(request.payload)
+
+    for k, v in request.__dict__.items():
+        assert v == parsed_request.__dict__[k]
+
+def test_parse_response_message():
+    TEST_ENDPOINT = 700
+    TEST_PAYLOAD = bytes.fromhex("BBAABBAADDAADDAA")
+
+    message_description = {
+        "wirepas": {
+            "get_configuration_data_item_resp": {
+                "header": {
+                    "req_id": REQUEST_ID,
+                    "gw_id": GATEWAY_ID,
+                    "sink_id": SINK_ID_2,
+                    "res": 0,
+                    "time_ms_epoch": RX_TIME_MS_EPOCH
+                },
+                "configuration_data_item": {
+                    "endpoint": TEST_ENDPOINT,
+                    # bytes objects are base64 encoded in protobuf JSON format
+                    "payload": base64.b64encode(TEST_PAYLOAD)
+                }
+            }
+        }
+    }
+
+    message = ParseDict(message_description, GenericMessage())
+    assert message.IsInitialized()
+
+    response = wirepas_mesh_messaging.GetConfigurationDataItemResponse.from_payload(
+        message.SerializeToString()
+    )
+
+    assert REQUEST_ID == response.req_id
+    assert GATEWAY_ID == response.gw_id
+    assert SINK_ID_2 == response.sink_id
+    assert RES_OK == response.res
+    assert RX_TIME_MS_EPOCH == response.time_ms_epoch
+    assert TEST_ENDPOINT == response.cdc_endpoint
+    assert TEST_PAYLOAD == response.cdc_payload
+
+def test_encode_response_message():
+    TEST_ENDPOINT = 9000
+    TEST_PAYLOAD = bytes.fromhex("FEFE1234")
+
+    request = wirepas_mesh_messaging.GetConfigurationDataItemResponse(
+        REQUEST_ID, GATEWAY_ID, RES_OK, SINK_ID, TEST_ENDPOINT, TEST_PAYLOAD
+    )
+
+    parsed_request = wirepas_mesh_messaging.GetConfigurationDataItemResponse.from_payload(request.payload)
+
+    for k, v in request.__dict__.items():
+        assert v == parsed_request.__dict__[k]
+
+def test_encode_response_with_no_item():
+    response = wirepas_mesh_messaging.GetConfigurationDataItemResponse(
+        REQUEST_ID, GATEWAY_ID, RES_KO, SINK_ID
+    )
+    parsed_response = wirepas_mesh_messaging.GetConfigurationDataItemResponse.from_payload(response.payload)
+
+    assert None == parsed_response.cdc_endpoint
+    assert None == parsed_response.cdc_payload
+
+def test_encode_response_with_empty_payload():
+    TEST_ENDPOINT = 543
+    TEST_PAYLOAD = bytes()
+    response = wirepas_mesh_messaging.GetConfigurationDataItemResponse(
+        REQUEST_ID, GATEWAY_ID, RES_KO, SINK_ID, TEST_ENDPOINT, TEST_PAYLOAD
+    )
+    parsed_response = wirepas_mesh_messaging.GetConfigurationDataItemResponse.from_payload(response.payload)
+
+    assert TEST_ENDPOINT == parsed_response.cdc_endpoint
+    assert TEST_PAYLOAD == parsed_response.cdc_payload
+
+def test_encoding_response_with_endpoint_but_no_payload_should_fail():
+    response = wirepas_mesh_messaging.GetConfigurationDataItemResponse(
+        REQUEST_ID, GATEWAY_ID, RES_KO, SINK_ID, 1000
+    )
+
+    with pytest.raises(Exception):
+        response.payload
+

tests/test_get_configs.py

@@ -26,5 +26,7 @@ def test_generate_parse_response():
         request.payload
     )
 
+    assert DUMMY_CONFIGS == request2.configs
     for k, v in request.__dict__.items():
         assert v == request2.__dict__[k]
+

tests/test_get_gw_info.py

@@ -26,12 +26,33 @@ def test_generate_parse_response():
         "Version x.y",
         max_scratchpad_size=512,
         implemented_api_version=IMPLEMENTED_API_VERSION,
+        gateway_features=[
+            wirepas_mesh_messaging.GatewayFeature.GW_FEATURE_SCRATCHPAD_CHUNK_V1,
+            wirepas_mesh_messaging.GatewayFeature.GW_FEATURE_CONFIGURATION_DATA_V1
+        ]
     )
 
     request2 = wirepas_mesh_messaging.GetGatewayInfoResponse.from_payload(
         request.payload
     )
 
+    expected_members = [
+        "gw_id",
+        "sink_id",
+        "req_id",
+        "res",
+        "time_ms_epoch",
+        "current_time_s_epoch",
+        "gateway_model",
+        "gateway_version",
+        "implemented_api_version",
+        "max_scratchpad_size",
+        "gateway_features"
+    ]
+    for field in expected_members:
+        assert field in request.__dict__
+        assert field in request2.__dict__
+
     for k, v in request.__dict__.items():
         assert v == request2.__dict__[k]
 
@@ -52,5 +73,4 @@ def test_generate_parse_response_not_all_optional():
     )
 
     for k, v in request.__dict__.items():
-        print(k)
         assert v == request2.__dict__[k]

tests/test_set_config.py

@@ -28,3 +28,18 @@ def test_generate_parse_response():
 
     for k, v in request.__dict__.items():
         assert v == request2.__dict__[k]
+
+
+def test_generate_parse_response_with_readonly_fields():
+    request = wirepas_mesh_messaging.SetConfigResponse(
+        REQUEST_ID, GATEWAY_ID, RES_OK, SINK_ID_2, NODE_CONFIG_2
+    )
+
+    request2 = wirepas_mesh_messaging.SetConfigResponse.from_payload(
+        request.payload
+    )
+
+    assert NODE_CONFIG_2 == request2.config
+    for k, v in request.__dict__.items():
+        assert v == request2.__dict__[k]
+

tests/test_set_config_data_item.py

@@ -0,0 +1,98 @@
+# flake8: noqa
+
+import pytest
+import base64
+import wirepas_mesh_messaging
+from wirepas_mesh_messaging.proto import GenericMessage
+from google.protobuf.json_format import ParseDict
+from default_value import *
+
+
+def test_parse_request_message():
+    TEST_REQ_ID = 300
+    TEST_ENDPOINT = 500
+    TEST_PAYLOAD = bytes.fromhex("AABBCCDDEEFF")
+
+    message_description = {
+        "wirepas": {
+            "set_configuration_data_item_req": {
+                "header": {"req_id": TEST_REQ_ID},
+                "configuration_data_item": {
+                    "endpoint": TEST_ENDPOINT,
+                    # bytes objects are base64 encoded in protobuf JSON format
+                    "payload": base64.b64encode(TEST_PAYLOAD)
+                }
+            }
+        }
+    }
+
+    message = ParseDict(message_description, GenericMessage())
+    assert message.IsInitialized()
+
+    request = wirepas_mesh_messaging.SetConfigurationDataItemRequest.from_payload(
+        message.SerializeToString()
+    )
+
+    assert TEST_REQ_ID == request.req_id
+    assert TEST_ENDPOINT == request.cdc_endpoint
+    assert TEST_PAYLOAD == request.cdc_payload
+
+def test_encode_request_message():
+    TEST_ENDPOINT = 1000
+    TEST_PAYLOAD = bytes.fromhex("102030")
+
+    request = wirepas_mesh_messaging.SetConfigurationDataItemRequest(SINK_ID, TEST_ENDPOINT, TEST_PAYLOAD)
+
+    parsed_request = wirepas_mesh_messaging.SetConfigurationDataItemRequest.from_payload(request.payload)
+
+    for k, v in request.__dict__.items():
+        assert v == parsed_request.__dict__[k]
+
+def test_encode_request_with_empty_payload():
+    request = wirepas_mesh_messaging.SetConfigurationDataItemRequest(SINK_ID, 100, bytes())
+    parsed_request = wirepas_mesh_messaging.SetConfigurationDataItemRequest.from_payload(request.payload)
+
+    assert bytes() == parsed_request.cdc_payload
+
+def test_encoding_request_with_none_payload_should_fail():
+    request = wirepas_mesh_messaging.SetConfigurationDataItemRequest(SINK_ID, 100, None)
+
+    with pytest.raises(TypeError):
+        request.payload
+
+def test_parse_response_message():
+    message_description = {
+        "wirepas": {
+            "set_configuration_data_item_resp": {
+                "header": {
+                    "req_id": REQUEST_ID,
+                    "gw_id": GATEWAY_ID,
+                    "sink_id": SINK_ID_2,
+                    "res": 0,
+                    "time_ms_epoch": RX_TIME_MS_EPOCH
+                }
+            }
+        }
+    }
+
+    message = ParseDict(message_description, GenericMessage())
+    assert message.IsInitialized()
+
+    response = wirepas_mesh_messaging.SetConfigurationDataItemResponse.from_payload(
+        message.SerializeToString()
+    )
+
+    assert REQUEST_ID == response.req_id
+    assert GATEWAY_ID == response.gw_id
+    assert SINK_ID_2 == response.sink_id
+    assert RES_OK == response.res
+    assert RX_TIME_MS_EPOCH == response.time_ms_epoch
+
+def test_encode_response_message():
+    request = wirepas_mesh_messaging.SetConfigurationDataItemResponse(REQUEST_ID, GATEWAY_ID, RES_OK, SINK_ID)
+
+    parsed_request = wirepas_mesh_messaging.SetConfigurationDataItemResponse.from_payload(request.payload)
+
+    for k, v in request.__dict__.items():
+        assert v == parsed_request.__dict__[k]
+

tests/test_status.py

@@ -1,5 +1,6 @@
 # flake8: noqa
 
+import pytest
 import wirepas_mesh_messaging
 from default_value import *
 
@@ -18,6 +19,7 @@ def test_generate_parse_event_complete():
 
     status2 = wirepas_mesh_messaging.StatusEvent.from_payload(status.payload)
 
+    assert DUMMY_CONFIGS == status2.sink_configs
     for k, v in status.__dict__.items():
         assert v == status2.__dict__[k]
 
@@ -28,3 +30,22 @@ def test_generate_parse_event_with_max_size():
 
     for k, v in status.__dict__.items():
         assert v == status2.__dict__[k]
+
+def test_generate_parse_event_with_feature_flags():
+    gw_features = [
+        wirepas_mesh_messaging.GatewayFeature.GW_FEATURE_SCRATCHPAD_CHUNK_V1,
+        wirepas_mesh_messaging.GatewayFeature.GW_FEATURE_CONFIGURATION_DATA_V1
+    ]
+    status = wirepas_mesh_messaging.StatusEvent(GATEWAY_ID, GATEWAY_STATE, gateway_features=gw_features)
+
+    status_parsed = wirepas_mesh_messaging.StatusEvent.from_payload(status.payload)
+
+    for k, v in status.__dict__.items():
+        assert v == status_parsed.__dict__[k]
+
+def test_encoding_message_with_invalid_feature_flag():
+    status = wirepas_mesh_messaging.StatusEvent(GATEWAY_ID, GATEWAY_STATE, gateway_features=["INVALID_VALUE12345"])
+
+    with pytest.raises(Exception):
+        status.payload
+

wirepas_mesh_messaging/__init__.py

@@ -32,6 +32,9 @@
 from .gateway_result_code import GatewayResultCode
 from .otap_helper import ScratchpadStatus, ScratchpadType, ScratchpadAction, ProcessingDelay
 from .wirepas_exceptions import GatewayAPIParsingException
+from .set_config_data_item import SetConfigurationDataItemRequest, SetConfigurationDataItemResponse
+from .get_config_data_item import GetConfigurationDataItemRequest, GetConfigurationDataItemResponse
+from .gateway_feature import GatewayFeature
 
 from google.protobuf.internal import api_implementation