go reports-service: Return from kafka subscription when context is ca…

…nceled

go/docker/logs_ingestion/Dockerfile

@@ -16,7 +16,7 @@ RUN --mount=type=cache,target=/go/pkg/mod/ \
 
 FROM build AS tests
 
-RUN echo "go env -w GOCACHE=/go/pkg/mod/ && go test -v ./services/logs_ingestion/..." > test.sh 
+RUN echo "go env -w GOCACHE=/go/pkg/mod/ && go test ./services/logs_ingestion/..." > test.sh 
 RUN chmod +x test.sh
 
 ENTRYPOINT ["bash", "./test.sh"]

go/docker/reports/Dockerfile

@@ -16,7 +16,7 @@ RUN --mount=type=cache,target=/go/pkg/mod/ \
 
 FROM build AS tests
 
-RUN echo "go env -w GOCACHE=/go/pkg/mod/ && go test ./services/reports/..." > test.sh 
+RUN echo "go env -w GOCACHE=/go/pkg/mod/ && go test -count=1 ./services/reports/..." > test.sh 
 RUN chmod +x test.sh
 
 ENTRYPOINT ["bash", "./test.sh"]

go/pkg/message-broker/json_kafka.go

@@ -52,7 +52,7 @@ func (b *KafkaJsonMessageBroker[T]) Subscribe(ctx context.Context, messages chan
 
 		case <-ctx.Done():
 			b.logger.Info("KafkaJsonMessageBroker conext was cancelled")
-			break
+			return
 		}
 
 	}

go/pkg/repositories/utils.go

@@ -91,7 +91,6 @@ func PrefillApplicationLogs(
 
 	for index := range indexes {
 		applicationLogsRepository.RemoveIndex(ctx, index)
-		// assert.NoError(t, err, "Failed to remove index")
 	}
 
 	insertedLogsIds := make([]string, 0, 0)

go/pkg/tests/tests.go

@@ -23,7 +23,5 @@ func RunTest[T any](test func(dependencies T), t *testing.T, appModule fx.Option
 	)
 
 	app.Start(context.Background())
-	// app.RequireStart()
-	// app.RequireStop()
 	app.Stop(context.Background())
 }

go/services/reports/pkg/config/modules.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"github.com/Magpie-Monitor/magpie-monitor/pkg/elasticsearch"
 	sharedrepositories "github.com/Magpie-Monitor/magpie-monitor/pkg/repositories"
-	// "github.com/Magpie-Monitor/magpie-monitor/pkg/routing"
 	"github.com/Magpie-Monitor/magpie-monitor/pkg/tests"
 	"github.com/Magpie-Monitor/magpie-monitor/services/reports/internal/brokers"
 	"github.com/Magpie-Monitor/magpie-monitor/services/reports/internal/database"
@@ -33,13 +32,9 @@ func init() {
 			}),
 			fx.Provide(
 
-				// routing.NewRootRouter,
 				services.NewReportsService,
-				// handlers.NewReportsRouter,
 				handlers.NewReportsHandler,
 
-				// handlers.NewIncidentsRouter,
-				// handlers.NewIncidentHandler,
 				services.NewNodeIncidentsService,
 				services.NewApplicationIncidentsService,
 
@@ -113,14 +108,9 @@ func init() {
 			}),
 			fx.Provide(
 
-				// insights.NewOpenAiInsightsGenerator,
-				// routing.NewRootRouter,
 				services.NewReportsService,
-				// handlers.NewReportsRouter,
 				handlers.NewReportsHandler,
 
-				// handlers.NewIncidentsRouter,
-				// handlers.NewIncidentHandler,
 				services.NewNodeIncidentsService,
 				services.NewApplicationIncidentsService,
 

go/services/reports/pkg/openai/client_test.go

@@ -110,7 +110,7 @@ func TestClientSplitCompletionReqestsByBatchSize(t *testing.T) {
 			dependencies.Client.BatchSizeBytes = tc.batchSizeBytes
 			splitted, err := dependencies.Client.SplitCompletionReqestsByBatchSize(tc.completionRequests)
 			assert.NoError(t, err, "Failed to split completion requests")
-			assert.Equal(t, tc.expectedSplittedCompletionRequests, splitted)
+			assert.ElementsMatch(t, tc.expectedSplittedCompletionRequests, splitted)
 
 		}
 	}