Fixes 4932: Add search module streams

pkg/api/rpms.go

@@ -92,6 +92,30 @@ type SearchRpmResponse struct {
 	Summary     string `json:"summary"`      // Summary of the package found
 }
 
+type SearchModuleStreamsRequest struct {
+	UUIDs    []string `json:"uuids" validate:"required"`     // List of repository UUIDs to search
+	RpmNames []string `json:"rpm_names" validate:"required"` // List of rpm names to search
+	SortBy   string   `json:"sort_by"`                       // SortBy sets the sort order of the result
+	Search   string   `json:"search"`                        // Search string to search rpm names
+	Offset   int      `json:"offset"`                        // Starting point for retrieving a subset of results.
+	Limit    *int     `json:"limit,omitempty"`               // Maximum number of records to return for the search
+}
+
+type Stream struct {
+	Name string `json:"name"` // Stream name
+}
+
+type SearchModuleStreams struct {
+	ModuleName string   `json:"module_name"` // Snapshot UUIDs to find modules and streams for
+	Streams    []Stream `json:"streams"`     // Package names to filter the above list by
+}
+
+type SearchModuleStreamsCollectionResponse struct {
+	Data  []SearchModuleStreams `json:"data"`  // Requested Data
+	Meta  ResponseMetadata      `json:"meta"`  // Metadata about the request
+	Links Links                 `json:"links"` // Links to other pages of results
+}
+
 type DetectRpmsResponse struct {
 	Found   []string `json:"found"`   // List of rpm names found in given repositories
 	Missing []string `json:"missing"` // List of rpm names not found in given repositories
@@ -105,6 +129,11 @@ func (r *RepositoryRpmCollectionResponse) SetMetadata(meta ResponseMetadata, lin
 	r.Links = links
 }
 
+func (r *SearchModuleStreamsCollectionResponse) SetMetadata(meta ResponseMetadata, links Links) {
+	r.Meta = meta
+	r.Links = links
+}
+
 func (r *SnapshotErrataCollectionResponse) SetMetadata(meta ResponseMetadata, links Links) {
 	r.Meta = meta
 	r.Links = links

pkg/dao/interfaces.go

@@ -78,6 +78,7 @@ type RpmDao interface {
 	List(ctx context.Context, orgID string, uuidRepo string, limit int, offset int, search string, sortBy string) (api.RepositoryRpmCollectionResponse, int64, error)
 	Search(ctx context.Context, orgID string, request api.ContentUnitSearchRequest) ([]api.SearchRpmResponse, error)
 	SearchSnapshotRpms(ctx context.Context, orgId string, request api.SnapshotSearchRpmRequest) ([]api.SearchRpmResponse, error)
+	SearchSnapshotModuleStreams(ctx context.Context, orgID string, request api.SearchModuleStreamsRequest) (api.SearchModuleStreamsCollectionResponse, error)
 	ListSnapshotRpms(ctx context.Context, orgId string, snapshotUUIDs []string, search string, pageOpts api.PaginationData) ([]api.SnapshotRpm, int, error)
 	DetectRpms(ctx context.Context, orgID string, request api.DetectRpmsRequest) (*api.DetectRpmsResponse, error)
 	ListSnapshotErrata(ctx context.Context, orgId string, snapshotUUIDs []string, filters tangy.ErrataListFilters, pageOpts api.PaginationData) ([]api.SnapshotErrata, int, error)

pkg/dao/rpms.go

@@ -30,6 +30,85 @@ func GetRpmDao(db *gorm.DB) RpmDao {
 	}
 }
 
+func (r *rpmDaoImpl) SearchSnapshotModuleStreams(ctx context.Context, orgID string, request api.SearchModuleStreamsRequest) (api.SearchModuleStreamsCollectionResponse, error) {
+	if orgID == "" {
+		return api.SearchModuleStreamsCollectionResponse{}, fmt.Errorf("orgID can not be an empty string")
+	}
+
+	if request.Limit == nil {
+		request.Limit = utils.Ptr(100)
+	}
+
+	if request.RpmNames == nil {
+		request.RpmNames = []string{}
+	}
+
+	if request.UUIDs == nil || len(request.UUIDs) == 0 {
+		return api.SearchModuleStreamsCollectionResponse{}, &ce.DaoError{
+			BadValidation: true,
+			Message:       "must contain at least 1 snapshot UUID",
+		}
+	}
+
+	response := []api.SearchModuleStreams{}
+
+	// Check that snapshot uuids exist
+	uuidsValid, uuid := checkForValidSnapshotUuids(ctx, request.UUIDs, r.db)
+	if !uuidsValid {
+		return api.SearchModuleStreamsCollectionResponse{}, &ce.DaoError{
+			NotFound: true,
+			Message:  "Could not find snapshot with UUID: " + uuid,
+		}
+	}
+
+	pulpHrefs := []string{}
+	res := readableSnapshots(r.db.WithContext(ctx), orgID).Where("snapshots.UUID in ?", UuidifyStrings(request.UUIDs)).Pluck("version_href", &pulpHrefs)
+	if res.Error != nil {
+		return api.SearchModuleStreamsCollectionResponse{}, fmt.Errorf("failed to query the db for snapshots: %w", res.Error)
+	}
+	if config.Tang == nil {
+		return api.SearchModuleStreamsCollectionResponse{}, fmt.Errorf("no tang configuration present")
+	}
+
+	if len(pulpHrefs) == 0 {
+		return api.SearchModuleStreamsCollectionResponse{}, nil
+	}
+
+	//Start here
+	pkgs, total, err := (*config.Tang).RpmRepositoryVersionModuleStreamsList(ctx, pulpHrefs,
+		tangy.ModuleStreamListFilters{RpmNames: request.RpmNames, Search: request.Search}, tangy.PageOptions{
+			Offset: request.Offset,
+			SortBy: request.SortBy,
+			Limit:  *request.Limit,
+		})
+
+	if err != nil {
+		return api.SearchModuleStreamsCollectionResponse{}, fmt.Errorf("error querying module streams in snapshots: %w", err)
+	}
+
+	for _, pkg := range pkgs {
+		streams := []api.Stream{}
+
+		for _, stream := range pkg.Streams {
+			streams = append(streams, api.Stream{Name: stream})
+		}
+
+		response = append(response, api.SearchModuleStreams{
+			ModuleName: pkg.ModuleName,
+			Streams:    streams,
+		})
+	}
+
+	return api.SearchModuleStreamsCollectionResponse{
+		Data: response,
+		Meta: api.ResponseMetadata{
+			Count:  int64(total),
+			Offset: request.Offset,
+			Limit:  *request.Limit,
+		},
+	}, nil
+}
+
 func (r *rpmDaoImpl) List(
 	ctx context.Context,
 	orgID string,

pkg/dao/rpms_test.go

@@ -971,6 +971,70 @@ func (s *RpmSuite) TestSearchRpmsForSnapshots() {
 	assert.Error(s.T(), err)
 }
 
+func (s *RpmSuite) TestSearchModulesForSnapshots() {
+	orgId := seeds.RandomOrgId()
+	mTangy, origTangy := mockTangy(s.T())
+	defer func() { config.Tang = origTangy }()
+	ctx := context.Background()
+
+	hrefs := []string{"some_pulp_version_href"}
+	expected := []tangy.ModuleStreams{{
+		ModuleName: "Foodidly",
+		Streams:    []string{},
+	}}
+
+	// Create a repo config, and snapshot, update its version_href to expected href
+	_, err := seeds.SeedRepositoryConfigurations(s.tx, 1, seeds.SeedOptions{
+		OrgID:     orgId,
+		BatchSize: 0,
+	})
+	require.NoError(s.T(), err)
+	repoConfig := models.RepositoryConfiguration{}
+	res := s.tx.Where("org_id = ?", orgId).First(&repoConfig)
+	require.NoError(s.T(), res.Error)
+	snaps, err := seeds.SeedSnapshots(s.tx, repoConfig.UUID, 1)
+	require.NoError(s.T(), err)
+	res = s.tx.Model(&models.Snapshot{}).Where("repository_configuration_uuid = ?", repoConfig.UUID).Update("version_href", hrefs[0])
+	require.NoError(s.T(), res.Error)
+	// pulpHrefs, request.Search, *request.Limit)
+	mTangy.On("RpmRepositoryVersionModuleStreamsList", ctx, hrefs, tangy.ModuleStreamListFilters{Search: "Foo", RpmNames: []string{}}, tangy.PageOptions{Offset: 0, Limit: 55, SortBy: ""}).Return(expected, 1, nil)
+	//ctx context.Context, hrefs []string, rpmNames []string, search string, pageOpts PageOption
+	dao := GetRpmDao(s.tx)
+
+	resp, err := dao.SearchSnapshotModuleStreams(ctx, orgId, api.SearchModuleStreamsRequest{
+		UUIDs:    []string{snaps[0].UUID},
+		RpmNames: []string(nil),
+		Search:   "Foo",
+		Limit:    utils.Ptr(55),
+	})
+
+	require.NoError(s.T(), err)
+
+	assert.Equal(s.T(),
+		[]api.SearchModuleStreams{{ModuleName: expected[0].ModuleName, Streams: []api.Stream{}}},
+		resp.Data,
+	)
+
+	// ensure error returned for invalid snapshot uuid
+	_, err = dao.SearchSnapshotModuleStreams(ctx, orgId, api.SearchModuleStreamsRequest{
+		UUIDs:  []string{"blerg!"},
+		Search: "Foo",
+		Limit:  utils.Ptr(55),
+	})
+
+	assert.Error(s.T(), err)
+
+	// ensure error returned for no uuids
+	_, err = dao.SearchSnapshotModuleStreams(ctx, orgId, api.SearchModuleStreamsRequest{
+		UUIDs:    []string{},
+		RpmNames: []string{},
+		Search:   "Foo",
+		Limit:    utils.Ptr(55),
+	})
+
+	assert.Error(s.T(), err)
+}
+
 func (s *RpmSuite) TestListRpmsAndErrataForSnapshots() {
 	orgId := seeds.RandomOrgId()
 	mTangy, origTangy := mockTangy(s.T())

pkg/handler/rpms.go

@@ -26,11 +26,46 @@ func RegisterRpmRoutes(engine *echo.Group, rDao *dao.DaoRegistry) {
 	addRepoRoute(engine, http.MethodGet, "/snapshots/:uuid/rpms", rh.listSnapshotRpm, rbac.RbacVerbRead)
 	addRepoRoute(engine, http.MethodGet, "/snapshots/:uuid/errata", rh.listSnapshotErrata, rbac.RbacVerbRead)
 	addRepoRoute(engine, http.MethodPost, "/snapshots/rpms/names", rh.searchSnapshotRPMs, rbac.RbacVerbRead)
+	addRepoRoute(engine, http.MethodPost, "/snapshots/module_streams/search", rh.searchSnapshotModuleStreams, rbac.RbacVerbRead)
 	addRepoRoute(engine, http.MethodPost, "/rpms/presence", rh.detectRpmsPresence, rbac.RbacVerbRead)
 	addTemplateRoute(engine, http.MethodGet, "/templates/:uuid/rpms", rh.listTemplateRpm, rbac.RbacVerbRead)
 	addTemplateRoute(engine, http.MethodGet, "/templates/:uuid/errata", rh.listTemplateErrata, rbac.RbacVerbRead)
 }
 
+// searchSnapshotModuleStreams godoc
+// @Summary      List modules and their streams for snapshots
+// @ID           searchSnapshotModuleStreams
+// @Description  List modules and their streams for snapshots
+// @Tags         snapshots
+// @Accept       json
+// @Produce      json
+// @Param        body  body   api.SearchModuleStreamsRequest  true  "request body"
+// @Param  	     uuids	path	[]string	true	"Snapshot IDs."
+// @Param  	     package_names	path	[]string	true	"Package Names"
+// @Success      200   {object}  api.SearchModuleStreamsCollectionResponse
+// @Failure      400 {object} ce.ErrorResponse
+// @Failure      401 {object} ce.ErrorResponse
+// @Failure      404 {object} ce.ErrorResponse
+// @Failure      500 {object} ce.ErrorResponse
+// @Router       /snapshots/module_streams/search [post]
+func (rh *RpmHandler) searchSnapshotModuleStreams(c echo.Context) error {
+	_, orgId := getAccountIdOrgId(c)
+
+	dataInput := api.SearchModuleStreamsRequest{}
+
+	if err := c.Bind(&dataInput); err != nil {
+		return ce.NewErrorResponse(http.StatusBadRequest, "Error binding parameters", err.Error())
+	}
+
+	apiResponse, err := rh.Dao.Rpm.SearchSnapshotModuleStreams(c.Request().Context(), orgId, dataInput)
+
+	if err != nil {
+		return ce.NewErrorResponse(ce.HttpCodeForDaoError(err), "Error searching modules streams", err.Error())
+	}
+
+	return c.JSON(200, apiResponse)
+}
+
 // searchRpmByName godoc
 // @Summary      Search RPMs
 // @ID           searchRpm

pkg/handler/rpms_test.go

@@ -544,6 +544,97 @@ func (suite *RpmSuite) TestSearchSnapshotRpmByName() {
 	}
 }
 
+func (suite *RpmSuite) TestSearchSnapshotModuleStreams() {
+	t := suite.T()
+
+	config.Load()
+	config.Get().Features.Snapshots.Enabled = true
+	config.Get().Features.Snapshots.Accounts = &[]string{test_handler.MockAccountNumber}
+	defer resetFeatures()
+
+	type TestCaseExpected struct {
+		Code int
+		Body string
+	}
+
+	type TestCaseGiven struct {
+		Method string
+		Body   string
+	}
+
+	type TestCase struct {
+		Name     string
+		Given    TestCaseGiven
+		Expected TestCaseExpected
+	}
+
+	var testCases []TestCase = []TestCase{
+		{
+			Name: "Success scenario",
+			Given: TestCaseGiven{
+				Method: http.MethodPost,
+				Body:   `{"uuids":["abcd"],"rpm_names":[],"search":"demo","limit":50}`,
+			},
+			Expected: TestCaseExpected{
+				Code: http.StatusOK,
+				Body: "{\"data\":[],\"meta\":{\"limit\":0,\"offset\":0,\"count\":0},\"links\":{\"first\":\"\",\"last\":\"\"}}\n",
+			},
+		},
+		{
+			Name: "Evoke a StatusBadRequest response",
+			Given: TestCaseGiven{
+				Method: http.MethodPost,
+				Body:   "{",
+			},
+			Expected: TestCaseExpected{
+				Code: http.StatusBadRequest,
+				Body: "{\"errors\":[{\"status\":400,\"title\":\"Error binding parameters\",\"detail\":\"code=400, message=unexpected EOF, internal=unexpected EOF\"}]}\n",
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Log(testCase.Name)
+
+		path := fmt.Sprintf("%s/snapshots/module_streams/search", api.FullRootPath())
+		switch {
+		case testCase.Expected.Code >= 200 && testCase.Expected.Code < 300:
+			{
+				var bodyRequest api.SearchModuleStreamsRequest
+				err := json.Unmarshal([]byte(testCase.Given.Body), &bodyRequest)
+				require.NoError(t, err)
+				suite.dao.Rpm.On("SearchSnapshotModuleStreams", mock.AnythingOfType("*context.valueCtx"), test_handler.MockOrgId, bodyRequest).
+					Return(api.SearchModuleStreamsCollectionResponse{
+						Data: []api.SearchModuleStreams{},
+					}, nil)
+			}
+		default:
+			{
+			}
+		}
+
+		var bodyRequest io.Reader
+		if testCase.Given.Body == "" {
+			bodyRequest = nil
+		} else {
+			bodyRequest = strings.NewReader(testCase.Given.Body)
+		}
+
+		// Prepare request
+		req := httptest.NewRequest(testCase.Given.Method, path, bodyRequest)
+		req.Header.Set(api.IdentityHeader, test_handler.EncodedIdentity(t))
+		req.Header.Set("Content-Type", "application/json")
+
+		// Execute the request
+		code, body, err := suite.serveRpmsRouter(req)
+
+		// Check results
+		assert.Equal(t, testCase.Expected.Code, code)
+		require.NoError(t, err)
+		assert.Equal(t, testCase.Expected.Body, string(body))
+	}
+}
+
 func (suite *RpmSuite) TestListSnapshotRpms() {
 	t := suite.T()