13832257596

Committed 13 Mar 2025 10:19AM UTC coverage: 90.57% (+0.4%) from 90.121%

Build # 13832257596

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #156 from UiPath/feature/network-package

Add network package with common http client

Run Details

478 of 505 new or added lines in 26 files covered. (94.65%)

10 existing lines in 6 files now uncovered.

5369 of 5928 relevant lines covered (90.57%)

1.02 hits per line

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

90.51

/plugin/orchestrator/upload_command.go

package orchestrator

import (
        "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"},
        }
        return network.NewHttpPutRequest(url, header, uploadReader, contentLength)
}

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.NewHttpGetRequest(uri, header)
}

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

UiPath / uipathcli / 13832257596

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