Skip to content

Commit 40dad68

Browse files
rverdilerverdile
authoredApr 16, 2024··
Fixes 3490: update distributions for create/update templates (#595)
* Fixes 3490: update distribution when creating/updating templates * address feedback * skip repos with no snapshot * change constants
1 parent 9b989ad commit 40dad68

26 files changed

+1068
-42
lines changed
 

‎cmd/content-sources/main.go

+4
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"time"
1212

1313
"github.com/content-services/content-sources-backend/pkg/config"
14+
"github.com/content-services/content-sources-backend/pkg/dao"
1415
"github.com/content-services/content-sources-backend/pkg/db"
1516
ce "github.com/content-services/content-sources-backend/pkg/errors"
1617
"github.com/content-services/content-sources-backend/pkg/handler"
@@ -45,6 +46,8 @@ func main() {
4546
}
4647
defer db.Close()
4748

49+
dao.SetupGormTableOrFail(db.DB)
50+
4851
if argsContain(args, "api") {
4952
err = config.ConfigureTang()
5053
if err != nil {
@@ -107,6 +110,7 @@ func kafkaConsumer(ctx context.Context, wg *sync.WaitGroup, metrics *m.Metrics)
107110
wrk.RegisterHandler(config.RepositorySnapshotTask, tasks.SnapshotHandler)
108111
wrk.RegisterHandler(config.DeleteRepositorySnapshotsTask, tasks.DeleteSnapshotHandler)
109112
wrk.RegisterHandler(config.DeleteTemplatesTask, tasks.DeleteTemplateHandler)
113+
wrk.RegisterHandler(config.UpdateTemplateDistributionsTask, tasks.UpdateTemplateDistributionsHandler)
110114
wrk.HeartbeatListener()
111115
go wrk.StartWorkers(ctx)
112116
<-ctx.Done()

‎cmd/dbmigrate/main.go

+3
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"os"
77
"time"
88

9+
"github.com/content-services/content-sources-backend/pkg/dao"
910
"github.com/content-services/content-sources-backend/pkg/db"
1011
"github.com/content-services/content-sources-backend/pkg/models"
1112
"github.com/content-services/content-sources-backend/pkg/seeds"
@@ -102,6 +103,8 @@ func main() {
102103
if err != nil {
103104
panic(err)
104105
}
106+
dao.SetupGormTableOrFail(db.DB)
107+
105108
if err = seeds.SeedRepositoryConfigurations(db.DB, 1000, seeds.SeedOptions{
106109
OrgID: "acme",
107110
}); err != nil {

‎cmd/external-repos/main.go

+2
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@ func main() {
3232
log.Panic().Err(err).Msg("Failed to connect to database")
3333
}
3434

35+
dao.SetupGormTableOrFail(db.DB)
36+
3537
if len(args) < 2 {
3638
log.Fatal().Msg("Requires arguments: download, import, introspect, snapshot, nightly-jobs")
3739
}

‎db/migrations.latest

+1-1
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
20240215104631
1+
20240403085352

‎db/migrations/20240403085352_add_distribution_href_to_templ_rc.down.sql

+5
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
BEGIN;
2+
ALTER TABLE templates_repository_configurations
3+
DROP COLUMN IF EXISTS distribution_href,
4+
DROP COLUMN IF EXISTS deleted_at;
5+
COMMIT;

‎db/migrations/20240403085352_add_distribution_href_to_templ_rc.up.sql

+5
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
BEGIN;
2+
ALTER TABLE templates_repository_configurations
3+
ADD COLUMN IF NOT EXISTS distribution_href VARCHAR (255) DEFAULT NULL,
4+
ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMP WITH TIME ZONE;
5+
COMMIT;

‎pkg/config/tasks.go

+6-5
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
package config
22

33
const (
4-
RepositorySnapshotTask = "snapshot" // Task to create a snapshot for a repository config
5-
DeleteRepositorySnapshotsTask = "delete-repository-snapshots" // Task to delete all snapshots for a repository config
6-
IntrospectTask = "introspect" // Task to introspect repository
7-
DeleteTemplatesTask = "delete-templates" // Task to delete all content templates marked for deletion
4+
RepositorySnapshotTask = "snapshot" // Task to create a snapshot for a repository config
5+
DeleteRepositorySnapshotsTask = "delete-repository-snapshots" // Task to delete all snapshots for a repository config
6+
IntrospectTask = "introspect" // Task to introspect repository
7+
DeleteTemplatesTask = "delete-templates" // Task to delete all content templates marked for deletion
8+
UpdateTemplateDistributionsTask = "update-template-content" // Task to update the pulp distributions of a template's snapshots
89
)
910

1011
const (
@@ -15,4 +16,4 @@ const (
1516
TaskStatusPending = "pending" // Task is waiting to be started
1617
)
1718

18-
var RequeueableTasks = []string{DeleteTemplatesTask, DeleteRepositorySnapshotsTask}
19+
var RequeueableTasks = []string{DeleteTemplatesTask, DeleteRepositorySnapshotsTask, UpdateTemplateDistributionsTask}

‎pkg/dao/interfaces.go

+16
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"github.com/content-services/content-sources-backend/pkg/models"
88
"github.com/content-services/content-sources-backend/pkg/pulp_client"
99
"github.com/content-services/yummy/pkg/yum"
10+
"github.com/rs/zerolog/log"
1011
"gorm.io/gorm"
1112
)
1213

@@ -52,6 +53,16 @@ func GetDaoRegistry(db *gorm.DB) *DaoRegistry {
5253
return &reg
5354
}
5455

56+
// SetupGormTableOrFail this is necessary to enable soft-delete
57+
// on the deleted_at column of the template_repository_configurations table.
58+
// More info here: https://gorm.io/docs/many_to_many.html#Customize-JoinTable
59+
func SetupGormTableOrFail(db *gorm.DB) {
60+
err := db.SetupJoinTable(models.Template{}, "RepositoryConfigurations", models.TemplateRepositoryConfiguration{})
61+
if err != nil {
62+
log.Logger.Fatal().Err(err).Msg("error setting up join table for templates_repository_configurations")
63+
}
64+
}
65+
5566
//go:generate mockery --name RepositoryConfigDao --filename repository_configs_mock.go --inpackage
5667
type RepositoryConfigDao interface {
5768
Create(newRepo api.RepositoryRequest) (api.RepositoryResponse, error)
@@ -106,6 +117,7 @@ type SnapshotDao interface {
106117
GetRepositoryConfigurationFile(orgID, snapshotUUID, host string) (string, error)
107118
WithContext(ctx context.Context) SnapshotDao
108119
Fetch(uuid string) (api.SnapshotResponse, error)
120+
FetchSnapshotsModelByDateAndRepository(orgID string, request api.ListSnapshotByDateRequest) ([]models.Snapshot, error)
109121
}
110122

111123
//go:generate mockery --name MetricsDao --filename metrics_mock.go --inpackage
@@ -163,4 +175,8 @@ type TemplateDao interface {
163175
Delete(orgID string, uuid string) error
164176
ClearDeletedAt(orgID string, uuid string) error
165177
Update(orgID string, uuid string, templParams api.TemplateUpdateRequest) (api.TemplateResponse, error)
178+
GetRepoChanges(templateUUID string, newRepoConfigUUIDs []string) ([]string, []string, []string, []string, error)
179+
GetDistributionHref(templateUUID string, repoConfigUUID string) (string, error)
180+
UpdateDistributionHrefs(templateUUID string, repoUUIDs []string, repoDistributionMap map[string]string) error
181+
DeleteTemplateRepoConfigs(templateUUID string, keepRepoConfigUUIDs []string) error
166182
}

‎pkg/dao/snapshots.go

+30-9
Original file line numberDiff line numberDiff line change
@@ -252,17 +252,26 @@ func (sDao *snapshotDaoImpl) Delete(snapUUID string) error {
252252
}
253253

254254
func (sDao *snapshotDaoImpl) FetchLatestSnapshot(repoConfigUUID string) (api.SnapshotResponse, error) {
255+
var snap models.Snapshot
256+
snap, err := sDao.fetchLatestSnapshot(repoConfigUUID)
257+
if err != nil {
258+
return api.SnapshotResponse{}, err
259+
}
260+
var apiSnap api.SnapshotResponse
261+
snapshotModelToApi(snap, &apiSnap)
262+
return apiSnap, nil
263+
}
264+
265+
func (sDao *snapshotDaoImpl) fetchLatestSnapshot(repoConfigUUID string) (models.Snapshot, error) {
255266
var snap models.Snapshot
256267
result := sDao.db.
257268
Where("snapshots.repository_configuration_uuid = ?", repoConfigUUID).
258269
Order("created_at DESC").
259270
First(&snap)
260271
if result.Error != nil {
261-
return api.SnapshotResponse{}, result.Error
272+
return models.Snapshot{}, result.Error
262273
}
263-
var apiSnap api.SnapshotResponse
264-
snapshotModelToApi(snap, &apiSnap)
265-
return apiSnap, nil
274+
return snap, nil
266275
}
267276

268277
func (sDao *snapshotDaoImpl) FetchSnapshotByVersionHref(repoConfigUUID string, versionHref string) (*api.SnapshotResponse, error) {
@@ -283,11 +292,9 @@ func (sDao *snapshotDaoImpl) FetchSnapshotByVersionHref(repoConfigUUID string, v
283292
return &apiSnap, nil
284293
}
285294

286-
// FetchSnapshotsByDateAndRepository returns a list of snapshots by date.
287-
func (sDao *snapshotDaoImpl) FetchSnapshotsByDateAndRepository(orgID string, request api.ListSnapshotByDateRequest) (api.ListSnapshotByDateResponse, error) {
295+
func (sDao *snapshotDaoImpl) FetchSnapshotsModelByDateAndRepository(orgID string, request api.ListSnapshotByDateRequest) ([]models.Snapshot, error) {
288296
snaps := []models.Snapshot{}
289-
layout := "2006-01-02"
290-
date, _ := time.Parse(layout, request.Date)
297+
date, _ := time.Parse(time.DateOnly, request.Date)
291298
date = date.AddDate(0, 0, 1) // Set the date to 24 hours later, inclusive of the current day
292299

293300
query := sDao.db.Raw(`
@@ -330,7 +337,21 @@ func (sDao *snapshotDaoImpl) FetchSnapshotsByDateAndRepository(orgID string, req
330337
Scan(&snaps)
331338

332339
if query.Error != nil {
333-
return api.ListSnapshotByDateResponse{}, query.Error
340+
return nil, query.Error
341+
}
342+
return snaps, nil
343+
}
344+
345+
// FetchSnapshotsByDateAndRepository returns a list of snapshots by date.
346+
func (sDao *snapshotDaoImpl) FetchSnapshotsByDateAndRepository(orgID string, request api.ListSnapshotByDateRequest) (api.ListSnapshotByDateResponse, error) {
347+
var snaps []models.Snapshot
348+
layout := "2006-01-02"
349+
date, _ := time.Parse(layout, request.Date)
350+
date = date.AddDate(0, 0, 1) // Set the date to 24 hours later, inclusive of the current day
351+
352+
snaps, err := sDao.FetchSnapshotsModelByDateAndRepository(orgID, request)
353+
if err != nil {
354+
return api.ListSnapshotByDateResponse{}, err
334355
}
335356

336357
repoUUIDCount := len(request.RepositoryUUIDS)

‎pkg/dao/snapshots_mock.go

+26
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎pkg/dao/suite_test.go

+4
Original file line numberDiff line numberDiff line change
@@ -167,5 +167,9 @@ func (s *DaoSuite) SetupTest() {
167167
LogLevel: logger.Info,
168168
}),
169169
})
170+
171+
SetupGormTableOrFail(db.DB)
172+
170173
s.tx = s.db.Begin()
174+
//s.tx = s.db
171175
}

‎pkg/dao/templates.go

+81-5
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@ import (
99
"github.com/content-services/content-sources-backend/pkg/event"
1010
"github.com/content-services/content-sources-backend/pkg/models"
1111
"github.com/jackc/pgx/v5/pgconn"
12+
"github.com/rs/zerolog/log"
13+
"golang.org/x/exp/slices"
1214
"gorm.io/gorm"
1315
"gorm.io/gorm/clause"
1416
)
@@ -86,6 +88,8 @@ func (t templateDaoImpl) create(tx *gorm.DB, reqTemplate api.TemplateRequest) (a
8688
return api.TemplateResponse{}, err
8789
}
8890

91+
log.Info().Msg("Create: insert into template repo configs")
92+
8993
templatesModelToApi(modelTemplate, &respTemplate)
9094
respTemplate.RepositoryUUIDS = reqTemplate.RepositoryUUIDS
9195

@@ -115,16 +119,16 @@ func (t templateDaoImpl) insertTemplateRepoConfigs(tx *gorm.DB, templateUUID str
115119

116120
err := tx.Clauses(clause.OnConflict{
117121
Columns: []clause.Column{{Name: "template_uuid"}, {Name: "repository_configuration_uuid"}},
118-
DoNothing: true,
122+
DoUpdates: clause.AssignmentColumns([]string{"deleted_at"}),
119123
}).Create(&templateRepoConfigs).Error
120124
if err != nil {
121125
return t.DBToApiError(err)
122126
}
123127
return nil
124128
}
125129

126-
func (t templateDaoImpl) removeTemplateRepoConfigs(tx *gorm.DB, templateUUID string, keepRepoConfigUUIDs []string) error {
127-
err := tx.Where("template_uuid = ? AND repository_configuration_uuid not in ?", UuidifyString(templateUUID), UuidifyStrings(keepRepoConfigUUIDs)).
130+
func (t templateDaoImpl) DeleteTemplateRepoConfigs(templateUUID string, keepRepoConfigUUIDs []string) error {
131+
err := t.db.Unscoped().Where("template_uuid = ? AND repository_configuration_uuid not in ?", UuidifyString(templateUUID), UuidifyStrings(keepRepoConfigUUIDs)).
128132
Delete(models.TemplateRepositoryConfiguration{}).Error
129133

130134
if err != nil {
@@ -133,6 +137,16 @@ func (t templateDaoImpl) removeTemplateRepoConfigs(tx *gorm.DB, templateUUID str
133137
return nil
134138
}
135139

140+
func (t templateDaoImpl) softDeleteTemplateRepoConfigs(tx *gorm.DB, templateUUID string, keepRepoConfigUUIDs []string) error {
141+
err := tx.Debug().Where("template_uuid = ? AND repository_configuration_uuid not in ?", UuidifyString(templateUUID), UuidifyStrings(keepRepoConfigUUIDs)).
142+
Delete(&models.TemplateRepositoryConfiguration{}).Error
143+
144+
if err != nil {
145+
return t.DBToApiError(err)
146+
}
147+
return nil
148+
}
149+
136150
func (t templateDaoImpl) Fetch(orgID string, uuid string) (api.TemplateResponse, error) {
137151
var respTemplate api.TemplateResponse
138152
modelTemplate, err := t.fetch(orgID, uuid)
@@ -185,17 +199,25 @@ func (t templateDaoImpl) update(tx *gorm.DB, orgID string, uuid string, templPar
185199

186200
templatesUpdateApiToModel(templParams, &dbTempl)
187201

188-
if err := tx.Model(&models.Template{}).Where("uuid = ?", UuidifyString(uuid)).Updates(dbTempl.MapForUpdate()).Error; err != nil {
202+
if err := tx.Model(&models.Template{}).Debug().Where("uuid = ?", UuidifyString(uuid)).Updates(dbTempl.MapForUpdate()).Error; err != nil {
189203
return DBErrorToApi(err)
190204
}
205+
206+
var existingRepoConfigUUIDs []string
207+
if err := tx.Model(&models.TemplateRepositoryConfiguration{}).Select("repository_configuration_uuid").Where("template_uuid = ?", dbTempl.UUID).Find(&existingRepoConfigUUIDs).Error; err != nil {
208+
return DBErrorToApi(err)
209+
}
210+
191211
if templParams.RepositoryUUIDS != nil {
192212
if err := t.validateRepositoryUUIDs(orgID, templParams.RepositoryUUIDS); err != nil {
193213
return err
194214
}
195-
err = t.removeTemplateRepoConfigs(tx, uuid, templParams.RepositoryUUIDS)
215+
216+
err = t.softDeleteTemplateRepoConfigs(tx, uuid, templParams.RepositoryUUIDS)
196217
if err != nil {
197218
return fmt.Errorf("could not remove uneeded template repositories %w", err)
198219
}
220+
199221
err = t.insertTemplateRepoConfigs(tx, uuid, templParams.RepositoryUUIDS)
200222
if err != nil {
201223
return fmt.Errorf("could not insert new template repositories %w", err)
@@ -320,6 +342,60 @@ func (t templateDaoImpl) ClearDeletedAt(orgID string, uuid string) error {
320342
return nil
321343
}
322344

345+
// GetRepoChanges given a template UUID and a slice of repo config uuids, returns the added/removed/unchanged/all between the existing and given repos
346+
func (t templateDaoImpl) GetRepoChanges(templateUUID string, newRepoConfigUUIDs []string) ([]string, []string, []string, []string, error) {
347+
var templateRepoConfigs []models.TemplateRepositoryConfiguration
348+
if err := t.db.Model(&models.TemplateRepositoryConfiguration{}).Unscoped().Where("template_uuid = ?", templateUUID).Find(&templateRepoConfigs).Error; err != nil {
349+
return nil, nil, nil, nil, t.DBToApiError(err)
350+
}
351+
352+
// if the repo is being added, it's in the request and the distribution_href is nil
353+
// if the repo is already part of the template, it's in request and distribution_href is not nil
354+
// if the repo is being removed, it's not in request but is in the table
355+
var added, unchanged, removed, all []string
356+
for _, v := range templateRepoConfigs {
357+
if v.DistributionHref == "" && slices.Contains(newRepoConfigUUIDs, v.RepositoryConfigurationUUID) {
358+
added = append(added, v.RepositoryConfigurationUUID)
359+
} else if slices.Contains(newRepoConfigUUIDs, v.RepositoryConfigurationUUID) {
360+
unchanged = append(unchanged, v.RepositoryConfigurationUUID)
361+
} else {
362+
removed = append(removed, v.RepositoryConfigurationUUID)
363+
}
364+
all = append(all, v.RepositoryConfigurationUUID)
365+
}
366+
367+
return added, removed, unchanged, all, nil
368+
}
369+
370+
func (t templateDaoImpl) GetDistributionHref(templateUUID string, repoConfigUUID string) (string, error) {
371+
var distributionHref string
372+
err := t.db.Model(&models.TemplateRepositoryConfiguration{}).Select("distribution_href").Unscoped().Where("template_uuid = ? AND repository_configuration_uuid = ?", templateUUID, repoConfigUUID).Find(&distributionHref).Error
373+
if err != nil {
374+
return "", err
375+
}
376+
return distributionHref, nil
377+
}
378+
379+
func (t templateDaoImpl) UpdateDistributionHrefs(templateUUID string, repoUUIDs []string, repoDistributionMap map[string]string) error {
380+
templateRepoConfigs := make([]models.TemplateRepositoryConfiguration, len(repoUUIDs))
381+
for i, repo := range repoUUIDs {
382+
templateRepoConfigs[i].TemplateUUID = templateUUID
383+
templateRepoConfigs[i].RepositoryConfigurationUUID = repo
384+
if repoDistributionMap != nil {
385+
templateRepoConfigs[i].DistributionHref = repoDistributionMap[repo]
386+
}
387+
}
388+
389+
err := t.db.Clauses(clause.OnConflict{
390+
Columns: []clause.Column{{Name: "template_uuid"}, {Name: "repository_configuration_uuid"}},
391+
DoUpdates: clause.AssignmentColumns([]string{"distribution_href"}),
392+
}).Create(&templateRepoConfigs).Error
393+
if err != nil {
394+
return t.DBToApiError(err)
395+
}
396+
return nil
397+
}
398+
323399
func templatesApiToModel(api api.TemplateRequest, model *models.Template) {
324400
if api.Name != nil {
325401
model.Name = *api.Name

‎pkg/dao/templates_mock.go

+105
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎pkg/dao/templates_test.go

+32
Original file line numberDiff line numberDiff line change
@@ -412,3 +412,35 @@ func (s *TemplateSuite) TestUpdate() {
412412
_, err = templateDao.Update(orgIDTest, found.UUID, api.TemplateUpdateRequest{RepositoryUUIDS: []string{"Notarealrepouuid"}})
413413
assert.Error(s.T(), err)
414414
}
415+
416+
func (s *TemplateSuite) TestGetRepoChanges() {
417+
err := seeds.SeedRepositoryConfigurations(s.tx, 3, seeds.SeedOptions{OrgID: orgIDTest})
418+
assert.NoError(s.T(), err)
419+
420+
var repoConfigs []models.RepositoryConfiguration
421+
s.tx.Model(&models.RepositoryConfiguration{}).Where("org_id = ?", orgIDTest).Find(&repoConfigs)
422+
423+
templateDao := templateDaoImpl{db: s.tx}
424+
req := api.TemplateRequest{
425+
Name: pointy.Pointer("test template"),
426+
RepositoryUUIDS: []string{repoConfigs[0].UUID, repoConfigs[1].UUID, repoConfigs[2].UUID},
427+
OrgID: pointy.Pointer(orgIDTest),
428+
}
429+
resp, err := templateDao.Create(req)
430+
assert.NoError(s.T(), err)
431+
assert.Equal(s.T(), resp.Name, "test template")
432+
433+
repoDistMap := map[string]string{}
434+
repoDistMap[repoConfigs[0].UUID] = "dist href"
435+
repoDistMap[repoConfigs[1].UUID] = "dist href"
436+
err = templateDao.UpdateDistributionHrefs(resp.UUID, resp.RepositoryUUIDS, repoDistMap)
437+
assert.NoError(s.T(), err)
438+
439+
added, removed, unchanged, all, err := templateDao.GetRepoChanges(resp.UUID, []string{
440+
repoConfigs[0].UUID, repoConfigs[2].UUID})
441+
assert.NoError(s.T(), err)
442+
assert.Equal(s.T(), []string{repoConfigs[2].UUID}, added)
443+
assert.Equal(s.T(), []string{repoConfigs[1].UUID}, removed)
444+
assert.Equal(s.T(), []string{repoConfigs[0].UUID}, unchanged)
445+
assert.ElementsMatch(s.T(), all, []string{repoConfigs[0].UUID, repoConfigs[1].UUID, repoConfigs[2].UUID})
446+
}

‎pkg/db/db.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,7 @@ func MigrateDB(dbURL string, direction string, steps ...int) error {
106106
return fmt.Errorf("migration setup failed: %w", err)
107107
}
108108

109-
err = checkLatestMigrationFile(m)
109+
err = checkLatestMigrationFile()
110110
if err != nil {
111111
return err
112112
}
@@ -180,7 +180,7 @@ func getPreviousMigrationVersion(m *migrate.Migrate) (int, error) {
180180

181181
const LatestMigrationFile = "./db/migrations.latest"
182182

183-
func checkLatestMigrationFile(m *migrate.Migrate) error {
183+
func checkLatestMigrationFile() error {
184184
migrationFileNames, err := getMigrationFiles()
185185
if err != nil {
186186
return err

‎pkg/handler/templates.go

+26-2
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
"github.com/content-services/content-sources-backend/pkg/rbac"
1111
"github.com/content-services/content-sources-backend/pkg/tasks"
1212
"github.com/content-services/content-sources-backend/pkg/tasks/client"
13+
"github.com/content-services/content-sources-backend/pkg/tasks/payloads"
1314
"github.com/content-services/content-sources-backend/pkg/tasks/queue"
1415
"github.com/labstack/echo/v4"
1516
"github.com/rs/zerolog/log"
@@ -72,6 +73,9 @@ func (th *TemplateHandler) createTemplate(c echo.Context) error {
7273
if err != nil {
7374
return ce.NewErrorResponse(ce.HttpCodeForDaoError(err), "Error creating template", err.Error())
7475
}
76+
77+
th.enqueueUpdateTemplateDistributionsEvent(c, orgID, respTemplate.UUID, respTemplate.RepositoryUUIDS)
78+
7579
return c.JSON(http.StatusCreated, respTemplate)
7680
}
7781

@@ -184,11 +188,14 @@ func (th *TemplateHandler) update(c echo.Context, fillDefaults bool) error {
184188
if fillDefaults {
185189
tempParams.FillDefaults()
186190
}
187-
apiTempl, err := th.DaoRegistry.Template.Update(orgID, uuid, tempParams)
191+
respTemplate, err := th.DaoRegistry.Template.Update(orgID, uuid, tempParams)
188192
if err != nil {
189193
return ce.NewErrorResponse(ce.HttpCodeForDaoError(err), "Error updating template", err.Error())
190194
}
191-
return c.JSON(http.StatusOK, apiTempl)
195+
196+
th.enqueueUpdateTemplateDistributionsEvent(c, orgID, respTemplate.UUID, tempParams.RepositoryUUIDS)
197+
198+
return c.JSON(http.StatusOK, respTemplate)
192199
}
193200

194201
func ParseTemplateFilters(c echo.Context) api.TemplateFilterData {
@@ -266,3 +273,20 @@ func (th *TemplateHandler) enqueueTemplateDeleteEvent(c echo.Context, orgID stri
266273

267274
return nil
268275
}
276+
277+
func (th *TemplateHandler) enqueueUpdateTemplateDistributionsEvent(c echo.Context, orgID, templateUUID string, repoConfigUUIDs []string) {
278+
accountID, _ := getAccountIdOrgId(c)
279+
payload := payloads.UpdateTemplateDistributionsPayload{TemplateUUID: templateUUID, RepoConfigUUIDs: repoConfigUUIDs}
280+
task := queue.Task{
281+
Typename: config.UpdateTemplateDistributionsTask,
282+
Payload: payload,
283+
OrgId: orgID,
284+
AccountId: accountID,
285+
RequestID: c.Response().Header().Get(config.HeaderRequestId),
286+
}
287+
taskID, err := th.TaskClient.Enqueue(task)
288+
if err != nil {
289+
logger := tasks.LogForTask(taskID.String(), task.Typename, task.RequestID)
290+
logger.Error().Msg("error enqueuing task")
291+
}
292+
}

‎pkg/handler/templates_test.go

+32-14
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import (
1717
"github.com/content-services/content-sources-backend/pkg/middleware"
1818
"github.com/content-services/content-sources-backend/pkg/tasks"
1919
"github.com/content-services/content-sources-backend/pkg/tasks/client"
20+
"github.com/content-services/content-sources-backend/pkg/tasks/payloads"
2021
"github.com/content-services/content-sources-backend/pkg/tasks/queue"
2122
test_handler "github.com/content-services/content-sources-backend/pkg/test/handler"
2223
"github.com/labstack/echo/v4"
@@ -76,16 +77,18 @@ func (suite *TemplatesSuite) TestCreate() {
7677
}
7778

7879
expected := api.TemplateResponse{
79-
UUID: "uuid",
80-
Name: "test template",
81-
OrgID: orgID,
82-
Description: "a new template",
83-
Arch: config.AARCH64,
84-
Version: config.El8,
85-
Date: time.Time{},
80+
UUID: "uuid",
81+
Name: "test template",
82+
OrgID: orgID,
83+
Description: "a new template",
84+
Arch: config.AARCH64,
85+
Version: config.El8,
86+
Date: time.Time{},
87+
RepositoryUUIDS: []string{"repo-uuid"},
8688
}
8789

8890
suite.reg.Template.On("Create", template).Return(expected, nil)
91+
mockUpdateTemplateDistributionsEvent(suite.tcMock, expected.UUID, expected.Date.String(), template.RepositoryUUIDS)
8992

9093
body, err := json.Marshal(template)
9194
require.NoError(suite.T(), err)
@@ -392,6 +395,18 @@ func mockTemplateDeleteEvent(tcMock *client.MockTaskClient, templateUUID string)
392395
}).Return(nil, nil)
393396
}
394397

398+
func mockUpdateTemplateDistributionsEvent(tcMock *client.MockTaskClient, templateUUID, templateDate string, repoConfigUUIDs []string) {
399+
tcMock.On("Enqueue", queue.Task{
400+
Typename: config.UpdateTemplateDistributionsTask,
401+
Payload: payloads.UpdateTemplateDistributionsPayload{
402+
TemplateUUID: templateUUID,
403+
RepoConfigUUIDs: repoConfigUUIDs,
404+
},
405+
OrgId: test_handler.MockOrgId,
406+
AccountId: test_handler.MockAccountNumber,
407+
}).Return(nil, nil)
408+
}
409+
395410
func (suite *TemplatesSuite) TestPartialUpdate() {
396411
uuid := "uuid"
397412
orgID := test_handler.MockOrgId
@@ -412,6 +427,7 @@ func (suite *TemplatesSuite) TestPartialUpdate() {
412427
}
413428

414429
suite.reg.Template.On("Update", orgID, uuid, template).Return(expected, nil)
430+
mockUpdateTemplateDistributionsEvent(suite.tcMock, expected.UUID, expected.Date.String(), template.RepositoryUUIDS)
415431

416432
body, err := json.Marshal(template)
417433
require.NoError(suite.T(), err)
@@ -444,16 +460,18 @@ func (suite *TemplatesSuite) TestFullUpdate() {
444460
templateExpected.FillDefaults()
445461

446462
expected := api.TemplateResponse{
447-
UUID: "uuid",
448-
Name: "test template",
449-
OrgID: orgID,
450-
Description: "a new template",
451-
Arch: config.AARCH64,
452-
Version: config.El8,
453-
Date: *templateExpected.Date,
463+
UUID: "uuid",
464+
Name: "test template",
465+
OrgID: orgID,
466+
Description: "a new template",
467+
Arch: config.AARCH64,
468+
Version: config.El8,
469+
Date: *templateExpected.Date,
470+
RepositoryUUIDS: []string{},
454471
}
455472

456473
suite.reg.Template.On("Update", orgID, uuid, templateExpected).Return(expected, nil)
474+
mockUpdateTemplateDistributionsEvent(suite.tcMock, expected.UUID, expected.Date.String(), expected.RepositoryUUIDS)
457475

458476
body, err := json.Marshal(template)
459477
require.NoError(suite.T(), err)

‎pkg/models/template_repository_configuration.go

+4-2
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,10 @@ import "gorm.io/gorm"
55
const TableNameTemplatesRepositoryConfigurations = "templates_repository_configurations"
66

77
type TemplateRepositoryConfiguration struct {
8-
RepositoryConfigurationUUID string `json:"repository_configuration_uuid" gorm:"not null"`
9-
TemplateUUID string `json:"template_uuid" gorm:"not null"`
8+
RepositoryConfigurationUUID string `json:"repository_configuration_uuid" gorm:"not null"`
9+
TemplateUUID string `json:"template_uuid" gorm:"not null"`
10+
DistributionHref string `json:"distribution_href"`
11+
DeletedAt gorm.DeletedAt `json:"deleted_at"`
1012
}
1113

1214
func (t *TemplateRepositoryConfiguration) BeforeCreate(db *gorm.DB) (err error) {

‎pkg/pulp_client/interfaces.go

+1
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ type PulpClient interface {
5858
CreateRpmDistribution(publicationHref string, name string, basePath string, contentGuardHref *string) (*string, error)
5959
FindDistributionByPath(path string) (*zest.RpmRpmDistributionResponse, error)
6060
DeleteRpmDistribution(rpmDistributionHref string) (string, error)
61+
UpdateRpmDistribution(rpmDistributionHref string, rpmPublicationHref string, distributionName string, basePath string) (string, error)
6162

6263
// Domains
6364
LookupOrCreateDomain(name string) (string, error)

‎pkg/pulp_client/pulp_client_mock.go

+25-1
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎pkg/pulp_client/pulp_global_client_mock.go

+1-1
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎pkg/pulp_client/rpm_distributions.go

+22
Original file line numberDiff line numberDiff line change
@@ -53,3 +53,25 @@ func (r *pulpDaoImpl) DeleteRpmDistribution(rpmDistributionHref string) (string,
5353
defer httpResp.Body.Close()
5454
return resp.Task, nil
5555
}
56+
57+
func (r *pulpDaoImpl) UpdateRpmDistribution(rpmDistributionHref string, rpmPublicationHref string, distributionName string, basePath string) (string, error) {
58+
patchedRpmDistribution := zest.PatchedrpmRpmDistribution{}
59+
60+
patchedRpmDistribution.Name = &distributionName
61+
patchedRpmDistribution.BasePath = &basePath
62+
63+
var pub zest.NullableString
64+
pub.Set(&rpmPublicationHref)
65+
patchedRpmDistribution.SetPublication(rpmPublicationHref)
66+
67+
resp, httpResp, err := r.client.DistributionsRpmAPI.DistributionsRpmRpmPartialUpdate(r.ctx, rpmDistributionHref).PatchedrpmRpmDistribution(patchedRpmDistribution).Execute()
68+
if httpResp != nil {
69+
defer httpResp.Body.Close()
70+
}
71+
if err != nil {
72+
return "", errorWithResponseBody("error listing rpm distributions", httpResp, err)
73+
}
74+
defer httpResp.Body.Close()
75+
76+
return resp.Task, nil
77+
}

‎pkg/tasks/payloads/update_template_distributions.go

+8
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
package payloads
2+
3+
const UpdateTemplateDistributions = "update-template-content"
4+
5+
type UpdateTemplateDistributionsPayload struct {
6+
TemplateUUID string
7+
RepoConfigUUIDs []string
8+
}

‎pkg/tasks/update_template_distributions.go

+331
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,331 @@
1+
package tasks
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"fmt"
7+
"net/url"
8+
"time"
9+
10+
"github.com/content-services/content-sources-backend/pkg/api"
11+
"github.com/content-services/content-sources-backend/pkg/config"
12+
"github.com/content-services/content-sources-backend/pkg/dao"
13+
"github.com/content-services/content-sources-backend/pkg/db"
14+
"github.com/content-services/content-sources-backend/pkg/models"
15+
"github.com/content-services/content-sources-backend/pkg/pulp_client"
16+
"github.com/content-services/content-sources-backend/pkg/tasks/payloads"
17+
"github.com/content-services/content-sources-backend/pkg/tasks/queue"
18+
zest "github.com/content-services/zest/release/v2024"
19+
"github.com/google/uuid"
20+
"github.com/rs/zerolog"
21+
"golang.org/x/exp/slices"
22+
)
23+
24+
func UpdateTemplateDistributionsHandler(ctx context.Context, task *models.TaskInfo, queue *queue.Queue) error {
25+
opts := payloads.UpdateTemplateDistributionsPayload{}
26+
if err := json.Unmarshal(task.Payload, &opts); err != nil {
27+
return fmt.Errorf("payload incorrect type for UpdateTemplateDistributions")
28+
}
29+
30+
logger := LogForTask(task.Id.String(), task.Typename, task.RequestID)
31+
ctxWithLogger := logger.WithContext(ctx)
32+
33+
daoReg := dao.GetDaoRegistry(db.DB)
34+
domainName, err := daoReg.Domain.Fetch(task.OrgId)
35+
if err != nil {
36+
return err
37+
}
38+
39+
rhDomainName, err := daoReg.Domain.Fetch(config.RedHatOrg)
40+
if err != nil {
41+
return err
42+
}
43+
44+
pulpClient := pulp_client.GetPulpClientWithDomain(ctxWithLogger, domainName)
45+
46+
t := UpdateTemplateDistributions{
47+
orgId: task.OrgId,
48+
domainName: domainName,
49+
rhDomainName: rhDomainName,
50+
repositoryUUID: task.RepositoryUUID,
51+
daoReg: daoReg,
52+
pulpClient: pulpClient,
53+
task: task,
54+
payload: &opts,
55+
queue: queue,
56+
ctx: ctx,
57+
logger: logger,
58+
}
59+
60+
return t.Run()
61+
}
62+
63+
type UpdateTemplateDistributions struct {
64+
orgId string
65+
domainName string
66+
rhDomainName string
67+
repositoryUUID uuid.UUID
68+
daoReg *dao.DaoRegistry
69+
pulpClient pulp_client.PulpClient
70+
payload *payloads.UpdateTemplateDistributionsPayload
71+
task *models.TaskInfo
72+
queue *queue.Queue
73+
ctx context.Context
74+
logger *zerolog.Logger
75+
}
76+
77+
func (t *UpdateTemplateDistributions) Run() error {
78+
if t.payload.RepoConfigUUIDs == nil {
79+
return nil
80+
}
81+
82+
repoConfigDistributionHref := map[string]string{} // mapping to associate each repo config to a distribution href
83+
84+
reposAdded, reposRemoved, reposUnchanged, allRepos, err := t.daoReg.Template.GetRepoChanges(t.payload.TemplateUUID, t.payload.RepoConfigUUIDs)
85+
if err != nil {
86+
return err
87+
}
88+
89+
template, err := t.daoReg.Template.Fetch(t.orgId, t.payload.TemplateUUID)
90+
if err != nil {
91+
return err
92+
}
93+
94+
date := template.Date.Format(time.DateOnly)
95+
l := api.ListSnapshotByDateRequest{Date: date, RepositoryUUIDS: allRepos}
96+
snapshots, err := t.daoReg.Snapshot.FetchSnapshotsModelByDateAndRepository(t.orgId, l)
97+
if err != nil {
98+
return err
99+
}
100+
101+
if reposAdded != nil {
102+
err := t.handleReposAdded(reposAdded, snapshots, repoConfigDistributionHref)
103+
if err != nil {
104+
return err
105+
}
106+
}
107+
108+
if reposRemoved != nil {
109+
err := t.handleReposRemoved(reposRemoved)
110+
if err != nil {
111+
return err
112+
}
113+
keepRepoConfigUUIDs := append(reposUnchanged, reposAdded...)
114+
err = t.daoReg.Template.DeleteTemplateRepoConfigs(t.payload.TemplateUUID, keepRepoConfigUUIDs)
115+
if err != nil {
116+
return err
117+
}
118+
}
119+
120+
if reposUnchanged != nil {
121+
err := t.handleReposUnchanged(reposUnchanged, snapshots, repoConfigDistributionHref)
122+
if err != nil {
123+
return err
124+
}
125+
}
126+
127+
err = t.daoReg.Template.UpdateDistributionHrefs(t.payload.TemplateUUID, t.payload.RepoConfigUUIDs, repoConfigDistributionHref)
128+
if err != nil {
129+
return err
130+
}
131+
132+
return nil
133+
}
134+
135+
func (t *UpdateTemplateDistributions) handleReposAdded(reposAdded []string, snapshots []models.Snapshot, repoConfigDistributionHref map[string]string) error {
136+
for _, repoConfigUUID := range reposAdded {
137+
repo, err := t.daoReg.RepositoryConfig.Fetch(t.orgId, repoConfigUUID)
138+
if err != nil {
139+
return err
140+
}
141+
if repo.LastSnapshot == nil {
142+
continue
143+
}
144+
145+
// Configure client for org
146+
if repo.OrgID == config.RedHatOrg {
147+
t.pulpClient = t.pulpClient.WithDomain(t.rhDomainName)
148+
} else {
149+
t.pulpClient = t.pulpClient.WithDomain(t.domainName)
150+
}
151+
152+
snapIndex := slices.IndexFunc(snapshots, func(s models.Snapshot) bool {
153+
return s.RepositoryConfigurationUUID == repoConfigUUID
154+
})
155+
156+
distPath, distName, err := getDistPathAndName(repo, t.payload.TemplateUUID, snapshots[snapIndex].UUID)
157+
if err != nil {
158+
return err
159+
}
160+
161+
distResp, err := t.createDistributionWithContentGuard(snapshots[snapIndex].PublicationHref, distName, distPath)
162+
if err != nil {
163+
return err
164+
}
165+
166+
distHrefPtr := pulp_client.SelectRpmDistributionHref(distResp)
167+
if distHrefPtr == nil {
168+
return fmt.Errorf("could not find a distribution href in task: %v", *distResp.PulpHref)
169+
}
170+
171+
repoConfigDistributionHref[repoConfigUUID] = *distHrefPtr
172+
}
173+
return nil
174+
}
175+
176+
func (t *UpdateTemplateDistributions) handleReposUnchanged(reposUnchanged []string, snapshots []models.Snapshot, repoConfigDistributionHref map[string]string) error {
177+
for _, repoConfigUUID := range reposUnchanged {
178+
repo, err := t.daoReg.RepositoryConfig.Fetch(t.orgId, repoConfigUUID)
179+
if err != nil {
180+
return err
181+
}
182+
if repo.LastSnapshot == nil {
183+
continue
184+
}
185+
186+
// Configure client for org
187+
if repo.OrgID == config.RedHatOrg {
188+
t.pulpClient = t.pulpClient.WithDomain(t.rhDomainName)
189+
} else {
190+
t.pulpClient = t.pulpClient.WithDomain(t.domainName)
191+
}
192+
193+
snapIndex := slices.IndexFunc(snapshots, func(s models.Snapshot) bool {
194+
return s.RepositoryConfigurationUUID == repoConfigUUID
195+
})
196+
if snapIndex == -1 {
197+
continue
198+
}
199+
200+
distPath, distName, err := getDistPathAndName(repo, t.payload.TemplateUUID, snapshots[snapIndex].UUID)
201+
if err != nil {
202+
return err
203+
}
204+
205+
distHref, err := t.daoReg.Template.GetDistributionHref(t.payload.TemplateUUID, repoConfigUUID)
206+
if err != nil {
207+
return err
208+
}
209+
210+
err = t.createOrUpdateDistribution(distHref, distName, distPath, snapshots[snapIndex].PublicationHref)
211+
if err != nil {
212+
return err
213+
}
214+
215+
distResp, err := t.pulpClient.FindDistributionByPath(distPath)
216+
if err != nil {
217+
return err
218+
}
219+
repoConfigDistributionHref[repoConfigUUID] = *distResp.PulpHref
220+
}
221+
return nil
222+
}
223+
224+
func (t *UpdateTemplateDistributions) handleReposRemoved(reposRemoved []string) error {
225+
for _, repoConfigUUID := range reposRemoved {
226+
repo, err := t.daoReg.RepositoryConfig.Fetch(t.orgId, repoConfigUUID)
227+
if err != nil {
228+
return err
229+
}
230+
if repo.LastSnapshot == nil {
231+
continue
232+
}
233+
234+
// Configure client for org
235+
if repo.OrgID == config.RedHatOrg {
236+
t.pulpClient = t.pulpClient.WithDomain(t.rhDomainName)
237+
} else {
238+
t.pulpClient = t.pulpClient.WithDomain(t.domainName)
239+
}
240+
241+
distHref, err := t.daoReg.Template.GetDistributionHref(t.payload.TemplateUUID, repoConfigUUID)
242+
if err != nil {
243+
return err
244+
}
245+
taskHref, err := t.pulpClient.DeleteRpmDistribution(distHref)
246+
if err != nil {
247+
return err
248+
}
249+
250+
if taskHref != "" {
251+
_, err = t.pulpClient.PollTask(taskHref)
252+
if err != nil {
253+
return err
254+
}
255+
}
256+
}
257+
return nil
258+
}
259+
260+
func (t *UpdateTemplateDistributions) createDistributionWithContentGuard(publicationHref, distName, distPath string) (*zest.TaskResponse, error) {
261+
// Create content guard
262+
var contentGuardHref *string
263+
if t.orgId != config.RedHatOrg && config.Get().Clients.Pulp.CustomRepoContentGuards {
264+
href, err := t.pulpClient.CreateOrUpdateGuardsForOrg(t.orgId)
265+
if err != nil {
266+
return nil, fmt.Errorf("could not fetch/create/update content guard: %w", err)
267+
}
268+
contentGuardHref = &href
269+
}
270+
271+
// Create distribution
272+
distTask, err := t.pulpClient.CreateRpmDistribution(publicationHref, distName, distPath, contentGuardHref)
273+
if err != nil {
274+
return nil, err
275+
}
276+
277+
distResp, err := t.pulpClient.PollTask(*distTask)
278+
if err != nil {
279+
return nil, err
280+
}
281+
282+
return distResp, nil
283+
}
284+
285+
func (t *UpdateTemplateDistributions) createOrUpdateDistribution(distHref, distName, distPath, publicationHref string) error {
286+
resp, err := t.pulpClient.FindDistributionByPath(distPath)
287+
if err != nil {
288+
return err
289+
}
290+
291+
if resp == nil {
292+
_, err := t.createDistributionWithContentGuard(publicationHref, distName, distPath)
293+
if err != nil {
294+
return err
295+
}
296+
} else {
297+
taskHref, err := t.pulpClient.UpdateRpmDistribution(distHref, publicationHref, distName, distPath)
298+
if err != nil {
299+
return err
300+
}
301+
302+
_, err = t.pulpClient.PollTask(taskHref)
303+
if err != nil {
304+
return err
305+
}
306+
}
307+
return nil
308+
}
309+
310+
func getDistPathAndName(repo api.RepositoryResponse, templateUUID string, snapshotUUID string) (distPath string, distName string, err error) {
311+
if repo.OrgID == config.RedHatOrg {
312+
path, err := getRHRepoContentPath(repo.URL)
313+
if err != nil {
314+
return "", "", err
315+
}
316+
distPath = fmt.Sprintf("templates/%v/%v", templateUUID, path)
317+
} else {
318+
distPath = fmt.Sprintf("templates/%v/%v", templateUUID, repo.UUID)
319+
}
320+
321+
distName = templateUUID + "/" + snapshotUUID
322+
return distPath, distName, nil
323+
}
324+
325+
func getRHRepoContentPath(rawURL string) (string, error) {
326+
u, err := url.Parse(rawURL)
327+
if err != nil {
328+
return "", err
329+
}
330+
return u.Path[1 : len(u.Path)-1], nil
331+
}

‎pkg/tasks/update_template_distributions_test.go

+41
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package tasks
2+
3+
import (
4+
"fmt"
5+
"testing"
6+
7+
"github.com/content-services/content-sources-backend/pkg/api"
8+
"github.com/content-services/content-sources-backend/pkg/config"
9+
"github.com/stretchr/testify/assert"
10+
"github.com/stretchr/testify/suite"
11+
)
12+
13+
type UpdateTemplateDistributionsSuite struct {
14+
suite.Suite
15+
}
16+
17+
func TestUpdateTemplateDistributionsSuite(t *testing.T) {
18+
suite.Run(t, new(UpdateTemplateDistributionsSuite))
19+
}
20+
21+
func (s *UpdateTemplateDistributionsSuite) TestGetDistributionPath() {
22+
repoUUID := "repo-uuid"
23+
templateUUID := "template-uuid"
24+
snapshotUUID := "snapshot-uuid"
25+
url := "http://example.com/red/hat/repo/path/"
26+
expectedRhPath := fmt.Sprintf("templates/%v/%v", templateUUID, "red/hat/repo/path")
27+
expectedCustomPath := fmt.Sprintf("templates/%v/%v", templateUUID, repoUUID)
28+
expectedName := templateUUID + "/" + snapshotUUID
29+
30+
repo := api.RepositoryResponse{UUID: repoUUID, URL: url, OrgID: config.RedHatOrg}
31+
distPath, distName, err := getDistPathAndName(repo, templateUUID, snapshotUUID)
32+
assert.NoError(s.T(), err)
33+
assert.Equal(s.T(), expectedRhPath, distPath)
34+
assert.Equal(s.T(), expectedName, distName)
35+
36+
repo = api.RepositoryResponse{UUID: repoUUID, URL: url, OrgID: "12345"}
37+
distPath, _, err = getDistPathAndName(repo, templateUUID, snapshotUUID)
38+
assert.NoError(s.T(), err)
39+
assert.Equal(s.T(), expectedCustomPath, distPath)
40+
assert.Equal(s.T(), expectedName, distName)
41+
}

‎test/integration/update_template_distributions_test.go

+255
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,255 @@
1+
package integration
2+
3+
import (
4+
"context"
5+
"encoding/base64"
6+
"encoding/json"
7+
"fmt"
8+
"net/http"
9+
"testing"
10+
"time"
11+
12+
"github.com/content-services/content-sources-backend/pkg/api"
13+
"github.com/content-services/content-sources-backend/pkg/config"
14+
"github.com/content-services/content-sources-backend/pkg/dao"
15+
"github.com/content-services/content-sources-backend/pkg/db"
16+
m "github.com/content-services/content-sources-backend/pkg/instrumentation"
17+
"github.com/content-services/content-sources-backend/pkg/models"
18+
"github.com/content-services/content-sources-backend/pkg/tasks"
19+
"github.com/content-services/content-sources-backend/pkg/tasks/client"
20+
"github.com/content-services/content-sources-backend/pkg/tasks/payloads"
21+
"github.com/content-services/content-sources-backend/pkg/tasks/queue"
22+
"github.com/content-services/content-sources-backend/pkg/tasks/worker"
23+
uuid2 "github.com/google/uuid"
24+
"github.com/openlyinc/pointy"
25+
"github.com/prometheus/client_golang/prometheus"
26+
"github.com/redhatinsights/platform-go-middlewares/v2/identity"
27+
"github.com/rs/zerolog/log"
28+
"github.com/stretchr/testify/assert"
29+
"github.com/stretchr/testify/require"
30+
"github.com/stretchr/testify/suite"
31+
)
32+
33+
type UpdateTemplateDistributionsSuite struct {
34+
Suite
35+
dao *dao.DaoRegistry
36+
queue queue.PgQueue
37+
taskClient client.TaskClient
38+
}
39+
40+
func (s *UpdateTemplateDistributionsSuite) SetupTest() {
41+
s.Suite.SetupTest()
42+
43+
wkrQueue, err := queue.NewPgQueue(db.GetUrl())
44+
require.NoError(s.T(), err)
45+
s.queue = wkrQueue
46+
47+
s.taskClient = client.NewTaskClient(&s.queue)
48+
49+
wrk := worker.NewTaskWorkerPool(&wkrQueue, m.NewMetrics(prometheus.NewRegistry()))
50+
wrk.RegisterHandler(config.RepositorySnapshotTask, tasks.SnapshotHandler)
51+
wrk.RegisterHandler(config.UpdateTemplateDistributionsTask, tasks.UpdateTemplateDistributionsHandler)
52+
wrk.RegisterHandler(config.DeleteRepositorySnapshotsTask, tasks.DeleteSnapshotHandler)
53+
wrk.HeartbeatListener()
54+
55+
wkrCtx := context.Background()
56+
go (wrk).StartWorkers(wkrCtx)
57+
go func() {
58+
<-wkrCtx.Done()
59+
wrk.Stop()
60+
}()
61+
// Force local storage for integration tests
62+
config.Get().Clients.Pulp.StorageType = "local"
63+
64+
// Force content guard setup
65+
config.Get().Clients.Pulp.CustomRepoContentGuards = true
66+
config.Get().Clients.Pulp.GuardSubjectDn = "warlin.door"
67+
}
68+
69+
func TestUpdateTemplateDistributionsSuite(t *testing.T) {
70+
suite.Run(t, new(UpdateTemplateDistributionsSuite))
71+
}
72+
73+
func (s *UpdateTemplateDistributionsSuite) TestUpdateTemplateDistributions() {
74+
s.dao = dao.GetDaoRegistry(db.DB)
75+
76+
orgID := uuid2.NewString()
77+
repo1 := s.createAndSyncRepository(orgID, "https://fixtures.pulpproject.org/rpm-unsigned/")
78+
repo2 := s.createAndSyncRepository(orgID, "https://rverdile.fedorapeople.org/dummy-repos/comps/repo1/")
79+
80+
repo3Name := uuid2.NewString()
81+
repoURL := "https://rverdile.fedorapeople.org/dummy-repos/comps/repo2/"
82+
repo3, err := s.dao.RepositoryConfig.Create(api.RepositoryRequest{
83+
Name: &repo3Name,
84+
URL: &repoURL,
85+
OrgID: &orgID,
86+
AccountID: &orgID,
87+
})
88+
assert.NoError(s.T(), err)
89+
90+
domainName, err := s.dao.Domain.Fetch(orgID)
91+
assert.NoError(s.T(), err)
92+
93+
reqTemplate := api.TemplateRequest{
94+
Name: pointy.Pointer("test template"),
95+
Description: pointy.Pointer("includes rpm unsigned"),
96+
RepositoryUUIDS: []string{repo1.UUID},
97+
OrgID: pointy.Pointer(repo1.OrgID),
98+
}
99+
tempResp, err := s.dao.Template.Create(reqTemplate)
100+
assert.NoError(s.T(), err)
101+
102+
s.updateTemplatesAndWait(orgID, tempResp.UUID, []string{repo1.UUID})
103+
distPath := fmt.Sprintf("%v/pulp/content/%s/templates/%v/%v", config.Get().Clients.Pulp.Server, domainName, tempResp.UUID, repo1.UUID)
104+
err = s.getRequest(distPath, identity.Identity{OrgID: repo1.OrgID, Internal: identity.Internal{OrgID: repo1.OrgID}}, 200)
105+
assert.NoError(s.T(), err)
106+
107+
updateReq := api.TemplateUpdateRequest{
108+
RepositoryUUIDS: []string{repo1.UUID, repo2.UUID, repo3.UUID},
109+
OrgID: &orgID,
110+
}
111+
_, err = s.dao.Template.Update(orgID, tempResp.UUID, updateReq)
112+
assert.NoError(s.T(), err)
113+
114+
s.updateTemplatesAndWait(orgID, tempResp.UUID, []string{repo1.UUID, repo2.UUID})
115+
distPath = fmt.Sprintf("%v/pulp/content/%s/templates/%v/%v", config.Get().Clients.Pulp.Server, domainName, tempResp.UUID, repo1.UUID)
116+
err = s.getRequest(distPath, identity.Identity{OrgID: orgID, Internal: identity.Internal{OrgID: orgID}}, 200)
117+
assert.NoError(s.T(), err)
118+
distPath = fmt.Sprintf("%v/pulp/content/%s/templates/%v/%v", config.Get().Clients.Pulp.Server, domainName, tempResp.UUID, repo2.UUID)
119+
err = s.getRequest(distPath, identity.Identity{OrgID: orgID, Internal: identity.Internal{OrgID: orgID}}, 200)
120+
assert.NoError(s.T(), err)
121+
distPath = fmt.Sprintf("%v/pulp/content/%s/templates/%v/%v", config.Get().Clients.Pulp.Server, domainName, tempResp.UUID, repo3.UUID)
122+
err = s.getRequest(distPath, identity.Identity{OrgID: orgID, Internal: identity.Internal{OrgID: orgID}}, 404)
123+
assert.NoError(s.T(), err)
124+
125+
updateReq = api.TemplateUpdateRequest{
126+
RepositoryUUIDS: []string{repo1.UUID},
127+
OrgID: &orgID,
128+
}
129+
_, err = s.dao.Template.Update(orgID, tempResp.UUID, updateReq)
130+
assert.NoError(s.T(), err)
131+
132+
s.updateTemplatesAndWait(orgID, tempResp.UUID, []string{repo1.UUID})
133+
distPath = fmt.Sprintf("%v/pulp/content/%s/templates/%v/%v", config.Get().Clients.Pulp.Server, domainName, tempResp.UUID, repo1.UUID)
134+
err = s.getRequest(distPath, identity.Identity{OrgID: orgID, Internal: identity.Internal{OrgID: orgID}}, 200)
135+
assert.NoError(s.T(), err)
136+
distPath = fmt.Sprintf("%v/pulp/content/%s/templates/%v/%v", config.Get().Clients.Pulp.Server, domainName, tempResp.UUID, repo2.UUID)
137+
err = s.getRequest(distPath, identity.Identity{OrgID: orgID, Internal: identity.Internal{OrgID: orgID}}, 404)
138+
assert.NoError(s.T(), err)
139+
distPath = fmt.Sprintf("%v/pulp/content/%s/templates/%v/%v", config.Get().Clients.Pulp.Server, domainName, tempResp.UUID, repo3.UUID)
140+
err = s.getRequest(distPath, identity.Identity{OrgID: orgID, Internal: identity.Internal{OrgID: orgID}}, 404)
141+
assert.NoError(s.T(), err)
142+
}
143+
144+
func (s *UpdateTemplateDistributionsSuite) updateTemplatesAndWait(orgId string, tempUUID string, repoConfigUUIDS []string) {
145+
var err error
146+
payload := payloads.UpdateTemplateDistributionsPayload{
147+
TemplateUUID: tempUUID,
148+
RepoConfigUUIDs: repoConfigUUIDS,
149+
}
150+
task := queue.Task{
151+
Typename: config.UpdateTemplateDistributionsTask,
152+
Payload: payload,
153+
OrgId: orgId,
154+
}
155+
156+
taskUUID, err := s.taskClient.Enqueue(task)
157+
assert.NoError(s.T(), err)
158+
159+
s.WaitOnTask(taskUUID)
160+
}
161+
162+
func (s *UpdateTemplateDistributionsSuite) snapshotAndWait(taskClient client.TaskClient, repo api.RepositoryResponse, repoUuid uuid2.UUID, orgId string) {
163+
var err error
164+
taskUuid, err := taskClient.Enqueue(queue.Task{Typename: config.RepositorySnapshotTask, Payload: payloads.SnapshotPayload{}, OrgId: repo.OrgID,
165+
RepositoryUUID: pointy.String(repoUuid.String())})
166+
assert.NoError(s.T(), err)
167+
168+
s.WaitOnTask(taskUuid)
169+
170+
// Verify the snapshot was created
171+
snaps, _, err := s.dao.Snapshot.List(repo.OrgID, repo.UUID, api.PaginationData{Limit: -1}, api.FilterData{})
172+
assert.NoError(s.T(), err)
173+
assert.NotEmpty(s.T(), snaps)
174+
time.Sleep(5 * time.Second)
175+
176+
// Fetch the repomd.xml to verify its being served
177+
distPath := fmt.Sprintf("%s/pulp/content/%s/repodata/repomd.xml",
178+
config.Get().Clients.Pulp.Server,
179+
snaps.Data[0].RepositoryPath)
180+
err = s.getRequest(distPath, identity.Identity{OrgID: repo.OrgID, Internal: identity.Internal{OrgID: repo.OrgID}}, 200)
181+
assert.NoError(s.T(), err)
182+
}
183+
184+
func (s *UpdateTemplateDistributionsSuite) WaitOnTask(taskUUID uuid2.UUID) {
185+
taskInfo := s.waitOnTask(taskUUID)
186+
if taskInfo.Error != nil {
187+
// if there is an error, throw and assertion so the error gets printed
188+
assert.Empty(s.T(), *taskInfo.Error)
189+
}
190+
assert.Equal(s.T(), config.TaskStatusCompleted, taskInfo.Status)
191+
}
192+
193+
func (s *UpdateTemplateDistributionsSuite) WaitOnCanceledTask(taskUUID uuid2.UUID) {
194+
taskInfo := s.waitOnTask(taskUUID)
195+
require.NotNil(s.T(), taskInfo.Error)
196+
assert.NotEmpty(s.T(), *taskInfo.Error)
197+
assert.Equal(s.T(), config.TaskStatusCanceled, taskInfo.Status)
198+
}
199+
200+
func (s *UpdateTemplateDistributionsSuite) waitOnTask(taskUUID uuid2.UUID) *models.TaskInfo {
201+
// Poll until the task is complete
202+
taskInfo, err := s.queue.Status(taskUUID)
203+
assert.NoError(s.T(), err)
204+
for {
205+
if taskInfo.Status == config.TaskStatusRunning || taskInfo.Status == config.TaskStatusPending {
206+
log.Logger.Error().Msg("SLEEPING")
207+
time.Sleep(1 * time.Second)
208+
} else {
209+
break
210+
}
211+
taskInfo, err = s.queue.Status(taskUUID)
212+
assert.NoError(s.T(), err)
213+
}
214+
return taskInfo
215+
}
216+
217+
func (s *UpdateTemplateDistributionsSuite) getRequest(url string, id identity.Identity, expectedCode int) error {
218+
client := http.Client{Transport: loggingTransport{}}
219+
req, err := http.NewRequest("GET", url, nil)
220+
if err != nil {
221+
return err
222+
}
223+
224+
js, err := json.Marshal(identity.XRHID{Identity: id})
225+
if err != nil {
226+
return err
227+
}
228+
req.Header = http.Header{}
229+
req.Header.Add(api.IdentityHeader, base64.StdEncoding.EncodeToString(js))
230+
res, err := client.Do(req)
231+
if err != nil {
232+
return err
233+
}
234+
defer res.Body.Close()
235+
assert.Equal(s.T(), expectedCode, res.StatusCode)
236+
237+
return nil
238+
}
239+
240+
func (s *UpdateTemplateDistributionsSuite) createAndSyncRepository(orgID string, url string) api.RepositoryResponse {
241+
// Setup the repository
242+
repo, err := s.dao.RepositoryConfig.Create(api.RepositoryRequest{
243+
Name: pointy.String(uuid2.NewString()),
244+
URL: pointy.String(url),
245+
AccountID: pointy.String(orgID),
246+
OrgID: pointy.String(orgID),
247+
})
248+
assert.NoError(s.T(), err)
249+
repoUuid, err := uuid2.Parse(repo.RepositoryUUID)
250+
assert.NoError(s.T(), err)
251+
252+
// Start the task
253+
s.snapshotAndWait(s.taskClient, repo, repoUuid, orgID)
254+
return repo
255+
}

0 commit comments

Comments
 (0)
Please sign in to comment.