107 otelevent as pydantic model (#108)

* 107 - Add test for pydantic model validation

* 107 - update tests to handle new implmentation of OTelEvent as pydantic model

* 107 - update OTelEvent to pydantic model

* 107 - Fix mypy erros with plugin

Author: FreddieMatherSmartDCSIT

.mypy.ini

@@ -1,4 +1,4 @@
 [mypy]
-plugins = sqlalchemy.ext.mypy.plugin
+plugins = sqlalchemy.ext.mypy.plugin, pydantic.mypy
 disable_error_code = import-untyped
 exclude = proof_of_concepts

tel2puml/otel_to_pv/otel_to_pv_types.py

@@ -1,8 +1,10 @@
 """This module contains types required for otel_to_pv package."""
-from typing import NamedTuple, Optional
+from typing import Optional
 
+from pydantic import BaseModel
 
-class OTelEvent(NamedTuple):
+
+class OTelEvent(BaseModel):
     """Named tuple for OTel event.
 
     :param job_name: The name of the job.

tests/tel2puml/otel_to_pv/conftest.py

@@ -1036,15 +1036,18 @@ def otel_jobs_multiple_job_names(
     for i, otel_events in enumerate(otel_jobs.values()):
         otel_jobs_updated[f"{i+5}"] = []
         for j, event in enumerate(reversed(otel_events)):
-            event = (
-                event._replace(job_name="test_name_1")
-                ._replace(job_id=f"test_id_{i+5}")
-                ._replace(event_id=f"{i+5}_{j}")
-            )
+            event = event.model_copy(update={
+                "job_name": "test_name_1", "job_id": f"test_id_{i+5}",
+                "event_id": f"{i+5}_{j}"
+            })
             if j == 0:
-                event = event._replace(child_event_ids=[f"{i+5}_{j+1}"])
+                event = event.model_copy(
+                    update={"child_event_ids": [f"{i+5}_{j+1}"]}
+                )
             else:
-                event = event._replace(parent_event_id=f"{i+5}_{j-1}")
+                event = event.model_copy(
+                    update={"parent_event_id": f"{i+5}_{j-1}"}
+                )
             otel_jobs_updated[f"{i+5}"].append(event)
 
     updated_otel_jobs = dict(otel_jobs_copy, **otel_jobs_updated)

tests/tel2puml/otel_to_pv/data_holders/sql_data_holder/test_sql_dataholder.py

@@ -753,7 +753,7 @@ def test_stream_data(
     )
 
     num_jobs = 10
-    events_per_job = 10000  # Total of 100,000 events
+    events_per_job = 1000  # Total of 10,000 events
 
     with large_sql_data_holder:
         for job_index in range(num_jobs):

tests/tel2puml/otel_to_pv/test_otel_to_pv_types.py

@@ -0,0 +1,65 @@
+"""Tests for otel_to_pv types."""
+
+import pytest
+from pydantic import ValidationError
+
+from tel2puml.otel_to_pv.otel_to_pv_types import OTelEvent
+
+
+def test_otel_event() -> None:
+    """Test OTelEvent."""
+    # test that setting with correct types works does not raise an error
+    input_dict = {
+        "job_name": "job_name",
+        "job_id": "job_id",
+        "event_type": "event_type",
+        "event_id": "event_id",
+        "start_timestamp": 1,
+        "end_timestamp": 2,
+        "application_name": "application_name",
+        "parent_event_id": "parent_event_id",
+        "child_event_ids": ["child_event_id"],
+    }
+    OTelEvent(**input_dict)
+    # test that setting with incorrect types raises a validation error
+    input_dict = {
+        "job_name": 1,
+        "job_id": 1,
+        "event_type": 1,
+        "event_id": 1,
+        "start_timestamp": "1",
+        "end_timestamp": "2",
+        "application_name": 1,
+        "parent_event_id": 1,
+        "child_event_ids": [1],
+    }
+    with pytest.raises(ValidationError):
+        OTelEvent(**input_dict)
+    # test that setting timestamp fields with a type that can be coerced works
+    input_dict = {
+        "job_name": "job_name",
+        "job_id": "job_id",
+        "event_type": "event_type",
+        "event_id": "event_id",
+        "start_timestamp": "1",
+        "end_timestamp": "2",
+        "application_name": "application_name",
+        "parent_event_id": "parent_event_id",
+        "child_event_ids": ["child_event_id"],
+    }
+    OTelEvent(**input_dict)
+    # test that setting timestamp fields with a type that cannot be coerced
+    # raises a validation error
+    input_dict = {
+        "job_name": "job_name",
+        "job_id": "job_id",
+        "event_type": "event_type",
+        "event_id": "event_id",
+        "start_timestamp": "a",
+        "end_timestamp": "b",
+        "application_name": "application_name",
+        "parent_event_id": "parent_event_id",
+        "child_event_ids": ["child_event_id"],
+    }
+    with pytest.raises(ValidationError):
+        OTelEvent(**input_dict)