feat: implement full native push with image reading and manifest upload

commit: 515ee12c6c9466923b5d22c8f609ea9e7145cea0
author: brice · 7mo ago · 2026-03-14 14:03 UTC
parent: a235943
signature: unsigned

3 files · +453 -4

M go.sum +19 -0 generated file: collapsed

M pkg/registrypush/push.go +174 -4


@@ -2,12 +2,19 @@ package registrypush
2	2
3	3		import (
4	4		"context"
	5	+	"crypto/sha256"
	6	+	"encoding/base64"
5	7		"fmt"
6	8		"io"
7	9		"net/http"
8	10		"net/url"
	11	+	"strings"
9	12
	13	+	"github.com/docker/docker/client"
	14	+	"github.com/google/go-containerregistry/pkg/authn"
	15	+	"github.com/google/go-containerregistry/pkg/name"
10	16		v1 "github.com/google/go-containerregistry/pkg/v1"
	17	+	"github.com/google/go-containerregistry/pkg/v1/tarball"
11	18		)
12	19
13	20		// DefaultChunkSize is 50MB - safely under Cloudflare's 100MB per-request limit

@@ -70,8 +77,38 @@ func New(opts ...Option) *Pusher {
70	77		//
71	78		// The ref must be fully qualified: registry.example.com/name:tag
72	79		func (p *Pusher) Push(ctx context.Context, ref string) error {
73		-	// Implementation in subsequent tasks
74		-	return nil
	80	+	parsedRef, err := name.ParseReference(ref)
	81	+	if err != nil {
	82	+	return fmt.Errorf("failed to parse image ref %s: %w", ref, err)
	83	+	}
	84	+
	85	+	tag, ok := parsedRef.(name.Tag)
	86	+	if !ok {
	87	+	return fmt.Errorf("image ref %s must include a tag", ref)
	88	+	}
	89	+
	90	+	authHeader, err := resolveAuthHeader(parsedRef)
	91	+	if err != nil {
	92	+	return err
	93	+	}
	94	+
	95	+	img, err := p.imageSource(ctx, ref)
	96	+	if err != nil {
	97	+	return err
	98	+	}
	99	+
	100	+	baseURL := registryBaseURL(parsedRef.Context().RegistryStr())
	101	+	repo := parsedRef.Context().RepositoryStr()
	102	+
	103	+	if err := p.uploadImageLayers(ctx, img, baseURL, repo, authHeader); err != nil {
	104	+	return err
	105	+	}
	106	+
	107	+	if err := p.uploadImageConfig(ctx, img, baseURL, repo, authHeader); err != nil {
	108	+	return err
	109	+	}
	110	+
	111	+	return p.uploadManifest(ctx, baseURL, repo, tag.Identifier(), img, authHeader)
75	112		}
76	113
77	114		// UploadBlob uploads a single blob to the registry using chunked uploads.

@@ -105,8 +142,141 @@ func (p *Pusher) UploadBlob(ctx context.Context, baseURL, repo, digest string, s
105	142
106	143		// defaultImageSource reads an image from the local Docker/Podman daemon.
107	144		func defaultImageSource(ctx context.Context, ref string) (v1.Image, error) {
108		-	// Implementation in Task 3
109		-	return nil, nil
	145	+	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
	146	+	if err != nil {
	147	+	return nil, fmt.Errorf("failed to create Docker client: %w", err)
	148	+	}
	149	+	defer cli.Close()
	150	+
	151	+	tag, err := name.NewTag(ref)
	152	+	if err != nil {
	153	+	return nil, fmt.Errorf("failed to parse image ref %s: %w", ref, err)
	154	+	}
	155	+
	156	+	rc, err := cli.ImageSave(ctx, []string{ref})
	157	+	if err != nil {
	158	+	return nil, fmt.Errorf("failed to export image %s: %w", ref, err)
	159	+	}
	160	+	defer rc.Close()
	161	+
	162	+	img, err := tarball.Image(func() (io.ReadCloser, error) {
	163	+	return rc, nil
	164	+	}, &tag)
	165	+	if err != nil {
	166	+	return nil, fmt.Errorf("failed to parse image tar for %s: %w", ref, err)
	167	+	}
	168	+	return img, nil
	169	+	}
	170	+
	171	+	func (p *Pusher) uploadImageLayers(ctx context.Context, img v1.Image, baseURL, repo, auth string) error {
	172	+	layers, err := img.Layers()
	173	+	if err != nil {
	174	+	return fmt.Errorf("failed to read image layers: %w", err)
	175	+	}
	176	+
	177	+	for _, layer := range layers {
	178	+	if err := p.uploadLayer(ctx, baseURL, repo, layer, auth); err != nil {
	179	+	return err
	180	+	}
	181	+	}
	182	+
	183	+	return nil
	184	+	}
	185	+
	186	+	func (p *Pusher) uploadLayer(ctx context.Context, baseURL, repo string, layer v1.Layer, auth string) error {
	187	+	digest, err := layer.Digest()
	188	+	if err != nil {
	189	+	return fmt.Errorf("failed to get layer digest: %w", err)
	190	+	}
	191	+
	192	+	size, err := layer.Size()
	193	+	if err != nil {
	194	+	return fmt.Errorf("failed to get layer size for %s: %w", digest, err)
	195	+	}
	196	+
	197	+	content, err := layer.Compressed()
	198	+	if err != nil {
	199	+	return fmt.Errorf("failed to read layer %s: %w", digest, err)
	200	+	}
	201	+	defer content.Close()
	202	+
	203	+	if err := p.UploadBlob(ctx, baseURL, repo, digest.String(), size, content, auth); err != nil {
	204	+	return fmt.Errorf("failed to upload layer %s: %w", digest, err)
	205	+	}
	206	+
	207	+	return nil
	208	+	}
	209	+
	210	+	func (p *Pusher) uploadImageConfig(ctx context.Context, img v1.Image, baseURL, repo, auth string) error {
	211	+	config, err := img.RawConfigFile()
	212	+	if err != nil {
	213	+	return fmt.Errorf("failed to read image config: %w", err)
	214	+	}
	215	+
	216	+	digest := fmt.Sprintf("sha256:%x", sha256.Sum256(config))
	217	+	if err := p.UploadBlob(ctx, baseURL, repo, digest, int64(len(config)), bytesReader(config), auth); err != nil {
	218	+	return fmt.Errorf("failed to upload image config %s: %w", digest, err)
	219	+	}
	220	+
	221	+	return nil
	222	+	}
	223	+
	224	+	func (p *Pusher) uploadManifest(ctx context.Context, baseURL, repo, tag string, img v1.Image, auth string) error {
	225	+	manifest, err := img.RawManifest()
	226	+	if err != nil {
	227	+	return fmt.Errorf("failed to read image manifest: %w", err)
	228	+	}
	229	+
	230	+	mediaType, err := img.MediaType()
	231	+	if err != nil {
	232	+	return fmt.Errorf("failed to read image media type: %w", err)
	233	+	}
	234	+
	235	+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, fmt.Sprintf("%s/v2/%s/manifests/%s", baseURL, repo, tag), bytesReader(manifest))
	236	+	if err != nil {
	237	+	return fmt.Errorf("failed to create manifest upload request: %w", err)
	238	+	}
	239	+	setAuthHeader(req, auth)
	240	+	req.Header.Set("Content-Type", string(mediaType))
	241	+	req.ContentLength = int64(len(manifest))
	242	+
	243	+	resp, err := p.httpClient().Do(req)
	244	+	if err != nil {
	245	+	return fmt.Errorf("failed to upload manifest: %w", err)
	246	+	}
	247	+	defer resp.Body.Close()
	248	+
	249	+	if resp.StatusCode != http.StatusCreated {
	250	+	return fmt.Errorf("manifest upload returned status %d", resp.StatusCode)
	251	+	}
	252	+
	253	+	return nil
	254	+	}
	255	+
	256	+	func resolveAuthHeader(ref name.Reference) (string, error) {
	257	+	authenticator, err := authn.DefaultKeychain.Resolve(ref.Context().Registry)
	258	+	if err != nil {
	259	+	return "", fmt.Errorf("failed to resolve registry credentials for %s: %w", ref.Context().RegistryStr(), err)
	260	+	}
	261	+
	262	+	authConfig, err := authenticator.Authorization()
	263	+	if err != nil {
	264	+	return "", fmt.Errorf("failed to authorize for registry %s: %w", ref.Context().RegistryStr(), err)
	265	+	}
	266	+
	267	+	if authConfig == nil \|\| (authConfig.Username == "" && authConfig.Password == "") {
	268	+	return "", nil
	269	+	}
	270	+
	271	+	encoded := base64.StdEncoding.EncodeToString([]byte(authConfig.Username + ":" + authConfig.Password))
	272	+	return "Basic " + encoded, nil
	273	+	}
	274	+
	275	+	func registryBaseURL(host string) string {
	276	+	if strings.HasPrefix(host, "localhost:") \|\| strings.HasPrefix(host, "127.0.0.1:") \|\| host == "localhost" \|\| host == "127.0.0.1" {
	277	+	return "http://" + host
	278	+	}
	279	+	return "https://" + host
110	280		}
111	281
112	282		var errBlobMissing = fmt.Errorf("blob missing")


@@ -2,12 +2,19 @@ package registrypush
2			2
3		import (	3		import (
4		"context"	4		"context"
			5	+	"crypto/sha256"
			6	+	"encoding/base64"
5		"fmt"	7		"fmt"
6		"io"	8		"io"
7		"net/http"	9		"net/http"
8		"net/url"	10		"net/url"
			11	+	"strings"
9			12
			13	+	"github.com/docker/docker/client"
			14	+	"github.com/google/go-containerregistry/pkg/authn"
			15	+	"github.com/google/go-containerregistry/pkg/name"
10		v1 "github.com/google/go-containerregistry/pkg/v1"	16		v1 "github.com/google/go-containerregistry/pkg/v1"
			17	+	"github.com/google/go-containerregistry/pkg/v1/tarball"
11		)	18		)
12			19
13		// DefaultChunkSize is 50MB - safely under Cloudflare's 100MB per-request limit	20		// DefaultChunkSize is 50MB - safely under Cloudflare's 100MB per-request limit

@@ -70,8 +77,38 @@ func New(opts ...Option) *Pusher {
70		//	77		//
71		// The ref must be fully qualified: registry.example.com/name:tag	78		// The ref must be fully qualified: registry.example.com/name:tag
72		func (p *Pusher) Push(ctx context.Context, ref string) error {	79		func (p *Pusher) Push(ctx context.Context, ref string) error {
73	-	// Implementation in subsequent tasks	80	+	parsedRef, err := name.ParseReference(ref)
74	-	return nil	81	+	if err != nil {
			82	+	return fmt.Errorf("failed to parse image ref %s: %w", ref, err)
			83	+	}
			84	+
			85	+	tag, ok := parsedRef.(name.Tag)
			86	+	if !ok {
			87	+	return fmt.Errorf("image ref %s must include a tag", ref)
			88	+	}
			89	+
			90	+	authHeader, err := resolveAuthHeader(parsedRef)
			91	+	if err != nil {
			92	+	return err
			93	+	}
			94	+
			95	+	img, err := p.imageSource(ctx, ref)
			96	+	if err != nil {
			97	+	return err
			98	+	}
			99	+
			100	+	baseURL := registryBaseURL(parsedRef.Context().RegistryStr())
			101	+	repo := parsedRef.Context().RepositoryStr()
			102	+
			103	+	if err := p.uploadImageLayers(ctx, img, baseURL, repo, authHeader); err != nil {
			104	+	return err
			105	+	}
			106	+
			107	+	if err := p.uploadImageConfig(ctx, img, baseURL, repo, authHeader); err != nil {
			108	+	return err
			109	+	}
			110	+
			111	+	return p.uploadManifest(ctx, baseURL, repo, tag.Identifier(), img, authHeader)
75		}	112		}
76			113
77		// UploadBlob uploads a single blob to the registry using chunked uploads.	114		// UploadBlob uploads a single blob to the registry using chunked uploads.

@@ -105,8 +142,141 @@ func (p *Pusher) UploadBlob(ctx context.Context, baseURL, repo, digest string, s
105			142
106		// defaultImageSource reads an image from the local Docker/Podman daemon.	143		// defaultImageSource reads an image from the local Docker/Podman daemon.
107		func defaultImageSource(ctx context.Context, ref string) (v1.Image, error) {	144		func defaultImageSource(ctx context.Context, ref string) (v1.Image, error) {
108	-	// Implementation in Task 3	145	+	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
109	-	return nil, nil	146	+	if err != nil {
			147	+	return nil, fmt.Errorf("failed to create Docker client: %w", err)
			148	+	}
			149	+	defer cli.Close()
			150	+
			151	+	tag, err := name.NewTag(ref)
			152	+	if err != nil {
			153	+	return nil, fmt.Errorf("failed to parse image ref %s: %w", ref, err)
			154	+	}
			155	+
			156	+	rc, err := cli.ImageSave(ctx, []string{ref})
			157	+	if err != nil {
			158	+	return nil, fmt.Errorf("failed to export image %s: %w", ref, err)
			159	+	}
			160	+	defer rc.Close()
			161	+
			162	+	img, err := tarball.Image(func() (io.ReadCloser, error) {
			163	+	return rc, nil
			164	+	}, &tag)
			165	+	if err != nil {
			166	+	return nil, fmt.Errorf("failed to parse image tar for %s: %w", ref, err)
			167	+	}
			168	+	return img, nil
			169	+	}
			170	+
			171	+	func (p *Pusher) uploadImageLayers(ctx context.Context, img v1.Image, baseURL, repo, auth string) error {
			172	+	layers, err := img.Layers()
			173	+	if err != nil {
			174	+	return fmt.Errorf("failed to read image layers: %w", err)
			175	+	}
			176	+
			177	+	for _, layer := range layers {
			178	+	if err := p.uploadLayer(ctx, baseURL, repo, layer, auth); err != nil {
			179	+	return err
			180	+	}
			181	+	}
			182	+
			183	+	return nil
			184	+	}
			185	+
			186	+	func (p *Pusher) uploadLayer(ctx context.Context, baseURL, repo string, layer v1.Layer, auth string) error {
			187	+	digest, err := layer.Digest()
			188	+	if err != nil {
			189	+	return fmt.Errorf("failed to get layer digest: %w", err)
			190	+	}
			191	+
			192	+	size, err := layer.Size()
			193	+	if err != nil {
			194	+	return fmt.Errorf("failed to get layer size for %s: %w", digest, err)
			195	+	}
			196	+
			197	+	content, err := layer.Compressed()
			198	+	if err != nil {
			199	+	return fmt.Errorf("failed to read layer %s: %w", digest, err)
			200	+	}
			201	+	defer content.Close()
			202	+
			203	+	if err := p.UploadBlob(ctx, baseURL, repo, digest.String(), size, content, auth); err != nil {
			204	+	return fmt.Errorf("failed to upload layer %s: %w", digest, err)
			205	+	}
			206	+
			207	+	return nil
			208	+	}
			209	+
			210	+	func (p *Pusher) uploadImageConfig(ctx context.Context, img v1.Image, baseURL, repo, auth string) error {
			211	+	config, err := img.RawConfigFile()
			212	+	if err != nil {
			213	+	return fmt.Errorf("failed to read image config: %w", err)
			214	+	}
			215	+
			216	+	digest := fmt.Sprintf("sha256:%x", sha256.Sum256(config))
			217	+	if err := p.UploadBlob(ctx, baseURL, repo, digest, int64(len(config)), bytesReader(config), auth); err != nil {
			218	+	return fmt.Errorf("failed to upload image config %s: %w", digest, err)
			219	+	}
			220	+
			221	+	return nil
			222	+	}
			223	+
			224	+	func (p *Pusher) uploadManifest(ctx context.Context, baseURL, repo, tag string, img v1.Image, auth string) error {
			225	+	manifest, err := img.RawManifest()
			226	+	if err != nil {
			227	+	return fmt.Errorf("failed to read image manifest: %w", err)
			228	+	}
			229	+
			230	+	mediaType, err := img.MediaType()
			231	+	if err != nil {
			232	+	return fmt.Errorf("failed to read image media type: %w", err)
			233	+	}
			234	+
			235	+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, fmt.Sprintf("%s/v2/%s/manifests/%s", baseURL, repo, tag), bytesReader(manifest))
			236	+	if err != nil {
			237	+	return fmt.Errorf("failed to create manifest upload request: %w", err)
			238	+	}
			239	+	setAuthHeader(req, auth)
			240	+	req.Header.Set("Content-Type", string(mediaType))
			241	+	req.ContentLength = int64(len(manifest))
			242	+
			243	+	resp, err := p.httpClient().Do(req)
			244	+	if err != nil {
			245	+	return fmt.Errorf("failed to upload manifest: %w", err)
			246	+	}
			247	+	defer resp.Body.Close()
			248	+
			249	+	if resp.StatusCode != http.StatusCreated {
			250	+	return fmt.Errorf("manifest upload returned status %d", resp.StatusCode)
			251	+	}
			252	+
			253	+	return nil
			254	+	}
			255	+
			256	+	func resolveAuthHeader(ref name.Reference) (string, error) {
			257	+	authenticator, err := authn.DefaultKeychain.Resolve(ref.Context().Registry)
			258	+	if err != nil {
			259	+	return "", fmt.Errorf("failed to resolve registry credentials for %s: %w", ref.Context().RegistryStr(), err)
			260	+	}
			261	+
			262	+	authConfig, err := authenticator.Authorization()
			263	+	if err != nil {
			264	+	return "", fmt.Errorf("failed to authorize for registry %s: %w", ref.Context().RegistryStr(), err)
			265	+	}
			266	+
			267	+	if authConfig == nil \|\| (authConfig.Username == "" && authConfig.Password == "") {
			268	+	return "", nil
			269	+	}
			270	+
			271	+	encoded := base64.StdEncoding.EncodeToString([]byte(authConfig.Username + ":" + authConfig.Password))
			272	+	return "Basic " + encoded, nil
			273	+	}
			274	+
			275	+	func registryBaseURL(host string) string {
			276	+	if strings.HasPrefix(host, "localhost:") \|\| strings.HasPrefix(host, "127.0.0.1:") \|\| host == "localhost" \|\| host == "127.0.0.1" {
			277	+	return "http://" + host
			278	+	}
			279	+	return "https://" + host
110		}	280		}
111			281
112		var errBlobMissing = fmt.Errorf("blob missing")	282		var errBlobMissing = fmt.Errorf("blob missing")

M pkg/registrypush/push_test.go +260 -0


@@ -3,14 +3,18 @@ package registrypush_test
3	3		import (
4	4		"bytes"
5	5		"context"
	6	+	"crypto/sha256"
6	7		"fmt"
7	8		"io"
8	9		"net/http"
9	10		"net/http/httptest"
	11	+	"net/url"
10	12		"strconv"
11	13		"strings"
12	14		"testing"
13	15
	16	+	"github.com/google/go-containerregistry/pkg/v1"
	17	+	"github.com/google/go-containerregistry/pkg/v1/random"
14	18		"github.com/stretchr/testify/assert"
15	19		"github.com/stretchr/testify/require"
16	20

@@ -119,6 +123,100 @@ func TestPusher_UploadBlob(t *testing.T) {
119	123		}
120	124		}
121	125
	126	+	func TestPusher_Push(t *testing.T) {
	127	+	testCases := []struct {
	128	+	name string
	129	+	imageSize int64
	130	+	layerCount int64
	131	+	existingBlobDigests func(t *testing.T, img v1.Image) map[string]bool
	132	+	expectedUploadedBlobCount int
	133	+	expectedManifestMediaTypes []string
	134	+	}{
	135	+	{
	136	+	name: "single-layer image uploads config layer and manifest",
	137	+	imageSize: 1024,
	138	+	layerCount: 1,
	139	+	existingBlobDigests: func(t *testing.T, img v1.Image) map[string]bool { return map[string]bool{} },
	140	+	expectedUploadedBlobCount: 2,
	141	+	expectedManifestMediaTypes: []string{
	142	+	"application/vnd.docker.distribution.manifest.v2+json",
	143	+	string(typesOCIManifestSchema1()),
	144	+	},
	145	+	},
	146	+	{
	147	+	name: "multi-layer image uploads all blobs and manifest",
	148	+	imageSize: 1024,
	149	+	layerCount: 2,
	150	+	existingBlobDigests: func(t *testing.T, img v1.Image) map[string]bool { return map[string]bool{} },
	151	+	expectedUploadedBlobCount: 3,
	152	+	expectedManifestMediaTypes: []string{
	153	+	"application/vnd.docker.distribution.manifest.v2+json",
	154	+	string(typesOCIManifestSchema1()),
	155	+	},
	156	+	},
	157	+	{
	158	+	name: "existing layer is skipped while others upload",
	159	+	imageSize: 1024,
	160	+	layerCount: 2,
	161	+	existingBlobDigests: func(t *testing.T, img v1.Image) map[string]bool {
	162	+	t.Helper()
	163	+	layers, err := img.Layers()
	164	+	require.NoError(t, err)
	165	+	digest, err := layers[0].Digest()
	166	+	require.NoError(t, err)
	167	+	return map[string]bool{digest.String(): true}
	168	+	},
	169	+	expectedUploadedBlobCount: 2,
	170	+	expectedManifestMediaTypes: []string{
	171	+	"application/vnd.docker.distribution.manifest.v2+json",
	172	+	string(typesOCIManifestSchema1()),
	173	+	},
	174	+	},
	175	+	}
	176	+
	177	+	for _, tc := range testCases {
	178	+	t.Run(tc.name, func(t *testing.T) {
	179	+	img, err := random.Image(tc.imageSize, tc.layerCount)
	180	+	require.NoError(t, err)
	181	+
	182	+	serverState := newFakePushRegistry(t, img, tc.existingBlobDigests(t, img))
	183	+	server := httptest.NewServer(serverState.handler())
	184	+	defer server.Close()
	185	+
	186	+	serverURL, err := url.Parse(server.URL)
	187	+	require.NoError(t, err)
	188	+	ref := serverURL.Host + "/demo/app:v1"
	189	+
	190	+	p := registrypush.New(
	191	+	registrypush.WithTransport(server.Client().Transport),
	192	+	registrypush.WithImageSource(func(context.Context, string) (v1.Image, error) {
	193	+	return img, nil
	194	+	}),
	195	+	)
	196	+
	197	+	err = p.Push(context.Background(), ref)
	198	+	require.NoError(t, err)
	199	+
	200	+	expectedDigests := make([]string, 0, len(serverState.expectedBlobs))
	201	+	for digest := range serverState.expectedBlobs {
	202	+	expectedDigests = append(expectedDigests, digest)
	203	+	assert.Contains(t, serverState.headChecks, digest)
	204	+	}
	205	+	assert.Len(t, serverState.headChecks, len(expectedDigests))
	206	+
	207	+	assert.Len(t, serverState.uploadedBlobs, tc.expectedUploadedBlobCount)
	208	+	for digest, content := range serverState.uploadedBlobs {
	209	+	assert.Equal(t, serverState.expectedBlobs[digest], content)
	210	+	}
	211	+
	212	+	manifestPut, ok := serverState.manifestPuts[serverState.expectedRepo+":"+serverState.expectedTag]
	213	+	require.True(t, ok)
	214	+	assert.Contains(t, tc.expectedManifestMediaTypes, manifestPut.contentType)
	215	+	assert.Equal(t, serverState.expectedManifest, manifestPut.body)
	216	+	})
	217	+	}
	218	+	}
	219	+
122	220		type fakeBlobUploadRegistry struct {
123	221		t *testing.T
124	222		expectedAuth string

@@ -226,3 +324,165 @@ func rangeHeaderValue(start, end int) string {
226	324		}
227	325		return fmt.Sprintf("%d-%d", start, end)
228	326		}
	327	+
	328	+	type fakePushRegistry struct {
	329	+	t *testing.T
	330	+	existingBlobs map[string]bool
	331	+	uploadedBlobs map[string][]byte
	332	+	manifestPuts map[string]fakeManifestPut
	333	+	uploads map[string]*fakeUploadSession
	334	+	nextUploadID int
	335	+	headChecks []string
	336	+	expectedBlobs map[string][]byte
	337	+	expectedRepo string
	338	+	expectedTag string
	339	+	expectedMT string
	340	+	expectedManifest []byte
	341	+	}
	342	+
	343	+	type fakeUploadSession struct {
	344	+	repo string
	345	+	data []byte
	346	+	}
	347	+
	348	+	type fakeManifestPut struct {
	349	+	contentType string
	350	+	body []byte
	351	+	}
	352	+
	353	+	func newFakePushRegistry(t testing.T, img v1.Image, existingBlobs map[string]bool) fakePushRegistry {
	354	+	t.Helper()
	355	+
	356	+	layers, err := img.Layers()
	357	+	require.NoError(t, err)
	358	+
	359	+	expectedBlobs := make(map[string][]byte, len(layers)+1)
	360	+	for _, layer := range layers {
	361	+	digest, err := layer.Digest()
	362	+	require.NoError(t, err)
	363	+	rc, err := layer.Compressed()
	364	+	require.NoError(t, err)
	365	+	content, err := io.ReadAll(rc)
	366	+	require.NoError(t, err)
	367	+	require.NoError(t, rc.Close())
	368	+	expectedBlobs[digest.String()] = content
	369	+	}
	370	+
	371	+	config, err := img.RawConfigFile()
	372	+	require.NoError(t, err)
	373	+	configDigest := fmt.Sprintf("sha256:%x", sha256.Sum256(config))
	374	+	expectedBlobs[configDigest] = config
	375	+
	376	+	manifest, err := img.RawManifest()
	377	+	require.NoError(t, err)
	378	+	mediaType, err := img.MediaType()
	379	+	require.NoError(t, err)
	380	+
	381	+	return &fakePushRegistry{
	382	+	t: t,
	383	+	existingBlobs: existingBlobs,
	384	+	uploadedBlobs: make(map[string][]byte),
	385	+	manifestPuts: make(map[string]fakeManifestPut),
	386	+	uploads: make(map[string]*fakeUploadSession),
	387	+	expectedBlobs: expectedBlobs,
	388	+	expectedRepo: "demo/app",
	389	+	expectedTag: "v1",
	390	+	expectedMT: string(mediaType),
	391	+	expectedManifest: manifest,
	392	+	}
	393	+	}
	394	+
	395	+	func (f *fakePushRegistry) handler() http.Handler {
	396	+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	397	+	switch {
	398	+	case r.Method == http.MethodHead && strings.HasPrefix(r.URL.Path, "/v2/") && strings.Contains(r.URL.Path, "/blobs/"):
	399	+	f.handleBlobHead(w, r)
	400	+	case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/blobs/uploads/"):
	401	+	f.handleBlobStart(w, r)
	402	+	case r.Method == http.MethodPatch && strings.Contains(r.URL.Path, "/blobs/uploads/"):
	403	+	f.handleBlobPatch(w, r)
	404	+	case r.Method == http.MethodPut && strings.Contains(r.URL.Path, "/blobs/uploads/"):
	405	+	f.handleBlobFinalize(w, r)
	406	+	case r.Method == http.MethodPut && strings.Contains(r.URL.Path, "/manifests/"):
	407	+	f.handleManifestPut(w, r)
	408	+	default:
	409	+	f.t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String())
	410	+	}
	411	+	})
	412	+	}
	413	+
	414	+	func (f fakePushRegistry) handleBlobHead(w http.ResponseWriter, r http.Request) {
	415	+	digest := pathTail(r.URL.Path)
	416	+	f.headChecks = append(f.headChecks, digest)
	417	+	if f.existingBlobs[digest] {
	418	+	w.WriteHeader(http.StatusOK)
	419	+	return
	420	+	}
	421	+	w.WriteHeader(http.StatusNotFound)
	422	+	}
	423	+
	424	+	func (f fakePushRegistry) handleBlobStart(w http.ResponseWriter, r http.Request) {
	425	+	repo := pathSegmentBetween(r.URL.Path, "/v2/", "/blobs/uploads/")
	426	+	assert.Equal(f.t, f.expectedRepo, repo)
	427	+	f.nextUploadID++
	428	+	uploadPath := fmt.Sprintf("/v2/%s/blobs/uploads/upload-%d", repo, f.nextUploadID)
	429	+	f.uploads[uploadPath] = &fakeUploadSession{repo: repo}
	430	+	w.Header().Set("Location", uploadPath)
	431	+	w.WriteHeader(http.StatusAccepted)
	432	+	}
	433	+
	434	+	func (f fakePushRegistry) handleBlobPatch(w http.ResponseWriter, r http.Request) {
	435	+	session := f.uploads[r.URL.Path]
	436	+	if session == nil {
	437	+	f.t.Fatalf("missing upload session for %s", r.URL.Path)
	438	+	}
	439	+	body, err := io.ReadAll(r.Body)
	440	+	require.NoError(f.t, err)
	441	+	session.data = append(session.data, body...)
	442	+	w.Header().Set("Location", r.URL.Path)
	443	+	w.WriteHeader(http.StatusAccepted)
	444	+	}
	445	+
	446	+	func (f fakePushRegistry) handleBlobFinalize(w http.ResponseWriter, r http.Request) {
	447	+	session := f.uploads[r.URL.Path]
	448	+	if session == nil {
	449	+	f.t.Fatalf("missing upload session for %s", r.URL.Path)
	450	+	}
	451	+	digest := r.URL.Query().Get("digest")
	452	+	f.uploadedBlobs[digest] = append([]byte(nil), session.data...)
	453	+	delete(f.uploads, r.URL.Path)
	454	+	w.WriteHeader(http.StatusCreated)
	455	+	}
	456	+
	457	+	func (f fakePushRegistry) handleManifestPut(w http.ResponseWriter, r http.Request) {
	458	+	repo := pathSegmentBetween(r.URL.Path, "/v2/", "/manifests/")
	459	+	tag := pathTail(r.URL.Path)
	460	+	assert.Equal(f.t, f.expectedRepo, repo)
	461	+	assert.Equal(f.t, f.expectedTag, tag)
	462	+	body, err := io.ReadAll(r.Body)
	463	+	require.NoError(f.t, err)
	464	+	f.manifestPuts[repo+":"+tag] = fakeManifestPut{contentType: r.Header.Get("Content-Type"), body: body}
	465	+	w.WriteHeader(http.StatusCreated)
	466	+	}
	467	+
	468	+	func pathTail(path string) string {
	469	+	parts := strings.Split(path, "/")
	470	+	return parts[len(parts)-1]
	471	+	}
	472	+
	473	+	func pathSegmentBetween(value, start, end string) string {
	474	+	startIdx := strings.Index(value, start)
	475	+	if startIdx == -1 {
	476	+	return ""
	477	+	}
	478	+	trimmed := value[startIdx+len(start):]
	479	+	endIdx := strings.Index(trimmed, end)
	480	+	if endIdx == -1 {
	481	+	return ""
	482	+	}
	483	+	return trimmed[:endIdx]
	484	+	}
	485	+
	486	+	func typesOCIManifestSchema1() string {
	487	+	return "application/vnd.oci.image.manifest.v1+json"
	488	+	}


@@ -3,14 +3,18 @@ package registrypush_test
3	import (	3		import (
4	"bytes"	4		"bytes"
5	"context"	5		"context"
		6	+	"crypto/sha256"
6	"fmt"	7		"fmt"
7	"io"	8		"io"
8	"net/http"	9		"net/http"
9	"net/http/httptest"	10		"net/http/httptest"
		11	+	"net/url"
10	"strconv"	12		"strconv"
11	"strings"	13		"strings"
12	"testing"	14		"testing"
13		15
		16	+	"github.com/google/go-containerregistry/pkg/v1"
		17	+	"github.com/google/go-containerregistry/pkg/v1/random"
14	"github.com/stretchr/testify/assert"	18		"github.com/stretchr/testify/assert"
15	"github.com/stretchr/testify/require"	19		"github.com/stretchr/testify/require"
16		20

@@ -119,6 +123,100 @@ func TestPusher_UploadBlob(t *testing.T) {
119	}	123		}
120	}	124		}
121		125
		126	+	func TestPusher_Push(t *testing.T) {
		127	+	testCases := []struct {
		128	+	name string
		129	+	imageSize int64
		130	+	layerCount int64
		131	+	existingBlobDigests func(t *testing.T, img v1.Image) map[string]bool
		132	+	expectedUploadedBlobCount int
		133	+	expectedManifestMediaTypes []string
		134	+	}{
		135	+	{
		136	+	name: "single-layer image uploads config layer and manifest",
		137	+	imageSize: 1024,
		138	+	layerCount: 1,
		139	+	existingBlobDigests: func(t *testing.T, img v1.Image) map[string]bool { return map[string]bool{} },
		140	+	expectedUploadedBlobCount: 2,
		141	+	expectedManifestMediaTypes: []string{
		142	+	"application/vnd.docker.distribution.manifest.v2+json",
		143	+	string(typesOCIManifestSchema1()),
		144	+	},
		145	+	},
		146	+	{
		147	+	name: "multi-layer image uploads all blobs and manifest",
		148	+	imageSize: 1024,
		149	+	layerCount: 2,
		150	+	existingBlobDigests: func(t *testing.T, img v1.Image) map[string]bool { return map[string]bool{} },
		151	+	expectedUploadedBlobCount: 3,
		152	+	expectedManifestMediaTypes: []string{
		153	+	"application/vnd.docker.distribution.manifest.v2+json",
		154	+	string(typesOCIManifestSchema1()),
		155	+	},
		156	+	},
		157	+	{
		158	+	name: "existing layer is skipped while others upload",
		159	+	imageSize: 1024,
		160	+	layerCount: 2,
		161	+	existingBlobDigests: func(t *testing.T, img v1.Image) map[string]bool {
		162	+	t.Helper()
		163	+	layers, err := img.Layers()
		164	+	require.NoError(t, err)
		165	+	digest, err := layers[0].Digest()
		166	+	require.NoError(t, err)
		167	+	return map[string]bool{digest.String(): true}
		168	+	},
		169	+	expectedUploadedBlobCount: 2,
		170	+	expectedManifestMediaTypes: []string{
		171	+	"application/vnd.docker.distribution.manifest.v2+json",
		172	+	string(typesOCIManifestSchema1()),
		173	+	},
		174	+	},
		175	+	}
		176	+
		177	+	for _, tc := range testCases {
		178	+	t.Run(tc.name, func(t *testing.T) {
		179	+	img, err := random.Image(tc.imageSize, tc.layerCount)
		180	+	require.NoError(t, err)
		181	+
		182	+	serverState := newFakePushRegistry(t, img, tc.existingBlobDigests(t, img))
		183	+	server := httptest.NewServer(serverState.handler())
		184	+	defer server.Close()
		185	+
		186	+	serverURL, err := url.Parse(server.URL)
		187	+	require.NoError(t, err)
		188	+	ref := serverURL.Host + "/demo/app:v1"
		189	+
		190	+	p := registrypush.New(
		191	+	registrypush.WithTransport(server.Client().Transport),
		192	+	registrypush.WithImageSource(func(context.Context, string) (v1.Image, error) {
		193	+	return img, nil
		194	+	}),
		195	+	)
		196	+
		197	+	err = p.Push(context.Background(), ref)
		198	+	require.NoError(t, err)
		199	+
		200	+	expectedDigests := make([]string, 0, len(serverState.expectedBlobs))
		201	+	for digest := range serverState.expectedBlobs {
		202	+	expectedDigests = append(expectedDigests, digest)
		203	+	assert.Contains(t, serverState.headChecks, digest)
		204	+	}
		205	+	assert.Len(t, serverState.headChecks, len(expectedDigests))
		206	+
		207	+	assert.Len(t, serverState.uploadedBlobs, tc.expectedUploadedBlobCount)
		208	+	for digest, content := range serverState.uploadedBlobs {
		209	+	assert.Equal(t, serverState.expectedBlobs[digest], content)
		210	+	}
		211	+
		212	+	manifestPut, ok := serverState.manifestPuts[serverState.expectedRepo+":"+serverState.expectedTag]
		213	+	require.True(t, ok)
		214	+	assert.Contains(t, tc.expectedManifestMediaTypes, manifestPut.contentType)
		215	+	assert.Equal(t, serverState.expectedManifest, manifestPut.body)
		216	+	})
		217	+	}
		218	+	}
		219	+
122	type fakeBlobUploadRegistry struct {	220		type fakeBlobUploadRegistry struct {
123	t *testing.T	221		t *testing.T
124	expectedAuth string	222		expectedAuth string

@@ -226,3 +324,165 @@ func rangeHeaderValue(start, end int) string {
226	}	324		}
227	return fmt.Sprintf("%d-%d", start, end)	325		return fmt.Sprintf("%d-%d", start, end)
228	}	326		}
		327	+
		328	+	type fakePushRegistry struct {
		329	+	t *testing.T
		330	+	existingBlobs map[string]bool
		331	+	uploadedBlobs map[string][]byte
		332	+	manifestPuts map[string]fakeManifestPut
		333	+	uploads map[string]*fakeUploadSession
		334	+	nextUploadID int
		335	+	headChecks []string
		336	+	expectedBlobs map[string][]byte
		337	+	expectedRepo string
		338	+	expectedTag string
		339	+	expectedMT string
		340	+	expectedManifest []byte
		341	+	}
		342	+
		343	+	type fakeUploadSession struct {
		344	+	repo string
		345	+	data []byte
		346	+	}
		347	+
		348	+	type fakeManifestPut struct {
		349	+	contentType string
		350	+	body []byte
		351	+	}
		352	+
		353	+	func newFakePushRegistry(t testing.T, img v1.Image, existingBlobs map[string]bool) fakePushRegistry {
		354	+	t.Helper()
		355	+
		356	+	layers, err := img.Layers()
		357	+	require.NoError(t, err)
		358	+
		359	+	expectedBlobs := make(map[string][]byte, len(layers)+1)
		360	+	for _, layer := range layers {
		361	+	digest, err := layer.Digest()
		362	+	require.NoError(t, err)
		363	+	rc, err := layer.Compressed()
		364	+	require.NoError(t, err)
		365	+	content, err := io.ReadAll(rc)
		366	+	require.NoError(t, err)
		367	+	require.NoError(t, rc.Close())
		368	+	expectedBlobs[digest.String()] = content
		369	+	}
		370	+
		371	+	config, err := img.RawConfigFile()
		372	+	require.NoError(t, err)
		373	+	configDigest := fmt.Sprintf("sha256:%x", sha256.Sum256(config))
		374	+	expectedBlobs[configDigest] = config
		375	+
		376	+	manifest, err := img.RawManifest()
		377	+	require.NoError(t, err)
		378	+	mediaType, err := img.MediaType()
		379	+	require.NoError(t, err)
		380	+
		381	+	return &fakePushRegistry{
		382	+	t: t,
		383	+	existingBlobs: existingBlobs,
		384	+	uploadedBlobs: make(map[string][]byte),
		385	+	manifestPuts: make(map[string]fakeManifestPut),
		386	+	uploads: make(map[string]*fakeUploadSession),
		387	+	expectedBlobs: expectedBlobs,
		388	+	expectedRepo: "demo/app",
		389	+	expectedTag: "v1",
		390	+	expectedMT: string(mediaType),
		391	+	expectedManifest: manifest,
		392	+	}
		393	+	}
		394	+
		395	+	func (f *fakePushRegistry) handler() http.Handler {
		396	+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		397	+	switch {
		398	+	case r.Method == http.MethodHead && strings.HasPrefix(r.URL.Path, "/v2/") && strings.Contains(r.URL.Path, "/blobs/"):
		399	+	f.handleBlobHead(w, r)
		400	+	case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/blobs/uploads/"):
		401	+	f.handleBlobStart(w, r)
		402	+	case r.Method == http.MethodPatch && strings.Contains(r.URL.Path, "/blobs/uploads/"):
		403	+	f.handleBlobPatch(w, r)
		404	+	case r.Method == http.MethodPut && strings.Contains(r.URL.Path, "/blobs/uploads/"):
		405	+	f.handleBlobFinalize(w, r)
		406	+	case r.Method == http.MethodPut && strings.Contains(r.URL.Path, "/manifests/"):
		407	+	f.handleManifestPut(w, r)
		408	+	default:
		409	+	f.t.Fatalf("unexpected request: %s %s", r.Method, r.URL.String())
		410	+	}
		411	+	})
		412	+	}
		413	+
		414	+	func (f fakePushRegistry) handleBlobHead(w http.ResponseWriter, r http.Request) {
		415	+	digest := pathTail(r.URL.Path)
		416	+	f.headChecks = append(f.headChecks, digest)
		417	+	if f.existingBlobs[digest] {
		418	+	w.WriteHeader(http.StatusOK)
		419	+	return
		420	+	}
		421	+	w.WriteHeader(http.StatusNotFound)
		422	+	}
		423	+
		424	+	func (f fakePushRegistry) handleBlobStart(w http.ResponseWriter, r http.Request) {
		425	+	repo := pathSegmentBetween(r.URL.Path, "/v2/", "/blobs/uploads/")
		426	+	assert.Equal(f.t, f.expectedRepo, repo)
		427	+	f.nextUploadID++
		428	+	uploadPath := fmt.Sprintf("/v2/%s/blobs/uploads/upload-%d", repo, f.nextUploadID)
		429	+	f.uploads[uploadPath] = &fakeUploadSession{repo: repo}
		430	+	w.Header().Set("Location", uploadPath)
		431	+	w.WriteHeader(http.StatusAccepted)
		432	+	}
		433	+
		434	+	func (f fakePushRegistry) handleBlobPatch(w http.ResponseWriter, r http.Request) {
		435	+	session := f.uploads[r.URL.Path]
		436	+	if session == nil {
		437	+	f.t.Fatalf("missing upload session for %s", r.URL.Path)
		438	+	}
		439	+	body, err := io.ReadAll(r.Body)
		440	+	require.NoError(f.t, err)
		441	+	session.data = append(session.data, body...)
		442	+	w.Header().Set("Location", r.URL.Path)
		443	+	w.WriteHeader(http.StatusAccepted)
		444	+	}
		445	+
		446	+	func (f fakePushRegistry) handleBlobFinalize(w http.ResponseWriter, r http.Request) {
		447	+	session := f.uploads[r.URL.Path]
		448	+	if session == nil {
		449	+	f.t.Fatalf("missing upload session for %s", r.URL.Path)
		450	+	}
		451	+	digest := r.URL.Query().Get("digest")
		452	+	f.uploadedBlobs[digest] = append([]byte(nil), session.data...)
		453	+	delete(f.uploads, r.URL.Path)
		454	+	w.WriteHeader(http.StatusCreated)
		455	+	}
		456	+
		457	+	func (f fakePushRegistry) handleManifestPut(w http.ResponseWriter, r http.Request) {
		458	+	repo := pathSegmentBetween(r.URL.Path, "/v2/", "/manifests/")
		459	+	tag := pathTail(r.URL.Path)
		460	+	assert.Equal(f.t, f.expectedRepo, repo)
		461	+	assert.Equal(f.t, f.expectedTag, tag)
		462	+	body, err := io.ReadAll(r.Body)
		463	+	require.NoError(f.t, err)
		464	+	f.manifestPuts[repo+":"+tag] = fakeManifestPut{contentType: r.Header.Get("Content-Type"), body: body}
		465	+	w.WriteHeader(http.StatusCreated)
		466	+	}
		467	+
		468	+	func pathTail(path string) string {
		469	+	parts := strings.Split(path, "/")
		470	+	return parts[len(parts)-1]
		471	+	}
		472	+
		473	+	func pathSegmentBetween(value, start, end string) string {
		474	+	startIdx := strings.Index(value, start)
		475	+	if startIdx == -1 {
		476	+	return ""
		477	+	}
		478	+	trimmed := value[startIdx+len(start):]
		479	+	endIdx := strings.Index(trimmed, end)
		480	+	if endIdx == -1 {
		481	+	return ""
		482	+	}
		483	+	return trimmed[:endIdx]
		484	+	}
		485	+
		486	+	func typesOCIManifestSchema1() string {
		487	+	return "application/vnd.oci.image.manifest.v1+json"
		488	+	}