RHINENG-23045: route events based on types + test updates

Author: TenSt

listener/common_test.go

@@ -89,10 +89,10 @@ func getOrCreateTestAccount(t *testing.T) int {
 }
 
 // nolint: unparam
-func createTestUploadEvent(orgID, inventoryID, reporter string, packages, yum bool) HostEvent {
+func createTestUploadEvent(orgID, inventoryID, reporter string, packages, yum bool, eventType string) HostEvent {
 	now := time.Now()
 	ev := HostEvent{
-		Type: "created",
+		Type: eventType,
 		Host: Host{
 			ID:       inventoryID,
 			OrgID:    &orgID,

listener/event_buffers.go

@@ -30,11 +30,6 @@ var createdEventsBuffer = eventBuffer{
 	Lock:       sync.Mutex{},
 }
 
-// var flushTimer = time.AfterFunc(87600*time.Hour, func() {
-// 	utils.LogInfo(FlushedTimeoutBuffer)
-// 	updatedEventsBuffer.flushEvalEvents()
-// })
-
 func (b *eventBuffer) initFlushTimer(w *mqueue.Writer) {
 	b.flushTimer = time.AfterFunc(87600*time.Hour, func() {
 		utils.LogInfo(FlushedTimeoutBuffer)

listener/events_test.go

@@ -27,7 +27,7 @@ func TestUpdateSystem(t *testing.T) {
 		DisplayName: id,
 	}).Error)
 
-	ev := createTestUploadEvent("1", id, "puptoo", false, false)
+	ev := createTestUploadEvent("1", id, "puptoo", false, false, "created")
 	name := "TEST_NAME"
 	ev.Host.DisplayName = &name
 	ev.Host.SystemProfile.InstalledPackages = &[]string{"kernel-0:4.18.0-193.1.2.el8_2.x86_64"}
@@ -112,7 +112,7 @@ func TestUploadAfterDelete(t *testing.T) {
 	assert.NoError(t, err)
 
 	// upload will be skipped and system won't be created
-	uploadEvent := createTestUploadEvent("1", id, "puptoo", true, false)
+	uploadEvent := createTestUploadEvent("1", id, "puptoo", true, false, "created")
 	err = HandleUpload(uploadEvent)
 	assert.NoError(t, err)
 	assertSystemNotInDB(t)
@@ -126,7 +126,7 @@ func TestCreateDeleteUpload(t *testing.T) {
 	configure()
 	deleteData(t)
 
-	uploadEvent := createTestUploadEvent("1", id, "puptoo", true, false)
+	uploadEvent := createTestUploadEvent("1", id, "puptoo", true, false, "created")
 	originalName := "UPLOADED"
 	uploadEvent.Host.DisplayName = &originalName
 	err := HandleUpload(uploadEvent)

listener/listener.go

@@ -40,11 +40,11 @@ func configure() {
 	core.ConfigureApp()
 	eventsTopic = utils.FailIfEmpty(utils.CoreCfg.EventsTopic, "EVENTS_TOPIC")
 	evalTopic := utils.FailIfEmpty(utils.CoreCfg.EvalTopic, "EVAL_TOPIC")
-	// createdTopic := utils.FailIfEmpty(utils.CoreCfg.CreatedSystemsTopic, "CREATED_SYSTEMS_TOPIC")
+	createdTopic := utils.FailIfEmpty(utils.CoreCfg.CreatedSystemsTopic, "CREATED_SYSTEMS_TOPIC")
 	ptTopic := utils.FailIfEmpty(utils.CoreCfg.PayloadTrackerTopic, "PAYLOAD_TRACKER_TOPIC")
 	evalWriter = mqueue.NewKafkaWriterFromEnv(evalTopic)
 	ptWriter = mqueue.NewKafkaWriterFromEnv(ptTopic)
-	// createdSystemsWriter = mqueue.NewKafkaWriterFromEnv(createdTopic)
+	createdSystemsWriter = mqueue.NewKafkaWriterFromEnv(createdTopic)
 
 	configureListener()
 }
@@ -76,7 +76,7 @@ func configureListener() {
 	validReporters = loadValidReporters()
 
 	updatedEventsBuffer.initFlushTimer(&evalWriter)
-	// createdEventsBuffer.initFlushTimer(createdSystemsWriter)
+	createdEventsBuffer.initFlushTimer(&createdSystemsWriter)
 }
 
 func loadValidReporters() map[string]int {

listener/upload.go

@@ -178,11 +178,11 @@ func HandleUpload(event HostEvent) error {
 	}
 
 	ptEvent.StatusMsg = ProcessingStatus
-	// if event.Type == "created" {
-	// 	// TODO: implement
-	// } else {
-	updatedEventsBuffer.bufferEvalEvents(sys.InventoryID, sys.RhAccountID, &ptEvent, &evalWriter)
-	// }
+	if event.Type == "created" {
+		createdEventsBuffer.bufferEvalEvents(sys.InventoryID, sys.RhAccountID, &ptEvent, &createdSystemsWriter)
+	} else {
+		updatedEventsBuffer.bufferEvalEvents(sys.InventoryID, sys.RhAccountID, &ptEvent, &evalWriter)
+	}
 	logAndObserve(UploadSuccess, ReceivedSuccess, &event, &ptEvent, tStart, SuccessStatus, false)
 	return nil
 }

listener/upload_test.go

@@ -113,47 +113,52 @@ func TestUpdateSystemPlatform(t *testing.T) {
 	deleteData(t)
 }
 
-func TestUploadHandler(t *testing.T) {
-	utils.SkipWithoutDB(t)
-	utils.SkipWithoutPlatform(t)
-	core.SetupTestEnvironment()
-	configure()
-	deleteData(t)
-
-	_ = getOrCreateTestAccount(t)
-	event := createTestUploadEvent(id, id, "puptoo", true, false)
-
-	event.Host.SystemProfile.OperatingSystem = inventory.OperatingSystem{Major: 8}
-	repos := append(event.Host.SystemProfile.GetYumRepos(), inventory.YumRepo{ID: "epel", Enabled: true})
-	event.Host.SystemProfile.YumRepos = &repos
-
-	err := HandleUpload(event)
-	assert.NoError(t, err)
-
-	reporterID := 1
-	assertSystemInDB(t, id, nil, &reporterID)
-
-	var sys models.SystemPlatform
-	assert.NoError(t, database.DB.Where("inventory_id = ?::uuid", id).Find(&sys).Error)
-	after := time.Now().Add(time.Hour)
-	sys.LastEvaluation = &after
-	assert.NoError(t, database.DB.Save(&sys).Error)
-	// Test that second upload did not cause re-evaluation
-	logHook := utils.NewTestLogHook()
-	log.AddHook(logHook)
-	err = HandleUpload(event)
-	assert.NoError(t, err)
-	assertInLogs(t, UploadSuccessNoEval, logHook.LogEntries...)
-	assertSystemReposInDB(t, sys.ID, []string{"epel-8"})
-	deleteData(t)
+func TestUploadHandlerCreatedSystem(t *testing.T) {
+	eventTypes := []string{"created", "updated"}
+	for _, eventType := range eventTypes {
+		t.Run(eventType, func(t *testing.T) {
+			utils.SkipWithoutDB(t)
+			utils.SkipWithoutPlatform(t)
+			core.SetupTestEnvironment()
+			configure()
+			deleteData(t)
+
+			_ = getOrCreateTestAccount(t)
+			event := createTestUploadEvent(id, id, "puptoo", true, false, eventType)
+
+			event.Host.SystemProfile.OperatingSystem = inventory.OperatingSystem{Major: 8}
+			repos := append(event.Host.SystemProfile.GetYumRepos(), inventory.YumRepo{ID: "epel", Enabled: true})
+			event.Host.SystemProfile.YumRepos = &repos
+
+			err := HandleUpload(event)
+			assert.NoError(t, err)
+
+			reporterID := 1
+			assertSystemInDB(t, id, nil, &reporterID)
+
+			var sys models.SystemPlatform
+			assert.NoError(t, database.DB.Where("inventory_id = ?::uuid", id).Find(&sys).Error)
+			after := time.Now().Add(time.Hour)
+			sys.LastEvaluation = &after
+			assert.NoError(t, database.DB.Save(&sys).Error)
+			// Test that second upload did not cause re-evaluation
+			logHook := utils.NewTestLogHook()
+			log.AddHook(logHook)
+			err = HandleUpload(event)
+			assert.NoError(t, err)
+			assertInLogs(t, UploadSuccessNoEval, logHook.LogEntries...)
+			assertSystemReposInDB(t, sys.ID, []string{"epel-8"})
+			deleteData(t)
+		})
+	}
 }
 
 func TestUploadHandlerWarn(t *testing.T) {
 	utils.SkipWithoutDB(t)
 	configure()
 	logHook := utils.NewTestLogHook()
 	log.AddHook(logHook)
-	noPkgsEvent := createTestUploadEvent("1", id, "puptoo", false, false)
+	noPkgsEvent := createTestUploadEvent("1", id, "puptoo", false, false, "created")
 	err := HandleUpload(noPkgsEvent)
 	if assert.Error(t, err) {
 		assert.ErrorIs(t, err, ErrNoPackages)
@@ -166,7 +171,7 @@ func TestUploadHandlerWarnSkipReporter(t *testing.T) {
 	configure()
 	logHook := utils.NewTestLogHook()
 	log.AddHook(logHook)
-	noPkgsEvent := createTestUploadEvent("1", id, "yupana", false, false)
+	noPkgsEvent := createTestUploadEvent("1", id, "yupana", false, false, "created")
 	err := HandleUpload(noPkgsEvent)
 	if assert.Error(t, err) {
 		assert.ErrorIs(t, err, ErrReporter)
@@ -179,7 +184,7 @@ func TestUploadHandlerWarnSkipHostType(t *testing.T) {
 	configure()
 	logHook := utils.NewTestLogHook()
 	log.AddHook(logHook)
-	event := createTestUploadEvent("1", id, "puptoo", true, false)
+	event := createTestUploadEvent("1", id, "puptoo", true, false, "created")
 	event.Host.SystemProfile.HostType = "edge"
 	err := HandleUpload(event)
 	if assert.Error(t, err) {
@@ -194,7 +199,7 @@ func TestUploadHandlerError1(t *testing.T) {
 	configure()
 	logHook := utils.NewTestLogHook()
 	log.AddHook(logHook)
-	event := createTestUploadEvent("1", id, "puptoo", true, false)
+	event := createTestUploadEvent("1", id, "puptoo", true, false, "created")
 	*event.Host.OrgID = ""
 	err := HandleUpload(event)
 	if assert.Error(t, err) {
@@ -216,10 +221,11 @@ func TestUploadHandlerError2(t *testing.T) {
 	configure()
 	deleteData(t)
 	evalWriter = &erroringWriter{}
+	createdSystemsWriter = &erroringWriter{}
 	logHook := utils.NewTestLogHook()
 	log.AddHook(logHook)
 	_ = getOrCreateTestAccount(t)
-	event := createTestUploadEvent("1", id, "puptoo", true, false)
+	event := createTestUploadEvent("1", id, "puptoo", true, false, "created")
 	err := HandleUpload(event)
 	assert.Nil(t, err)
 	time.Sleep(2 * uploadEvalTimeout)
@@ -316,7 +322,7 @@ func TestUpdateSystemPlatformYumUpdates(t *testing.T) {
 		HTTPClient: &http.Client{},
 		Debug:      true,
 	}
-	hostEvent := createTestUploadEvent("1", id, "puptoo", false, true)
+	hostEvent := createTestUploadEvent("1", id, "puptoo", false, true, "created")
 	yumUpdates, err := getYumUpdates(hostEvent, httpClient)
 	assert.Nil(t, err)