diff --git a/.coveragerc b/.coveragerc index d55c0ad..53d94b1 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1,6 +1,6 @@ [run] -omit = uprotocol/core/*,uprotocol/v1/*, uprotocol/uoptions_pb2.py, uprotocol/cloudevent/*_pb2.py, tests/*, */__init__.py +omit = uprotocol/core/*,uprotocol/v1/*, uprotocol/uoptions_pb2.py, uprotocol/uoptions_pb2_grpc.py, uprotocol/cloudevent/*_pb2.py, tests/*, */__init__.py [report] exclude_lines = pragma: no cover - .*pass.* \ No newline at end of file + .*pass.* diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index 116b983..1873476 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -21,33 +21,42 @@ jobs: - name: Set up Python uses: actions/setup-python@82c7e631bb3cdc910f68e0081d67478d79c6982d # v5.1.0 with: - python-version: '3.x' + python-version: '3.8' - name: Install Poetry run: | python -m pip install --upgrade pip python -m pip install poetry - - name: Install dependencies + - name: Checkout submodules run: | - poetry install + git submodule update --init --recursive - - name: Run prebuild script + - name: Set up Python dependencies run: | - cd scripts - # Run the script within the Poetry virtual environment - poetry run python pull_and_compile_protos.py + python -m pip install --upgrade pip + python -m pip install .[dev] + + - name: Generate protobufs + run: | + python generate_proto.py + + - name: Verify generated protobuf files + run: | + ls -lhR uprotocol/v1 + - name: Run tests with coverage run: | 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 + coverage run --source=uprotocol -m pytest -x -o log_cli=true 2>&1 | tee test-output.log + 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 + coverage html + timeout-minutes: 3 + - name: Upload coverage report uses: actions/upload-artifact@5d5d22a31266ced268874388b861e4b58bb5c2f3 # v4.3.1 diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..4b9f47b --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "up-spec"] + path = up-spec + url = https://github.com/eclipse-uprotocol/up-spec.git diff --git a/README.adoc b/README.adoc index 45a492c..2beb28c 100644 --- a/README.adoc +++ b/README.adoc @@ -18,37 +18,35 @@ image:https://raw.githubusercontent.com/eclipse-uprotocol/up-spec/main/up_librar === Prerequisites Before proceeding with the setup of this project, ensure that the following prerequisites are met: -* Maven is installed and configured in your environment. You can verify this by running the following command in your terminal: -[,bash] ----- -mvn -version ----- -If Maven is properly installed, you should see information about the Maven version and configuration. + +* `git` is installed and configured in your environment. -NOTE: Ensure you are using Java 17 with your Maven installation before continuing with the next steps. Other versions of Java may not be supported. +* Python 3.8+ is installed and configured in your environment. === Importing the Library To set up SDK, follow the steps below: -. Clone the code from the GitHub repository: +. Clone the code from the GitHub repository , The repository contains the reference to up-spec main branch as the git submodule: + [source] ---- -git clone https://github.com/eclipse-uprotocol/up-python.git +git clone --recurse-submodules https://github.com/eclipse-uprotocol/up-python.git +cd up-python ---- - -. Execute the `pull_and_compile_protos.py` script using the following commands: +If you have already cloned without `--recurse-submodules`, you can initialize and update using: + [source] ---- -cd scripts -python pull_and_compile_protos.py +git submodule update --init --recursive ---- -This script automates the following tasks: -1. **Cloning and Compilation of Protos:** - Clones the `up-core-api` protos from the `up-spec` repository, compiles them, and generates Python protofiles in the protos folder. +. Generate Python proto bindings using the provided script: ++ +[source] +---- +python generate_proto.py +---- +This script will compile the `up-spec` protos locally and generate Python proto files under the appropriate package folders for local use in your environment. . Install up-python + diff --git a/clean_project.py b/clean_project.py index 0f0c265..5ca6244 100644 --- a/clean_project.py +++ b/clean_project.py @@ -1,5 +1,5 @@ """ -SPDX-FileCopyrightText: Copyright (c) 2023 Contributors to the +SPDX-FileCopyrightText: Copyright (c) 2025 Contributors to the Eclipse Foundation See the NOTICE file(s) distributed with this work for additional @@ -23,22 +23,53 @@ import os import shutil +from config import TRACK_FILE + def clean_project(): - # Remove build/ directory - if os.path.exists('build'): - shutil.rmtree('build') + # ----------------------------------- + # Remove known build and cache folders + # ----------------------------------- + directories_to_remove = ["build", "dist", "htmlcov", ".pytest_cache"] + directories_to_remove.extend([d for d in os.listdir() if d.endswith('.egg-info')]) + + for directory in directories_to_remove: + if os.path.exists(directory): + shutil.rmtree(directory) + print(f"Removed directory: {directory}/") + + # ----------------------------------- + # Remove generated proto files + # ----------------------------------- + if os.path.exists(TRACK_FILE): + with open(TRACK_FILE, "r") as f: + files = [line.strip() for line in f if line.strip()] + + for file in files: + # Normalize path to avoid './' issues + file_path = file.lstrip("./") + if os.path.exists(file_path): + os.remove(file_path) + print(f"Deleted file: {file_path}") + else: + print(f"File not found, skipping: {file_path}") - # Remove dist/ directory - if os.path.exists('dist'): - shutil.rmtree('dist') + # Remove the tracking file itself + os.remove(TRACK_FILE) + print(f"Removed tracking file: {TRACK_FILE}") + else: + print(f"No {TRACK_FILE} found, skipping proto cleanup.") - # Remove *.egg-info/ directories - egg_info_directories = [d for d in os.listdir() if d.endswith('.egg-info')] - for egg_info_directory in egg_info_directories: - shutil.rmtree(egg_info_directory) + # ----------------------------------- + # Remove all __pycache__ directories + # ----------------------------------- + for root, dirs, _ in os.walk("."): + for d in dirs: + if d == "__pycache__": + cache_path = os.path.join(root, d) + shutil.rmtree(cache_path) + print(f"Removed __pycache__: {cache_path}") if __name__ == "__main__": clean_project() - print("Cleanup complete.") diff --git a/config.py b/config.py new file mode 100644 index 0000000..a699e32 --- /dev/null +++ b/config.py @@ -0,0 +1,30 @@ +""" +SPDX-FileCopyrightText: Copyright (c) 2025 Contributors to the +Eclipse Foundation + +See the NOTICE file(s) distributed with this work for additional +information regarding copyright ownership. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +SPDX-FileType: SOURCE +SPDX-License-Identifier: Apache-2.0 +""" + +# Input location of protofile +PROTO_DIR = "up-spec/up-core-api" + +# Output location of the proto +OUTPUT_DIR = "." + +# The list of generated files +TRACK_FILE = "generated_proto_files.txt" diff --git a/generate_proto.py b/generate_proto.py new file mode 100644 index 0000000..288397f --- /dev/null +++ b/generate_proto.py @@ -0,0 +1,78 @@ +""" +SPDX-FileCopyrightText: 2025 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 os + +import pkg_resources +from grpc_tools import protoc + +from config import OUTPUT_DIR, PROTO_DIR, TRACK_FILE + +# Add google protobuf include path +PROTOBUF_INCLUDE = pkg_resources.resource_filename('grpc_tools', '_proto') + + +def generate_all_protos(): + if not os.path.exists(OUTPUT_DIR): + os.makedirs(OUTPUT_DIR) + + generated_files = [] + + for root, _, files in os.walk(PROTO_DIR): + for file in files: + if file.endswith(".proto"): + proto_file = os.path.join(root, file) + print(f"Compiling {proto_file}...") + + result = protoc.main( + [ + '', + f'-I{PROTO_DIR}', + f'-I{PROTOBUF_INCLUDE}', + f'--python_out={OUTPUT_DIR}', + f'--grpc_python_out={OUTPUT_DIR}', + proto_file, + ] + ) + + if result != 0: + print(f"Failed to compile {proto_file}") + else: + print(f"Compiled {proto_file} successfully.") + base_name = os.path.splitext(file)[0] + rel_dir = os.path.relpath(root, PROTO_DIR) + output_dir = os.path.join(OUTPUT_DIR, rel_dir) + + generated_files.append(os.path.join(output_dir, f"{base_name}_pb2.py")) + generated_files.append(os.path.join(output_dir, f"{base_name}_pb2_grpc.py")) + + # Ensure __init__.py in all generated proto folders + for root, dirs, _ in os.walk(OUTPUT_DIR): + if '.git' in root or os.path.commonpath([root, PROTO_DIR]) == PROTO_DIR: + continue + init_file = os.path.join(root, '__init__.py') + if not os.path.exists(init_file): + open(init_file, 'a').close() + print(f"Created {init_file}") + generated_files.append(init_file) + + # Write generated files to track file + with open(TRACK_FILE, "w") as f: + for path in generated_files: + f.write(f"{path}\n") + print(f"Tracked {len(generated_files)} generated files in {TRACK_FILE}") + + +if __name__ == "__main__": + generate_all_protos() diff --git a/pyproject.toml b/pyproject.toml index 48ce3f8..3b669a8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,29 +1,33 @@ -[tool.poetry] +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] name = "up-python" version = "0.2.0-dev" description = "Language specific uProtocol library for building and using UUri, UUID, UAttributes, UTransport, and more." -authors = ["Neelam Kushwah "] -license = "The Apache License, Version 2.0" -readme = "README.adoc" -repository = "https://github.com/eclipse-uprotocol/up-python" -packages = [{ include = "uprotocol" }, - { include = "tests" }, - { include = "scripts" } +authors = [{ name = "Eclipse Foundation uProtocol Project", email = "uprotocol-dev@eclipse.org" }] +license = { file = "LICENSE" } +readme = { file = "README.adoc", content-type = "text/asciidoc" } +dependencies = [ + "cloudevents", + "googleapis-common-protos>=1.56.4", + "protobuf>=4.24.2", + "grpcio>=1.60.0", + "grpcio-tools>=1.60.0" ] -[tool.poetry.dependencies] -python = "^3.8" -cloudevents = "*" -gitpython = ">=3.1.41" -googleapis-common-protos = ">=1.56.4" -protobuf = "4.24.2" -pytest = ">=6.2.5" -pytest-asyncio = ">=0.15.1" -coverage = ">=6.5.0" -pytest-timeout = ">=1.4.2" +[project.urls] +Repository = "https://github.com/eclipse-uprotocol/up-python" +[project.optional-dependencies] +dev = [ + "pytest>=6.2.5", + "pytest-asyncio>=0.15.1", + "coverage>=6.5.0", + "pytest-timeout>=1.4.2" +] -[build-system] -requires = ["poetry-core"] -build-backend = "poetry.core.masonry.api" \ No newline at end of file +[tool.setuptools.packages.find] +include = ["uprotocol*", "tests*", "scripts*"] diff --git a/scripts/pull_and_compile_protos.py b/scripts/pull_and_compile_protos.py deleted file mode 100644 index 3a8fff4..0000000 --- a/scripts/pull_and_compile_protos.py +++ /dev/null @@ -1,105 +0,0 @@ -""" -SPDX-FileCopyrightText: 2023 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 os -import re -import shutil -import subprocess - -import git -from git import Repo - -REPO_URL = "https://github.com/eclipse-uprotocol/up-spec.git" -PROTO_REPO_DIR = os.path.abspath("../target") -TAG_NAME = "v1.6.0-alpha.2" -PROTO_OUTPUT_DIR = os.path.abspath("../uprotocol/") - - -def clone_or_pull(repo_url, proto_repo_dir): - try: - repo = Repo.clone_from(repo_url, proto_repo_dir) - print(f"Repository cloned successfully from {repo_url} to {proto_repo_dir}") - # Checkout the specific tag - repo.git.checkout(TAG_NAME) - except git.exc.GitCommandError: - try: - git_pull_command = ["git", "pull", "origin", TAG_NAME] - subprocess.run(git_pull_command, cwd=proto_repo_dir, check=True) - print("Git pull successful after clone failure.") - except subprocess.CalledProcessError as pull_error: - print(f"Error during Git pull: {pull_error}") - - -def execute_maven_command(project_dir, command): - try: - with subprocess.Popen( - command, - cwd=os.path.join(os.getcwd(), project_dir), - shell=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) as process: - stdout, stderr = process.communicate() - print(stdout) - - if process.returncode != 0: - print(f"Error: {stderr}") - else: - print("Maven command executed successfully.") - src_directory = os.path.join( - os.getcwd(), project_dir, "target", "generated-sources", "protobuf", "python", "uprotocol" - ) - - shutil.copytree(src_directory, PROTO_OUTPUT_DIR, dirs_exist_ok=True) - process_python_protofiles(PROTO_OUTPUT_DIR) - except Exception as e: - print(f"Error executing Maven command: {e}") - - -def replace_in_file(file_path, search_pattern, replace_pattern): - with open(file_path, 'r') as file: - file_content = file.read() - - updated_content = re.sub(search_pattern, replace_pattern, file_content) - - with open(file_path, 'w') as file: - file.write(updated_content) - - -def process_python_protofiles(directory): - for root, dirs, files in os.walk(directory): - create_init_py(root) - - -def create_init_py(directory): - init_file_path = os.path.join(directory, "__init__.py") - - # Check if the file already exists - if not os.path.exists(init_file_path): - # Create an empty __init__.py file - with open(init_file_path, "w"): - pass - - -def execute(): - clone_or_pull(REPO_URL, PROTO_REPO_DIR) - - # Execute mvn compile-python - maven_command = "mvn protobuf:compile-python" - execute_maven_command(PROTO_REPO_DIR, maven_command) - - -if __name__ == "__main__": - execute() diff --git a/tests/test_client/test_usubscription/test_v3/test_inmemoryusubcriptionclient.py b/tests/test_client/test_usubscription/test_v3/test_inmemoryusubcriptionclient.py index b0410f8..51842c5 100644 --- a/tests/test_client/test_usubscription/test_v3/test_inmemoryusubcriptionclient.py +++ b/tests/test_client/test_usubscription/test_v3/test_inmemoryusubcriptionclient.py @@ -17,8 +17,13 @@ from unittest.mock import AsyncMock, MagicMock from tests.test_communication.mock_utransport import MockUTransport -from uprotocol.client.usubscription.v3.inmemoryusubcriptionclient import InMemoryUSubscriptionClient -from uprotocol.client.usubscription.v3.subscriptionchangehandler import SubscriptionChangeHandler +from uprotocol.client.usubscription.v3.inmemoryusubcriptionclient import ( + InMemoryUSubscriptionClient, + MyNotificationListener, +) +from uprotocol.client.usubscription.v3.subscriptionchangehandler import ( + SubscriptionChangeHandler, +) from uprotocol.communication.calloptions import CallOptions from uprotocol.communication.inmemoryrpcclient import InMemoryRpcClient from uprotocol.communication.simplenotifier import SimpleNotifier @@ -36,6 +41,8 @@ from uprotocol.transport.builder.umessagebuilder import UMessageBuilder from uprotocol.transport.ulistener import UListener from uprotocol.transport.utransport import UTransport +from uprotocol.uri.serializer.uriserializer import UriSerializer +from uprotocol.v1 import uattributes_pb2 from uprotocol.v1.ucode_pb2 import UCode from uprotocol.v1.umessage_pb2 import UMessage from uprotocol.v1.uri_pb2 import UUri @@ -79,7 +86,6 @@ async def test_simple_mock_of_rpc_client_and_notifier(self): self.rpc_client.invoke_method.assert_called_once() self.notifier.register_notification_listener.assert_called_once() self.transport.register_listener.assert_called_once() - self.transport.get_source.assert_called_once() async def test_simple_mock_of_rpc_client_and_notifier_returned_subscribe_pending(self): response = SubscriptionResponse( @@ -105,7 +111,6 @@ async def test_simple_mock_of_rpc_client_and_notifier_returned_subscribe_pending self.rpc_client.invoke_method.assert_called_once() self.notifier.register_notification_listener.assert_called_once() self.transport.register_listener.assert_called_once() - self.transport.get_source.assert_called_once() async def test_simple_mock_of_rpc_client_and_notifier_returned_unsubscribed(self): response = SubscriptionResponse( @@ -131,7 +136,6 @@ async def test_simple_mock_of_rpc_client_and_notifier_returned_unsubscribed(self self.rpc_client.invoke_method.assert_called_once() self.notifier.register_notification_listener.assert_called_once() self.transport.register_listener.assert_not_called() - self.transport.get_source.assert_called_once() async def test_subscribe_using_mock_rpc_client_and_simplernotifier_when_invokemethod_return_an_exception(self): self.transport.get_source.return_value = self.source @@ -151,7 +155,6 @@ async def test_subscribe_using_mock_rpc_client_and_simplernotifier_when_invokeme self.rpc_client.invoke_method.assert_called_once() self.notifier.register_notification_listener.assert_called_once() self.transport.register_listener.assert_not_called() - self.transport.get_source.assert_called_once() async def test_subscribe_when_register_notification_listener_return_failed_status(self): self.transport.get_source.return_value = self.source @@ -185,7 +188,6 @@ async def test_subscribe_using_mock_rpc_client_and_simplernotifier_when_invokeme self.rpc_client.invoke_method.assert_called_once() self.notifier.register_notification_listener.assert_called_once() self.transport.register_listener.assert_not_called() - self.transport.get_source.assert_called_once() async def test_subscribe_when_we_pass_a_subscription_change_notification_handler(self): self.transport.get_source.return_value = self.source @@ -212,7 +214,6 @@ async def test_subscribe_when_we_pass_a_subscription_change_notification_handler self.rpc_client.invoke_method.assert_called_once() self.notifier.register_notification_listener.assert_called_once() self.transport.register_listener.assert_called_once() - self.transport.get_source.assert_called_once() async def test_subscribe_when_we_try_to_subscribe_to_the_same_topic_twice_with_same_notification_handler(self): self.transport.get_source.return_value = self.source @@ -240,7 +241,6 @@ async def test_subscribe_when_we_try_to_subscribe_to_the_same_topic_twice_with_s self.assertEqual(self.rpc_client.invoke_method.call_count, 2) self.notifier.register_notification_listener.assert_called_once() - self.assertEqual(self.transport.get_source.call_count, 2) async def test_subscribe_when_we_try_to_subscribe_to_the_same_topic_twice_with_different_notification_handler(self): self.transport.get_source.return_value = self.source @@ -269,7 +269,6 @@ async def test_subscribe_when_we_try_to_subscribe_to_the_same_topic_twice_with_d self.assertEqual(2, self.rpc_client.invoke_method.call_count) self.notifier.register_notification_listener.assert_called_once() - self.assertEqual(self.transport.get_source.call_count, 2) async def test_unsubscribe_using_mock_rpcclient_and_simplernotifier(self): self.transport.get_source.return_value = self.source @@ -284,7 +283,7 @@ async def test_unsubscribe_using_mock_rpcclient_and_simplernotifier(self): response = await subscriber.unsubscribe(self.topic, self.listener) self.assertEqual(response.message, "") self.assertEqual(response.code, UCode.OK) - subscriber.close() + await subscriber.close() self.rpc_client.invoke_method.assert_called_once() self.notifier.unregister_notification_listener.assert_called_once() self.transport.unregister_listener.assert_called_once() @@ -302,7 +301,7 @@ async def test_unsubscribe_when_invokemethod_return_an_exception(self): response = await subscriber.unsubscribe(self.topic, self.listener) self.assertEqual(response.message, "Operation cancelled") self.assertEqual(response.code, UCode.CANCELLED) - subscriber.close() + await subscriber.close() self.rpc_client.invoke_method.assert_called_once() self.notifier.unregister_notification_listener.assert_called_once() self.transport.unregister_listener.assert_not_called() @@ -320,7 +319,7 @@ async def test_unsubscribe_when_invokemethod_returned_ok_but_we_failed_to_unregi response = await subscriber.unsubscribe(self.topic, self.listener) self.assertEqual(response.status.code, UCode.ABORTED) self.assertEqual(response.status.message, "aborted") - subscriber.close() + await subscriber.close() self.rpc_client.invoke_method.assert_called_once() self.notifier.unregister_notification_listener.assert_called_once() self.transport.unregister_listener.assert_called_once() @@ -354,7 +353,6 @@ async def register_notification_listener(uri, listener): self.rpc_client.invoke_method.assert_called_once() self.notifier.register_notification_listener.assert_called_once() self.transport.register_listener.assert_called_once() - self.transport.get_source.assert_called_once() async def test_unregister_listener_missing_topic(self): notifier = MagicMock(spec=SimpleNotifier) @@ -377,11 +375,11 @@ async def on_receive(self, umsg: UMessage) -> None: pass listener = MyListener() - notifier = MagicMock(spec=SimpleNotifier) - self.transport.unregister_listener.return_value = UStatus(code=UCode.OK) - - subscriber = InMemoryUSubscriptionClient(self.transport, self.rpc_client, notifier) - # with self.assertRaises(ValueError) as context: + self.transport = MockUTransport() + subscriber = InMemoryUSubscriptionClient(self.transport) + # Register the listener before attempting to unregister + await self.transport.register_listener(self.topic, listener) + # Now attempt to unregister status = await subscriber.unregister_listener(self.topic, listener) self.assertEqual(UCode.OK, status.code) @@ -538,7 +536,6 @@ async def test_register_for_notifications_to_the_same_topic_twice_with_same_noti self.assertTrue(result is not None) self.assertEqual(self.rpc_client.invoke_method.call_count, 2) - self.assertEqual(self.transport.get_source.call_count, 2) async def test_register_for_notifications_to_the_same_topic_twice_with_different_notification_handler(self): self.transport.get_source.return_value = self.source @@ -566,7 +563,6 @@ async def test_register_for_notifications_to_the_same_topic_twice_with_different self.assertEqual("Handler already registered", context.exception.status.message) self.assertEqual(self.rpc_client.invoke_method.call_count, 2) - self.assertEqual(self.transport.get_source.call_count, 2) async def test_unregister_notification_api_for_the_happy_path(self): handler = MagicMock(spec=SubscriptionChangeHandler) @@ -620,6 +616,37 @@ async def test_unregister_notification_api_options_none(self): await subscriber.unregister_for_notifications(self.topic, handler, None) self.assertEqual("CallOptions missing", str(error.exception)) + async def test_my_notification_listener_dispatches_correctly(self): + mock_handler = MagicMock(spec=SubscriptionChangeHandler) + topic_str = UriSerializer.serialize(self.topic) + handlers = {topic_str: mock_handler} + listener = MyNotificationListener(handlers) + update = Update(topic=self.topic, status=SubscriptionStatus(state=SubscriptionStatus.State.SUBSCRIBED)) + umsg = UMessageBuilder.notification(self.topic, self.source).build_from_upayload(UPayload.pack(update)) + await listener.on_receive(umsg) + mock_handler.handle_subscription_change.assert_called_once_with(update.topic, update.status) + + async def test_my_notification_listener_ignores_wrong_message_type(self): + mock_handler = MagicMock(spec=SubscriptionChangeHandler) + topic_str = UriSerializer.serialize(self.topic) + handlers = {topic_str: mock_handler} + listener = MyNotificationListener(handlers) + umsg = UMessage() + umsg.attributes.type = uattributes_pb2.UMESSAGE_TYPE_REQUEST + await listener.on_receive(umsg) + mock_handler.handle_subscription_change.assert_not_called() + + async def test_my_notification_listener_handles_handler_exception(self): + mock_handler = MagicMock(spec=SubscriptionChangeHandler) + mock_handler.handle_subscription_change.side_effect = RuntimeError("Simulated handler error") + topic_str = UriSerializer.serialize(self.topic) + handlers = {topic_str: mock_handler} + listener = MyNotificationListener(handlers) + update = Update(topic=self.topic, status=SubscriptionStatus(state=SubscriptionStatus.State.SUBSCRIBED)) + umsg = UMessageBuilder.notification(self.topic, self.source).build_from_upayload(UPayload.pack(update)) + await listener.on_receive(umsg) + mock_handler.handle_subscription_change.assert_called_once_with(update.topic, update.status) + async def test_register_notification_api_options_none(self): handler = MagicMock(spec=SubscriptionChangeHandler) handler.handle_subscription_change.return_value = NotImplementedError( diff --git a/tests/test_communication/mock_utransport.py b/tests/test_communication/mock_utransport.py index 527c8aa..26f23b0 100644 --- a/tests/test_communication/mock_utransport.py +++ b/tests/test_communication/mock_utransport.py @@ -99,9 +99,9 @@ async def unregister_listener(self, source: UUri, listener: UListener, sink: UUr return UStatus(code=UCode.OK) return UStatus(code=UCode.NOT_FOUND) - def close(self): - self.listeners.clear() - self.executor.shutdown() + async def close(self): + await self.listeners.clear() + await self.executor.shutdown() class TimeoutUTransport(MockUTransport, ABC): diff --git a/scripts/__init__.py b/tests/test_validation/__init__.py similarity index 100% rename from scripts/__init__.py rename to tests/test_validation/__init__.py diff --git a/up-spec b/up-spec new file mode 160000 index 0000000..7af90bd --- /dev/null +++ b/up-spec @@ -0,0 +1 @@ +Subproject commit 7af90bd9aee517e79b7a41427d37de28af0f6de0 diff --git a/uprotocol/client/usubscription/v3/inmemoryusubcriptionclient.py b/uprotocol/client/usubscription/v3/inmemoryusubcriptionclient.py index 0a11746..c09794d 100644 --- a/uprotocol/client/usubscription/v3/inmemoryusubcriptionclient.py +++ b/uprotocol/client/usubscription/v3/inmemoryusubcriptionclient.py @@ -32,7 +32,6 @@ FetchSubscriptionsResponse, NotificationsRequest, NotificationsResponse, - SubscriberInfo, SubscriptionRequest, SubscriptionResponse, SubscriptionStatus, @@ -170,7 +169,7 @@ async def subscribe( raise UStatusError.from_code_message(status.code, "Failed to register listener for rpc client") self.is_listener_registered = True - request = SubscriptionRequest(topic=topic, subscriber=SubscriberInfo(uri=self.transport.get_source())) + request = SubscriptionRequest(topic=topic) # Send the subscription request and handle the response future_result = self.rpc_client.invoke_method(self.subscribe_uri, UPayload.pack(request), options) @@ -218,10 +217,8 @@ async def unsubscribe( raise ValueError("Listener missing") if not options: raise ValueError("CallOptions missing") - unsubscribe_request = UnsubscribeRequest( - topic=topic, subscriber=SubscriberInfo(uri=self.transport.get_source()) - ) - future_result = self.rpc_client.invoke_method(self.unsubscribe_uri, UPayload.pack(unsubscribe_request), options) + request = UnsubscribeRequest(topic=topic) + future_result = self.rpc_client.invoke_method(self.unsubscribe_uri, UPayload.pack(request), options) response = await RpcMapper.map_response_to_result(future_result, UnsubscribeResponse) if response.is_success(): @@ -249,12 +246,12 @@ async def unregister_listener(self, topic: UUri, listener: UListener) -> UStatus self.handlers.pop(UriSerializer.serialize(topic), None) return status - def close(self): + async def close(self): """ Close the InMemoryRpcClient by clearing stored requests and unregistering the listener. """ self.handlers.clear() - self.notifier.unregister_notification_listener(self.notification_uri, self.notification_handler) + await self.notifier.unregister_notification_listener(self.notification_uri, self.notification_handler) async def register_for_notifications( self, topic: UUri, handler: SubscriptionChangeHandler, options: Optional[CallOptions] = CallOptions.DEFAULT @@ -280,7 +277,7 @@ async def register_for_notifications( if not options: raise ValueError("CallOptions missing") - request = NotificationsRequest(topic=topic, subscriber=SubscriberInfo(uri=self.transport.get_source())) + request = NotificationsRequest(topic=topic) response = self.rpc_client.invoke_method(self.register_for_notification_uri, UPayload.pack(request), options) notifications_response = await RpcMapper.map_response(response, NotificationsResponse) @@ -312,7 +309,7 @@ async def unregister_for_notifications( if not options: raise ValueError("CallOptions missing") - request = NotificationsRequest(topic=topic, subscriber=SubscriberInfo(uri=self.transport.get_source())) + request = NotificationsRequest(topic=topic) response = self.rpc_client.invoke_method(self.unregister_for_notification_uri, UPayload.pack(request), options) notifications_response = await RpcMapper.map_response(response, NotificationsResponse)