quyun-v2/backend/app/services/media_asset_test.go

package services

import (
	"database/sql"
	"errors"
	"testing"
	"time"

	"quyun/v2/app/commands/testx"
	"quyun/v2/app/errorx"
	tenant_dto "quyun/v2/app/http/tenant/dto"
	jobs_args "quyun/v2/app/jobs/args"
	"quyun/v2/database"
	"quyun/v2/database/models"
	"quyun/v2/pkg/consts"

	. "github.com/smartystreets/goconvey/convey"
	"github.com/stretchr/testify/suite"

	_ "go.ipao.vip/atom"
	"go.ipao.vip/atom/contracts"
	"go.ipao.vip/gen/types"
	"go.uber.org/dig"
)

type MediaAssetTestSuiteInjectParams struct {
	dig.In

	DB       *sql.DB
	Initials []contracts.Initial `group:"initials"` // nolint:structcheck
}

type MediaAssetTestSuite struct {
	suite.Suite
	MediaAssetTestSuiteInjectParams
}

func Test_MediaAsset(t *testing.T) {
	providers := testx.Default().With(Provide)

	testx.Serve(providers, t, func(p MediaAssetTestSuiteInjectParams) {
		suite.Run(t, &MediaAssetTestSuite{MediaAssetTestSuiteInjectParams: p})
	})
}

func (s *MediaAssetTestSuite) Test_AdminUploadInit_VariantAndSource() {
	Convey("MediaAsset.AdminUploadInit variant/source_asset_id", s.T(), func() {
		ctx := s.T().Context()
		now := time.Now().UTC()
		tenantID := int64(1)
		userID := int64(2)

		database.Truncate(ctx, s.DB, models.TableNameMediaAsset)

		Convey("main variant 不允许 source_asset_id", func() {
			src := int64(123)
			v := consts.MediaAssetVariantMain
			_, err := MediaAsset.AdminUploadInit(ctx, tenantID, userID, &tenant_dto.AdminMediaAssetUploadInitForm{
				Type:          "video",
				Variant:       &v,
				SourceAssetID: &src,
			}, now)
			So(err, ShouldNotBeNil)
			var appErr *errorx.AppError
			So(errors.As(err, &appErr), ShouldBeTrue)
			So(appErr.Code, ShouldEqual, errorx.ErrInvalidParameter.Code)
		})

		Convey("preview variant 必须带 source_asset_id", func() {
			v := consts.MediaAssetVariantPreview
			_, err := MediaAsset.AdminUploadInit(ctx, tenantID, userID, &tenant_dto.AdminMediaAssetUploadInitForm{
				Type:    "video",
				Variant: &v,
			}, now)
			So(err, ShouldNotBeNil)
			var appErr *errorx.AppError
			So(errors.As(err, &appErr), ShouldBeTrue)
			So(appErr.Code, ShouldEqual, errorx.ErrInvalidParameter.Code)
		})

		Convey("preview variant 的 source_asset_id 必须存在且为 main variant", func() {
			src := &models.MediaAsset{
				TenantID:  tenantID,
				UserID:    userID,
				Type:      consts.MediaAssetTypeVideo,
				Status:    consts.MediaAssetStatusReady,
				Provider:  "test",
				Bucket:    "b",
				ObjectKey: "k",
				Meta:      types.JSON([]byte("{}")),
				CreatedAt: now,
				UpdatedAt: now,
			}
			So(src.Create(ctx), ShouldBeNil)

			// 将来源资源标记为 preview，模拟“来源不是 main”的非法情况。
			_, err := s.DB.ExecContext(ctx, "UPDATE media_assets SET variant = 'preview' WHERE tenant_id = $1 AND id = $2", tenantID, src.ID)
			So(err, ShouldBeNil)

			v := consts.MediaAssetVariantPreview
			_, err = MediaAsset.AdminUploadInit(ctx, tenantID, userID, &tenant_dto.AdminMediaAssetUploadInitForm{
				Type:          "video",
				Variant:       &v,
				SourceAssetID: &src.ID,
			}, now)
			So(err, ShouldNotBeNil)
			var appErr *errorx.AppError
			So(errors.As(err, &appErr), ShouldBeTrue)
			So(appErr.Code, ShouldEqual, errorx.ErrPreconditionFailed.Code)
		})
	})
}

func (s *MediaAssetTestSuite) Test_AdminUploadComplete_EnqueueAndProcess() {
	Convey("MediaAsset.AdminUploadComplete enqueue job and worker process", s.T(), func() {
		ctx := s.T().Context()
		now := time.Now().UTC()
		tenantID := int64(1)
		userID := int64(2)

		database.Truncate(ctx, s.DB, "river_job", models.TableNameMediaAsset)

		asset := &models.MediaAsset{
			TenantID:  tenantID,
			UserID:    userID,
			Type:      consts.MediaAssetTypeVideo,
			Status:    consts.MediaAssetStatusUploaded,
			Provider:  "test",
			Bucket:    "b",
			ObjectKey: "k",
			Meta:      []byte("{}"),
			CreatedAt: now,
			UpdatedAt: now,
		}
		So(asset.Create(ctx), ShouldBeNil)

		Convey("首次 upload_complete：uploaded -> processing，并入队一次", func() {
			out, err := MediaAsset.AdminUploadComplete(ctx, tenantID, userID, asset.ID, nil, now)
			So(err, ShouldBeNil)
			So(out.Status, ShouldEqual, consts.MediaAssetStatusProcessing)

			var cnt int
			err = s.DB.QueryRowContext(ctx, "SELECT COUNT(1) FROM river_job WHERE kind = $1", jobs_args.MediaAssetProcessJob{}.Kind()).Scan(&cnt)
			So(err, ShouldBeNil)
			So(cnt, ShouldEqual, 1)

			Convey("重复 upload_complete：仍可触发入队，但不会产生重复任务", func() {
				out2, err := MediaAsset.AdminUploadComplete(ctx, tenantID, userID, asset.ID, nil, now.Add(1*time.Second))
				So(err, ShouldBeNil)
				So(out2.Status, ShouldEqual, consts.MediaAssetStatusProcessing)

				err = s.DB.QueryRowContext(ctx, "SELECT COUNT(1) FROM river_job WHERE kind = $1", jobs_args.MediaAssetProcessJob{}.Kind()).Scan(&cnt)
				So(err, ShouldBeNil)
				So(cnt, ShouldEqual, 1)
			})
		})
	})
}