added tests to improve coverage

Signed-off-by: Andreas Gkizas <[email protected]>

Author: gizas

exporter/exporterhelper/internal/queue/persistent_queue_test.go

@@ -58,6 +58,42 @@ func (int64Encoding) Unmarshal(bytes []byte) (context.Context, int64, error) {
 	return context.Background(), val, nil
 }
 
+func TestPersistentQueue_CorruptedItemOnRead(t *testing.T) {
+	ext := storagetest.NewMockStorageExtension(nil)
+	pq := createTestPersistentQueueWithRequestsSizer(t, ext, 100)
+
+	// Add one item
+	require.NoError(t, pq.Offer(context.Background(), int64(123)))
+	assert.EqualValues(t, 1, pq.Size())
+	assert.EqualValues(t, 123, pq.metadata.ItemsSize)
+	assert.EqualValues(t, 1230, pq.metadata.BytesSize)
+
+	// Corrupt the item in storage
+	require.NoError(t, pq.client.Set(context.Background(), getItemKey(0), []byte("invalid data")))
+
+	// Try to read the item. It should fail to unmarshal.
+	// The loop in Read continues, and since there are no more items, Read will block.
+	readFinished := make(chan struct{})
+	go func() {
+		_, _, _, ok := pq.Read(context.Background())
+		// Read returns false because the queue is stopped.
+		assert.False(t, ok)
+		close(readFinished)
+	}()
+
+	// Wait a bit to ensure Read has processed the corrupted item and is now waiting.
+	assert.Eventually(t, func() bool {
+		pq.mu.Lock()
+		defer pq.mu.Unlock()
+		// After the failed read, the queue should be empty and sizes should be reset.
+		return pq.requestSize() == 0 && pq.metadata.BytesSize == 0 && pq.metadata.ItemsSize == 0
+	}, time.Second, 10*time.Millisecond, "queue size and byte/item sizes should be reset to 0")
+
+	// Shutdown the queue to unblock Read.
+	require.NoError(t, pq.Shutdown(context.Background()))
+	<-readFinished
+}
+
 func newFakeBoundedStorageClient(maxSizeInBytes int) *fakeBoundedStorageClient {
 	return &fakeBoundedStorageClient{
 		st:             map[string][]byte{},

exporter/exporterhelper/logs_test.go

@@ -68,9 +68,9 @@ func TestLogsEncoding_UnmarshalError(t *testing.T) {
 
 	enc := logsEncoding{}
 	ctx, req, err := enc.Unmarshal([]byte("!invalid-proto"))
-	assert.Error(t, err)
-	assert.Nil(t, req)
-	assert.NotNil(t, ctx)
+	require.Error(t, err)
+	require.Nil(t, req)
+	require.NotNil(t, ctx)
 }
 
 func TestLogs_InvalidName(t *testing.T) {

exporter/exporterhelper/metrics_test.go

@@ -68,9 +68,9 @@ func TestMetricsEncoding_UnmarshalError(t *testing.T) {
 
 	enc := metricsEncoding{}
 	ctx, req, err := enc.Unmarshal([]byte("!invalid-proto"))
-	assert.Error(t, err)
-	assert.Nil(t, req)
-	assert.NotNil(t, ctx)
+	require.Error(t, err)
+	require.Nil(t, req)
+	require.NotNil(t, ctx)
 }
 
 func TestMetrics_NilConfig(t *testing.T) {

exporter/exporterhelper/traces_test.go

@@ -64,9 +64,9 @@ func TestTracesEncoding_UnmarshalError(t *testing.T) {
 
 	enc := tracesEncoding{}
 	ctx, req, err := enc.Unmarshal([]byte("!invalid-proto"))
-	assert.Error(t, err)
-	assert.Nil(t, req)
-	assert.NotNil(t, ctx)
+	require.Error(t, err)
+	require.Nil(t, req)
+	require.NotNil(t, ctx)
 }
 
 func TestTraces_InvalidName(t *testing.T) {

exporter/exporterhelper/xexporterhelper/profiles_test.go

@@ -57,6 +57,20 @@ func TestProfilesRequest(t *testing.T) {
 	)
 }
 
+func TestProfilesEncoding_UnmarshalError(t *testing.T) {
+	fgOrigReadState := queue.PersistRequestContextOnRead
+	queue.PersistRequestContextOnRead = func() bool { return false }
+	t.Cleanup(func() {
+		queue.PersistRequestContextOnRead = fgOrigReadState
+	})
+
+	enc := profilesEncoding{}
+	ctx, req, err := enc.Unmarshal([]byte("!invalid-proto"))
+	require.Error(t, err)
+	require.Nil(t, req)
+	require.NotNil(t, ctx)
+}
+
 func TestProfilesExporter_InvalidName(t *testing.T) {
 	le, err := NewProfiles(context.Background(), exportertest.NewNopSettings(exportertest.NopType), nil, newPushProfilesData(nil))
 	require.Nil(t, le)