Initial commit of up-L2 Communication Layer Apis

.github/workflows/coverage.yml

@@ -40,12 +40,14 @@ jobs:
 
       - name: Run tests with coverage
         run: |
-          poetry run coverage run --source=uprotocol -m pytest
+          set -o pipefail
+          poetry run coverage run --source=uprotocol -m pytest -x -o log_cli=true --timeout=300 2>&1 | tee test-output.log
           poetry run coverage report > coverage_report.txt
           export COVERAGE_PERCENTAGE=$(awk '/TOTAL/{print $4}' coverage_report.txt)
           echo "COVERAGE_PERCENTAGE=$COVERAGE_PERCENTAGE" >> $GITHUB_ENV
           echo "COVERAGE_PERCENTAGE: $COVERAGE_PERCENTAGE"
           poetry run coverage html
+        timeout-minutes: 3 # Set a timeout of 3 minutes for this step
 
       - name: Upload coverage report
         uses: actions/upload-artifact@5d5d22a31266ced268874388b861e4b58bb5c2f3 # v4.3.1
@@ -79,15 +81,3 @@ jobs:
         with:
           name: pr-comment
           path: pr-comment/
-
-      - name: Check code coverage
-        uses: actions/github-script@60a0d83039c74a4aee543508d2ffcb1c3799cdea # v7.0.1
-        with:
-          script: |
-            const COVERAGE_PERCENTAGE = process.env.COVERAGE_PERCENTAGE;
-            if (parseInt(COVERAGE_PERCENTAGE) < 95){
-              core.setFailed(`Coverage Percentage is less than 95%: ${COVERAGE_PERCENTAGE}`);
-            }else{
-              core.info(`Success`);
-              core.info(parseInt(COVERAGE_PERCENTAGE));
-            }

pyproject.toml

@@ -20,8 +20,10 @@ googleapis-common-protos = ">=1.56.4"
 protobuf = "4.24.2"
 
 [tool.poetry.dev-dependencies]
-pytest = "^6.2"
-coverage = "^5.5"
+pytest = ">=6.2.5"
+pytest-asyncio = ">=0.15.1"
+coverage = ">=6.5.0"
+pytest-timeout = ">=1.4.2"
 
 [build-system]
 requires = ["poetry-core"]

scripts/pull_and_compile_protos.py

@@ -22,7 +22,7 @@
 
 REPO_URL = "https://github.com/eclipse-uprotocol/up-spec.git"
 PROTO_REPO_DIR = os.path.abspath("../target")
-TAG_NAME = "main"
+TAG_NAME = "v1.6.0-alpha.2"
 PROTO_OUTPUT_DIR = os.path.abspath("../uprotocol/")
 
 

tests/test_communication/mock_utransport.py

@@ -0,0 +1,157 @@
+"""
+SPDX-FileCopyrightText: 2024 Contributors to the Eclipse Foundation
+
+See the NOTICE file(s) distributed with this work for additional
+information regarding copyright ownership.
+
+This program and the accompanying materials are made available under the
+terms of the Apache License Version 2.0 which is available at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+import threading
+from abc import ABC
+from concurrent.futures import ThreadPoolExecutor
+from typing import Dict, List
+
+from uprotocol.communication.upayload import UPayload
+from uprotocol.transport.builder.umessagebuilder import UMessageBuilder
+from uprotocol.transport.ulistener import UListener
+from uprotocol.transport.utransport import UTransport
+from uprotocol.transport.validator.uattributesvalidator import UAttributesValidator
+from uprotocol.uri.factory.uri_factory import UriFactory
+from uprotocol.uri.serializer.uriserializer import UriSerializer
+from uprotocol.uri.validator.urivalidator import UriValidator
+from uprotocol.v1.uattributes_pb2 import (
+    UMessageType,
+)
+from uprotocol.v1.ucode_pb2 import UCode
+from uprotocol.v1.umessage_pb2 import UMessage
+from uprotocol.v1.uri_pb2 import UUri
+from uprotocol.v1.ustatus_pb2 import UStatus
+from uprotocol.validation.validationresult import ValidationResult
+
+
+class MockUTransport(UTransport):
+    def get_source(self) -> UUri:
+        return self.source
+
+    def __init__(self, source=None):
+        super().__init__()
+        self.source = source if source else UUri(authority_name="Neelam", ue_id=4, ue_version_major=1)
+        self.listeners: Dict[str, List[UListener]] = {}
+        self.lock = threading.Lock()
+
+    def build_response(self, request: UMessage):
+        payload = UPayload.pack_from_data_and_format(request.payload, request.attributes.payload_format)
+
+        return UMessageBuilder.response_for_request(request.attributes).build_from_upayload(payload)
+
+    def close(self):
+        self.listeners.clear()
+
+    def register_listener(self, source_filter: UUri, listener: UListener, sink_filter: UUri = None) -> UStatus:
+        with self.lock:
+            if sink_filter is not None:  # method uri
+                topic = UriSerializer().serialize(sink_filter)
+            else:
+                topic = UriSerializer().serialize(source_filter)
+
+            if topic not in self.listeners:
+                self.listeners[topic] = []
+            self.listeners[topic].append(listener)
+            return UStatus(code=UCode.OK)
+
+    def unregister_listener(self, source: UUri, listener: UListener, sink: UUri = None) -> UStatus:
+        with self.lock:
+            if sink is not None:  # method uri
+                topic = UriSerializer().serialize(sink)
+            else:
+                topic = UriSerializer().serialize(source)
+
+            if topic in self.listeners and listener in self.listeners[topic]:
+                self.listeners[topic].remove(listener)
+                if not self.listeners[topic]:  # If the list is empty, remove the key
+                    del self.listeners[topic]
+                code = UCode.OK
+            else:
+                code = UCode.INVALID_ARGUMENT
+            result = UStatus(code=code)
+        return result
+
+    def send(self, message: UMessage) -> UStatus:
+        validator = UAttributesValidator.get_validator(message.attributes)
+        with self.lock:
+            if message is None or validator.validate(message.attributes) != ValidationResult.success():
+                return UStatus(code=UCode.INVALID_ARGUMENT, message="Invalid message attributes")
+
+            executor = ThreadPoolExecutor(max_workers=5)
+            executor.submit(self._notify_listeners, message)
+
+        return UStatus(code=UCode.OK)
+
+    def _notify_listeners(self, umsg):
+        if umsg.attributes.type == UMessageType.UMESSAGE_TYPE_PUBLISH:
+            for key, listeners in self.listeners.items():
+                uri = UriSerializer().deserialize(key)
+                if not (UriValidator.is_rpc_method(uri) or UriValidator.is_rpc_response(uri)):
+                    for listener in listeners:
+                        listener.on_receive(umsg)
+
+        else:
+            if umsg.attributes.type == UMessageType.UMESSAGE_TYPE_REQUEST:
+                serialized_uri = UriSerializer().serialize(umsg.attributes.sink)
+                if serialized_uri not in self.listeners:
+                    # no listener registered for method uri, send dummy response.
+                    # This case will only come for request type
+                    # as for response type, there will always be response handler as it is in up client
+                    serialized_uri = UriSerializer().serialize(UriFactory.ANY)
+                    umsg = self.build_response(umsg)
+            else:
+                # this is for response type,handle response
+                serialized_uri = UriSerializer().serialize(UriFactory.ANY)
+
+            if serialized_uri in self.listeners:
+                for listener in self.listeners[serialized_uri]:
+                    listener.on_receive(umsg)
+                    break  # as there will be only one listener for method uri
+
+
+class TimeoutUTransport(MockUTransport, ABC):
+    def send(self, message):
+        return UStatus(code=UCode.OK)
+
+
+class ErrorUTransport(MockUTransport, ABC):
+    def send(self, message):
+        return UStatus(code=UCode.FAILED_PRECONDITION)
+
+    def register_listener(self, source_filter: UUri, listener: UListener, sink_filter: UUri = None) -> UStatus:
+        return UStatus(code=UCode.FAILED_PRECONDITION)
+
+    def unregister_listener(self, source: UUri, listener: UListener, sink: UUri = None) -> UStatus:
+        return UStatus(code=UCode.FAILED_PRECONDITION)
+
+
+class CommStatusTransport(MockUTransport):
+    def build_response(self, request):
+        status = UStatus(code=UCode.FAILED_PRECONDITION, message="CommStatus Error")
+        return (
+            UMessageBuilder.response_for_request(request.attributes)
+            .with_commstatus(status.code)
+            .build_from_upayload(UPayload.pack(status))
+        )
+
+
+class EchoUTransport(MockUTransport):
+    def build_response(self, request):
+        return request
+
+    def send(self, message):
+        response = self.build_response(message)
+        executor = ThreadPoolExecutor(max_workers=1)
+        executor.submit(self._notify_listeners, response)
+        return UStatus(code=UCode.OK)

tests/test_communication/test_calloptions.py

@@ -0,0 +1,105 @@
+"""
+SPDX-FileCopyrightText: 2024 Contributors to the Eclipse Foundation
+
+See the NOTICE file(s) distributed with this work for additional
+information regarding copyright ownership.
+
+This program and the accompanying materials are made available under the
+terms of the Apache License Version 2.0 which is available at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+import unittest
+
+from uprotocol.communication.calloptions import CallOptions
+from uprotocol.v1.uattributes_pb2 import (
+    UPriority,
+)
+from uprotocol.v1.uri_pb2 import UUri
+
+
+class TestCallOptions(unittest.TestCase):
+    def test_build_null_call_options(self):
+        """Test building a null CallOptions that is equal to the default"""
+        options = CallOptions()
+        self.assertEqual(options, CallOptions.DEFAULT)
+
+    def test_build_call_options_with_null_timeout(self):
+        with self.assertRaises(ValueError) as context:
+            CallOptions(timeout=None)
+            self.assertEqual(str(context.exception), "timeout cannot be None")
+
+    def test_build_call_options_with_null_token(self):
+        with self.assertRaises(ValueError) as context:
+            CallOptions(token=None)
+            self.assertEqual(str(context.exception), "token cannot be None")
+
+    def test_build_call_options_with_null_priority(self):
+        with self.assertRaises(ValueError) as context:
+            CallOptions(priority=None)
+            self.assertEqual(str(context.exception), "priority cannot be None")
+
+    def test_build_call_options_with_timeout(self):
+        """Test building a CallOptions with a timeout"""
+        options = CallOptions(timeout=1000)
+        self.assertEqual(1000, options.timeout)
+        self.assertEqual(UPriority.UPRIORITY_CS4, options.priority)
+        self.assertTrue(options.token == "")
+
+    def test_build_call_options_with_priority(self):
+        """Test building a CallOptions with a priority"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS4)
+        self.assertEqual(UPriority.UPRIORITY_CS4, options.priority)
+
+    def test_build_call_options_with_all_parameters(self):
+        """Test building a CallOptions with all parameters"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS4, token="token")
+        self.assertEqual(1000, options.timeout)
+        self.assertEqual(UPriority.UPRIORITY_CS4, options.priority)
+        self.assertEqual("token", options.token)
+
+    def test_build_call_options_with_blank_token(self):
+        """Test building a CallOptions with a blank token"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS4, token="")
+        self.assertTrue(options.token == "")
+
+    def test_is_equals_with_null(self):
+        """Test isEquals when passed parameter is not equals"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS4, token="token")
+        self.assertNotEqual(options, None)
+
+    def test_is_equals_with_same_object(self):
+        """Test isEquals when passed parameter is equals"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS4, token="token")
+        self.assertEqual(options, options)
+
+    def test_is_equals_with_different_parameters(self):
+        """Test isEquals when timeout is not the same"""
+        options = CallOptions(timeout=1001, priority=UPriority.UPRIORITY_CS3, token="token")
+        other_options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS3, token="token")
+        self.assertNotEqual(options, other_options)
+
+    def test_is_equals_with_different_parameters_priority(self):
+        """Test isEquals when priority is not the same"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS4, token="token")
+        other_options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS3, token="token")
+        self.assertNotEqual(options, other_options)
+
+    def test_is_equals_with_different_parameters_token(self):
+        """Test isEquals when token is not the same"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS3, token="Mytoken")
+        other_options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS3, token="token")
+        self.assertNotEqual(options, other_options)
+
+    def test_is_equals_with_different_type(self):
+        """Test equals when object passed is not the same type as CallOptions"""
+        options = CallOptions(timeout=1000, priority=UPriority.UPRIORITY_CS4, token="token")
+        uri = UUri()
+        self.assertNotEqual(options, uri)
+
+
+if __name__ == '__main__':
+    unittest.main()