test exc

.github/workflows/tests.yml

@@ -40,10 +40,6 @@ jobs:
       env:
         COVERAGE_FILE: ".coverage.${{ matrix.python-version }}"
         PYTEST_ARGS: "--cov=karapace --cov-append"
-    - run: make integration-tests
-      env:
-        COVERAGE_FILE: ".coverage.${{ matrix.python-version }}"
-        PYTEST_ARGS: "--cov=karapace --cov-append --random-order"
 
     - name: Archive logs
       uses: actions/upload-artifact@v4

karapace/protobuf/io.py

@@ -240,7 +240,7 @@ def read(self, bio: BytesIO) -> dict:
         return read_in_forked_process_multiprocess_process(self.config, self._writer_schema, self._reader_schema, bio)
 
 
-_WriterQueue: TypeAlias = "Queue[bytes | BaseException]"
+_WriterQueue: TypeAlias = "Queue[bytes | str | BaseException]"
 
 
 def writer_process(
@@ -253,17 +253,19 @@ def writer_process(
     try:
         class_instance = get_protobuf_class_instance(writer_schema, message_name, config)
         dict_to_protobuf(class_instance, datum)
+        result = class_instance.SerializeToString()
+        writer_queue.put(result)
     # Writing happens in the forked process, catch is broad so exception will get communicated
     # back to calling process.
-    except Exception as bare_exception:  # pylint: disable=broad-exception-caught
-        try:
-            raise ProtobufTypeException(writer_schema, datum) from bare_exception
-        except ProtobufTypeException as protobuf_exception:
-            writer_queue.put(protobuf_exception)
-            raise protobuf_exception
-    except BaseException as base_exception:  # pylint: disable=broad-exception-caught
-        writer_queue.put(base_exception)
-    writer_queue.put(class_instance.SerializeToString())
+    except BaseException as bare_exception:  # pylint: disable=broad-exception-caught
+        #try:
+        #    raise ProtobufTypeException(writer_schema, datum) from bare_exception
+        #except ProtobufTypeException as protobuf_exception:
+        #    writer_queue.put(protobuf_exception)
+        #    raise protobuf_exception
+        import traceback
+        writer_queue.put(traceback.format_exc())
+        raise bare_exception
 
 
 def write_in_forked_process(
@@ -289,6 +291,8 @@ def write_in_forked_process(
         finally:
             p.join()
             writer_queue.close()
+        if isinstance(result, str):
+            raise Exception(result)
         if isinstance(result, bytes):
             return result
         if isinstance(result, BaseException):

tests/unit/test_protobuf_serialization.py

@@ -226,14 +226,14 @@ async def test_serialization_fails(default_config_path: Path):
     mock_protobuf_registry_client.get_schema.return_value = get_latest_schema_future
 
     serializer = await make_ser_deser(default_config_path, mock_protobuf_registry_client)
-    with pytest.raises(InvalidMessageSchema):
+    with pytest.raises(RuntimeError):
         schema = await serializer.get_schema_for_subject("top")
         await serializer.serialize(schema, test_fail_objects_protobuf[0])
 
     assert mock_protobuf_registry_client.method_calls == [call.get_schema("top")]
     mock_protobuf_registry_client.reset_mock()
 
-    with pytest.raises(InvalidMessageSchema):
+    with pytest.raises(KeyError):
         schema = await serializer.get_schema_for_subject("top")
         await serializer.serialize(schema, test_fail_objects_protobuf[1])