Copy streamformatter and progress packages from github.com/moby/moby

cli/command/container/run_test.go

@@ -12,10 +12,10 @@ import (
 	"github.com/creack/pty"
 	"github.com/docker/cli/cli"
 	"github.com/docker/cli/cli/streams"
+	"github.com/docker/cli/internal/progress"
+	"github.com/docker/cli/internal/streamformatter"
 	"github.com/docker/cli/internal/test"
 	"github.com/docker/cli/internal/test/notary"
-	"github.com/moby/moby/api/pkg/progress"
-	"github.com/moby/moby/api/pkg/streamformatter"
 	"github.com/moby/moby/api/types"
 	"github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/api/types/network"

cli/command/image/build.go

@@ -19,10 +19,10 @@ import (
 	"github.com/docker/cli/cli/command/image/build"
 	"github.com/docker/cli/cli/streams"
 	"github.com/docker/cli/internal/jsonstream"
+	"github.com/docker/cli/internal/progress"
+	"github.com/docker/cli/internal/streamformatter"
 	"github.com/docker/cli/opts"
 	"github.com/moby/go-archive"
-	"github.com/moby/moby/api/pkg/progress"
-	"github.com/moby/moby/api/pkg/streamformatter"
 	buildtypes "github.com/moby/moby/api/types/build"
 	"github.com/moby/moby/api/types/container"
 	registrytypes "github.com/moby/moby/api/types/registry"

cli/command/image/build/context.go

@@ -18,10 +18,10 @@ import (
 	"time"
 
 	"github.com/docker/cli/cli/command/image/build/internal/git"
+	"github.com/docker/cli/internal/progress"
+	"github.com/docker/cli/internal/streamformatter"
 	"github.com/moby/go-archive"
 	"github.com/moby/go-archive/compression"
-	"github.com/moby/moby/api/pkg/progress"
-	"github.com/moby/moby/api/pkg/streamformatter"
 	"github.com/moby/patternmatcher"
 )
 

cli/command/service/progress/progress.go

@@ -12,8 +12,8 @@ import (
 	"time"
 
 	"github.com/docker/cli/cli/command/formatter"
-	"github.com/moby/moby/api/pkg/progress"
-	"github.com/moby/moby/api/pkg/streamformatter"
+	"github.com/docker/cli/internal/progress"
+	"github.com/docker/cli/internal/streamformatter"
 	"github.com/moby/moby/api/types/swarm"
 	"github.com/moby/moby/client"
 )

cli/command/service/progress/progress_test.go

@@ -5,7 +5,7 @@ import (
 	"strconv"
 	"testing"
 
-	"github.com/moby/moby/api/pkg/progress"
+	"github.com/docker/cli/internal/progress"
 	"github.com/moby/moby/api/types/swarm"
 	"gotest.tools/v3/assert"
 	is "gotest.tools/v3/assert/cmp"

cli/command/swarm/progress/root_rotation.go

@@ -8,8 +8,8 @@ import (
 	"os/signal"
 	"time"
 
-	"github.com/moby/moby/api/pkg/progress"
-	"github.com/moby/moby/api/pkg/streamformatter"
+	"github.com/docker/cli/internal/progress"
+	"github.com/docker/cli/internal/streamformatter"
 	"github.com/moby/moby/api/types/swarm"
 	"github.com/moby/moby/client"
 	"github.com/opencontainers/go-digest"

internal/progress/progressreader_test.go

@@ -0,0 +1,74 @@
+package progress
+
+import (
+	"bytes"
+	"io"
+	"testing"
+)
+
+func TestOutputOnPrematureClose(t *testing.T) {
+	content := []byte("TESTING")
+	reader := io.NopCloser(bytes.NewReader(content))
+	progressChan := make(chan Progress, 10)
+
+	pr := NewProgressReader(reader, ChanOutput(progressChan), int64(len(content)), "Test", "Read")
+
+	part := make([]byte, 4)
+	_, err := io.ReadFull(pr, part)
+	if err != nil {
+		pr.Close()
+		t.Fatal(err)
+	}
+
+drainLoop:
+	for {
+		select {
+		case <-progressChan:
+		default:
+			break drainLoop
+		}
+	}
+
+	pr.Close()
+
+	select {
+	case <-progressChan:
+	default:
+		t.Fatalf("Expected some output when closing prematurely")
+	}
+}
+
+func TestCompleteSilently(t *testing.T) {
+	content := []byte("TESTING")
+	reader := io.NopCloser(bytes.NewReader(content))
+	progressChan := make(chan Progress, 10)
+
+	pr := NewProgressReader(reader, ChanOutput(progressChan), int64(len(content)), "Test", "Read")
+
+	out, err := io.ReadAll(pr)
+	if err != nil {
+		pr.Close()
+		t.Fatal(err)
+	}
+	if string(out) != "TESTING" {
+		pr.Close()
+		t.Fatalf("Unexpected output %q from reader", string(out))
+	}
+
+drainLoop:
+	for {
+		select {
+		case <-progressChan:
+		default:
+			break drainLoop
+		}
+	}
+
+	pr.Close()
+
+	select {
+	case <-progressChan:
+		t.Fatalf("Should have closed silently when read is complete")
+	default:
+	}
+}

vendor/github.com/moby/moby/api/pkg/streamformatter/streamformatter.go → internal/streamformatter/streamformatter.go

@@ -9,8 +9,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/docker/cli/internal/progress"
 	"github.com/docker/go-units"
-	"github.com/moby/moby/api/pkg/progress"
 	"github.com/moby/moby/api/types/jsonstream"
 )
 
@@ -63,14 +63,14 @@ func FormatError(err error) []byte {
 	return []byte(`{"error":"format error"}` + streamNewline)
 }
 
-func (sf *jsonProgressFormatter) formatStatus(id, format string, a ...any) []byte {
+func (*jsonProgressFormatter) formatStatus(id, format string, a ...any) []byte {
 	return FormatStatus(id, format, a...)
 }
 
 // formatProgress formats the progress information for a specified action.
-func (sf *jsonProgressFormatter) formatProgress(id, action string, progress *jsonstream.Progress, aux any) []byte {
-	if progress == nil {
-		progress = &jsonstream.Progress{}
+func (*jsonProgressFormatter) formatProgress(id, action string, p *jsonstream.Progress, aux any) []byte {
+	if p == nil {
+		p = &jsonstream.Progress{}
 	}
 	var auxJSON *json.RawMessage
 	if aux != nil {
@@ -83,7 +83,7 @@ func (sf *jsonProgressFormatter) formatProgress(id, action string, progress *jso
 	}
 	b, err := json.Marshal(&jsonMessage{
 		Status:   action,
-		Progress: progress,
+		Progress: p,
 		ID:       id,
 		Aux:      auxJSON,
 	})
@@ -95,7 +95,7 @@ func (sf *jsonProgressFormatter) formatProgress(id, action string, progress *jso
 
 type rawProgressFormatter struct{}
 
-func (sf *rawProgressFormatter) formatStatus(id, format string, a ...any) []byte {
+func (*rawProgressFormatter) formatStatus(id, format string, a ...any) []byte {
 	return []byte(fmt.Sprintf(format, a...) + streamNewline)
 }
 
@@ -155,12 +155,12 @@ func rawProgressString(p *jsonstream.Progress) string {
 	return pbBox + numbersBox + timeLeftBox
 }
 
-func (sf *rawProgressFormatter) formatProgress(id, action string, progress *jsonstream.Progress, aux any) []byte {
-	if progress == nil {
-		progress = &jsonstream.Progress{}
+func (*rawProgressFormatter) formatProgress(id, action string, p *jsonstream.Progress, aux any) []byte {
+	if p == nil {
+		p = &jsonstream.Progress{}
 	}
 	endl := "\r"
-	out := rawProgressString(progress)
+	out := rawProgressString(p)
 	if out == "" {
 		endl += "\n"
 	}
@@ -181,7 +181,7 @@ func NewJSONProgressOutput(out io.Writer, newLines bool) progress.Output {
 
 type formatProgress interface {
 	formatStatus(id, format string, a ...any) []byte
-	formatProgress(id, action string, progress *jsonstream.Progress, aux any) []byte
+	formatProgress(id, action string, p *jsonstream.Progress, aux any) []byte
 }
 
 type progressOutput struct {

internal/streamformatter/streamformatter_test.go

@@ -0,0 +1,110 @@
+package streamformatter
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/moby/moby/api/types/jsonstream"
+	"gotest.tools/v3/assert"
+	is "gotest.tools/v3/assert/cmp"
+)
+
+func TestRawProgressFormatterFormatStatus(t *testing.T) {
+	sf := rawProgressFormatter{}
+	res := sf.formatStatus("ID", "%s%d", "a", 1)
+	assert.Check(t, is.Equal("a1\r\n", string(res)))
+}
+
+func TestRawProgressFormatterFormatProgress(t *testing.T) {
+	sf := rawProgressFormatter{}
+	jsonProgress := &jsonstream.Progress{
+		Current: 15,
+		Total:   30,
+		Start:   1,
+	}
+	res := sf.formatProgress("id", "action", jsonProgress, nil)
+	out := string(res)
+	assert.Check(t, strings.HasPrefix(out, "action [===="))
+	assert.Check(t, is.Contains(out, "15B/30B"))
+	assert.Check(t, strings.HasSuffix(out, "\r"))
+}
+
+func TestFormatStatus(t *testing.T) {
+	res := FormatStatus("ID", "%s%d", "a", 1)
+	expected := `{"status":"a1","id":"ID"}` + streamNewline
+	assert.Check(t, is.Equal(expected, string(res)))
+}
+
+func TestFormatError(t *testing.T) {
+	res := FormatError(errors.New("Error for formatter"))
+	expected := `{"errorDetail":{"message":"Error for formatter"},"error":"Error for formatter"}` + "\r\n"
+	assert.Check(t, is.Equal(expected, string(res)))
+}
+
+func TestFormatJSONError(t *testing.T) {
+	err := &jsonstream.Error{Code: 50, Message: "Json error"}
+	res := FormatError(err)
+	expected := `{"errorDetail":{"code":50,"message":"Json error"},"error":"Json error"}` + streamNewline
+	assert.Check(t, is.Equal(expected, string(res)))
+}
+
+func TestJsonProgressFormatterFormatProgress(t *testing.T) {
+	sf := &jsonProgressFormatter{}
+	jsonProgress := &jsonstream.Progress{
+		Current: 15,
+		Total:   30,
+		Start:   1,
+	}
+	aux := "aux message"
+	res := sf.formatProgress("id", "action", jsonProgress, aux)
+	msg := &jsonMessage{}
+
+	assert.NilError(t, json.Unmarshal(res, msg))
+
+	rawAux := json.RawMessage(`"` + aux + `"`)
+	expected := &jsonMessage{
+		ID:       "id",
+		Status:   "action",
+		Aux:      &rawAux,
+		Progress: jsonProgress,
+	}
+	assert.DeepEqual(t, msg, expected, cmpJSONMessageOpt())
+}
+
+func cmpJSONMessageOpt() cmp.Option {
+	progressMessagePath := func(path cmp.Path) bool {
+		return path.String() == "ProgressMessage"
+	}
+	return cmp.Options{
+		// Ignore deprecated property that is a derivative of Progress
+		cmp.FilterPath(progressMessagePath, cmp.Ignore()),
+	}
+}
+
+func TestJsonProgressFormatterFormatStatus(t *testing.T) {
+	sf := jsonProgressFormatter{}
+	res := sf.formatStatus("ID", "%s%d", "a", 1)
+	assert.Check(t, is.Equal(`{"status":"a1","id":"ID"}`+streamNewline, string(res)))
+}
+
+func TestNewJSONProgressOutput(t *testing.T) {
+	b := bytes.Buffer{}
+	b.Write(FormatStatus("id", "Downloading"))
+	_ = NewJSONProgressOutput(&b, false)
+	assert.Check(t, is.Equal(`{"status":"Downloading","id":"id"}`+streamNewline, b.String()))
+}
+
+func TestAuxFormatterEmit(t *testing.T) {
+	b := bytes.Buffer{}
+	aux := &AuxFormatter{Writer: &b}
+	sampleAux := &struct {
+		Data string
+	}{"Additional data"}
+	err := aux.Emit("", sampleAux)
+	assert.NilError(t, err)
+	assert.Check(t, is.Equal(`{"aux":{"Data":"Additional data"}}`+streamNewline, b.String()))
+}

vendor.mod

@@ -60,6 +60,7 @@ require (
 	golang.org/x/sys v0.33.0
 	golang.org/x/term v0.32.0
 	golang.org/x/text v0.26.0
+	golang.org/x/time v0.11.0
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/v3 v3.5.2
 	tags.cncf.io/container-device-interface v1.0.1
@@ -102,7 +103,6 @@ require (
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
 	golang.org/x/crypto v0.39.0 // indirect
 	golang.org/x/net v0.39.0 // indirect
-	golang.org/x/time v0.11.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250218202821-56aae31c358a // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a // indirect
 	google.golang.org/grpc v1.72.2 // indirect