13786392533

Committed 11 Mar 2025 10:59AM UTC coverage: 90.225% (+0.1%) from 90.121%

Build # 13786392533

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

399 of 419 new or added lines in 22 files covered. (95.23%)

5 existing lines in 3 files now uncovered.

5372 of 5954 relevant lines covered (90.23%)

1.01 hits per line

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

87.65

/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, err := c.createUploadRequest(ctx, url, uploadBar, cancel)
        if err != nil {
                return err
        }
        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) (*http.Request, error) {
        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)

        request, err := http.NewRequest("PUT", url, uploadReader)
        if err != nil {
                return nil, err
        }
        request.ContentLength = contentLength
        request.Header.Add("Content-Type", contentType)
        request.Header.Add("x-ms-blob-type", "BlockBlob")
        return request, nil
}

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
        }
        progressReader := visualization.NewProgressReader(reader, func(progress visualization.Progress) {
                displayText := text
                if progress.Completed {
                        displayText = completedText
                }
                progressBar.UpdateProgress(displayText, progress.BytesRead, length, progress.BytesPerSecond)
        })
        return progressReader
}

func (c UploadCommand) getWriteUrl(ctx plugin.ExecutionContext, logger log.Logger) (string, error) {
        request, err := c.createWriteUrlRequest(ctx)
        if err != nil {
                return "", err
        }
        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) (*http.Request, error) {
        if ctx.Organization == "" {
                return nil, errors.New("Organization is not set")
        }
        if ctx.Tenant == "" {
                return nil, errors.New("Tenant is not set")
        }
        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)
        request, err := http.NewRequest("GET", uri, &bytes.Buffer{})
        if err != nil {
                return nil, err
        }
        request.Header.Add("X-UiPath-OrganizationUnitId", fmt.Sprintf("%d", folderId))
        for key, value := range ctx.Auth.Header {
                request.Header.Add(key, value)
        }
        return request, nil
}

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, err := c.createUploadRequest(ctx, url, uploadBar, cancel)	1✔
50	if err != nil {	1✔
51	return err	×
52	}	×
53	client := network.NewHttpClient(logger, c.httpClientSettings(ctx))	1✔
54	response, err := client.SendWithContext(request, context)	1✔
55	if err != nil {	2✔
56	return err	1✔
57	}	1✔
58	defer response.Body.Close()	1✔
59	body, err := io.ReadAll(response.Body)	1✔
60	if err != nil {	1✔
61	return fmt.Errorf("Error reading response: %w", err)	×
62	}	×
63	if response.StatusCode != http.StatusCreated {	1✔
UNCOV 64	return fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))	×
65	}	×
66	return nil	1✔
67	}
68
69	func (c UploadCommand) createUploadRequest(ctx plugin.ExecutionContext, url string, uploadBar visualization.ProgressBar, cancel context.CancelCauseFunc) (http.Request, error) {	1✔
70	file := ctx.Input	1✔
71	if file == nil {	2✔
72	file = c.getFileParameter(ctx.Parameters)	1✔
73	}	1✔
74	bodyReader, bodyWriter := io.Pipe()	1✔
75	contentType, contentLength := c.writeBody(bodyWriter, file, cancel)	1✔
76	uploadReader := c.progressReader("uploading...", "completing ", bodyReader, contentLength, uploadBar)	1✔
77		1✔
78	request, err := http.NewRequest("PUT", url, uploadReader)	1✔
79	if err != nil {	1✔
80	return nil, err	×
81	}	×
82	request.ContentLength = contentLength	1✔
83	request.Header.Add("Content-Type", contentType)	1✔
84	request.Header.Add("x-ms-blob-type", "BlockBlob")	1✔
85	return request, nil	1✔
86	}
87
88	func (c UploadCommand) writeBody(bodyWriter *io.PipeWriter, input stream.Stream, cancel context.CancelCauseFunc) (string, int64) {	1✔
89	go func() {	2✔
90	defer bodyWriter.Close()	1✔
91	data, err := input.Data()	1✔
92	if err != nil {	2✔
93	cancel(err)	1✔
94	return	1✔
95	}	1✔
96	defer data.Close()	1✔
97	_, err = io.Copy(bodyWriter, data)	1✔
98	if err != nil {	1✔
NEW 99	cancel(err)	×
100	return	×
101	}	×
102	}()
103	size, _ := input.Size()	1✔
104	return "application/octet-stream", size	1✔
105	}
106
107	func (c UploadCommand) progressReader(text string, completedText string, reader io.Reader, length int64, progressBar *visualization.ProgressBar) io.Reader {	1✔
108	if length < 1010241024 {	2✔
109	return reader	1✔
110	}	1✔
111	progressReader := visualization.NewProgressReader(reader, func(progress visualization.Progress) {	2✔
112	displayText := text	1✔
113	if progress.Completed {	2✔
114	displayText = completedText	1✔
115	}	1✔
116	progressBar.UpdateProgress(displayText, progress.BytesRead, length, progress.BytesPerSecond)	1✔
117	})
118	return progressReader	1✔
119	}
120
121	func (c UploadCommand) getWriteUrl(ctx plugin.ExecutionContext, logger log.Logger) (string, error) {	1✔
122	request, err := c.createWriteUrlRequest(ctx)	1✔
123	if err != nil {	2✔
124	return "", err	1✔
125	}	1✔
126	client := network.NewHttpClient(logger, c.httpClientSettings(ctx))	1✔
127	response, err := client.Send(request)	1✔
128	if err != nil {	1✔
NEW 129	return "", err	×
130	}	×
131	defer response.Body.Close()	1✔
132	body, err := io.ReadAll(response.Body)	1✔
133	if err != nil {	1✔
134	return "", fmt.Errorf("Error reading response: %w", err)	×
135	}	×
136	if response.StatusCode != http.StatusOK {	2✔
137	return "", fmt.Errorf("Orchestrator returned status code '%v' and body '%v'", response.StatusCode, string(body))	1✔
138	}	1✔
139	var result urlResponse	1✔
140	err = json.Unmarshal(body, &result)	1✔
141	if err != nil {	1✔
142	return "", fmt.Errorf("Error parsing json response: %w", err)	×
143	}	×
144	return result.Uri, nil	1✔
145	}
146
147	func (c UploadCommand) createWriteUrlRequest(ctx plugin.ExecutionContext) (*http.Request, error) {	1✔
148	if ctx.Organization == "" {	2✔
149	return nil, errors.New("Organization is not set")	1✔
150	}	1✔
151	if ctx.Tenant == "" {	2✔
152	return nil, errors.New("Tenant is not set")	1✔
153	}	1✔
154	folderId := c.getIntParameter("folder-id", ctx.Parameters)	1✔
155	bucketId := c.getIntParameter("key", ctx.Parameters)	1✔
156	path := c.getStringParameter("path", ctx.Parameters)	1✔
157		1✔
158	uri := c.formatUri(ctx.BaseUri, ctx.Organization, ctx.Tenant) + fmt.Sprintf("/odata/Buckets(%d)/UiPath.Server.Configuration.OData.GetWriteUri?path=%s", bucketId, path)	1✔
159	request, err := http.NewRequest("GET", uri, &bytes.Buffer{})	1✔
160	if err != nil {	1✔
161	return nil, err	×
162	}	×
163	request.Header.Add("X-UiPath-OrganizationUnitId", fmt.Sprintf("%d", folderId))	1✔
164	for key, value := range ctx.Auth.Header {	1✔
165	request.Header.Add(key, value)	×
166	}	×
167	return request, nil	1✔
168	}
169
170	func (c UploadCommand) formatUri(baseUri url.URL, org string, tenant string) string {	1✔
171	path := baseUri.Path	1✔
172	if baseUri.Path == "" {	2✔
173	path = "/{organization}/{tenant}/orchestrator_"	1✔
174	}	1✔
175	path = strings.ReplaceAll(path, "{organization}", org)	1✔
176	path = strings.ReplaceAll(path, "{tenant}", tenant)	1✔
177	path = strings.TrimSuffix(path, "/")	1✔
178	return fmt.Sprintf("%s://%s%s", baseUri.Scheme, baseUri.Host, path)	1✔
179	}
180
181	func (c UploadCommand) getStringParameter(name string, parameters []plugin.ExecutionParameter) string {	1✔
182	result := ""	1✔
183	for _, p := range parameters {	2✔
184	if p.Name == name {	2✔
185	if data, ok := p.Value.(string); ok {	2✔
186	result = data	1✔
187	break	1✔
188	}
189	}
190	}
191	return result	1✔
192	}
193
194	func (c UploadCommand) getIntParameter(name string, parameters []plugin.ExecutionParameter) int {	1✔
195	result := 0	1✔
196	for _, p := range parameters {	2✔
197	if p.Name == name {	2✔
198	if data, ok := p.Value.(int); ok {	2✔
199	result = data	1✔
200	break	1✔
201	}
202	}
203	}
204	return result	1✔
205	}
206
207	func (c UploadCommand) getFileParameter(parameters []plugin.ExecutionParameter) stream.Stream {	1✔
208	var result stream.Stream	1✔
209	for _, p := range parameters {	2✔
210	if p.Name == "file" {	2✔
211	if stream, ok := p.Value.(stream.Stream); ok {	2✔
212	result = stream	1✔
213	break	1✔
214	}
215	}
216	}
217	return result	1✔
218	}
219
220	func (c UploadCommand) httpClientSettings(ctx plugin.ExecutionContext) network.HttpClientSettings {	1✔
221	return *network.NewHttpClientSettings(	1✔
222	ctx.Debug,	1✔
223	ctx.Settings.OperationId,	1✔
224	ctx.Settings.Timeout,	1✔
225	ctx.Settings.MaxAttempts,	1✔
226	ctx.Settings.Insecure)	1✔
227	}	1✔
228
229	func NewUploadCommand() *UploadCommand {	1✔
230	return &UploadCommand{}	1✔
231	}	1✔

UiPath / uipathcli / 13786392533

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