feat(rpc): ConnectRPC ReleaseService with streamed asset upload and download
- commit
- bfabb0c8d3f4fb919852a6475d24e5ceaf3b497e
- author
- brice · · 2026-10-21 15:02 UTC
- committer
- brice · · 2026-10-21 17:53 UTC
- parent
- b8fe0a4
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
6 files · +825 -3
M internal/adapters/rpc/api.go +9 -0
| @@ -59,6 +59,8 @@ type Config struct { | |||
| 59 | 59 | SigningKey ports.SigningKeyService | |
| 60 | 60 | // Issues serves IssueService. | |
| 61 | 61 | Issues ports.IssueService | |
| 62 | + | // Releases serves ReleaseService when set; leave unset rather than a typed nil. | |
| 63 | + | Releases ports.ReleaseService | |
| 62 | 64 | // IssueModeration serves the moderation calls of IssueService. | |
| 63 | 65 | IssueModeration ports.IssueModerationService | |
| 64 | 66 | Version string | |
| @@ -142,6 +144,9 @@ func New(cfg Config) *API { | |||
| 142 | 144 | if artifacts { | |
| 143 | 145 | features = append(features, FeatureCIArtifacts) | |
| 144 | 146 | } | |
| 147 | + | if cfg.Releases != nil { | |
| 148 | + | features = append(features, FeatureReleases) | |
| 149 | + | } | |
| 145 | 150 | if cfg.OIDCKeys != nil && cfg.OIDCKeys.Enabled() { | |
| 146 | 151 | features = append(features, FeatureCIOIDC) | |
| 147 | 152 | } | |
| @@ -162,6 +167,10 @@ func New(cfg Config) *API { | |||
| 162 | 167 | mux.Handle(ermiteconnect.NewVariableServiceHandler(&variableService{cfg.Variables}, opts...)) | |
| 163 | 168 | mux.Handle(ermiteconnect.NewEnvironmentServiceHandler(&environmentService{cfg.Environments}, opts...)) | |
| 164 | 169 | } | |
| 170 | + | if cfg.Releases != nil { | |
| 171 | + | releaseOpts := append(slices.Clone(opts), connect.WithReadMaxBytes(RunnerMaxRequestBytes)) | |
| 172 | + | mux.Handle(ermiteconnect.NewReleaseServiceHandler(&releaseService{cfg.Releases}, releaseOpts...)) | |
| 173 | + | } | |
| 165 | 174 | if cfg.Runner != nil { | |
| 166 | 175 | runnerOpts := append(slices.Clone(opts), connect.WithReadMaxBytes(RunnerMaxRequestBytes)) | |
| 167 | 176 | runners := &runnerService{runners: cfg.Runner, limiter: cfg.Limiter} | |
| @@ -59,6 +59,8 @@ type Config struct { | |||||
| 59 | SigningKey ports.SigningKeyService | 59 | SigningKey ports.SigningKeyService | ||
| 60 | // Issues serves IssueService. | 60 | // Issues serves IssueService. | ||
| 61 | Issues ports.IssueService | 61 | Issues ports.IssueService | ||
| 62 | + | // Releases serves ReleaseService when set; leave unset rather than a typed nil. | |||
| 63 | + | Releases ports.ReleaseService | |||
| 62 | // IssueModeration serves the moderation calls of IssueService. | 64 | // IssueModeration serves the moderation calls of IssueService. | ||
| 63 | IssueModeration ports.IssueModerationService | 65 | IssueModeration ports.IssueModerationService | ||
| 64 | Version string | 66 | Version string | ||
| @@ -142,6 +144,9 @@ func New(cfg Config) *API { | |||||
| 142 | if artifacts { | 144 | if artifacts { | ||
| 143 | features = append(features, FeatureCIArtifacts) | 145 | features = append(features, FeatureCIArtifacts) | ||
| 144 | } | 146 | } | ||
| 147 | + | if cfg.Releases != nil { | |||
| 148 | + | features = append(features, FeatureReleases) | |||
| 149 | + | } | |||
| 145 | if cfg.OIDCKeys != nil && cfg.OIDCKeys.Enabled() { | 150 | if cfg.OIDCKeys != nil && cfg.OIDCKeys.Enabled() { | ||
| 146 | features = append(features, FeatureCIOIDC) | 151 | features = append(features, FeatureCIOIDC) | ||
| 147 | } | 152 | } | ||
| @@ -162,6 +167,10 @@ func New(cfg Config) *API { | |||||
| 162 | mux.Handle(ermiteconnect.NewVariableServiceHandler(&variableService{cfg.Variables}, opts...)) | 167 | mux.Handle(ermiteconnect.NewVariableServiceHandler(&variableService{cfg.Variables}, opts...)) | ||
| 163 | mux.Handle(ermiteconnect.NewEnvironmentServiceHandler(&environmentService{cfg.Environments}, opts...)) | 168 | mux.Handle(ermiteconnect.NewEnvironmentServiceHandler(&environmentService{cfg.Environments}, opts...)) | ||
| 164 | } | 169 | } | ||
| 170 | + | if cfg.Releases != nil { | |||
| 171 | + | releaseOpts := append(slices.Clone(opts), connect.WithReadMaxBytes(RunnerMaxRequestBytes)) | |||
| 172 | + | mux.Handle(ermiteconnect.NewReleaseServiceHandler(&releaseService{cfg.Releases}, releaseOpts...)) | |||
| 173 | + | } | |||
| 165 | if cfg.Runner != nil { | 174 | if cfg.Runner != nil { | ||
| 166 | runnerOpts := append(slices.Clone(opts), connect.WithReadMaxBytes(RunnerMaxRequestBytes)) | 175 | runnerOpts := append(slices.Clone(opts), connect.WithReadMaxBytes(RunnerMaxRequestBytes)) | ||
| 167 | runners := &runnerService{runners: cfg.Runner, limiter: cfg.Limiter} | 176 | runners := &runnerService{runners: cfg.Runner, limiter: cfg.Limiter} | ||
M internal/adapters/rpc/convert.go +27 -3
| @@ -538,9 +538,9 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event { | |||
| 538 | 538 | out.Time = timestamp(ev.At) | |
| 539 | 539 | out.Kind = &pb.Event_MirrorRecovered{MirrorRecovered: &pb.MirrorRecovered{Repo: ev.Repo.String()}} | |
| 540 | 540 | default: | |
| 541 | - | // Web-only events, such as domain.IssueReferenced and | |
| 542 | - | // domain.IssuePendingRemoved: the API has no messages for them. | |
| 543 | - | return nil | |
| 541 | + | // Release lifecycle conversions live alongside the event service. | |
| 542 | + | // Web-only events have no representation in the API. | |
| 543 | + | return releaseEventToPB(we) | |
| 544 | 544 | } | |
| 545 | 545 | return out | |
| 546 | 546 | } | |
| @@ -567,3 +567,27 @@ func passkeysToPB(list []domain.Passkey) []*pb.Passkey { | |||
| 567 | 567 | } | |
| 568 | 568 | return out | |
| 569 | 569 | } | |
| 570 | + | ||
| 571 | + | func releaseToPB(r domain.Release) *pb.Release { | |
| 572 | + | return &pb.Release{ | |
| 573 | + | Id: r.ID, Repo: r.Repo.String(), Tag: r.Tag, Title: r.Title, Notes: r.Notes, | |
| 574 | + | Draft: r.Draft, Prerelease: r.Prerelease, TargetCommitish: r.TargetCommitish, Author: r.Author, | |
| 575 | + | CreatedAt: timestamp(r.CreatedAt), PublishedAt: timestamp(r.PublishedAt), UpdatedAt: timestamp(r.UpdatedAt), | |
| 576 | + | } | |
| 577 | + | } | |
| 578 | + | ||
| 579 | + | func assetToPB(a domain.ReleaseAsset) *pb.ReleaseAsset { | |
| 580 | + | return &pb.ReleaseAsset{ | |
| 581 | + | Id: a.ID, ReleaseId: a.ReleaseID, Name: a.Name, Label: a.Label, ContentType: a.ContentType, | |
| 582 | + | Size: a.Size, Digest: a.Digest, Uploader: a.Uploader, | |
| 583 | + | CreatedAt: timestamp(a.CreatedAt), UpdatedAt: timestamp(a.UpdatedAt), | |
| 584 | + | } | |
| 585 | + | } | |
| 586 | + | ||
| 587 | + | func assetsToPB(assets []domain.ReleaseAsset) []*pb.ReleaseAsset { | |
| 588 | + | out := make([]*pb.ReleaseAsset, 0, len(assets)) | |
| 589 | + | for _, a := range assets { | |
| 590 | + | out = append(out, assetToPB(a)) | |
| 591 | + | } | |
| 592 | + | return out | |
| 593 | + | } | |
| @@ -538,9 +538,9 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event { | |||||
| 538 | out.Time = timestamp(ev.At) | 538 | out.Time = timestamp(ev.At) | ||
| 539 | out.Kind = &pb.Event_MirrorRecovered{MirrorRecovered: &pb.MirrorRecovered{Repo: ev.Repo.String()}} | 539 | out.Kind = &pb.Event_MirrorRecovered{MirrorRecovered: &pb.MirrorRecovered{Repo: ev.Repo.String()}} | ||
| 540 | default: | 540 | default: | ||
| 541 | - | // Web-only events, such as domain.IssueReferenced and | 541 | + | // Release lifecycle conversions live alongside the event service. |
| 542 | - | // domain.IssuePendingRemoved: the API has no messages for them. | 542 | + | // Web-only events have no representation in the API. |
| 543 | - | return nil | 543 | + | return releaseEventToPB(we) |
| 544 | } | 544 | } | ||
| 545 | return out | 545 | return out | ||
| 546 | } | 546 | } | ||
| @@ -567,3 +567,27 @@ func passkeysToPB(list []domain.Passkey) []*pb.Passkey { | |||||
| 567 | } | 567 | } | ||
| 568 | return out | 568 | return out | ||
| 569 | } | 569 | } | ||
| 570 | + | ||||
| 571 | + | func releaseToPB(r domain.Release) *pb.Release { | |||
| 572 | + | return &pb.Release{ | |||
| 573 | + | Id: r.ID, Repo: r.Repo.String(), Tag: r.Tag, Title: r.Title, Notes: r.Notes, | |||
| 574 | + | Draft: r.Draft, Prerelease: r.Prerelease, TargetCommitish: r.TargetCommitish, Author: r.Author, | |||
| 575 | + | CreatedAt: timestamp(r.CreatedAt), PublishedAt: timestamp(r.PublishedAt), UpdatedAt: timestamp(r.UpdatedAt), | |||
| 576 | + | } | |||
| 577 | + | } | |||
| 578 | + | ||||
| 579 | + | func assetToPB(a domain.ReleaseAsset) *pb.ReleaseAsset { | |||
| 580 | + | return &pb.ReleaseAsset{ | |||
| 581 | + | Id: a.ID, ReleaseId: a.ReleaseID, Name: a.Name, Label: a.Label, ContentType: a.ContentType, | |||
| 582 | + | Size: a.Size, Digest: a.Digest, Uploader: a.Uploader, | |||
| 583 | + | CreatedAt: timestamp(a.CreatedAt), UpdatedAt: timestamp(a.UpdatedAt), | |||
| 584 | + | } | |||
| 585 | + | } | |||
| 586 | + | ||||
| 587 | + | func assetsToPB(assets []domain.ReleaseAsset) []*pb.ReleaseAsset { | |||
| 588 | + | out := make([]*pb.ReleaseAsset, 0, len(assets)) | |||
| 589 | + | for _, a := range assets { | |||
| 590 | + | out = append(out, assetToPB(a)) | |||
| 591 | + | } | |||
| 592 | + | return out | |||
| 593 | + | } | |||
M internal/adapters/rpc/errors.go +2 -0
| @@ -55,6 +55,8 @@ func codeOf(err error) (connect.Code, bool) { | |||
| 55 | 55 | return connect.CodeAborted, true | |
| 56 | 56 | case errors.Is(err, domain.ErrResourceExhausted): | |
| 57 | 57 | return connect.CodeResourceExhausted, true | |
| 58 | + | case errors.Is(err, domain.ErrOutOfRange): | |
| 59 | + | return connect.CodeOutOfRange, true | |
| 58 | 60 | case errors.Is(err, domain.ErrUnavailable): | |
| 59 | 61 | return connect.CodeUnavailable, true | |
| 60 | 62 | } | |
| @@ -55,6 +55,8 @@ func codeOf(err error) (connect.Code, bool) { | |||||
| 55 | return connect.CodeAborted, true | 55 | return connect.CodeAborted, true | ||
| 56 | case errors.Is(err, domain.ErrResourceExhausted): | 56 | case errors.Is(err, domain.ErrResourceExhausted): | ||
| 57 | return connect.CodeResourceExhausted, true | 57 | return connect.CodeResourceExhausted, true | ||
| 58 | + | case errors.Is(err, domain.ErrOutOfRange): | |||
| 59 | + | return connect.CodeOutOfRange, true | |||
| 58 | case errors.Is(err, domain.ErrUnavailable): | 60 | case errors.Is(err, domain.ErrUnavailable): | ||
| 59 | return connect.CodeUnavailable, true | 61 | return connect.CodeUnavailable, true | ||
| 60 | } | 62 | } | ||
M internal/adapters/rpc/event.go +26 -0
| @@ -8,6 +8,7 @@ import ( | |||
| 8 | 8 | "connectrpc.com/connect" | |
| 9 | 9 | ||
| 10 | 10 | pb "bnema.dev/ermite-api/gen/ermite" | |
| 11 | + | "bnema.dev/ermite/internal/domain" | |
| 11 | 12 | "bnema.dev/ermite/internal/ports" | |
| 12 | 13 | ) | |
| 13 | 14 | ||
| @@ -37,3 +38,28 @@ func (s *eventService) Watch(ctx context.Context, req *connect.Request[pb.WatchR | |||
| 37 | 38 | }) | |
| 38 | 39 | return connectError(ctx, err) | |
| 39 | 40 | } | |
| 41 | + | ||
| 42 | + | func releaseEventToPB(we ports.WatchEvent) *pb.Event { | |
| 43 | + | out := &pb.Event{Id: we.ID} | |
| 44 | + | switch ev := we.Event.(type) { | |
| 45 | + | case domain.ReleasePublished: | |
| 46 | + | r := ev.Release | |
| 47 | + | out.Time = timestamp(ev.At) | |
| 48 | + | out.Kind = &pb.Event_ReleasePublished{ReleasePublished: &pb.ReleasePublished{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}} | |
| 49 | + | case domain.ReleaseUpdated: | |
| 50 | + | r := ev.Release | |
| 51 | + | out.Time = timestamp(ev.At) | |
| 52 | + | out.Kind = &pb.Event_ReleaseUpdated{ReleaseUpdated: &pb.ReleaseUpdated{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}} | |
| 53 | + | case domain.ReleaseDeleted: | |
| 54 | + | r := ev.Release | |
| 55 | + | out.Time = timestamp(ev.At) | |
| 56 | + | out.Kind = &pb.Event_ReleaseDeleted{ReleaseDeleted: &pb.ReleaseDeleted{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag}} | |
| 57 | + | case domain.ReleaseUnpublished: | |
| 58 | + | r := ev.Release | |
| 59 | + | out.Time = timestamp(ev.At) | |
| 60 | + | out.Kind = &pb.Event_ReleaseUnpublished{ReleaseUnpublished: &pb.ReleaseUnpublished{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}} | |
| 61 | + | default: | |
| 62 | + | return nil | |
| 63 | + | } | |
| 64 | + | return out | |
| 65 | + | } | |
| @@ -8,6 +8,7 @@ import ( | |||||
| 8 | "connectrpc.com/connect" | 8 | "connectrpc.com/connect" | ||
| 9 | 9 | ||||
| 10 | pb "bnema.dev/ermite-api/gen/ermite" | 10 | pb "bnema.dev/ermite-api/gen/ermite" | ||
| 11 | + | "bnema.dev/ermite/internal/domain" | |||
| 11 | "bnema.dev/ermite/internal/ports" | 12 | "bnema.dev/ermite/internal/ports" | ||
| 12 | ) | 13 | ) | ||
| 13 | 14 | ||||
| @@ -37,3 +38,28 @@ func (s *eventService) Watch(ctx context.Context, req *connect.Request[pb.WatchR | |||||
| 37 | }) | 38 | }) | ||
| 38 | return connectError(ctx, err) | 39 | return connectError(ctx, err) | ||
| 39 | } | 40 | } | ||
| 41 | + | ||||
| 42 | + | func releaseEventToPB(we ports.WatchEvent) *pb.Event { | |||
| 43 | + | out := &pb.Event{Id: we.ID} | |||
| 44 | + | switch ev := we.Event.(type) { | |||
| 45 | + | case domain.ReleasePublished: | |||
| 46 | + | r := ev.Release | |||
| 47 | + | out.Time = timestamp(ev.At) | |||
| 48 | + | out.Kind = &pb.Event_ReleasePublished{ReleasePublished: &pb.ReleasePublished{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}} | |||
| 49 | + | case domain.ReleaseUpdated: | |||
| 50 | + | r := ev.Release | |||
| 51 | + | out.Time = timestamp(ev.At) | |||
| 52 | + | out.Kind = &pb.Event_ReleaseUpdated{ReleaseUpdated: &pb.ReleaseUpdated{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}} | |||
| 53 | + | case domain.ReleaseDeleted: | |||
| 54 | + | r := ev.Release | |||
| 55 | + | out.Time = timestamp(ev.At) | |||
| 56 | + | out.Kind = &pb.Event_ReleaseDeleted{ReleaseDeleted: &pb.ReleaseDeleted{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag}} | |||
| 57 | + | case domain.ReleaseUnpublished: | |||
| 58 | + | r := ev.Release | |||
| 59 | + | out.Time = timestamp(ev.At) | |||
| 60 | + | out.Kind = &pb.Event_ReleaseUnpublished{ReleaseUnpublished: &pb.ReleaseUnpublished{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}} | |||
| 61 | + | default: | |||
| 62 | + | return nil | |||
| 63 | + | } | |||
| 64 | + | return out | |||
| 65 | + | } | |||
A internal/adapters/rpc/release.go +339 -0
| @@ -0,0 +1,339 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | package rpc | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "errors" | |
| 8 | + | "fmt" | |
| 9 | + | "io" | |
| 10 | + | ||
| 11 | + | "connectrpc.com/connect" | |
| 12 | + | ||
| 13 | + | pb "bnema.dev/ermite-api/gen/ermite" | |
| 14 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |
| 15 | + | "bnema.dev/ermite/internal/domain" | |
| 16 | + | "bnema.dev/ermite/internal/ports" | |
| 17 | + | ) | |
| 18 | + | ||
| 19 | + | // FeatureReleases is listed by GetVersion when ReleaseService is served. | |
| 20 | + | const FeatureReleases = "releases" | |
| 21 | + | ||
| 22 | + | // releaseService implements ermiteconnect.ReleaseServiceHandler. | |
| 23 | + | type releaseService struct{ rel ports.ReleaseService } | |
| 24 | + | ||
| 25 | + | var _ ermiteconnect.ReleaseServiceHandler = (*releaseService)(nil) | |
| 26 | + | ||
| 27 | + | // releaseMessage fills the computed latest flag and uploaded assets using the | |
| 28 | + | // same caller's read permissions. No latest release is a normal repository state. | |
| 29 | + | func (s *releaseService) releaseMessage(ctx context.Context, repo domain.RepoName, r domain.Release) (*pb.Release, error) { | |
| 30 | + | latest, err := s.rel.GetRelease(ctx, caller(ctx), repo, ports.ReleaseRef{Latest: true}) | |
| 31 | + | if err != nil && !errors.Is(err, domain.ErrNotFound) { | |
| 32 | + | return nil, err | |
| 33 | + | } | |
| 34 | + | assets, err := s.rel.ListAssets(ctx, caller(ctx), repo, r.ID) | |
| 35 | + | if err != nil { | |
| 36 | + | return nil, err | |
| 37 | + | } | |
| 38 | + | out := releaseToPB(r) | |
| 39 | + | out.Latest = latest.ID != 0 && latest.ID == r.ID | |
| 40 | + | out.Assets = assetsToPB(assets) | |
| 41 | + | return out, nil | |
| 42 | + | } | |
| 43 | + | ||
| 44 | + | func (s *releaseService) CreateRelease(ctx context.Context, req *connect.Request[pb.CreateReleaseRequest]) (*connect.Response[pb.CreateReleaseResponse], error) { | |
| 45 | + | m := req.Msg | |
| 46 | + | repo, err := repoOf(m.GetRepo()) | |
| 47 | + | if err != nil { | |
| 48 | + | return nil, connectError(ctx, err) | |
| 49 | + | } | |
| 50 | + | r, err := s.rel.CreateRelease(ctx, caller(ctx), repo, ports.CreateReleaseParams{Tag: m.GetTag(), TargetCommitish: m.GetTargetCommitish(), Title: m.GetTitle(), Notes: m.GetNotes(), Draft: m.GetDraft(), Prerelease: m.GetPrerelease(), MakeLatest: m.GetMakeLatest()}) | |
| 51 | + | if err != nil { | |
| 52 | + | return nil, connectError(ctx, err) | |
| 53 | + | } | |
| 54 | + | out, err := s.releaseMessage(ctx, repo, r) | |
| 55 | + | if err != nil { | |
| 56 | + | return nil, connectError(ctx, err) | |
| 57 | + | } | |
| 58 | + | return connect.NewResponse(&pb.CreateReleaseResponse{Release: out}), nil | |
| 59 | + | } | |
| 60 | + | ||
| 61 | + | func releaseRef(m *pb.GetReleaseRequest) (ports.ReleaseRef, error) { | |
| 62 | + | switch v := m.GetSelector().(type) { | |
| 63 | + | case *pb.GetReleaseRequest_Id: | |
| 64 | + | if v.Id > 0 { | |
| 65 | + | return ports.ReleaseRef{ID: v.Id}, nil | |
| 66 | + | } | |
| 67 | + | case *pb.GetReleaseRequest_Tag: | |
| 68 | + | if v.Tag != "" { | |
| 69 | + | return ports.ReleaseRef{Tag: v.Tag}, nil | |
| 70 | + | } | |
| 71 | + | case *pb.GetReleaseRequest_Latest: | |
| 72 | + | if v.Latest { | |
| 73 | + | return ports.ReleaseRef{Latest: true}, nil | |
| 74 | + | } | |
| 75 | + | } | |
| 76 | + | return ports.ReleaseRef{}, fmt.Errorf("%w: select an id, tag or latest=true", domain.ErrInvalid) | |
| 77 | + | } | |
| 78 | + | ||
| 79 | + | func (s *releaseService) GetRelease(ctx context.Context, req *connect.Request[pb.GetReleaseRequest]) (*connect.Response[pb.GetReleaseResponse], error) { | |
| 80 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 81 | + | if err != nil { | |
| 82 | + | return nil, connectError(ctx, err) | |
| 83 | + | } | |
| 84 | + | ref, err := releaseRef(req.Msg) | |
| 85 | + | if err != nil { | |
| 86 | + | return nil, connectError(ctx, err) | |
| 87 | + | } | |
| 88 | + | r, err := s.rel.GetRelease(ctx, caller(ctx), repo, ref) | |
| 89 | + | if err != nil { | |
| 90 | + | return nil, connectError(ctx, err) | |
| 91 | + | } | |
| 92 | + | out, err := s.releaseMessage(ctx, repo, r) | |
| 93 | + | if err != nil { | |
| 94 | + | return nil, connectError(ctx, err) | |
| 95 | + | } | |
| 96 | + | return connect.NewResponse(&pb.GetReleaseResponse{Release: out}), nil | |
| 97 | + | } | |
| 98 | + | ||
| 99 | + | func (s *releaseService) ListReleases(ctx context.Context, req *connect.Request[pb.ListReleasesRequest]) (*connect.Response[pb.ListReleasesResponse], error) { | |
| 100 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 101 | + | if err != nil { | |
| 102 | + | return nil, connectError(ctx, err) | |
| 103 | + | } | |
| 104 | + | page, err := s.rel.ListReleases(ctx, caller(ctx), repo, int(req.Msg.GetPageSize()), req.Msg.GetPageToken()) | |
| 105 | + | if err != nil { | |
| 106 | + | return nil, connectError(ctx, err) | |
| 107 | + | } | |
| 108 | + | out := &pb.ListReleasesResponse{Releases: make([]*pb.Release, 0, len(page.Releases)), NextPageToken: page.NextPageToken} | |
| 109 | + | for _, r := range page.Releases { | |
| 110 | + | m, err := s.releaseMessage(ctx, repo, r) | |
| 111 | + | if err != nil { | |
| 112 | + | return nil, connectError(ctx, err) | |
| 113 | + | } | |
| 114 | + | out.Releases = append(out.Releases, m) | |
| 115 | + | } | |
| 116 | + | return connect.NewResponse(out), nil | |
| 117 | + | } | |
| 118 | + | ||
| 119 | + | func releaseUpdateParams(m *pb.UpdateReleaseRequest) (ports.UpdateReleaseParams, error) { | |
| 120 | + | var p ports.UpdateReleaseParams | |
| 121 | + | if len(m.GetUpdateMask().GetPaths()) == 0 { | |
| 122 | + | return p, fmt.Errorf("%w: update_mask is empty", domain.ErrInvalid) | |
| 123 | + | } | |
| 124 | + | for _, path := range m.GetUpdateMask().GetPaths() { | |
| 125 | + | switch path { | |
| 126 | + | case "tag": | |
| 127 | + | p.Tag = new(m.GetTag()) | |
| 128 | + | case "target_commitish": | |
| 129 | + | p.TargetCommitish = new(m.GetTargetCommitish()) | |
| 130 | + | case "title": | |
| 131 | + | p.Title = new(m.GetTitle()) | |
| 132 | + | case "notes": | |
| 133 | + | p.Notes = new(m.GetNotes()) | |
| 134 | + | case "draft": | |
| 135 | + | p.Draft = new(m.GetDraft()) | |
| 136 | + | case "prerelease": | |
| 137 | + | p.Prerelease = new(m.GetPrerelease()) | |
| 138 | + | case "make_latest": | |
| 139 | + | p.MakeLatest = new(m.GetMakeLatest()) | |
| 140 | + | default: | |
| 141 | + | return p, fmt.Errorf("%w: unknown update_mask path %q", domain.ErrInvalid, path) | |
| 142 | + | } | |
| 143 | + | } | |
| 144 | + | return p, nil | |
| 145 | + | } | |
| 146 | + | ||
| 147 | + | func (s *releaseService) UpdateRelease(ctx context.Context, req *connect.Request[pb.UpdateReleaseRequest]) (*connect.Response[pb.UpdateReleaseResponse], error) { | |
| 148 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 149 | + | if err != nil { | |
| 150 | + | return nil, connectError(ctx, err) | |
| 151 | + | } | |
| 152 | + | p, err := releaseUpdateParams(req.Msg) | |
| 153 | + | if err != nil { | |
| 154 | + | return nil, connectError(ctx, err) | |
| 155 | + | } | |
| 156 | + | r, err := s.rel.UpdateRelease(ctx, caller(ctx), repo, req.Msg.GetId(), p) | |
| 157 | + | if err != nil { | |
| 158 | + | return nil, connectError(ctx, err) | |
| 159 | + | } | |
| 160 | + | out, err := s.releaseMessage(ctx, repo, r) | |
| 161 | + | if err != nil { | |
| 162 | + | return nil, connectError(ctx, err) | |
| 163 | + | } | |
| 164 | + | return connect.NewResponse(&pb.UpdateReleaseResponse{Release: out}), nil | |
| 165 | + | } | |
| 166 | + | ||
| 167 | + | func (s *releaseService) DeleteRelease(ctx context.Context, req *connect.Request[pb.DeleteReleaseRequest]) (*connect.Response[pb.DeleteReleaseResponse], error) { | |
| 168 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 169 | + | if err != nil { | |
| 170 | + | return nil, connectError(ctx, err) | |
| 171 | + | } | |
| 172 | + | if err := s.rel.DeleteRelease(ctx, caller(ctx), repo, req.Msg.GetId()); err != nil { | |
| 173 | + | return nil, connectError(ctx, err) | |
| 174 | + | } | |
| 175 | + | return connect.NewResponse(&pb.DeleteReleaseResponse{}), nil | |
| 176 | + | } | |
| 177 | + | ||
| 178 | + | // chunkReader retains at most one message; metadata belongs to the first only. | |
| 179 | + | type chunkReader struct { | |
| 180 | + | stream *connect.ClientStream[pb.UploadReleaseAssetRequest] | |
| 181 | + | buf []byte | |
| 182 | + | } | |
| 183 | + | ||
| 184 | + | func (r *chunkReader) Read(p []byte) (int, error) { | |
| 185 | + | if len(p) == 0 { | |
| 186 | + | return 0, nil | |
| 187 | + | } | |
| 188 | + | for len(r.buf) == 0 { | |
| 189 | + | if !r.stream.Receive() { | |
| 190 | + | if err := r.stream.Err(); err != nil { | |
| 191 | + | return 0, err | |
| 192 | + | } | |
| 193 | + | return 0, io.EOF | |
| 194 | + | } | |
| 195 | + | m := r.stream.Msg() | |
| 196 | + | if m.GetRepo() != "" || m.GetReleaseId() != 0 || m.GetName() != "" || m.GetLabel() != "" || m.GetContentType() != "" || m.GetSize() != 0 || m.GetReplace() { | |
| 197 | + | return 0, fmt.Errorf("%w: upload metadata is in the first message only", domain.ErrInvalid) | |
| 198 | + | } | |
| 199 | + | if err := checkBlockData(m.GetData()); err != nil { | |
| 200 | + | return 0, err | |
| 201 | + | } | |
| 202 | + | r.buf = m.GetData() | |
| 203 | + | } | |
| 204 | + | n := copy(p, r.buf) | |
| 205 | + | r.buf = r.buf[n:] | |
| 206 | + | return n, nil | |
| 207 | + | } | |
| 208 | + | ||
| 209 | + | func (s *releaseService) UploadReleaseAsset(ctx context.Context, stream *connect.ClientStream[pb.UploadReleaseAssetRequest]) (*connect.Response[pb.UploadReleaseAssetResponse], error) { | |
| 210 | + | if !stream.Receive() { | |
| 211 | + | err := stream.Err() | |
| 212 | + | if err == nil { | |
| 213 | + | err = fmt.Errorf("%w: the stream has no message", domain.ErrInvalid) | |
| 214 | + | } | |
| 215 | + | return nil, connectError(ctx, err) | |
| 216 | + | } | |
| 217 | + | m := stream.Msg() | |
| 218 | + | repo, err := repoOf(m.GetRepo()) | |
| 219 | + | if err != nil { | |
| 220 | + | return nil, connectError(ctx, err) | |
| 221 | + | } | |
| 222 | + | if m.GetSize() < 0 { | |
| 223 | + | return nil, connectError(ctx, fmt.Errorf("%w: size must be nonnegative", domain.ErrInvalid)) | |
| 224 | + | } | |
| 225 | + | if err := checkBlockData(m.GetData()); err != nil { | |
| 226 | + | return nil, connectError(ctx, err) | |
| 227 | + | } | |
| 228 | + | a, err := s.rel.UploadAsset(ctx, caller(ctx), repo, ports.UploadAssetParams{ReleaseID: m.GetReleaseId(), Name: m.GetName(), Label: m.GetLabel(), ContentType: m.GetContentType(), Size: m.GetSize(), Replace: m.GetReplace()}, &chunkReader{stream: stream, buf: m.GetData()}) | |
| 229 | + | if err != nil { | |
| 230 | + | return nil, connectError(ctx, err) | |
| 231 | + | } | |
| 232 | + | return connect.NewResponse(&pb.UploadReleaseAssetResponse{Asset: assetToPB(a)}), nil | |
| 233 | + | } | |
| 234 | + | ||
| 235 | + | func assetUpdateParams(m *pb.UpdateReleaseAssetRequest) (ports.UpdateAssetParams, error) { | |
| 236 | + | var p ports.UpdateAssetParams | |
| 237 | + | if len(m.GetUpdateMask().GetPaths()) == 0 { | |
| 238 | + | return p, fmt.Errorf("%w: update_mask is empty", domain.ErrInvalid) | |
| 239 | + | } | |
| 240 | + | for _, path := range m.GetUpdateMask().GetPaths() { | |
| 241 | + | switch path { | |
| 242 | + | case "name": | |
| 243 | + | p.Name = new(m.GetName()) | |
| 244 | + | case "label": | |
| 245 | + | p.Label = new(m.GetLabel()) | |
| 246 | + | default: | |
| 247 | + | return p, fmt.Errorf("%w: unknown update_mask path %q", domain.ErrInvalid, path) | |
| 248 | + | } | |
| 249 | + | } | |
| 250 | + | return p, nil | |
| 251 | + | } | |
| 252 | + | ||
| 253 | + | func (s *releaseService) UpdateReleaseAsset(ctx context.Context, req *connect.Request[pb.UpdateReleaseAssetRequest]) (*connect.Response[pb.UpdateReleaseAssetResponse], error) { | |
| 254 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 255 | + | if err != nil { | |
| 256 | + | return nil, connectError(ctx, err) | |
| 257 | + | } | |
| 258 | + | p, err := assetUpdateParams(req.Msg) | |
| 259 | + | if err != nil { | |
| 260 | + | return nil, connectError(ctx, err) | |
| 261 | + | } | |
| 262 | + | a, err := s.rel.UpdateAsset(ctx, caller(ctx), repo, req.Msg.GetAssetId(), p) | |
| 263 | + | if err != nil { | |
| 264 | + | return nil, connectError(ctx, err) | |
| 265 | + | } | |
| 266 | + | return connect.NewResponse(&pb.UpdateReleaseAssetResponse{Asset: assetToPB(a)}), nil | |
| 267 | + | } | |
| 268 | + | ||
| 269 | + | func (s *releaseService) DeleteReleaseAsset(ctx context.Context, req *connect.Request[pb.DeleteReleaseAssetRequest]) (*connect.Response[pb.DeleteReleaseAssetResponse], error) { | |
| 270 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 271 | + | if err != nil { | |
| 272 | + | return nil, connectError(ctx, err) | |
| 273 | + | } | |
| 274 | + | if err := s.rel.DeleteAsset(ctx, caller(ctx), repo, req.Msg.GetAssetId()); err != nil { | |
| 275 | + | return nil, connectError(ctx, err) | |
| 276 | + | } | |
| 277 | + | return connect.NewResponse(&pb.DeleteReleaseAssetResponse{}), nil | |
| 278 | + | } | |
| 279 | + | ||
| 280 | + | func assetRef(m *pb.DownloadReleaseAssetRequest) (ports.AssetRef, error) { | |
| 281 | + | switch v := m.GetSelector().(type) { | |
| 282 | + | case *pb.DownloadReleaseAssetRequest_AssetId: | |
| 283 | + | if v.AssetId > 0 { | |
| 284 | + | return ports.AssetRef{ID: v.AssetId}, nil | |
| 285 | + | } | |
| 286 | + | case *pb.DownloadReleaseAssetRequest_Tag: | |
| 287 | + | if v.Tag.GetTag() != "" && v.Tag.GetName() != "" { | |
| 288 | + | return ports.AssetRef{Tag: v.Tag.GetTag(), Name: v.Tag.GetName()}, nil | |
| 289 | + | } | |
| 290 | + | case *pb.DownloadReleaseAssetRequest_Latest: | |
| 291 | + | if v.Latest.GetLatest() && v.Latest.GetName() != "" { | |
| 292 | + | return ports.AssetRef{Latest: true, Name: v.Latest.GetName()}, nil | |
| 293 | + | } | |
| 294 | + | } | |
| 295 | + | return ports.AssetRef{}, fmt.Errorf("%w: select an asset id, tag/name or latest=true/name", domain.ErrInvalid) | |
| 296 | + | } | |
| 297 | + | ||
| 298 | + | func (s *releaseService) DownloadReleaseAsset(ctx context.Context, req *connect.Request[pb.DownloadReleaseAssetRequest], stream *connect.ServerStream[pb.DownloadReleaseAssetResponse]) error { | |
| 299 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 300 | + | if err != nil { | |
| 301 | + | return connectError(ctx, err) | |
| 302 | + | } | |
| 303 | + | ref, err := assetRef(req.Msg) | |
| 304 | + | if err != nil { | |
| 305 | + | return connectError(ctx, err) | |
| 306 | + | } | |
| 307 | + | err = s.rel.DownloadAsset(ctx, caller(ctx), repo, ref, req.Msg.GetOffset(), -1, func(c ports.AssetChunk) error { | |
| 308 | + | m := &pb.DownloadReleaseAssetResponse{Chunk: c.Data} | |
| 309 | + | if c.Asset != nil { | |
| 310 | + | m.Asset = assetToPB(*c.Asset) | |
| 311 | + | } | |
| 312 | + | return stream.Send(m) | |
| 313 | + | }) | |
| 314 | + | return connectError(ctx, err) | |
| 315 | + | } | |
| 316 | + | ||
| 317 | + | func (s *releaseService) ListReleaseAssets(ctx context.Context, req *connect.Request[pb.ListReleaseAssetsRequest]) (*connect.Response[pb.ListReleaseAssetsResponse], error) { | |
| 318 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 319 | + | if err != nil { | |
| 320 | + | return nil, connectError(ctx, err) | |
| 321 | + | } | |
| 322 | + | assets, err := s.rel.ListAssets(ctx, caller(ctx), repo, req.Msg.GetReleaseId()) | |
| 323 | + | if err != nil { | |
| 324 | + | return nil, connectError(ctx, err) | |
| 325 | + | } | |
| 326 | + | return connect.NewResponse(&pb.ListReleaseAssetsResponse{Assets: assetsToPB(assets)}), nil | |
| 327 | + | } | |
| 328 | + | ||
| 329 | + | func (s *releaseService) GetReleaseUsage(ctx context.Context, req *connect.Request[pb.GetReleaseUsageRequest]) (*connect.Response[pb.GetReleaseUsageResponse], error) { | |
| 330 | + | repo, err := repoOf(req.Msg.GetRepo()) | |
| 331 | + | if err != nil { | |
| 332 | + | return nil, connectError(ctx, err) | |
| 333 | + | } | |
| 334 | + | u, err := s.rel.Usage(ctx, caller(ctx), repo) | |
| 335 | + | if err != nil { | |
| 336 | + | return nil, connectError(ctx, err) | |
| 337 | + | } | |
| 338 | + | return connect.NewResponse(&pb.GetReleaseUsageResponse{UsedBytes: u.Used, QuotaBytes: u.Limits.Quota, AssetMaxBytes: u.Limits.AssetMax, AssetsMax: int32(u.Limits.AssetsMax)}), nil | |
| 339 | + | } | |
| @@ -0,0 +1,339 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | package rpc | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "errors" | |||
| 8 | + | "fmt" | |||
| 9 | + | "io" | |||
| 10 | + | ||||
| 11 | + | "connectrpc.com/connect" | |||
| 12 | + | ||||
| 13 | + | pb "bnema.dev/ermite-api/gen/ermite" | |||
| 14 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |||
| 15 | + | "bnema.dev/ermite/internal/domain" | |||
| 16 | + | "bnema.dev/ermite/internal/ports" | |||
| 17 | + | ) | |||
| 18 | + | ||||
| 19 | + | // FeatureReleases is listed by GetVersion when ReleaseService is served. | |||
| 20 | + | const FeatureReleases = "releases" | |||
| 21 | + | ||||
| 22 | + | // releaseService implements ermiteconnect.ReleaseServiceHandler. | |||
| 23 | + | type releaseService struct{ rel ports.ReleaseService } | |||
| 24 | + | ||||
| 25 | + | var _ ermiteconnect.ReleaseServiceHandler = (*releaseService)(nil) | |||
| 26 | + | ||||
| 27 | + | // releaseMessage fills the computed latest flag and uploaded assets using the | |||
| 28 | + | // same caller's read permissions. No latest release is a normal repository state. | |||
| 29 | + | func (s *releaseService) releaseMessage(ctx context.Context, repo domain.RepoName, r domain.Release) (*pb.Release, error) { | |||
| 30 | + | latest, err := s.rel.GetRelease(ctx, caller(ctx), repo, ports.ReleaseRef{Latest: true}) | |||
| 31 | + | if err != nil && !errors.Is(err, domain.ErrNotFound) { | |||
| 32 | + | return nil, err | |||
| 33 | + | } | |||
| 34 | + | assets, err := s.rel.ListAssets(ctx, caller(ctx), repo, r.ID) | |||
| 35 | + | if err != nil { | |||
| 36 | + | return nil, err | |||
| 37 | + | } | |||
| 38 | + | out := releaseToPB(r) | |||
| 39 | + | out.Latest = latest.ID != 0 && latest.ID == r.ID | |||
| 40 | + | out.Assets = assetsToPB(assets) | |||
| 41 | + | return out, nil | |||
| 42 | + | } | |||
| 43 | + | ||||
| 44 | + | func (s *releaseService) CreateRelease(ctx context.Context, req *connect.Request[pb.CreateReleaseRequest]) (*connect.Response[pb.CreateReleaseResponse], error) { | |||
| 45 | + | m := req.Msg | |||
| 46 | + | repo, err := repoOf(m.GetRepo()) | |||
| 47 | + | if err != nil { | |||
| 48 | + | return nil, connectError(ctx, err) | |||
| 49 | + | } | |||
| 50 | + | r, err := s.rel.CreateRelease(ctx, caller(ctx), repo, ports.CreateReleaseParams{Tag: m.GetTag(), TargetCommitish: m.GetTargetCommitish(), Title: m.GetTitle(), Notes: m.GetNotes(), Draft: m.GetDraft(), Prerelease: m.GetPrerelease(), MakeLatest: m.GetMakeLatest()}) | |||
| 51 | + | if err != nil { | |||
| 52 | + | return nil, connectError(ctx, err) | |||
| 53 | + | } | |||
| 54 | + | out, err := s.releaseMessage(ctx, repo, r) | |||
| 55 | + | if err != nil { | |||
| 56 | + | return nil, connectError(ctx, err) | |||
| 57 | + | } | |||
| 58 | + | return connect.NewResponse(&pb.CreateReleaseResponse{Release: out}), nil | |||
| 59 | + | } | |||
| 60 | + | ||||
| 61 | + | func releaseRef(m *pb.GetReleaseRequest) (ports.ReleaseRef, error) { | |||
| 62 | + | switch v := m.GetSelector().(type) { | |||
| 63 | + | case *pb.GetReleaseRequest_Id: | |||
| 64 | + | if v.Id > 0 { | |||
| 65 | + | return ports.ReleaseRef{ID: v.Id}, nil | |||
| 66 | + | } | |||
| 67 | + | case *pb.GetReleaseRequest_Tag: | |||
| 68 | + | if v.Tag != "" { | |||
| 69 | + | return ports.ReleaseRef{Tag: v.Tag}, nil | |||
| 70 | + | } | |||
| 71 | + | case *pb.GetReleaseRequest_Latest: | |||
| 72 | + | if v.Latest { | |||
| 73 | + | return ports.ReleaseRef{Latest: true}, nil | |||
| 74 | + | } | |||
| 75 | + | } | |||
| 76 | + | return ports.ReleaseRef{}, fmt.Errorf("%w: select an id, tag or latest=true", domain.ErrInvalid) | |||
| 77 | + | } | |||
| 78 | + | ||||
| 79 | + | func (s *releaseService) GetRelease(ctx context.Context, req *connect.Request[pb.GetReleaseRequest]) (*connect.Response[pb.GetReleaseResponse], error) { | |||
| 80 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 81 | + | if err != nil { | |||
| 82 | + | return nil, connectError(ctx, err) | |||
| 83 | + | } | |||
| 84 | + | ref, err := releaseRef(req.Msg) | |||
| 85 | + | if err != nil { | |||
| 86 | + | return nil, connectError(ctx, err) | |||
| 87 | + | } | |||
| 88 | + | r, err := s.rel.GetRelease(ctx, caller(ctx), repo, ref) | |||
| 89 | + | if err != nil { | |||
| 90 | + | return nil, connectError(ctx, err) | |||
| 91 | + | } | |||
| 92 | + | out, err := s.releaseMessage(ctx, repo, r) | |||
| 93 | + | if err != nil { | |||
| 94 | + | return nil, connectError(ctx, err) | |||
| 95 | + | } | |||
| 96 | + | return connect.NewResponse(&pb.GetReleaseResponse{Release: out}), nil | |||
| 97 | + | } | |||
| 98 | + | ||||
| 99 | + | func (s *releaseService) ListReleases(ctx context.Context, req *connect.Request[pb.ListReleasesRequest]) (*connect.Response[pb.ListReleasesResponse], error) { | |||
| 100 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 101 | + | if err != nil { | |||
| 102 | + | return nil, connectError(ctx, err) | |||
| 103 | + | } | |||
| 104 | + | page, err := s.rel.ListReleases(ctx, caller(ctx), repo, int(req.Msg.GetPageSize()), req.Msg.GetPageToken()) | |||
| 105 | + | if err != nil { | |||
| 106 | + | return nil, connectError(ctx, err) | |||
| 107 | + | } | |||
| 108 | + | out := &pb.ListReleasesResponse{Releases: make([]*pb.Release, 0, len(page.Releases)), NextPageToken: page.NextPageToken} | |||
| 109 | + | for _, r := range page.Releases { | |||
| 110 | + | m, err := s.releaseMessage(ctx, repo, r) | |||
| 111 | + | if err != nil { | |||
| 112 | + | return nil, connectError(ctx, err) | |||
| 113 | + | } | |||
| 114 | + | out.Releases = append(out.Releases, m) | |||
| 115 | + | } | |||
| 116 | + | return connect.NewResponse(out), nil | |||
| 117 | + | } | |||
| 118 | + | ||||
| 119 | + | func releaseUpdateParams(m *pb.UpdateReleaseRequest) (ports.UpdateReleaseParams, error) { | |||
| 120 | + | var p ports.UpdateReleaseParams | |||
| 121 | + | if len(m.GetUpdateMask().GetPaths()) == 0 { | |||
| 122 | + | return p, fmt.Errorf("%w: update_mask is empty", domain.ErrInvalid) | |||
| 123 | + | } | |||
| 124 | + | for _, path := range m.GetUpdateMask().GetPaths() { | |||
| 125 | + | switch path { | |||
| 126 | + | case "tag": | |||
| 127 | + | p.Tag = new(m.GetTag()) | |||
| 128 | + | case "target_commitish": | |||
| 129 | + | p.TargetCommitish = new(m.GetTargetCommitish()) | |||
| 130 | + | case "title": | |||
| 131 | + | p.Title = new(m.GetTitle()) | |||
| 132 | + | case "notes": | |||
| 133 | + | p.Notes = new(m.GetNotes()) | |||
| 134 | + | case "draft": | |||
| 135 | + | p.Draft = new(m.GetDraft()) | |||
| 136 | + | case "prerelease": | |||
| 137 | + | p.Prerelease = new(m.GetPrerelease()) | |||
| 138 | + | case "make_latest": | |||
| 139 | + | p.MakeLatest = new(m.GetMakeLatest()) | |||
| 140 | + | default: | |||
| 141 | + | return p, fmt.Errorf("%w: unknown update_mask path %q", domain.ErrInvalid, path) | |||
| 142 | + | } | |||
| 143 | + | } | |||
| 144 | + | return p, nil | |||
| 145 | + | } | |||
| 146 | + | ||||
| 147 | + | func (s *releaseService) UpdateRelease(ctx context.Context, req *connect.Request[pb.UpdateReleaseRequest]) (*connect.Response[pb.UpdateReleaseResponse], error) { | |||
| 148 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 149 | + | if err != nil { | |||
| 150 | + | return nil, connectError(ctx, err) | |||
| 151 | + | } | |||
| 152 | + | p, err := releaseUpdateParams(req.Msg) | |||
| 153 | + | if err != nil { | |||
| 154 | + | return nil, connectError(ctx, err) | |||
| 155 | + | } | |||
| 156 | + | r, err := s.rel.UpdateRelease(ctx, caller(ctx), repo, req.Msg.GetId(), p) | |||
| 157 | + | if err != nil { | |||
| 158 | + | return nil, connectError(ctx, err) | |||
| 159 | + | } | |||
| 160 | + | out, err := s.releaseMessage(ctx, repo, r) | |||
| 161 | + | if err != nil { | |||
| 162 | + | return nil, connectError(ctx, err) | |||
| 163 | + | } | |||
| 164 | + | return connect.NewResponse(&pb.UpdateReleaseResponse{Release: out}), nil | |||
| 165 | + | } | |||
| 166 | + | ||||
| 167 | + | func (s *releaseService) DeleteRelease(ctx context.Context, req *connect.Request[pb.DeleteReleaseRequest]) (*connect.Response[pb.DeleteReleaseResponse], error) { | |||
| 168 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 169 | + | if err != nil { | |||
| 170 | + | return nil, connectError(ctx, err) | |||
| 171 | + | } | |||
| 172 | + | if err := s.rel.DeleteRelease(ctx, caller(ctx), repo, req.Msg.GetId()); err != nil { | |||
| 173 | + | return nil, connectError(ctx, err) | |||
| 174 | + | } | |||
| 175 | + | return connect.NewResponse(&pb.DeleteReleaseResponse{}), nil | |||
| 176 | + | } | |||
| 177 | + | ||||
| 178 | + | // chunkReader retains at most one message; metadata belongs to the first only. | |||
| 179 | + | type chunkReader struct { | |||
| 180 | + | stream *connect.ClientStream[pb.UploadReleaseAssetRequest] | |||
| 181 | + | buf []byte | |||
| 182 | + | } | |||
| 183 | + | ||||
| 184 | + | func (r *chunkReader) Read(p []byte) (int, error) { | |||
| 185 | + | if len(p) == 0 { | |||
| 186 | + | return 0, nil | |||
| 187 | + | } | |||
| 188 | + | for len(r.buf) == 0 { | |||
| 189 | + | if !r.stream.Receive() { | |||
| 190 | + | if err := r.stream.Err(); err != nil { | |||
| 191 | + | return 0, err | |||
| 192 | + | } | |||
| 193 | + | return 0, io.EOF | |||
| 194 | + | } | |||
| 195 | + | m := r.stream.Msg() | |||
| 196 | + | if m.GetRepo() != "" || m.GetReleaseId() != 0 || m.GetName() != "" || m.GetLabel() != "" || m.GetContentType() != "" || m.GetSize() != 0 || m.GetReplace() { | |||
| 197 | + | return 0, fmt.Errorf("%w: upload metadata is in the first message only", domain.ErrInvalid) | |||
| 198 | + | } | |||
| 199 | + | if err := checkBlockData(m.GetData()); err != nil { | |||
| 200 | + | return 0, err | |||
| 201 | + | } | |||
| 202 | + | r.buf = m.GetData() | |||
| 203 | + | } | |||
| 204 | + | n := copy(p, r.buf) | |||
| 205 | + | r.buf = r.buf[n:] | |||
| 206 | + | return n, nil | |||
| 207 | + | } | |||
| 208 | + | ||||
| 209 | + | func (s *releaseService) UploadReleaseAsset(ctx context.Context, stream *connect.ClientStream[pb.UploadReleaseAssetRequest]) (*connect.Response[pb.UploadReleaseAssetResponse], error) { | |||
| 210 | + | if !stream.Receive() { | |||
| 211 | + | err := stream.Err() | |||
| 212 | + | if err == nil { | |||
| 213 | + | err = fmt.Errorf("%w: the stream has no message", domain.ErrInvalid) | |||
| 214 | + | } | |||
| 215 | + | return nil, connectError(ctx, err) | |||
| 216 | + | } | |||
| 217 | + | m := stream.Msg() | |||
| 218 | + | repo, err := repoOf(m.GetRepo()) | |||
| 219 | + | if err != nil { | |||
| 220 | + | return nil, connectError(ctx, err) | |||
| 221 | + | } | |||
| 222 | + | if m.GetSize() < 0 { | |||
| 223 | + | return nil, connectError(ctx, fmt.Errorf("%w: size must be nonnegative", domain.ErrInvalid)) | |||
| 224 | + | } | |||
| 225 | + | if err := checkBlockData(m.GetData()); err != nil { | |||
| 226 | + | return nil, connectError(ctx, err) | |||
| 227 | + | } | |||
| 228 | + | a, err := s.rel.UploadAsset(ctx, caller(ctx), repo, ports.UploadAssetParams{ReleaseID: m.GetReleaseId(), Name: m.GetName(), Label: m.GetLabel(), ContentType: m.GetContentType(), Size: m.GetSize(), Replace: m.GetReplace()}, &chunkReader{stream: stream, buf: m.GetData()}) | |||
| 229 | + | if err != nil { | |||
| 230 | + | return nil, connectError(ctx, err) | |||
| 231 | + | } | |||
| 232 | + | return connect.NewResponse(&pb.UploadReleaseAssetResponse{Asset: assetToPB(a)}), nil | |||
| 233 | + | } | |||
| 234 | + | ||||
| 235 | + | func assetUpdateParams(m *pb.UpdateReleaseAssetRequest) (ports.UpdateAssetParams, error) { | |||
| 236 | + | var p ports.UpdateAssetParams | |||
| 237 | + | if len(m.GetUpdateMask().GetPaths()) == 0 { | |||
| 238 | + | return p, fmt.Errorf("%w: update_mask is empty", domain.ErrInvalid) | |||
| 239 | + | } | |||
| 240 | + | for _, path := range m.GetUpdateMask().GetPaths() { | |||
| 241 | + | switch path { | |||
| 242 | + | case "name": | |||
| 243 | + | p.Name = new(m.GetName()) | |||
| 244 | + | case "label": | |||
| 245 | + | p.Label = new(m.GetLabel()) | |||
| 246 | + | default: | |||
| 247 | + | return p, fmt.Errorf("%w: unknown update_mask path %q", domain.ErrInvalid, path) | |||
| 248 | + | } | |||
| 249 | + | } | |||
| 250 | + | return p, nil | |||
| 251 | + | } | |||
| 252 | + | ||||
| 253 | + | func (s *releaseService) UpdateReleaseAsset(ctx context.Context, req *connect.Request[pb.UpdateReleaseAssetRequest]) (*connect.Response[pb.UpdateReleaseAssetResponse], error) { | |||
| 254 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 255 | + | if err != nil { | |||
| 256 | + | return nil, connectError(ctx, err) | |||
| 257 | + | } | |||
| 258 | + | p, err := assetUpdateParams(req.Msg) | |||
| 259 | + | if err != nil { | |||
| 260 | + | return nil, connectError(ctx, err) | |||
| 261 | + | } | |||
| 262 | + | a, err := s.rel.UpdateAsset(ctx, caller(ctx), repo, req.Msg.GetAssetId(), p) | |||
| 263 | + | if err != nil { | |||
| 264 | + | return nil, connectError(ctx, err) | |||
| 265 | + | } | |||
| 266 | + | return connect.NewResponse(&pb.UpdateReleaseAssetResponse{Asset: assetToPB(a)}), nil | |||
| 267 | + | } | |||
| 268 | + | ||||
| 269 | + | func (s *releaseService) DeleteReleaseAsset(ctx context.Context, req *connect.Request[pb.DeleteReleaseAssetRequest]) (*connect.Response[pb.DeleteReleaseAssetResponse], error) { | |||
| 270 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 271 | + | if err != nil { | |||
| 272 | + | return nil, connectError(ctx, err) | |||
| 273 | + | } | |||
| 274 | + | if err := s.rel.DeleteAsset(ctx, caller(ctx), repo, req.Msg.GetAssetId()); err != nil { | |||
| 275 | + | return nil, connectError(ctx, err) | |||
| 276 | + | } | |||
| 277 | + | return connect.NewResponse(&pb.DeleteReleaseAssetResponse{}), nil | |||
| 278 | + | } | |||
| 279 | + | ||||
| 280 | + | func assetRef(m *pb.DownloadReleaseAssetRequest) (ports.AssetRef, error) { | |||
| 281 | + | switch v := m.GetSelector().(type) { | |||
| 282 | + | case *pb.DownloadReleaseAssetRequest_AssetId: | |||
| 283 | + | if v.AssetId > 0 { | |||
| 284 | + | return ports.AssetRef{ID: v.AssetId}, nil | |||
| 285 | + | } | |||
| 286 | + | case *pb.DownloadReleaseAssetRequest_Tag: | |||
| 287 | + | if v.Tag.GetTag() != "" && v.Tag.GetName() != "" { | |||
| 288 | + | return ports.AssetRef{Tag: v.Tag.GetTag(), Name: v.Tag.GetName()}, nil | |||
| 289 | + | } | |||
| 290 | + | case *pb.DownloadReleaseAssetRequest_Latest: | |||
| 291 | + | if v.Latest.GetLatest() && v.Latest.GetName() != "" { | |||
| 292 | + | return ports.AssetRef{Latest: true, Name: v.Latest.GetName()}, nil | |||
| 293 | + | } | |||
| 294 | + | } | |||
| 295 | + | return ports.AssetRef{}, fmt.Errorf("%w: select an asset id, tag/name or latest=true/name", domain.ErrInvalid) | |||
| 296 | + | } | |||
| 297 | + | ||||
| 298 | + | func (s *releaseService) DownloadReleaseAsset(ctx context.Context, req *connect.Request[pb.DownloadReleaseAssetRequest], stream *connect.ServerStream[pb.DownloadReleaseAssetResponse]) error { | |||
| 299 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 300 | + | if err != nil { | |||
| 301 | + | return connectError(ctx, err) | |||
| 302 | + | } | |||
| 303 | + | ref, err := assetRef(req.Msg) | |||
| 304 | + | if err != nil { | |||
| 305 | + | return connectError(ctx, err) | |||
| 306 | + | } | |||
| 307 | + | err = s.rel.DownloadAsset(ctx, caller(ctx), repo, ref, req.Msg.GetOffset(), -1, func(c ports.AssetChunk) error { | |||
| 308 | + | m := &pb.DownloadReleaseAssetResponse{Chunk: c.Data} | |||
| 309 | + | if c.Asset != nil { | |||
| 310 | + | m.Asset = assetToPB(*c.Asset) | |||
| 311 | + | } | |||
| 312 | + | return stream.Send(m) | |||
| 313 | + | }) | |||
| 314 | + | return connectError(ctx, err) | |||
| 315 | + | } | |||
| 316 | + | ||||
| 317 | + | func (s *releaseService) ListReleaseAssets(ctx context.Context, req *connect.Request[pb.ListReleaseAssetsRequest]) (*connect.Response[pb.ListReleaseAssetsResponse], error) { | |||
| 318 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 319 | + | if err != nil { | |||
| 320 | + | return nil, connectError(ctx, err) | |||
| 321 | + | } | |||
| 322 | + | assets, err := s.rel.ListAssets(ctx, caller(ctx), repo, req.Msg.GetReleaseId()) | |||
| 323 | + | if err != nil { | |||
| 324 | + | return nil, connectError(ctx, err) | |||
| 325 | + | } | |||
| 326 | + | return connect.NewResponse(&pb.ListReleaseAssetsResponse{Assets: assetsToPB(assets)}), nil | |||
| 327 | + | } | |||
| 328 | + | ||||
| 329 | + | func (s *releaseService) GetReleaseUsage(ctx context.Context, req *connect.Request[pb.GetReleaseUsageRequest]) (*connect.Response[pb.GetReleaseUsageResponse], error) { | |||
| 330 | + | repo, err := repoOf(req.Msg.GetRepo()) | |||
| 331 | + | if err != nil { | |||
| 332 | + | return nil, connectError(ctx, err) | |||
| 333 | + | } | |||
| 334 | + | u, err := s.rel.Usage(ctx, caller(ctx), repo) | |||
| 335 | + | if err != nil { | |||
| 336 | + | return nil, connectError(ctx, err) | |||
| 337 | + | } | |||
| 338 | + | return connect.NewResponse(&pb.GetReleaseUsageResponse{UsedBytes: u.Used, QuotaBytes: u.Limits.Quota, AssetMaxBytes: u.Limits.AssetMax, AssetsMax: int32(u.Limits.AssetsMax)}), nil | |||
| 339 | + | } | |||
A internal/adapters/rpc/release_test.go +422 -0
| @@ -0,0 +1,422 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | package rpc | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "bytes" | |
| 7 | + | "context" | |
| 8 | + | "io" | |
| 9 | + | "net/http/httptest" | |
| 10 | + | "testing" | |
| 11 | + | "time" | |
| 12 | + | ||
| 13 | + | "connectrpc.com/connect" | |
| 14 | + | "github.com/stretchr/testify/assert" | |
| 15 | + | "github.com/stretchr/testify/mock" | |
| 16 | + | "github.com/stretchr/testify/require" | |
| 17 | + | "google.golang.org/protobuf/proto" | |
| 18 | + | "google.golang.org/protobuf/types/known/fieldmaskpb" | |
| 19 | + | ||
| 20 | + | pb "bnema.dev/ermite-api/gen/ermite" | |
| 21 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |
| 22 | + | "bnema.dev/ermite/internal/domain" | |
| 23 | + | "bnema.dev/ermite/internal/ports" | |
| 24 | + | "bnema.dev/ermite/internal/ports/mocks" | |
| 25 | + | ) | |
| 26 | + | ||
| 27 | + | func releaseClient(t *testing.T, rel ports.ReleaseService) (ermiteconnect.ReleaseServiceClient, ermiteconnect.ServerServiceClient) { | |
| 28 | + | t.Helper() | |
| 29 | + | srv := httptest.NewUnstartedServer(New(Config{Releases: rel, Sessions: mocks.NewSessionService(t), SigningKey: mocks.NewSigningKeyService(t)}).Handler()) | |
| 30 | + | srv.EnableHTTP2 = true | |
| 31 | + | srv.StartTLS() | |
| 32 | + | t.Cleanup(srv.Close) | |
| 33 | + | return ermiteconnect.NewReleaseServiceClient(srv.Client(), srv.URL), ermiteconnect.NewServerServiceClient(srv.Client(), srv.URL) | |
| 34 | + | } | |
| 35 | + | ||
| 36 | + | func TestReleaseRPCUnary(t *testing.T) { | |
| 37 | + | ctx := context.Background() | |
| 38 | + | repo := mustRepoName(t, "demo") | |
| 39 | + | now := time.Now().UTC().Truncate(time.Second) | |
| 40 | + | r := domain.Release{ID: 7, Repo: repo, Tag: "v1", Title: "title", Notes: "notes", Prerelease: true, TargetCommitish: "abc", Author: "owner", CreatedAt: now, PublishedAt: now, UpdatedAt: now} | |
| 41 | + | a := domain.ReleaseAsset{ID: 9, ReleaseID: 7, Repo: repo, Name: "a.zip", Label: "label", ContentType: "application/zip", Size: 3, Digest: "sha256:123", Uploader: "owner", CreatedAt: now, UpdatedAt: now} | |
| 42 | + | expected := releaseToPB(r) | |
| 43 | + | expected.Latest = true | |
| 44 | + | expected.Assets = []*pb.ReleaseAsset{assetToPB(a)} | |
| 45 | + | for _, method := range []string{"create", "get id", "get tag", "get latest", "list", "update", "delete", "update asset", "delete asset", "list assets", "usage"} { | |
| 46 | + | t.Run(method, func(t *testing.T) { | |
| 47 | + | rel := mocks.NewReleaseService(t) | |
| 48 | + | c, _ := releaseClient(t, rel) | |
| 49 | + | p := domain.Anonymous() | |
| 50 | + | hydrate := func() { | |
| 51 | + | rel.EXPECT().GetRelease(mock.Anything, p, repo, ports.ReleaseRef{Latest: true}).Return(r, nil).Once() | |
| 52 | + | rel.EXPECT().ListAssets(mock.Anything, p, repo, r.ID).Return([]domain.ReleaseAsset{a}, nil).Once() | |
| 53 | + | } | |
| 54 | + | switch method { | |
| 55 | + | case "create": | |
| 56 | + | rel.EXPECT().CreateRelease(mock.Anything, p, repo, ports.CreateReleaseParams{Tag: "v1", TargetCommitish: "abc", Title: "title", Notes: "notes", Draft: true, Prerelease: true, MakeLatest: true}).Return(r, nil).Once() | |
| 57 | + | hydrate() | |
| 58 | + | res, err := c.CreateRelease(ctx, connect.NewRequest(&pb.CreateReleaseRequest{Repo: "demo", Tag: "v1", TargetCommitish: "abc", Title: "title", Notes: "notes", Draft: true, Prerelease: true, MakeLatest: true})) | |
| 59 | + | require.NoError(t, err) | |
| 60 | + | assert.True(t, proto.Equal(expected, res.Msg.GetRelease())) | |
| 61 | + | case "get id", "get tag", "get latest": | |
| 62 | + | req := &pb.GetReleaseRequest{Repo: "demo"} | |
| 63 | + | ref := ports.ReleaseRef{} | |
| 64 | + | switch method { | |
| 65 | + | case "get id": | |
| 66 | + | req.Selector = &pb.GetReleaseRequest_Id{Id: 7} | |
| 67 | + | ref.ID = 7 | |
| 68 | + | case "get tag": | |
| 69 | + | req.Selector = &pb.GetReleaseRequest_Tag{Tag: "v1"} | |
| 70 | + | ref.Tag = "v1" | |
| 71 | + | default: | |
| 72 | + | req.Selector = &pb.GetReleaseRequest_Latest{Latest: true} | |
| 73 | + | ref.Latest = true | |
| 74 | + | } | |
| 75 | + | rel.EXPECT().GetRelease(mock.Anything, p, repo, ref).Return(r, nil).Once() | |
| 76 | + | hydrate() | |
| 77 | + | res, err := c.GetRelease(ctx, connect.NewRequest(req)) | |
| 78 | + | require.NoError(t, err) | |
| 79 | + | assert.True(t, proto.Equal(expected, res.Msg.GetRelease())) | |
| 80 | + | case "list": | |
| 81 | + | rel.EXPECT().ListReleases(mock.Anything, p, repo, 17, "cursor").Return(ports.ReleasePage{Releases: []domain.Release{r}, NextPageToken: "next"}, nil).Once() | |
| 82 | + | hydrate() | |
| 83 | + | res, err := c.ListReleases(ctx, connect.NewRequest(&pb.ListReleasesRequest{Repo: "demo", PageSize: 17, PageToken: "cursor"})) | |
| 84 | + | require.NoError(t, err) | |
| 85 | + | require.Len(t, res.Msg.Releases, 1) | |
| 86 | + | assert.True(t, proto.Equal(expected, res.Msg.Releases[0])) | |
| 87 | + | assert.Equal(t, "next", res.Msg.NextPageToken) | |
| 88 | + | case "update": | |
| 89 | + | rel.EXPECT().UpdateRelease(mock.Anything, p, repo, int64(7), ports.UpdateReleaseParams{Title: new(""), Draft: new(false)}).Return(r, nil).Once() | |
| 90 | + | hydrate() | |
| 91 | + | res, err := c.UpdateRelease(ctx, connect.NewRequest(&pb.UpdateReleaseRequest{Repo: "demo", Id: 7, Tag: "ignored", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"title", "draft"}}})) | |
| 92 | + | require.NoError(t, err) | |
| 93 | + | assert.True(t, proto.Equal(expected, res.Msg.GetRelease())) | |
| 94 | + | case "delete": | |
| 95 | + | rel.EXPECT().DeleteRelease(mock.Anything, p, repo, int64(7)).Return(nil).Once() | |
| 96 | + | _, err := c.DeleteRelease(ctx, connect.NewRequest(&pb.DeleteReleaseRequest{Repo: "demo", Id: 7})) | |
| 97 | + | require.NoError(t, err) | |
| 98 | + | case "update asset": | |
| 99 | + | rel.EXPECT().UpdateAsset(mock.Anything, p, repo, int64(9), ports.UpdateAssetParams{Name: new("new.zip"), Label: new("")}).Return(a, nil).Once() | |
| 100 | + | res, err := c.UpdateReleaseAsset(ctx, connect.NewRequest(&pb.UpdateReleaseAssetRequest{Repo: "demo", AssetId: 9, Name: "new.zip", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"name", "label"}}})) | |
| 101 | + | require.NoError(t, err) | |
| 102 | + | assert.True(t, proto.Equal(assetToPB(a), res.Msg.GetAsset())) | |
| 103 | + | case "delete asset": | |
| 104 | + | rel.EXPECT().DeleteAsset(mock.Anything, p, repo, int64(9)).Return(nil).Once() | |
| 105 | + | _, err := c.DeleteReleaseAsset(ctx, connect.NewRequest(&pb.DeleteReleaseAssetRequest{Repo: "demo", AssetId: 9})) | |
| 106 | + | require.NoError(t, err) | |
| 107 | + | case "list assets": | |
| 108 | + | rel.EXPECT().ListAssets(mock.Anything, p, repo, int64(7)).Return([]domain.ReleaseAsset{a}, nil).Once() | |
| 109 | + | res, err := c.ListReleaseAssets(ctx, connect.NewRequest(&pb.ListReleaseAssetsRequest{Repo: "demo", ReleaseId: 7})) | |
| 110 | + | require.NoError(t, err) | |
| 111 | + | require.Len(t, res.Msg.Assets, 1) | |
| 112 | + | assert.True(t, proto.Equal(assetToPB(a), res.Msg.Assets[0])) | |
| 113 | + | case "usage": | |
| 114 | + | rel.EXPECT().Usage(mock.Anything, p, repo).Return(ports.ReleaseUsage{Used: 1, Limits: ports.ReleaseLimits{Quota: 2, AssetMax: 3, AssetsMax: 4}}, nil).Once() | |
| 115 | + | res, err := c.GetReleaseUsage(ctx, connect.NewRequest(&pb.GetReleaseUsageRequest{Repo: "demo"})) | |
| 116 | + | require.NoError(t, err) | |
| 117 | + | assert.True(t, proto.Equal(&pb.GetReleaseUsageResponse{UsedBytes: 1, QuotaBytes: 2, AssetMaxBytes: 3, AssetsMax: 4}, res.Msg)) | |
| 118 | + | } | |
| 119 | + | }) | |
| 120 | + | } | |
| 121 | + | } | |
| 122 | + | ||
| 123 | + | func TestReleaseFeatureAndMount(t *testing.T) { | |
| 124 | + | for _, enabled := range []bool{false, true} { | |
| 125 | + | t.Run(map[bool]string{false: "absent", true: "present"}[enabled], func(t *testing.T) { | |
| 126 | + | var rel ports.ReleaseService | |
| 127 | + | if enabled { | |
| 128 | + | rel = mocks.NewReleaseService(t) | |
| 129 | + | } | |
| 130 | + | c, server := releaseClient(t, rel) | |
| 131 | + | res, err := server.GetVersion(context.Background(), connect.NewRequest(&pb.GetVersionRequest{})) | |
| 132 | + | require.NoError(t, err) | |
| 133 | + | assert.Equal(t, enabled, containsReleaseFeature(res.Msg.Features)) | |
| 134 | + | assert.NotContains(t, Features, FeatureReleases) | |
| 135 | + | if !enabled { | |
| 136 | + | _, err := c.GetReleaseUsage(context.Background(), connect.NewRequest(&pb.GetReleaseUsageRequest{Repo: "demo"})) | |
| 137 | + | requireCode(t, connect.CodeUnimplemented, err) | |
| 138 | + | } | |
| 139 | + | }) | |
| 140 | + | } | |
| 141 | + | } | |
| 142 | + | func containsReleaseFeature(features []string) bool { | |
| 143 | + | for _, f := range features { | |
| 144 | + | if f == FeatureReleases { | |
| 145 | + | return true | |
| 146 | + | } | |
| 147 | + | } | |
| 148 | + | return false | |
| 149 | + | } | |
| 150 | + | ||
| 151 | + | func TestReleaseRPCUpload(t *testing.T) { | |
| 152 | + | for _, firstData := range []bool{false, true} { | |
| 153 | + | t.Run(map[bool]string{false: "header only", true: "header with data"}[firstData], func(t *testing.T) { | |
| 154 | + | rel := mocks.NewReleaseService(t) | |
| 155 | + | c, _ := releaseClient(t, rel) | |
| 156 | + | full := bytes.Repeat([]byte("x"), domain.CIArtifactChunk) | |
| 157 | + | tail := []byte("tail") | |
| 158 | + | want := append(bytes.Clone(full), tail...) | |
| 159 | + | rel.EXPECT().UploadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.UploadAssetParams{ReleaseID: 7, Name: "a", Label: "l", ContentType: "application/octet-stream", Size: int64(len(want)), Replace: true}, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.UploadAssetParams, r io.Reader) (domain.ReleaseAsset, error) { | |
| 160 | + | got, err := io.ReadAll(r) | |
| 161 | + | assert.NoError(t, err) | |
| 162 | + | assert.Equal(t, want, got) | |
| 163 | + | return domain.ReleaseAsset{ID: 9, Size: int64(len(got))}, err | |
| 164 | + | }).Once() | |
| 165 | + | stream := c.UploadReleaseAsset(context.Background()) | |
| 166 | + | first := &pb.UploadReleaseAssetRequest{Repo: "demo", ReleaseId: 7, Name: "a", Label: "l", ContentType: "application/octet-stream", Size: int64(len(want)), Replace: true} | |
| 167 | + | if firstData { | |
| 168 | + | first.Data = full | |
| 169 | + | } | |
| 170 | + | require.NoError(t, stream.Send(first)) | |
| 171 | + | if !firstData { | |
| 172 | + | require.NoError(t, stream.Send(&pb.UploadReleaseAssetRequest{Data: full})) | |
| 173 | + | } | |
| 174 | + | require.NoError(t, stream.Send(&pb.UploadReleaseAssetRequest{Data: tail})) | |
| 175 | + | res, err := stream.CloseAndReceive() | |
| 176 | + | require.NoError(t, err) | |
| 177 | + | assert.EqualValues(t, 9, res.Msg.Asset.Id) | |
| 178 | + | assert.EqualValues(t, len(want), res.Msg.Asset.Size) | |
| 179 | + | }) | |
| 180 | + | } | |
| 181 | + | } | |
| 182 | + | ||
| 183 | + | func TestReleaseRPCUploadInvalid(t *testing.T) { | |
| 184 | + | tests := []struct { | |
| 185 | + | name string | |
| 186 | + | messages []*pb.UploadReleaseAssetRequest | |
| 187 | + | read bool | |
| 188 | + | code connect.Code | |
| 189 | + | }{ | |
| 190 | + | {name: "empty", code: connect.CodeInvalidArgument}, | |
| 191 | + | {name: "negative size", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo", Size: -1}}, code: connect.CodeInvalidArgument}, | |
| 192 | + | {name: "bad repo", messages: []*pb.UploadReleaseAssetRequest{{Repo: "../bad"}}, code: connect.CodeInvalidArgument}, | |
| 193 | + | {name: "first chunk too big", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo", Data: make([]byte, domain.CIArtifactChunk+1)}}, code: connect.CodeInvalidArgument}, | |
| 194 | + | {name: "later chunk too big", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo"}, {Data: make([]byte, domain.CIArtifactChunk+1)}}, read: true, code: connect.CodeInvalidArgument}, | |
| 195 | + | {name: "message exceeds envelope", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo", Data: make([]byte, RunnerMaxRequestBytes+1)}}, code: connect.CodeResourceExhausted}, | |
| 196 | + | } | |
| 197 | + | for _, m := range []*pb.UploadReleaseAssetRequest{{Repo: "demo"}, {ReleaseId: 1}, {Name: "a"}, {Label: "l"}, {ContentType: "text/plain"}, {Size: 1}, {Replace: true}} { | |
| 198 | + | tests = append(tests, struct { | |
| 199 | + | name string | |
| 200 | + | messages []*pb.UploadReleaseAssetRequest | |
| 201 | + | read bool | |
| 202 | + | code connect.Code | |
| 203 | + | }{name: "repeated metadata " + m.String(), messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo"}, m}, read: true, code: connect.CodeInvalidArgument}) | |
| 204 | + | } | |
| 205 | + | for _, tt := range tests { | |
| 206 | + | t.Run(tt.name, func(t *testing.T) { | |
| 207 | + | rel := mocks.NewReleaseService(t) | |
| 208 | + | c, _ := releaseClient(t, rel) | |
| 209 | + | if tt.read { | |
| 210 | + | rel.EXPECT().UploadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.UploadAssetParams, r io.Reader) (domain.ReleaseAsset, error) { | |
| 211 | + | _, err := io.ReadAll(r) | |
| 212 | + | return domain.ReleaseAsset{}, err | |
| 213 | + | }).Once() | |
| 214 | + | } | |
| 215 | + | stream := c.UploadReleaseAsset(context.Background()) | |
| 216 | + | for _, m := range tt.messages { | |
| 217 | + | if err := stream.Send(m); err != nil { | |
| 218 | + | break | |
| 219 | + | } | |
| 220 | + | } | |
| 221 | + | _, err := stream.CloseAndReceive() | |
| 222 | + | requireCode(t, tt.code, err) | |
| 223 | + | }) | |
| 224 | + | } | |
| 225 | + | } | |
| 226 | + | ||
| 227 | + | func TestReleaseRPCDownload(t *testing.T) { | |
| 228 | + | for _, selector := range []string{"id", "tag", "latest"} { | |
| 229 | + | t.Run(selector, func(t *testing.T) { | |
| 230 | + | rel := mocks.NewReleaseService(t) | |
| 231 | + | c, _ := releaseClient(t, rel) | |
| 232 | + | req := &pb.DownloadReleaseAssetRequest{Repo: "demo", Offset: 3} | |
| 233 | + | ref := ports.AssetRef{} | |
| 234 | + | switch selector { | |
| 235 | + | case "id": | |
| 236 | + | req.Selector = &pb.DownloadReleaseAssetRequest_AssetId{AssetId: 9} | |
| 237 | + | ref.ID = 9 | |
| 238 | + | case "tag": | |
| 239 | + | req.Selector = &pb.DownloadReleaseAssetRequest_Tag{Tag: &pb.ReleaseAssetTag{Tag: "v1", Name: "a"}} | |
| 240 | + | ref.Tag = "v1" | |
| 241 | + | ref.Name = "a" | |
| 242 | + | default: | |
| 243 | + | req.Selector = &pb.DownloadReleaseAssetRequest_Latest{Latest: &pb.ReleaseAssetLatest{Latest: true, Name: "a"}} | |
| 244 | + | ref.Latest = true | |
| 245 | + | ref.Name = "a" | |
| 246 | + | } | |
| 247 | + | a := domain.ReleaseAsset{ID: 9, ReleaseID: 7, Name: "a", Size: 8} | |
| 248 | + | rel.EXPECT().DownloadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ref, int64(3), int64(-1), mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.AssetRef, _, _ int64, send func(ports.AssetChunk) error) error { | |
| 249 | + | if err := send(ports.AssetChunk{Asset: &a}); err != nil { | |
| 250 | + | return err | |
| 251 | + | } | |
| 252 | + | if err := send(ports.AssetChunk{Data: []byte("12")}); err != nil { | |
| 253 | + | return err | |
| 254 | + | } | |
| 255 | + | return send(ports.AssetChunk{Data: []byte("345")}) | |
| 256 | + | }).Once() | |
| 257 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(req)) | |
| 258 | + | require.NoError(t, err) | |
| 259 | + | defer stream.Close() | |
| 260 | + | require.True(t, stream.Receive()) | |
| 261 | + | assert.True(t, proto.Equal(assetToPB(a), stream.Msg().Asset)) | |
| 262 | + | assert.Empty(t, stream.Msg().Chunk) | |
| 263 | + | var got []byte | |
| 264 | + | for stream.Receive() { | |
| 265 | + | assert.Nil(t, stream.Msg().Asset) | |
| 266 | + | got = append(got, stream.Msg().Chunk...) | |
| 267 | + | } | |
| 268 | + | require.NoError(t, stream.Err()) | |
| 269 | + | assert.Equal(t, "12345", string(got)) | |
| 270 | + | }) | |
| 271 | + | } | |
| 272 | + | } | |
| 273 | + | ||
| 274 | + | func TestReleaseRPCErrors(t *testing.T) { | |
| 275 | + | for _, tt := range []struct { | |
| 276 | + | err error | |
| 277 | + | code connect.Code | |
| 278 | + | }{{domain.ErrNotFound, connect.CodeNotFound}, {domain.ErrAlreadyExists, connect.CodeAlreadyExists}, {domain.ErrResourceExhausted, connect.CodeResourceExhausted}, {domain.ErrForbidden, connect.CodePermissionDenied}, {domain.ErrOutOfRange, connect.CodeOutOfRange}, {domain.ErrPreconditionFailed, connect.CodeFailedPrecondition}, {domain.ErrInvalid, connect.CodeInvalidArgument}} { | |
| 279 | + | t.Run(tt.code.String(), func(t *testing.T) { | |
| 280 | + | rel := mocks.NewReleaseService(t) | |
| 281 | + | c, _ := releaseClient(t, rel) | |
| 282 | + | rel.EXPECT().GetRelease(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.ReleaseRef{ID: 7}).Return(domain.Release{}, tt.err).Once() | |
| 283 | + | _, err := c.GetRelease(context.Background(), connect.NewRequest(&pb.GetReleaseRequest{Repo: "demo", Selector: &pb.GetReleaseRequest_Id{Id: 7}})) | |
| 284 | + | requireCode(t, tt.code, err) | |
| 285 | + | rel.EXPECT().DownloadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.AssetRef{ID: 9}, int64(0), int64(-1), mock.Anything).Return(tt.err).Once() | |
| 286 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(&pb.DownloadReleaseAssetRequest{Repo: "demo", Selector: &pb.DownloadReleaseAssetRequest_AssetId{AssetId: 9}})) | |
| 287 | + | if err == nil { | |
| 288 | + | defer stream.Close() | |
| 289 | + | assert.False(t, stream.Receive()) | |
| 290 | + | err = stream.Err() | |
| 291 | + | } | |
| 292 | + | requireCode(t, tt.code, err) | |
| 293 | + | }) | |
| 294 | + | } | |
| 295 | + | } | |
| 296 | + | ||
| 297 | + | func TestReleaseMasksAndSelectors(t *testing.T) { | |
| 298 | + | paths := []string{"tag", "target_commitish", "title", "notes", "draft", "prerelease", "make_latest"} | |
| 299 | + | p, err := releaseUpdateParams(&pb.UpdateReleaseRequest{UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}}) | |
| 300 | + | require.NoError(t, err) | |
| 301 | + | assert.Equal(t, ports.UpdateReleaseParams{Tag: new(""), TargetCommitish: new(""), Title: new(""), Notes: new(""), Draft: new(false), Prerelease: new(false), MakeLatest: new(false)}, p) | |
| 302 | + | for _, paths := range [][]string{nil, {"unknown"}, {"*"}, {"title", "assets"}} { | |
| 303 | + | _, err := releaseUpdateParams(&pb.UpdateReleaseRequest{UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}}) | |
| 304 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |
| 305 | + | _, err = assetUpdateParams(&pb.UpdateReleaseAssetRequest{UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}}) | |
| 306 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |
| 307 | + | } | |
| 308 | + | for _, req := range []*pb.GetReleaseRequest{{}, {Selector: &pb.GetReleaseRequest_Id{Id: -1}}, {Selector: &pb.GetReleaseRequest_Tag{}}, {Selector: &pb.GetReleaseRequest_Latest{Latest: false}}} { | |
| 309 | + | _, err := releaseRef(req) | |
| 310 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |
| 311 | + | } | |
| 312 | + | for _, req := range []*pb.DownloadReleaseAssetRequest{{}, {Selector: &pb.DownloadReleaseAssetRequest_AssetId{}}, {Selector: &pb.DownloadReleaseAssetRequest_Tag{}}, {Selector: &pb.DownloadReleaseAssetRequest_Tag{Tag: &pb.ReleaseAssetTag{Tag: "v1"}}}, {Selector: &pb.DownloadReleaseAssetRequest_Latest{Latest: &pb.ReleaseAssetLatest{Name: "a"}}}} { | |
| 313 | + | _, err := assetRef(req) | |
| 314 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |
| 315 | + | } | |
| 316 | + | } | |
| 317 | + | ||
| 318 | + | func TestReleaseEvents(t *testing.T) { | |
| 319 | + | r := domain.Release{ID: 7, Repo: mustRepoName(t, "demo"), Tag: "v1", Draft: true, Prerelease: true} | |
| 320 | + | at := time.Now().UTC() | |
| 321 | + | events := []domain.Event{domain.ReleasePublished{Release: r, At: at}, domain.ReleaseUpdated{Release: r, At: at}, domain.ReleaseDeleted{Release: r, At: at}, domain.ReleaseUnpublished{Release: r, At: at}} | |
| 322 | + | expected := []*pb.Event{ | |
| 323 | + | {Kind: &pb.Event_ReleasePublished{ReleasePublished: &pb.ReleasePublished{Repo: "demo", Id: 7, Tag: "v1", Draft: true, Prerelease: true}}}, | |
| 324 | + | {Kind: &pb.Event_ReleaseUpdated{ReleaseUpdated: &pb.ReleaseUpdated{Repo: "demo", Id: 7, Tag: "v1", Draft: true, Prerelease: true}}}, | |
| 325 | + | {Kind: &pb.Event_ReleaseDeleted{ReleaseDeleted: &pb.ReleaseDeleted{Repo: "demo", Id: 7, Tag: "v1"}}}, | |
| 326 | + | {Kind: &pb.Event_ReleaseUnpublished{ReleaseUnpublished: &pb.ReleaseUnpublished{Repo: "demo", Id: 7, Tag: "v1", Draft: true, Prerelease: true}}}, | |
| 327 | + | } | |
| 328 | + | for i, ev := range events { | |
| 329 | + | t.Run(ev.EventKind(), func(t *testing.T) { | |
| 330 | + | expected[i].Id = "event" | |
| 331 | + | expected[i].Time = timestamp(at) | |
| 332 | + | assert.True(t, proto.Equal(expected[i], watchEventToPB(ports.WatchEvent{ID: "event", Event: ev}))) | |
| 333 | + | }) | |
| 334 | + | } | |
| 335 | + | } | |
| 336 | + | ||
| 337 | + | func TestReleaseResponseEnrichment(t *testing.T) { | |
| 338 | + | repo := mustRepoName(t, "demo") | |
| 339 | + | for _, tt := range []struct { | |
| 340 | + | name string | |
| 341 | + | latest domain.Release | |
| 342 | + | latestErr, assetsErr error | |
| 343 | + | code connect.Code | |
| 344 | + | wantLatest bool | |
| 345 | + | }{ | |
| 346 | + | {name: "draft without latest", latestErr: domain.ErrNotFound}, | |
| 347 | + | {name: "different latest", latest: domain.Release{ID: 8}}, | |
| 348 | + | {name: "current latest", latest: domain.Release{ID: 7}, wantLatest: true}, | |
| 349 | + | {name: "latest lookup fails", latestErr: domain.ErrUnavailable, code: connect.CodeUnavailable}, | |
| 350 | + | {name: "assets lookup fails", assetsErr: domain.ErrNotFound, code: connect.CodeNotFound}, | |
| 351 | + | } { | |
| 352 | + | t.Run(tt.name, func(t *testing.T) { | |
| 353 | + | rel := mocks.NewReleaseService(t) | |
| 354 | + | rel.EXPECT().GetRelease(mock.Anything, domain.Owner(), repo, ports.ReleaseRef{ID: 7}).Return(domain.Release{ID: 7, Repo: repo, Draft: true}, nil).Once() | |
| 355 | + | rel.EXPECT().GetRelease(mock.Anything, domain.Owner(), repo, ports.ReleaseRef{Latest: true}).Return(tt.latest, tt.latestErr).Once() | |
| 356 | + | if tt.code != connect.CodeUnavailable { | |
| 357 | + | rel.EXPECT().ListAssets(mock.Anything, domain.Owner(), repo, int64(7)).Return(nil, tt.assetsErr).Once() | |
| 358 | + | } | |
| 359 | + | res, err := (&releaseService{rel}).GetRelease(ownerCtx(), connect.NewRequest(&pb.GetReleaseRequest{Repo: "demo", Selector: &pb.GetReleaseRequest_Id{Id: 7}})) | |
| 360 | + | if tt.code != 0 { | |
| 361 | + | requireCode(t, tt.code, err) | |
| 362 | + | return | |
| 363 | + | } | |
| 364 | + | require.NoError(t, err) | |
| 365 | + | assert.Equal(t, tt.wantLatest, res.Msg.Release.Latest) | |
| 366 | + | assert.Empty(t, res.Msg.Release.Assets) | |
| 367 | + | assert.Nil(t, res.Msg.Release.PublishedAt) | |
| 368 | + | }) | |
| 369 | + | } | |
| 370 | + | } | |
| 371 | + | ||
| 372 | + | func TestReleaseRPCAuthenticatedCaller(t *testing.T) { | |
| 373 | + | rel := mocks.NewReleaseService(t) | |
| 374 | + | auth := mocks.NewAuthService(t) | |
| 375 | + | auth.EXPECT().Authenticate(mock.Anything, "secret").Return(domain.Owner(), nil).Once() | |
| 376 | + | rel.EXPECT().DeleteRelease(mock.Anything, domain.Owner(), mustRepoName(t, "demo"), int64(7)).Return(nil).Once() | |
| 377 | + | srv := httptest.NewServer(New(Config{Releases: rel, Auth: auth, Sessions: mocks.NewSessionService(t)}).Handler()) | |
| 378 | + | defer srv.Close() | |
| 379 | + | c := ermiteconnect.NewReleaseServiceClient(srv.Client(), srv.URL) | |
| 380 | + | req := connect.NewRequest(&pb.DeleteReleaseRequest{Repo: "demo", Id: 7}) | |
| 381 | + | req.Header().Set("Authorization", "Bearer secret") | |
| 382 | + | _, err := c.DeleteRelease(context.Background(), req) | |
| 383 | + | require.NoError(t, err) | |
| 384 | + | } | |
| 385 | + | ||
| 386 | + | func TestReleaseRPCDownloadFullChunk(t *testing.T) { | |
| 387 | + | rel := mocks.NewReleaseService(t) | |
| 388 | + | c, _ := releaseClient(t, rel) | |
| 389 | + | data := bytes.Repeat([]byte("x"), domain.CIArtifactChunk) | |
| 390 | + | rel.EXPECT().DownloadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.AssetRef{ID: 9}, int64(0), int64(-1), mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.AssetRef, _, _ int64, send func(ports.AssetChunk) error) error { | |
| 391 | + | if err := send(ports.AssetChunk{Asset: &domain.ReleaseAsset{ID: 9, Size: int64(len(data))}}); err != nil { | |
| 392 | + | return err | |
| 393 | + | } | |
| 394 | + | return send(ports.AssetChunk{Data: data}) | |
| 395 | + | }).Once() | |
| 396 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(&pb.DownloadReleaseAssetRequest{Repo: "demo", Selector: &pb.DownloadReleaseAssetRequest_AssetId{AssetId: 9}})) | |
| 397 | + | require.NoError(t, err) | |
| 398 | + | defer stream.Close() | |
| 399 | + | require.True(t, stream.Receive()) | |
| 400 | + | require.NotNil(t, stream.Msg().Asset) | |
| 401 | + | require.True(t, stream.Receive()) | |
| 402 | + | assert.Equal(t, data, stream.Msg().Chunk) | |
| 403 | + | assert.False(t, stream.Receive()) | |
| 404 | + | require.NoError(t, stream.Err()) | |
| 405 | + | } | |
| 406 | + | ||
| 407 | + | func TestReleaseRPCRejectsInvalidMasksAndSelectors(t *testing.T) { | |
| 408 | + | c, _ := releaseClient(t, mocks.NewReleaseService(t)) | |
| 409 | + | _, err := c.UpdateRelease(context.Background(), connect.NewRequest(&pb.UpdateReleaseRequest{Repo: "demo", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"unknown"}}})) | |
| 410 | + | requireCode(t, connect.CodeInvalidArgument, err) | |
| 411 | + | _, err = c.UpdateReleaseAsset(context.Background(), connect.NewRequest(&pb.UpdateReleaseAssetRequest{Repo: "demo"})) | |
| 412 | + | requireCode(t, connect.CodeInvalidArgument, err) | |
| 413 | + | _, err = c.GetRelease(context.Background(), connect.NewRequest(&pb.GetReleaseRequest{Repo: "demo", Selector: &pb.GetReleaseRequest_Latest{Latest: false}})) | |
| 414 | + | requireCode(t, connect.CodeInvalidArgument, err) | |
| 415 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(&pb.DownloadReleaseAssetRequest{Repo: "demo", Selector: &pb.DownloadReleaseAssetRequest_Latest{Latest: &pb.ReleaseAssetLatest{Name: "a"}}})) | |
| 416 | + | if err == nil { | |
| 417 | + | defer stream.Close() | |
| 418 | + | assert.False(t, stream.Receive()) | |
| 419 | + | err = stream.Err() | |
| 420 | + | } | |
| 421 | + | requireCode(t, connect.CodeInvalidArgument, err) | |
| 422 | + | } | |
| @@ -0,0 +1,422 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | package rpc | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "bytes" | |||
| 7 | + | "context" | |||
| 8 | + | "io" | |||
| 9 | + | "net/http/httptest" | |||
| 10 | + | "testing" | |||
| 11 | + | "time" | |||
| 12 | + | ||||
| 13 | + | "connectrpc.com/connect" | |||
| 14 | + | "github.com/stretchr/testify/assert" | |||
| 15 | + | "github.com/stretchr/testify/mock" | |||
| 16 | + | "github.com/stretchr/testify/require" | |||
| 17 | + | "google.golang.org/protobuf/proto" | |||
| 18 | + | "google.golang.org/protobuf/types/known/fieldmaskpb" | |||
| 19 | + | ||||
| 20 | + | pb "bnema.dev/ermite-api/gen/ermite" | |||
| 21 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |||
| 22 | + | "bnema.dev/ermite/internal/domain" | |||
| 23 | + | "bnema.dev/ermite/internal/ports" | |||
| 24 | + | "bnema.dev/ermite/internal/ports/mocks" | |||
| 25 | + | ) | |||
| 26 | + | ||||
| 27 | + | func releaseClient(t *testing.T, rel ports.ReleaseService) (ermiteconnect.ReleaseServiceClient, ermiteconnect.ServerServiceClient) { | |||
| 28 | + | t.Helper() | |||
| 29 | + | srv := httptest.NewUnstartedServer(New(Config{Releases: rel, Sessions: mocks.NewSessionService(t), SigningKey: mocks.NewSigningKeyService(t)}).Handler()) | |||
| 30 | + | srv.EnableHTTP2 = true | |||
| 31 | + | srv.StartTLS() | |||
| 32 | + | t.Cleanup(srv.Close) | |||
| 33 | + | return ermiteconnect.NewReleaseServiceClient(srv.Client(), srv.URL), ermiteconnect.NewServerServiceClient(srv.Client(), srv.URL) | |||
| 34 | + | } | |||
| 35 | + | ||||
| 36 | + | func TestReleaseRPCUnary(t *testing.T) { | |||
| 37 | + | ctx := context.Background() | |||
| 38 | + | repo := mustRepoName(t, "demo") | |||
| 39 | + | now := time.Now().UTC().Truncate(time.Second) | |||
| 40 | + | r := domain.Release{ID: 7, Repo: repo, Tag: "v1", Title: "title", Notes: "notes", Prerelease: true, TargetCommitish: "abc", Author: "owner", CreatedAt: now, PublishedAt: now, UpdatedAt: now} | |||
| 41 | + | a := domain.ReleaseAsset{ID: 9, ReleaseID: 7, Repo: repo, Name: "a.zip", Label: "label", ContentType: "application/zip", Size: 3, Digest: "sha256:123", Uploader: "owner", CreatedAt: now, UpdatedAt: now} | |||
| 42 | + | expected := releaseToPB(r) | |||
| 43 | + | expected.Latest = true | |||
| 44 | + | expected.Assets = []*pb.ReleaseAsset{assetToPB(a)} | |||
| 45 | + | for _, method := range []string{"create", "get id", "get tag", "get latest", "list", "update", "delete", "update asset", "delete asset", "list assets", "usage"} { | |||
| 46 | + | t.Run(method, func(t *testing.T) { | |||
| 47 | + | rel := mocks.NewReleaseService(t) | |||
| 48 | + | c, _ := releaseClient(t, rel) | |||
| 49 | + | p := domain.Anonymous() | |||
| 50 | + | hydrate := func() { | |||
| 51 | + | rel.EXPECT().GetRelease(mock.Anything, p, repo, ports.ReleaseRef{Latest: true}).Return(r, nil).Once() | |||
| 52 | + | rel.EXPECT().ListAssets(mock.Anything, p, repo, r.ID).Return([]domain.ReleaseAsset{a}, nil).Once() | |||
| 53 | + | } | |||
| 54 | + | switch method { | |||
| 55 | + | case "create": | |||
| 56 | + | rel.EXPECT().CreateRelease(mock.Anything, p, repo, ports.CreateReleaseParams{Tag: "v1", TargetCommitish: "abc", Title: "title", Notes: "notes", Draft: true, Prerelease: true, MakeLatest: true}).Return(r, nil).Once() | |||
| 57 | + | hydrate() | |||
| 58 | + | res, err := c.CreateRelease(ctx, connect.NewRequest(&pb.CreateReleaseRequest{Repo: "demo", Tag: "v1", TargetCommitish: "abc", Title: "title", Notes: "notes", Draft: true, Prerelease: true, MakeLatest: true})) | |||
| 59 | + | require.NoError(t, err) | |||
| 60 | + | assert.True(t, proto.Equal(expected, res.Msg.GetRelease())) | |||
| 61 | + | case "get id", "get tag", "get latest": | |||
| 62 | + | req := &pb.GetReleaseRequest{Repo: "demo"} | |||
| 63 | + | ref := ports.ReleaseRef{} | |||
| 64 | + | switch method { | |||
| 65 | + | case "get id": | |||
| 66 | + | req.Selector = &pb.GetReleaseRequest_Id{Id: 7} | |||
| 67 | + | ref.ID = 7 | |||
| 68 | + | case "get tag": | |||
| 69 | + | req.Selector = &pb.GetReleaseRequest_Tag{Tag: "v1"} | |||
| 70 | + | ref.Tag = "v1" | |||
| 71 | + | default: | |||
| 72 | + | req.Selector = &pb.GetReleaseRequest_Latest{Latest: true} | |||
| 73 | + | ref.Latest = true | |||
| 74 | + | } | |||
| 75 | + | rel.EXPECT().GetRelease(mock.Anything, p, repo, ref).Return(r, nil).Once() | |||
| 76 | + | hydrate() | |||
| 77 | + | res, err := c.GetRelease(ctx, connect.NewRequest(req)) | |||
| 78 | + | require.NoError(t, err) | |||
| 79 | + | assert.True(t, proto.Equal(expected, res.Msg.GetRelease())) | |||
| 80 | + | case "list": | |||
| 81 | + | rel.EXPECT().ListReleases(mock.Anything, p, repo, 17, "cursor").Return(ports.ReleasePage{Releases: []domain.Release{r}, NextPageToken: "next"}, nil).Once() | |||
| 82 | + | hydrate() | |||
| 83 | + | res, err := c.ListReleases(ctx, connect.NewRequest(&pb.ListReleasesRequest{Repo: "demo", PageSize: 17, PageToken: "cursor"})) | |||
| 84 | + | require.NoError(t, err) | |||
| 85 | + | require.Len(t, res.Msg.Releases, 1) | |||
| 86 | + | assert.True(t, proto.Equal(expected, res.Msg.Releases[0])) | |||
| 87 | + | assert.Equal(t, "next", res.Msg.NextPageToken) | |||
| 88 | + | case "update": | |||
| 89 | + | rel.EXPECT().UpdateRelease(mock.Anything, p, repo, int64(7), ports.UpdateReleaseParams{Title: new(""), Draft: new(false)}).Return(r, nil).Once() | |||
| 90 | + | hydrate() | |||
| 91 | + | res, err := c.UpdateRelease(ctx, connect.NewRequest(&pb.UpdateReleaseRequest{Repo: "demo", Id: 7, Tag: "ignored", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"title", "draft"}}})) | |||
| 92 | + | require.NoError(t, err) | |||
| 93 | + | assert.True(t, proto.Equal(expected, res.Msg.GetRelease())) | |||
| 94 | + | case "delete": | |||
| 95 | + | rel.EXPECT().DeleteRelease(mock.Anything, p, repo, int64(7)).Return(nil).Once() | |||
| 96 | + | _, err := c.DeleteRelease(ctx, connect.NewRequest(&pb.DeleteReleaseRequest{Repo: "demo", Id: 7})) | |||
| 97 | + | require.NoError(t, err) | |||
| 98 | + | case "update asset": | |||
| 99 | + | rel.EXPECT().UpdateAsset(mock.Anything, p, repo, int64(9), ports.UpdateAssetParams{Name: new("new.zip"), Label: new("")}).Return(a, nil).Once() | |||
| 100 | + | res, err := c.UpdateReleaseAsset(ctx, connect.NewRequest(&pb.UpdateReleaseAssetRequest{Repo: "demo", AssetId: 9, Name: "new.zip", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"name", "label"}}})) | |||
| 101 | + | require.NoError(t, err) | |||
| 102 | + | assert.True(t, proto.Equal(assetToPB(a), res.Msg.GetAsset())) | |||
| 103 | + | case "delete asset": | |||
| 104 | + | rel.EXPECT().DeleteAsset(mock.Anything, p, repo, int64(9)).Return(nil).Once() | |||
| 105 | + | _, err := c.DeleteReleaseAsset(ctx, connect.NewRequest(&pb.DeleteReleaseAssetRequest{Repo: "demo", AssetId: 9})) | |||
| 106 | + | require.NoError(t, err) | |||
| 107 | + | case "list assets": | |||
| 108 | + | rel.EXPECT().ListAssets(mock.Anything, p, repo, int64(7)).Return([]domain.ReleaseAsset{a}, nil).Once() | |||
| 109 | + | res, err := c.ListReleaseAssets(ctx, connect.NewRequest(&pb.ListReleaseAssetsRequest{Repo: "demo", ReleaseId: 7})) | |||
| 110 | + | require.NoError(t, err) | |||
| 111 | + | require.Len(t, res.Msg.Assets, 1) | |||
| 112 | + | assert.True(t, proto.Equal(assetToPB(a), res.Msg.Assets[0])) | |||
| 113 | + | case "usage": | |||
| 114 | + | rel.EXPECT().Usage(mock.Anything, p, repo).Return(ports.ReleaseUsage{Used: 1, Limits: ports.ReleaseLimits{Quota: 2, AssetMax: 3, AssetsMax: 4}}, nil).Once() | |||
| 115 | + | res, err := c.GetReleaseUsage(ctx, connect.NewRequest(&pb.GetReleaseUsageRequest{Repo: "demo"})) | |||
| 116 | + | require.NoError(t, err) | |||
| 117 | + | assert.True(t, proto.Equal(&pb.GetReleaseUsageResponse{UsedBytes: 1, QuotaBytes: 2, AssetMaxBytes: 3, AssetsMax: 4}, res.Msg)) | |||
| 118 | + | } | |||
| 119 | + | }) | |||
| 120 | + | } | |||
| 121 | + | } | |||
| 122 | + | ||||
| 123 | + | func TestReleaseFeatureAndMount(t *testing.T) { | |||
| 124 | + | for _, enabled := range []bool{false, true} { | |||
| 125 | + | t.Run(map[bool]string{false: "absent", true: "present"}[enabled], func(t *testing.T) { | |||
| 126 | + | var rel ports.ReleaseService | |||
| 127 | + | if enabled { | |||
| 128 | + | rel = mocks.NewReleaseService(t) | |||
| 129 | + | } | |||
| 130 | + | c, server := releaseClient(t, rel) | |||
| 131 | + | res, err := server.GetVersion(context.Background(), connect.NewRequest(&pb.GetVersionRequest{})) | |||
| 132 | + | require.NoError(t, err) | |||
| 133 | + | assert.Equal(t, enabled, containsReleaseFeature(res.Msg.Features)) | |||
| 134 | + | assert.NotContains(t, Features, FeatureReleases) | |||
| 135 | + | if !enabled { | |||
| 136 | + | _, err := c.GetReleaseUsage(context.Background(), connect.NewRequest(&pb.GetReleaseUsageRequest{Repo: "demo"})) | |||
| 137 | + | requireCode(t, connect.CodeUnimplemented, err) | |||
| 138 | + | } | |||
| 139 | + | }) | |||
| 140 | + | } | |||
| 141 | + | } | |||
| 142 | + | func containsReleaseFeature(features []string) bool { | |||
| 143 | + | for _, f := range features { | |||
| 144 | + | if f == FeatureReleases { | |||
| 145 | + | return true | |||
| 146 | + | } | |||
| 147 | + | } | |||
| 148 | + | return false | |||
| 149 | + | } | |||
| 150 | + | ||||
| 151 | + | func TestReleaseRPCUpload(t *testing.T) { | |||
| 152 | + | for _, firstData := range []bool{false, true} { | |||
| 153 | + | t.Run(map[bool]string{false: "header only", true: "header with data"}[firstData], func(t *testing.T) { | |||
| 154 | + | rel := mocks.NewReleaseService(t) | |||
| 155 | + | c, _ := releaseClient(t, rel) | |||
| 156 | + | full := bytes.Repeat([]byte("x"), domain.CIArtifactChunk) | |||
| 157 | + | tail := []byte("tail") | |||
| 158 | + | want := append(bytes.Clone(full), tail...) | |||
| 159 | + | rel.EXPECT().UploadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.UploadAssetParams{ReleaseID: 7, Name: "a", Label: "l", ContentType: "application/octet-stream", Size: int64(len(want)), Replace: true}, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.UploadAssetParams, r io.Reader) (domain.ReleaseAsset, error) { | |||
| 160 | + | got, err := io.ReadAll(r) | |||
| 161 | + | assert.NoError(t, err) | |||
| 162 | + | assert.Equal(t, want, got) | |||
| 163 | + | return domain.ReleaseAsset{ID: 9, Size: int64(len(got))}, err | |||
| 164 | + | }).Once() | |||
| 165 | + | stream := c.UploadReleaseAsset(context.Background()) | |||
| 166 | + | first := &pb.UploadReleaseAssetRequest{Repo: "demo", ReleaseId: 7, Name: "a", Label: "l", ContentType: "application/octet-stream", Size: int64(len(want)), Replace: true} | |||
| 167 | + | if firstData { | |||
| 168 | + | first.Data = full | |||
| 169 | + | } | |||
| 170 | + | require.NoError(t, stream.Send(first)) | |||
| 171 | + | if !firstData { | |||
| 172 | + | require.NoError(t, stream.Send(&pb.UploadReleaseAssetRequest{Data: full})) | |||
| 173 | + | } | |||
| 174 | + | require.NoError(t, stream.Send(&pb.UploadReleaseAssetRequest{Data: tail})) | |||
| 175 | + | res, err := stream.CloseAndReceive() | |||
| 176 | + | require.NoError(t, err) | |||
| 177 | + | assert.EqualValues(t, 9, res.Msg.Asset.Id) | |||
| 178 | + | assert.EqualValues(t, len(want), res.Msg.Asset.Size) | |||
| 179 | + | }) | |||
| 180 | + | } | |||
| 181 | + | } | |||
| 182 | + | ||||
| 183 | + | func TestReleaseRPCUploadInvalid(t *testing.T) { | |||
| 184 | + | tests := []struct { | |||
| 185 | + | name string | |||
| 186 | + | messages []*pb.UploadReleaseAssetRequest | |||
| 187 | + | read bool | |||
| 188 | + | code connect.Code | |||
| 189 | + | }{ | |||
| 190 | + | {name: "empty", code: connect.CodeInvalidArgument}, | |||
| 191 | + | {name: "negative size", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo", Size: -1}}, code: connect.CodeInvalidArgument}, | |||
| 192 | + | {name: "bad repo", messages: []*pb.UploadReleaseAssetRequest{{Repo: "../bad"}}, code: connect.CodeInvalidArgument}, | |||
| 193 | + | {name: "first chunk too big", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo", Data: make([]byte, domain.CIArtifactChunk+1)}}, code: connect.CodeInvalidArgument}, | |||
| 194 | + | {name: "later chunk too big", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo"}, {Data: make([]byte, domain.CIArtifactChunk+1)}}, read: true, code: connect.CodeInvalidArgument}, | |||
| 195 | + | {name: "message exceeds envelope", messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo", Data: make([]byte, RunnerMaxRequestBytes+1)}}, code: connect.CodeResourceExhausted}, | |||
| 196 | + | } | |||
| 197 | + | for _, m := range []*pb.UploadReleaseAssetRequest{{Repo: "demo"}, {ReleaseId: 1}, {Name: "a"}, {Label: "l"}, {ContentType: "text/plain"}, {Size: 1}, {Replace: true}} { | |||
| 198 | + | tests = append(tests, struct { | |||
| 199 | + | name string | |||
| 200 | + | messages []*pb.UploadReleaseAssetRequest | |||
| 201 | + | read bool | |||
| 202 | + | code connect.Code | |||
| 203 | + | }{name: "repeated metadata " + m.String(), messages: []*pb.UploadReleaseAssetRequest{{Repo: "demo"}, m}, read: true, code: connect.CodeInvalidArgument}) | |||
| 204 | + | } | |||
| 205 | + | for _, tt := range tests { | |||
| 206 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 207 | + | rel := mocks.NewReleaseService(t) | |||
| 208 | + | c, _ := releaseClient(t, rel) | |||
| 209 | + | if tt.read { | |||
| 210 | + | rel.EXPECT().UploadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.UploadAssetParams, r io.Reader) (domain.ReleaseAsset, error) { | |||
| 211 | + | _, err := io.ReadAll(r) | |||
| 212 | + | return domain.ReleaseAsset{}, err | |||
| 213 | + | }).Once() | |||
| 214 | + | } | |||
| 215 | + | stream := c.UploadReleaseAsset(context.Background()) | |||
| 216 | + | for _, m := range tt.messages { | |||
| 217 | + | if err := stream.Send(m); err != nil { | |||
| 218 | + | break | |||
| 219 | + | } | |||
| 220 | + | } | |||
| 221 | + | _, err := stream.CloseAndReceive() | |||
| 222 | + | requireCode(t, tt.code, err) | |||
| 223 | + | }) | |||
| 224 | + | } | |||
| 225 | + | } | |||
| 226 | + | ||||
| 227 | + | func TestReleaseRPCDownload(t *testing.T) { | |||
| 228 | + | for _, selector := range []string{"id", "tag", "latest"} { | |||
| 229 | + | t.Run(selector, func(t *testing.T) { | |||
| 230 | + | rel := mocks.NewReleaseService(t) | |||
| 231 | + | c, _ := releaseClient(t, rel) | |||
| 232 | + | req := &pb.DownloadReleaseAssetRequest{Repo: "demo", Offset: 3} | |||
| 233 | + | ref := ports.AssetRef{} | |||
| 234 | + | switch selector { | |||
| 235 | + | case "id": | |||
| 236 | + | req.Selector = &pb.DownloadReleaseAssetRequest_AssetId{AssetId: 9} | |||
| 237 | + | ref.ID = 9 | |||
| 238 | + | case "tag": | |||
| 239 | + | req.Selector = &pb.DownloadReleaseAssetRequest_Tag{Tag: &pb.ReleaseAssetTag{Tag: "v1", Name: "a"}} | |||
| 240 | + | ref.Tag = "v1" | |||
| 241 | + | ref.Name = "a" | |||
| 242 | + | default: | |||
| 243 | + | req.Selector = &pb.DownloadReleaseAssetRequest_Latest{Latest: &pb.ReleaseAssetLatest{Latest: true, Name: "a"}} | |||
| 244 | + | ref.Latest = true | |||
| 245 | + | ref.Name = "a" | |||
| 246 | + | } | |||
| 247 | + | a := domain.ReleaseAsset{ID: 9, ReleaseID: 7, Name: "a", Size: 8} | |||
| 248 | + | rel.EXPECT().DownloadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ref, int64(3), int64(-1), mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.AssetRef, _, _ int64, send func(ports.AssetChunk) error) error { | |||
| 249 | + | if err := send(ports.AssetChunk{Asset: &a}); err != nil { | |||
| 250 | + | return err | |||
| 251 | + | } | |||
| 252 | + | if err := send(ports.AssetChunk{Data: []byte("12")}); err != nil { | |||
| 253 | + | return err | |||
| 254 | + | } | |||
| 255 | + | return send(ports.AssetChunk{Data: []byte("345")}) | |||
| 256 | + | }).Once() | |||
| 257 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(req)) | |||
| 258 | + | require.NoError(t, err) | |||
| 259 | + | defer stream.Close() | |||
| 260 | + | require.True(t, stream.Receive()) | |||
| 261 | + | assert.True(t, proto.Equal(assetToPB(a), stream.Msg().Asset)) | |||
| 262 | + | assert.Empty(t, stream.Msg().Chunk) | |||
| 263 | + | var got []byte | |||
| 264 | + | for stream.Receive() { | |||
| 265 | + | assert.Nil(t, stream.Msg().Asset) | |||
| 266 | + | got = append(got, stream.Msg().Chunk...) | |||
| 267 | + | } | |||
| 268 | + | require.NoError(t, stream.Err()) | |||
| 269 | + | assert.Equal(t, "12345", string(got)) | |||
| 270 | + | }) | |||
| 271 | + | } | |||
| 272 | + | } | |||
| 273 | + | ||||
| 274 | + | func TestReleaseRPCErrors(t *testing.T) { | |||
| 275 | + | for _, tt := range []struct { | |||
| 276 | + | err error | |||
| 277 | + | code connect.Code | |||
| 278 | + | }{{domain.ErrNotFound, connect.CodeNotFound}, {domain.ErrAlreadyExists, connect.CodeAlreadyExists}, {domain.ErrResourceExhausted, connect.CodeResourceExhausted}, {domain.ErrForbidden, connect.CodePermissionDenied}, {domain.ErrOutOfRange, connect.CodeOutOfRange}, {domain.ErrPreconditionFailed, connect.CodeFailedPrecondition}, {domain.ErrInvalid, connect.CodeInvalidArgument}} { | |||
| 279 | + | t.Run(tt.code.String(), func(t *testing.T) { | |||
| 280 | + | rel := mocks.NewReleaseService(t) | |||
| 281 | + | c, _ := releaseClient(t, rel) | |||
| 282 | + | rel.EXPECT().GetRelease(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.ReleaseRef{ID: 7}).Return(domain.Release{}, tt.err).Once() | |||
| 283 | + | _, err := c.GetRelease(context.Background(), connect.NewRequest(&pb.GetReleaseRequest{Repo: "demo", Selector: &pb.GetReleaseRequest_Id{Id: 7}})) | |||
| 284 | + | requireCode(t, tt.code, err) | |||
| 285 | + | rel.EXPECT().DownloadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.AssetRef{ID: 9}, int64(0), int64(-1), mock.Anything).Return(tt.err).Once() | |||
| 286 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(&pb.DownloadReleaseAssetRequest{Repo: "demo", Selector: &pb.DownloadReleaseAssetRequest_AssetId{AssetId: 9}})) | |||
| 287 | + | if err == nil { | |||
| 288 | + | defer stream.Close() | |||
| 289 | + | assert.False(t, stream.Receive()) | |||
| 290 | + | err = stream.Err() | |||
| 291 | + | } | |||
| 292 | + | requireCode(t, tt.code, err) | |||
| 293 | + | }) | |||
| 294 | + | } | |||
| 295 | + | } | |||
| 296 | + | ||||
| 297 | + | func TestReleaseMasksAndSelectors(t *testing.T) { | |||
| 298 | + | paths := []string{"tag", "target_commitish", "title", "notes", "draft", "prerelease", "make_latest"} | |||
| 299 | + | p, err := releaseUpdateParams(&pb.UpdateReleaseRequest{UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}}) | |||
| 300 | + | require.NoError(t, err) | |||
| 301 | + | assert.Equal(t, ports.UpdateReleaseParams{Tag: new(""), TargetCommitish: new(""), Title: new(""), Notes: new(""), Draft: new(false), Prerelease: new(false), MakeLatest: new(false)}, p) | |||
| 302 | + | for _, paths := range [][]string{nil, {"unknown"}, {"*"}, {"title", "assets"}} { | |||
| 303 | + | _, err := releaseUpdateParams(&pb.UpdateReleaseRequest{UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}}) | |||
| 304 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |||
| 305 | + | _, err = assetUpdateParams(&pb.UpdateReleaseAssetRequest{UpdateMask: &fieldmaskpb.FieldMask{Paths: paths}}) | |||
| 306 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |||
| 307 | + | } | |||
| 308 | + | for _, req := range []*pb.GetReleaseRequest{{}, {Selector: &pb.GetReleaseRequest_Id{Id: -1}}, {Selector: &pb.GetReleaseRequest_Tag{}}, {Selector: &pb.GetReleaseRequest_Latest{Latest: false}}} { | |||
| 309 | + | _, err := releaseRef(req) | |||
| 310 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |||
| 311 | + | } | |||
| 312 | + | for _, req := range []*pb.DownloadReleaseAssetRequest{{}, {Selector: &pb.DownloadReleaseAssetRequest_AssetId{}}, {Selector: &pb.DownloadReleaseAssetRequest_Tag{}}, {Selector: &pb.DownloadReleaseAssetRequest_Tag{Tag: &pb.ReleaseAssetTag{Tag: "v1"}}}, {Selector: &pb.DownloadReleaseAssetRequest_Latest{Latest: &pb.ReleaseAssetLatest{Name: "a"}}}} { | |||
| 313 | + | _, err := assetRef(req) | |||
| 314 | + | assert.ErrorIs(t, err, domain.ErrInvalid) | |||
| 315 | + | } | |||
| 316 | + | } | |||
| 317 | + | ||||
| 318 | + | func TestReleaseEvents(t *testing.T) { | |||
| 319 | + | r := domain.Release{ID: 7, Repo: mustRepoName(t, "demo"), Tag: "v1", Draft: true, Prerelease: true} | |||
| 320 | + | at := time.Now().UTC() | |||
| 321 | + | events := []domain.Event{domain.ReleasePublished{Release: r, At: at}, domain.ReleaseUpdated{Release: r, At: at}, domain.ReleaseDeleted{Release: r, At: at}, domain.ReleaseUnpublished{Release: r, At: at}} | |||
| 322 | + | expected := []*pb.Event{ | |||
| 323 | + | {Kind: &pb.Event_ReleasePublished{ReleasePublished: &pb.ReleasePublished{Repo: "demo", Id: 7, Tag: "v1", Draft: true, Prerelease: true}}}, | |||
| 324 | + | {Kind: &pb.Event_ReleaseUpdated{ReleaseUpdated: &pb.ReleaseUpdated{Repo: "demo", Id: 7, Tag: "v1", Draft: true, Prerelease: true}}}, | |||
| 325 | + | {Kind: &pb.Event_ReleaseDeleted{ReleaseDeleted: &pb.ReleaseDeleted{Repo: "demo", Id: 7, Tag: "v1"}}}, | |||
| 326 | + | {Kind: &pb.Event_ReleaseUnpublished{ReleaseUnpublished: &pb.ReleaseUnpublished{Repo: "demo", Id: 7, Tag: "v1", Draft: true, Prerelease: true}}}, | |||
| 327 | + | } | |||
| 328 | + | for i, ev := range events { | |||
| 329 | + | t.Run(ev.EventKind(), func(t *testing.T) { | |||
| 330 | + | expected[i].Id = "event" | |||
| 331 | + | expected[i].Time = timestamp(at) | |||
| 332 | + | assert.True(t, proto.Equal(expected[i], watchEventToPB(ports.WatchEvent{ID: "event", Event: ev}))) | |||
| 333 | + | }) | |||
| 334 | + | } | |||
| 335 | + | } | |||
| 336 | + | ||||
| 337 | + | func TestReleaseResponseEnrichment(t *testing.T) { | |||
| 338 | + | repo := mustRepoName(t, "demo") | |||
| 339 | + | for _, tt := range []struct { | |||
| 340 | + | name string | |||
| 341 | + | latest domain.Release | |||
| 342 | + | latestErr, assetsErr error | |||
| 343 | + | code connect.Code | |||
| 344 | + | wantLatest bool | |||
| 345 | + | }{ | |||
| 346 | + | {name: "draft without latest", latestErr: domain.ErrNotFound}, | |||
| 347 | + | {name: "different latest", latest: domain.Release{ID: 8}}, | |||
| 348 | + | {name: "current latest", latest: domain.Release{ID: 7}, wantLatest: true}, | |||
| 349 | + | {name: "latest lookup fails", latestErr: domain.ErrUnavailable, code: connect.CodeUnavailable}, | |||
| 350 | + | {name: "assets lookup fails", assetsErr: domain.ErrNotFound, code: connect.CodeNotFound}, | |||
| 351 | + | } { | |||
| 352 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 353 | + | rel := mocks.NewReleaseService(t) | |||
| 354 | + | rel.EXPECT().GetRelease(mock.Anything, domain.Owner(), repo, ports.ReleaseRef{ID: 7}).Return(domain.Release{ID: 7, Repo: repo, Draft: true}, nil).Once() | |||
| 355 | + | rel.EXPECT().GetRelease(mock.Anything, domain.Owner(), repo, ports.ReleaseRef{Latest: true}).Return(tt.latest, tt.latestErr).Once() | |||
| 356 | + | if tt.code != connect.CodeUnavailable { | |||
| 357 | + | rel.EXPECT().ListAssets(mock.Anything, domain.Owner(), repo, int64(7)).Return(nil, tt.assetsErr).Once() | |||
| 358 | + | } | |||
| 359 | + | res, err := (&releaseService{rel}).GetRelease(ownerCtx(), connect.NewRequest(&pb.GetReleaseRequest{Repo: "demo", Selector: &pb.GetReleaseRequest_Id{Id: 7}})) | |||
| 360 | + | if tt.code != 0 { | |||
| 361 | + | requireCode(t, tt.code, err) | |||
| 362 | + | return | |||
| 363 | + | } | |||
| 364 | + | require.NoError(t, err) | |||
| 365 | + | assert.Equal(t, tt.wantLatest, res.Msg.Release.Latest) | |||
| 366 | + | assert.Empty(t, res.Msg.Release.Assets) | |||
| 367 | + | assert.Nil(t, res.Msg.Release.PublishedAt) | |||
| 368 | + | }) | |||
| 369 | + | } | |||
| 370 | + | } | |||
| 371 | + | ||||
| 372 | + | func TestReleaseRPCAuthenticatedCaller(t *testing.T) { | |||
| 373 | + | rel := mocks.NewReleaseService(t) | |||
| 374 | + | auth := mocks.NewAuthService(t) | |||
| 375 | + | auth.EXPECT().Authenticate(mock.Anything, "secret").Return(domain.Owner(), nil).Once() | |||
| 376 | + | rel.EXPECT().DeleteRelease(mock.Anything, domain.Owner(), mustRepoName(t, "demo"), int64(7)).Return(nil).Once() | |||
| 377 | + | srv := httptest.NewServer(New(Config{Releases: rel, Auth: auth, Sessions: mocks.NewSessionService(t)}).Handler()) | |||
| 378 | + | defer srv.Close() | |||
| 379 | + | c := ermiteconnect.NewReleaseServiceClient(srv.Client(), srv.URL) | |||
| 380 | + | req := connect.NewRequest(&pb.DeleteReleaseRequest{Repo: "demo", Id: 7}) | |||
| 381 | + | req.Header().Set("Authorization", "Bearer secret") | |||
| 382 | + | _, err := c.DeleteRelease(context.Background(), req) | |||
| 383 | + | require.NoError(t, err) | |||
| 384 | + | } | |||
| 385 | + | ||||
| 386 | + | func TestReleaseRPCDownloadFullChunk(t *testing.T) { | |||
| 387 | + | rel := mocks.NewReleaseService(t) | |||
| 388 | + | c, _ := releaseClient(t, rel) | |||
| 389 | + | data := bytes.Repeat([]byte("x"), domain.CIArtifactChunk) | |||
| 390 | + | rel.EXPECT().DownloadAsset(mock.Anything, domain.Anonymous(), mustRepoName(t, "demo"), ports.AssetRef{ID: 9}, int64(0), int64(-1), mock.Anything).RunAndReturn(func(_ context.Context, _ domain.Principal, _ domain.RepoName, _ ports.AssetRef, _, _ int64, send func(ports.AssetChunk) error) error { | |||
| 391 | + | if err := send(ports.AssetChunk{Asset: &domain.ReleaseAsset{ID: 9, Size: int64(len(data))}}); err != nil { | |||
| 392 | + | return err | |||
| 393 | + | } | |||
| 394 | + | return send(ports.AssetChunk{Data: data}) | |||
| 395 | + | }).Once() | |||
| 396 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(&pb.DownloadReleaseAssetRequest{Repo: "demo", Selector: &pb.DownloadReleaseAssetRequest_AssetId{AssetId: 9}})) | |||
| 397 | + | require.NoError(t, err) | |||
| 398 | + | defer stream.Close() | |||
| 399 | + | require.True(t, stream.Receive()) | |||
| 400 | + | require.NotNil(t, stream.Msg().Asset) | |||
| 401 | + | require.True(t, stream.Receive()) | |||
| 402 | + | assert.Equal(t, data, stream.Msg().Chunk) | |||
| 403 | + | assert.False(t, stream.Receive()) | |||
| 404 | + | require.NoError(t, stream.Err()) | |||
| 405 | + | } | |||
| 406 | + | ||||
| 407 | + | func TestReleaseRPCRejectsInvalidMasksAndSelectors(t *testing.T) { | |||
| 408 | + | c, _ := releaseClient(t, mocks.NewReleaseService(t)) | |||
| 409 | + | _, err := c.UpdateRelease(context.Background(), connect.NewRequest(&pb.UpdateReleaseRequest{Repo: "demo", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"unknown"}}})) | |||
| 410 | + | requireCode(t, connect.CodeInvalidArgument, err) | |||
| 411 | + | _, err = c.UpdateReleaseAsset(context.Background(), connect.NewRequest(&pb.UpdateReleaseAssetRequest{Repo: "demo"})) | |||
| 412 | + | requireCode(t, connect.CodeInvalidArgument, err) | |||
| 413 | + | _, err = c.GetRelease(context.Background(), connect.NewRequest(&pb.GetReleaseRequest{Repo: "demo", Selector: &pb.GetReleaseRequest_Latest{Latest: false}})) | |||
| 414 | + | requireCode(t, connect.CodeInvalidArgument, err) | |||
| 415 | + | stream, err := c.DownloadReleaseAsset(context.Background(), connect.NewRequest(&pb.DownloadReleaseAssetRequest{Repo: "demo", Selector: &pb.DownloadReleaseAssetRequest_Latest{Latest: &pb.ReleaseAssetLatest{Name: "a"}}})) | |||
| 416 | + | if err == nil { | |||
| 417 | + | defer stream.Close() | |||
| 418 | + | assert.False(t, stream.Receive()) | |||
| 419 | + | err = stream.Err() | |||
| 420 | + | } | |||
| 421 | + | requireCode(t, connect.CodeInvalidArgument, err) | |||
| 422 | + | } | |||