Add workflow to test code coverage and generate coverage report

.github/workflows/coverage.yml

@@ -0,0 +1,101 @@
+name: Python Test and Coverage
+
+on:
+  pull_request:
+    branches:
+      - main
+
+
+jobs:
+  test-and-coverage:
+    name: Test with coverage
+    runs-on: ubuntu-latest
+    permissions:
+      pull-requests: write
+
+    steps:
+      - run: |
+          git config --global user.name 'eclipse-uprotocol-bot'
+          git config --global user.email 'uprotocol-bot@eclipse.org'
+
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Set up Apache Maven Central
+        uses: actions/setup-java@v3
+        with: # configure settings.xml
+          distribution: 'temurin'
+          java-version: '11'
+          server-id: ossrh
+          server-username: OSSRH_USER
+          server-password: OSSRH_PASSWORD
+
+      - name: Set up Python
+        uses: actions/setup-python@v3
+        with:
+          python-version: '3.x'
+
+      - name: Install Poetry
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install poetry
+
+      - name: Install dependencies
+        run: |
+          poetry install
+
+      - name: Run prebuild script
+        run: |
+          cd scripts
+          # Run the script within the Poetry virtual environment
+          poetry run python pull_and_compile_protos.py
+
+      - name: Run tests with coverage
+        run: |
+          poetry run coverage run --source=uprotocol --omit=uprotocol/proto/*,uprotocol/cloudevent/*_pb2.py,tests/*,*/__init__.py -m pytest
+          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
+
+
+      - name: Upload coverage report
+        uses: actions/upload-artifact@v2
+        with:
+          name: coverage-report
+          path: htmlcov/
+
+      - name: Check code coverage
+        uses: actions/github-script@v6
+        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));
+            }
+
+
+#      - name: Comment PR with coverage results
+#        uses: actions/github-script@v6
+#        with:
+#          github-token: ${{ secrets.GITHUB_TOKEN  }}
+#
+#          script: |
+#            const COVERAGE_PERCENTAGE = process.env.COVERAGE_PERCENTAGE;;
+#            const COVERAGE_REPORT_PATH = `https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}/`;
+#             github.rest.issues.createComment({
+#              issue_number: context.issue.number,
+#              owner: context.repo.owner,
+#              repo: context.repo.repo,
+#              body: `
+#              Code coverage report is ready! :chart_with_upwards_trend:
+#
+#              - **Code Coverage Percentage:** ${COVERAGE_PERCENTAGE}
+#              - **Code Coverage Report:** [View Coverage Report](${COVERAGE_REPORT_PATH})
+#              `
+#              });
+#

.gitignore

@@ -8,4 +8,6 @@
 .coverage
 target
 #**/proto
-poetry.lock
+poetry.lock
+htmlcov
+coverage_report.txt

README.adoc

@@ -112,4 +112,4 @@ Clean up by running the command:
 Requires coverage to be installed first, that can be done by running `pip install coverage`
 
 then you run:
-`python -m coverage run --source tests/ -m unittest discover`
+`python -m coverage run --source tests/ -m unittest discover`

pyproject.toml

@@ -20,6 +20,10 @@ gitpython = ">=3.1.41"
 googleapis-common-protos = ">=1.56.4"
 protobuf = "4.24.2"
 
+[tool.poetry.dev-dependencies]
+pytest = "^6.2"
+coverage = "^5.5"
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

tests/test_cloudevent/test_datamodel/test_ucloudevent.py

@@ -24,10 +24,13 @@
 #
 # -------------------------------------------------------------------------
 
+from datetime import datetime, timezone, timedelta
+
 from uprotocol.proto.uri_pb2 import UUri, UEntity, UResource
 from uprotocol.uri.serializer.longuriserializer import LongUriSerializer
 from uprotocol.cloudevent.cloudevents_pb2 import CloudEvent
 from uprotocol.proto.uattributes_pb2 import UMessageType, UPriority
+from uprotocol.proto.upayload_pb2 import UPayloadFormat 
 from uprotocol.cloudevent.factory.cloudeventfactory import CloudEventFactory
 from uprotocol.proto.ustatus_pb2 import UCode
 from uprotocol.uuid.factory.uuidfactory import Factories
@@ -90,6 +93,31 @@ def build_cloud_event_for_test():
     return cloud_event
 
 
+def build_cloud_event_for_test_with_id(id):
+    source = build_uri_for_test()
+    proto_payload = build_proto_payload_for_test()
+    # additional attributes
+    u_cloud_event_attributes = (
+        UCloudEventAttributesBuilder()
+        .with_hash("somehash")
+        .with_priority(UPriority.UPRIORITY_CS1)
+        .with_ttl(3)
+        .with_token("someOAuthToken")
+        .build()
+    )
+
+    # build the cloud event
+    cloud_event = CloudEventFactory.build_base_cloud_event(
+        id,
+        source,
+        proto_payload.SerializeToString(),
+        proto_payload.type_url,
+        u_cloud_event_attributes,
+        UCloudEvent.get_event_type(UMessageType.UMESSAGE_TYPE_PUBLISH),
+    )
+    return cloud_event
+
+
 class TestUCloudEvent(unittest.TestCase):
     DATA_CONTENT_TYPE = "application/x-protobuf"
 
@@ -102,6 +130,7 @@ def test_extract_sink_from_cloudevent_when_sink_exists(self):
         sink = "//bo.cloud/petapp/1/rpc.response"
         cloud_event = build_cloud_event_for_test()
         cloud_event.__setitem__("sink", sink)
+        cloud_event.__setitem__("plevel", 4)
         self.assertEqual(sink, UCloudEvent.get_sink(cloud_event))
 
     def test_extract_sink_from_cloudevent_when_sink_does_not_exist(self):
@@ -193,6 +222,82 @@ def test_extract_platform_error_from_cloudevent_when_platform_error_exists_in_wr
             UCode.OK, UCloudEvent.get_communication_status(cloud_event)
         )
 
+    def test_extract_platform_error_from_cloudevent_when_platform_error_exists(
+        self,
+    ):
+        cloud_event = build_cloud_event_for_test()
+        cloud_event.__setitem__("commstatus", UCode.INVALID_ARGUMENT)
+        self.assertEqual(
+            UCode.INVALID_ARGUMENT,
+            UCloudEvent.get_communication_status(cloud_event),
+        )
+
+    def test_extract_platform_error_from_cloudevent_when_platform_error_does_not_exist(
+        self,
+    ):
+        cloud_event = build_cloud_event_for_test()
+        self.assertEqual(
+            UCode.OK, UCloudEvent.get_communication_status(cloud_event)
+        )
+
+    def test_adding_platform_error_to_existing_cloudevent(
+        self,
+    ):
+        cloud_event = build_cloud_event_for_test()
+        self.assertEqual(
+            UCode.OK, UCloudEvent.get_communication_status(cloud_event)
+        )
+
+        cloud_event_1 = UCloudEvent.add_communication_status(
+            cloud_event, UCode.DEADLINE_EXCEEDED
+        )
+
+        self.assertEqual(
+            UCode.OK, UCloudEvent.get_communication_status(cloud_event)
+        )
+
+        self.assertEqual(
+            UCode.DEADLINE_EXCEEDED,
+            UCloudEvent.get_communication_status(cloud_event_1),
+        )
+
+    def test_adding_empty_platform_error_to_existing_cloudevent(
+        self,
+    ):
+        cloud_event = build_cloud_event_for_test()
+        self.assertEqual(
+            UCode.OK, UCloudEvent.get_communication_status(cloud_event)
+        )
+
+        cloud_event_1 = UCloudEvent.add_communication_status(cloud_event, None)
+
+        self.assertEqual(
+            UCode.OK, UCloudEvent.get_communication_status(cloud_event)
+        )
+
+        self.assertEqual(cloud_event, cloud_event_1)
+
+    def test_extract_creation_timestamp_from_cloudevent_UUID_Id_when_not_a_UUIDV8_id(
+        self,
+    ):
+        cloud_event = build_cloud_event_for_test()
+        self.assertEqual(None, UCloudEvent.get_creation_timestamp(cloud_event))
+
+    def test_extract_creation_timestamp_from_cloudevent_UUIDV8_Id_when_UUIDV8_id_is_valid(
+        self,
+    ):
+        uuid = Factories.UPROTOCOL.create()
+        str_uuid = LongUuidSerializer.instance().serialize(uuid)
+        cloud_event = build_cloud_event_for_test_with_id(str_uuid)
+        maybe_creation_timestamp = UCloudEvent.get_creation_timestamp(
+            cloud_event
+        )
+        self.assertIsNotNone(maybe_creation_timestamp)
+        creation_timestamp = maybe_creation_timestamp / 1000
+
+        now_timestamp = datetime.now(timezone.utc).timestamp()
+        self.assertAlmostEqual(creation_timestamp, now_timestamp, delta=1)
+
     def test_cloudevent_is_not_expired_cd_when_no_ttl_configured(self):
         cloud_event = build_cloud_event_for_test()
         cloud_event.__delitem__("ttl")
@@ -214,6 +319,14 @@ def test_cloudevent_is_not_expired_cd_when_ttl_is_minus_one(self):
             UCloudEvent.is_expired_by_cloud_event_creation_date(cloud_event)
         )
 
+    def test_cloudevent_is_expired_cd_when_ttl_is_one(self):
+        cloud_event = build_cloud_event_for_test()
+        cloud_event.__setitem__("ttl", 1)
+        time.sleep(0.002)
+        self.assertTrue(
+            UCloudEvent.is_expired_by_cloud_event_creation_date(cloud_event)
+        )
+
     def test_cloudevent_is_expired_when_ttl_1_mili(self):
         uuid = Factories.UPROTOCOL.create()
         str_uuid = LongUuidSerializer.instance().serialize(uuid)
@@ -294,6 +407,8 @@ def test_from_message_with_valid_message(self):
             UCloudEvent.get_type(cloud_event1),
         )
 
+
+
     def test_to_from_message_from_request_cloudevent(self):
         # additional attributes
         u_cloud_event_attributes = (
@@ -444,6 +559,8 @@ def test_to_from_message_from_UCP_cloudevent(self):
             u_cloud_event_attributes,
         )
         cloud_event.__setitem__("priority", "CS4")
+        cloud_event.__setitem__("commstatus", 16)
+        cloud_event.__setitem__("permission_level", 4)
 
         result = UCloudEvent.toMessage(cloud_event)
         self.assertIsNotNone(result)
@@ -458,3 +575,39 @@ def test_from_message_with_null_message(self):
         with self.assertRaises(ValueError) as context:
             UCloudEvent.fromMessage(None)
             self.assertTrue("message cannot be null." in context.exception)
+
+    def test_cloud_event_to_string(self):
+        u_cloud_event_attributes = (
+            UCloudEventAttributesBuilder()
+            .with_ttl(3)
+            .with_token("someOAuthToken")
+            .build()
+        )
+
+        cloud_event = CloudEventFactory.request(
+            build_uri_for_test(),
+            "//bo.cloud/petapp/1/rpc.response",
+            CloudEventFactory.generate_cloud_event_id(),
+            build_proto_payload_for_test(),
+            u_cloud_event_attributes,
+        )
+        cloud_event_string = UCloudEvent.to_string(cloud_event)
+        self.assertTrue(
+            "source='/body.access//door.front_left#Door', sink='//bo.cloud/petapp/1/rpc.response', type='req.v1'}"
+            in cloud_event_string
+        )
+
+    def test_cloud_event_to_string_none(self):
+        cloud_event_string = UCloudEvent.to_string(None)
+        self.assertEqual(
+            cloud_event_string, "null"
+        )
+
+    def test_get_upayload_format_from_content_type(self):
+        new_format = UCloudEvent().get_upayload_format_from_content_type("application/json")
+        self.assertEqual(new_format, UPayloadFormat.UPAYLOAD_FORMAT_JSON)
+
+    def test_to_message_none_entry(self):
+        with self.assertRaises(ValueError) as context:
+            UCloudEvent().toMessage(None)
+            self.assertTrue("Cloud Event can't be None" in context.exception)

tests/test_cloudevent/test_datamodel/test_ucloudeventattributes.py

@@ -98,6 +98,29 @@ def test_is_empty_function_permutations(self):
         u_cloud_event_attributes5 = UCloudEventAttributesBuilder().with_ttl(8).build()
         self.assertFalse(u_cloud_event_attributes5.is_empty())
 
+    def test__eq__is_same(self):
+        u_cloud_event_attributes = UCloudEventAttributesBuilder().with_hash("  ").with_token("  ").build()
+        self.assertTrue(u_cloud_event_attributes.__eq__(u_cloud_event_attributes))
+
+    def test__eq__is_equal(self):
+        u_cloud_event_attributes_1 = UCloudEventAttributesBuilder().with_hash("  ").with_token("  ").build()
+        u_cloud_event_attributes_2 = UCloudEventAttributesBuilder().with_hash("  ").with_token("  ").build()
+        self.assertTrue(u_cloud_event_attributes_1.__eq__(u_cloud_event_attributes_2))
+
+    def test__eq__is_not_equal(self):
+        u_cloud_event_attributes_1 = UCloudEventAttributesBuilder().with_hash("  ").with_token("  ").build()
+        u_cloud_event_attributes_2 = UCloudEventAttributesBuilder().with_hash("  ").with_token("12345").build()
+        self.assertFalse(u_cloud_event_attributes_1.__eq__(u_cloud_event_attributes_2))
+
+    def test__hash__same(self):
+        u_cloud_event_attributes_1 = UCloudEventAttributesBuilder().with_hash("  ").with_token("  ").build()
+        self.assertEqual(hash(u_cloud_event_attributes_1), hash(u_cloud_event_attributes_1))
+
+    def test__hash__different(self):
+        u_cloud_event_attributes_1 = UCloudEventAttributesBuilder().with_hash("  ").with_token("  ").build()
+        u_cloud_event_attributes_2 = UCloudEventAttributesBuilder().with_hash("  ").with_token("12345").build()
+        self.assertNotEqual(hash(u_cloud_event_attributes_1), hash(u_cloud_event_attributes_2))
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_cloudevent/test_validator/test_cloudeventvalidator.py

@@ -640,4 +640,4 @@ def fetching_the_notification_validator(self):
         validator = CloudEventValidator.get_validator(cloud_event)
         status = validator.validate_type(cloud_event).to_status()
         self.assertEqual(status, ValidationResult.STATUS_SUCCESS)
-        self.assertEqual("CloudEventValidator.Notification", str(validator))
+        self.assertEqual("CloudEventValidator.Notification", str(validator))