13790794198

Committed 11 Mar 2025 02:38PM UTC coverage: 90.575% (+0.5%) from 90.121%

Build # 13790794198

Build Type

push

github

Committed by

thschmitt

Commit Message

Add network package with common http client

Moved common code for handling HTTP requests in a new shared
network package to avoid duplication and ensure that each plugin
handles network requests the same way.

The common HTTP client code takes care of a couple of cross-cutting
concerns needed by all plugins and the main HTTP executor:

- Logs request and response when debug flag is set. Added a new
  resettableReader which preserves the request and response bodies while
  they are being read and forwards them to the debug logger.

- Added retries for all HTTP requests. This also leverages the
  resettableReader to ensure the request body can be replayed.

- The `CommandBuilder` generates now an operation id which is set on every
  request the uipathcli performs. The same operation id is kept for the
  duration of the whole command execution which makes it easier to
  correlate multiple requests performed by a single command.

- The `HttpClient` also sets transport-related settings like certificate
  validation and response header timeout.

- Using the built-in context instead of a custom requestError channel.

Run Details

466 of 493 new or added lines in 26 files covered. (94.52%)

10 existing lines in 6 files now uncovered.

5372 of 5931 relevant lines covered (90.57%)

1.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.63

/plugin/orchestrator/upload_command.go

package orchestrator

import (
        "bytes"
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "net/http"
        "net/url"
        "strings"

        "github.com/UiPath/uipathcli/log"
        "github.com/UiPath/uipathcli/output"
        "github.com/UiPath/uipathcli/plugin"
        "github.com/UiPath/uipathcli/utils/network"
        "github.com/UiPath/uipathcli/utils/stream"
        "github.com/UiPath/uipathcli/utils/visualization"
)

// The UploadCommand is a custom command for the orchestrator service which makes uploading
// files more convenient. It provides a wrapper over retrieving the write url and actually
// performing the upload.
type UploadCommand struct{}

func (c UploadCommand) Command() plugin.Command {
        return *plugin.NewCommand("orchestrator").
                WithCategory("buckets", "Orchestrator Buckets", "Buckets provide a per-folder storage solution for RPA developers to leverage in creating automation projects.").
                WithOperation("upload", "Upload file", "Uploads the provided file to the bucket").
                WithParameter("folder-id", plugin.ParameterTypeInteger, "Folder/OrganizationUnit Id", true).
                WithParameter("key", plugin.ParameterTypeInteger, "The Bucket Id", true).
                WithParameter("path", plugin.ParameterTypeString, "The BlobFile full path", true).
                WithParameter("file", plugin.ParameterTypeBinary, "The file to upload", true)
}

func (c UploadCommand) Execute(ctx plugin.ExecutionContext, writer output.OutputWriter, logger log.Logger) error {
        writeUrl, err := c.getWriteUrl(ctx, logger)
        if err != nil {
                return err
        }
        return c.upload(ctx, logger, writeUrl)
}

func (c UploadCommand) upload(ctx plugin.ExecutionContext, logger log.Logger, url string) error {
        uploadBar := visualization.NewProgressBar(logger)
        defer uploadBar.Remove()
        context, cancel := context.WithCancelCause(context.Background())
        request := c.createUploadRequest(ctx, url, uploadBar, cancel)
        client := network.NewHttpClient(logger, c.httpClientSettings(ctx))
        response, err := client.SendWithContext(request, context)
        if err != nil {
                return err
        }
        defer response.Body.Close()
        body, err := io.ReadAll(response.Body)
        if err != nil {
                return fmt.Errorf("Error reading response: %w", err)
        }
        if response.StatusCode != http.StatusCreated {
                return fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))
        }
        return nil
}

func (c UploadCommand) createUploadRequest(ctx plugin.ExecutionContext, url string, uploadBar *visualization.ProgressBar, cancel context.CancelCauseFunc) *network.HttpRequest {
        file := ctx.Input
        if file == nil {
                file = c.getFileParameter(ctx.Parameters)
        }
        bodyReader, bodyWriter := io.Pipe()
        contentType, contentLength := c.writeBody(bodyWriter, file, cancel)
        uploadReader := c.progressReader("uploading...", "completing  ", bodyReader, contentLength, uploadBar)

        header := http.Header{
                "Content-Type":   {contentType},
                "x-ms-blob-type": {"BlockBlob"},
        }
        request := network.NewHttpRequest(http.MethodPut, url, header, uploadReader)
        request.ContentLength = contentLength
        return request
}

func (c UploadCommand) writeBody(bodyWriter *io.PipeWriter, input stream.Stream, cancel context.CancelCauseFunc) (string, int64) {
        go func() {
                defer bodyWriter.Close()
                data, err := input.Data()
                if err != nil {
                        cancel(err)
                        return
                }
                defer data.Close()
                _, err = io.Copy(bodyWriter, data)
                if err != nil {
                        cancel(err)
                        return
                }
        }()
        size, _ := input.Size()
        return "application/octet-stream", size
}

func (c UploadCommand) progressReader(text string, completedText string, reader io.Reader, length int64, progressBar *visualization.ProgressBar) io.Reader {
        if length < 10*1024*1024 {
                return reader
        }
        return visualization.NewProgressReader(reader, func(progress visualization.Progress) {
                displayText := text
                if progress.Completed {
                        displayText = completedText
                }
                progressBar.UpdateProgress(displayText, progress.BytesRead, length, progress.BytesPerSecond)
        })
}

func (c UploadCommand) getWriteUrl(ctx plugin.ExecutionContext, logger log.Logger) (string, error) {
        if ctx.Organization == "" {
                return "", errors.New("Organization is not set")
        }
        if ctx.Tenant == "" {
                return "", errors.New("Tenant is not set")
        }
        request := c.createWriteUrlRequest(ctx)
        client := network.NewHttpClient(logger, c.httpClientSettings(ctx))
        response, err := client.Send(request)
        if err != nil {
                return "", err
        }
        defer response.Body.Close()
        body, err := io.ReadAll(response.Body)
        if err != nil {
                return "", fmt.Errorf("Error reading response: %w", err)
        }
        if response.StatusCode != http.StatusOK {
                return "", fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))
        }
        var result urlResponse
        err = json.Unmarshal(body, &result)
        if err != nil {
                return "", fmt.Errorf("Error parsing json response: %w", err)
        }
        return result.Uri, nil
}

func (c UploadCommand) createWriteUrlRequest(ctx plugin.ExecutionContext) *network.HttpRequest {
        folderId := c.getIntParameter("folder-id", ctx.Parameters)
        bucketId := c.getIntParameter("key", ctx.Parameters)
        path := c.getStringParameter("path", ctx.Parameters)

        uri := c.formatUri(ctx.BaseUri, ctx.Organization, ctx.Tenant) + fmt.Sprintf("/odata/Buckets(%d)/UiPath.Server.Configuration.OData.GetWriteUri?path=%s", bucketId, path)
        header := http.Header{
                "X-UiPath-OrganizationUnitId": {fmt.Sprintf("%d", folderId)},
        }
        for key, value := range ctx.Auth.Header {
                header.Set(key, value)
        }
        return network.NewHttpRequest(http.MethodGet, uri, header, &bytes.Buffer{})
}

func (c UploadCommand) formatUri(baseUri url.URL, org string, tenant string) string {
        path := baseUri.Path
        if baseUri.Path == "" {
                path = "/{organization}/{tenant}/orchestrator_"
        }
        path = strings.ReplaceAll(path, "{organization}", org)
        path = strings.ReplaceAll(path, "{tenant}", tenant)
        path = strings.TrimSuffix(path, "/")
        return fmt.Sprintf("%s://%s%s", baseUri.Scheme, baseUri.Host, path)
}

func (c UploadCommand) getStringParameter(name string, parameters []plugin.ExecutionParameter) string {
        result := ""
        for _, p := range parameters {
                if p.Name == name {
                        if data, ok := p.Value.(string); ok {
                                result = data
                                break
                        }
                }
        }
        return result
}

func (c UploadCommand) getIntParameter(name string, parameters []plugin.ExecutionParameter) int {
        result := 0
        for _, p := range parameters {
                if p.Name == name {
                        if data, ok := p.Value.(int); ok {
                                result = data
                                break
                        }
                }
        }
        return result
}

func (c UploadCommand) getFileParameter(parameters []plugin.ExecutionParameter) stream.Stream {
        var result stream.Stream
        for _, p := range parameters {
                if p.Name == "file" {
                        if stream, ok := p.Value.(stream.Stream); ok {
                                result = stream
                                break
                        }
                }
        }
        return result
}

func (c UploadCommand) httpClientSettings(ctx plugin.ExecutionContext) network.HttpClientSettings {
        return *network.NewHttpClientSettings(
                ctx.Debug,
                ctx.Settings.OperationId,
                ctx.Settings.Timeout,
                ctx.Settings.MaxAttempts,
                ctx.Settings.Insecure)
}

func NewUploadCommand() *UploadCommand {
        return &UploadCommand{}
}

1	package orchestrator
2
3	import (
4	"bytes"
5	"context"
6	"encoding/json"
7	"errors"
8	"fmt"
9	"io"
10	"net/http"
11	"net/url"
12	"strings"
13
14	"github.com/UiPath/uipathcli/log"
15	"github.com/UiPath/uipathcli/output"
16	"github.com/UiPath/uipathcli/plugin"
17	"github.com/UiPath/uipathcli/utils/network"
18	"github.com/UiPath/uipathcli/utils/stream"
19	"github.com/UiPath/uipathcli/utils/visualization"
20	)
21
22	// The UploadCommand is a custom command for the orchestrator service which makes uploading
23	// files more convenient. It provides a wrapper over retrieving the write url and actually
24	// performing the upload.
25	type UploadCommand struct{}
26
27	func (c UploadCommand) Command() plugin.Command {	1✔
28	return *plugin.NewCommand("orchestrator").	1✔
29	WithCategory("buckets", "Orchestrator Buckets", "Buckets provide a per-folder storage solution for RPA developers to leverage in creating automation projects.").	1✔
30	WithOperation("upload", "Upload file", "Uploads the provided file to the bucket").	1✔
31	WithParameter("folder-id", plugin.ParameterTypeInteger, "Folder/OrganizationUnit Id", true).	1✔
32	WithParameter("key", plugin.ParameterTypeInteger, "The Bucket Id", true).	1✔
33	WithParameter("path", plugin.ParameterTypeString, "The BlobFile full path", true).	1✔
34	WithParameter("file", plugin.ParameterTypeBinary, "The file to upload", true)	1✔
35	}	1✔
36
37	func (c UploadCommand) Execute(ctx plugin.ExecutionContext, writer output.OutputWriter, logger log.Logger) error {	1✔
38	writeUrl, err := c.getWriteUrl(ctx, logger)	1✔
39	if err != nil {	2✔
40	return err	1✔
41	}	1✔
42	return c.upload(ctx, logger, writeUrl)	1✔
43	}
44
45	func (c UploadCommand) upload(ctx plugin.ExecutionContext, logger log.Logger, url string) error {	1✔
46	uploadBar := visualization.NewProgressBar(logger)	1✔
47	defer uploadBar.Remove()	1✔
48	context, cancel := context.WithCancelCause(context.Background())	1✔
49	request := c.createUploadRequest(ctx, url, uploadBar, cancel)	1✔
50	client := network.NewHttpClient(logger, c.httpClientSettings(ctx))	1✔
51	response, err := client.SendWithContext(request, context)	1✔
52	if err != nil {	2✔
53	return err	1✔
54	}	1✔
55	defer response.Body.Close()	1✔
56	body, err := io.ReadAll(response.Body)	1✔
57	if err != nil {	1✔
58	return fmt.Errorf("Error reading response: %w", err)	×
59	}	×
60	if response.StatusCode != http.StatusCreated {	1✔
UNCOV 61	return fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))	×
62	}	×
63	return nil	1✔
64	}
65
66	func (c UploadCommand) createUploadRequest(ctx plugin.ExecutionContext, url string, uploadBar visualization.ProgressBar, cancel context.CancelCauseFunc) network.HttpRequest {	1✔
67	file := ctx.Input	1✔
68	if file == nil {	2✔
69	file = c.getFileParameter(ctx.Parameters)	1✔
70	}	1✔
71	bodyReader, bodyWriter := io.Pipe()	1✔
72	contentType, contentLength := c.writeBody(bodyWriter, file, cancel)	1✔
73	uploadReader := c.progressReader("uploading...", "completing ", bodyReader, contentLength, uploadBar)	1✔
74		1✔
75	header := http.Header{	1✔
76	"Content-Type": {contentType},	1✔
77	"x-ms-blob-type": {"BlockBlob"},	1✔
78	}	1✔
79	request := network.NewHttpRequest(http.MethodPut, url, header, uploadReader)	1✔
80	request.ContentLength = contentLength	1✔
81	return request	1✔
82	}
83
84	func (c UploadCommand) writeBody(bodyWriter *io.PipeWriter, input stream.Stream, cancel context.CancelCauseFunc) (string, int64) {	1✔
85	go func() {	2✔
86	defer bodyWriter.Close()	1✔
87	data, err := input.Data()	1✔
88	if err != nil {	2✔
89	cancel(err)	1✔
90	return	1✔
91	}	1✔
92	defer data.Close()	1✔
93	_, err = io.Copy(bodyWriter, data)	1✔
94	if err != nil {	1✔
NEW 95	cancel(err)	×
96	return	×
97	}	×
98	}()
99	size, _ := input.Size()	1✔
100	return "application/octet-stream", size	1✔
101	}
102
103	func (c UploadCommand) progressReader(text string, completedText string, reader io.Reader, length int64, progressBar *visualization.ProgressBar) io.Reader {	1✔
104	if length < 1010241024 {	2✔
105	return reader	1✔
106	}	1✔
107	return visualization.NewProgressReader(reader, func(progress visualization.Progress) {	2✔
108	displayText := text	1✔
109	if progress.Completed {	2✔
110	displayText = completedText	1✔
111	}	1✔
112	progressBar.UpdateProgress(displayText, progress.BytesRead, length, progress.BytesPerSecond)	1✔
113	})
114	}
115
116	func (c UploadCommand) getWriteUrl(ctx plugin.ExecutionContext, logger log.Logger) (string, error) {	1✔
117	if ctx.Organization == "" {	2✔
118	return "", errors.New("Organization is not set")	1✔
119	}	1✔
120	if ctx.Tenant == "" {	2✔
121	return "", errors.New("Tenant is not set")	1✔
122	}	1✔
123	request := c.createWriteUrlRequest(ctx)	1✔
124	client := network.NewHttpClient(logger, c.httpClientSettings(ctx))	1✔
125	response, err := client.Send(request)	1✔
126	if err != nil {	1✔
NEW 127	return "", err	×
128	}	×
129	defer response.Body.Close()	1✔
130	body, err := io.ReadAll(response.Body)	1✔
131	if err != nil {	1✔
132	return "", fmt.Errorf("Error reading response: %w", err)	×
133	}	×
134	if response.StatusCode != http.StatusOK {	2✔
135	return "", fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))	1✔
136	}	1✔
137	var result urlResponse	1✔
138	err = json.Unmarshal(body, &result)	1✔
139	if err != nil {	1✔
140	return "", fmt.Errorf("Error parsing json response: %w", err)	×
141	}	×
142	return result.Uri, nil	1✔
143	}
144
145	func (c UploadCommand) createWriteUrlRequest(ctx plugin.ExecutionContext) *network.HttpRequest {	1✔
146	folderId := c.getIntParameter("folder-id", ctx.Parameters)	1✔
147	bucketId := c.getIntParameter("key", ctx.Parameters)	1✔
148	path := c.getStringParameter("path", ctx.Parameters)	1✔
149		1✔
150	uri := c.formatUri(ctx.BaseUri, ctx.Organization, ctx.Tenant) + fmt.Sprintf("/odata/Buckets(%d)/UiPath.Server.Configuration.OData.GetWriteUri?path=%s", bucketId, path)	1✔
151	header := http.Header{	1✔
152	"X-UiPath-OrganizationUnitId": {fmt.Sprintf("%d", folderId)},	1✔
153	}	1✔
154	for key, value := range ctx.Auth.Header {	1✔
NEW 155	header.Set(key, value)	×
UNCOV 156	}	×
157	return network.NewHttpRequest(http.MethodGet, uri, header, &bytes.Buffer{})	1✔
158	}
159
160	func (c UploadCommand) formatUri(baseUri url.URL, org string, tenant string) string {	1✔
161	path := baseUri.Path	1✔
162	if baseUri.Path == "" {	2✔
163	path = "/{organization}/{tenant}/orchestrator_"	1✔
164	}	1✔
165	path = strings.ReplaceAll(path, "{organization}", org)	1✔
166	path = strings.ReplaceAll(path, "{tenant}", tenant)	1✔
167	path = strings.TrimSuffix(path, "/")	1✔
168	return fmt.Sprintf("%s://%s%s", baseUri.Scheme, baseUri.Host, path)	1✔
169	}
170
171	func (c UploadCommand) getStringParameter(name string, parameters []plugin.ExecutionParameter) string {	1✔
172	result := ""	1✔
173	for _, p := range parameters {	2✔
174	if p.Name == name {	2✔
175	if data, ok := p.Value.(string); ok {	2✔
176	result = data	1✔
177	break	1✔
178	}
179	}
180	}
181	return result	1✔
182	}
183
184	func (c UploadCommand) getIntParameter(name string, parameters []plugin.ExecutionParameter) int {	1✔
185	result := 0	1✔
186	for _, p := range parameters {	2✔
187	if p.Name == name {	2✔
188	if data, ok := p.Value.(int); ok {	2✔
189	result = data	1✔
190	break	1✔
191	}
192	}
193	}
194	return result	1✔
195	}
196
197	func (c UploadCommand) getFileParameter(parameters []plugin.ExecutionParameter) stream.Stream {	1✔
198	var result stream.Stream	1✔
199	for _, p := range parameters {	2✔
200	if p.Name == "file" {	2✔
201	if stream, ok := p.Value.(stream.Stream); ok {	2✔
202	result = stream	1✔
203	break	1✔
204	}
205	}
206	}
207	return result	1✔
208	}
209
210	func (c UploadCommand) httpClientSettings(ctx plugin.ExecutionContext) network.HttpClientSettings {	1✔
211	return *network.NewHttpClientSettings(	1✔
212	ctx.Debug,	1✔
213	ctx.Settings.OperationId,	1✔
214	ctx.Settings.Timeout,	1✔
215	ctx.Settings.MaxAttempts,	1✔
216	ctx.Settings.Insecure)	1✔
217	}	1✔
218
219	func NewUploadCommand() *UploadCommand {	1✔
220	return &UploadCommand{}	1✔
221	}	1✔

UiPath / uipathcli / 13790794198

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous