Merge commit 'f5c5d2ca86e1ebd8750563e1c1ea71a6fbed2670' into HEAD

README.md

@@ -106,7 +106,7 @@ Authorization: Bearer <Token>
 }
 ```
 
-### `GET` `/{pipeline}`
+### `GET` `/pipelines/{pipeline}`
 
 Show pipeline information.
 

main.go

@@ -1,7 +1,9 @@
 package main
 
 import (
+	"bytes"
 	"encoding/json"
+	"io"
 	"net/http"
 
 	"github.com/geplauder/lithium/middlewares"
@@ -39,14 +41,108 @@ func IndexHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func RegisterPipelineRoutes(r *mux.Router, pipelines []pipelines.IPipeline, storageProvider storage.IStorageProvider) {
-	for _, pipeline := range pipelines {
-		r.HandleFunc("/"+pipeline.GetSlug(), func(w http.ResponseWriter, r *http.Request) {
-			PipelineHandler(pipeline, storageProvider, w, r)
-		})
+func writeError(w http.ResponseWriter, status int, errStr string) {
+	w.WriteHeader(status)
+	json.NewEncoder(w).Encode(struct {
+		Error string `json:"error"`
+	}{errStr})
+}
+
+func UploadHandler(w http.ResponseWriter, r *http.Request, pipes []pipelines.IPipeline, storageProvider storage.IStorageProvider) {
+	// open file handler
+	formFile, handler, err := r.FormFile("file")
+	if err != nil {
+		writeError(w, http.StatusUnprocessableEntity, err.Error())
+		return
+	}
+
+	defer formFile.Close()
+
+	// check pipelines form param
+	formPipeline := r.FormValue("pipeline")
+	if formPipeline == "" {
+		writeError(w, http.StatusUnprocessableEntity, "pipeline parameter missing")
+		return
+	}
+
+	var execPipe pipelines.IPipeline
+	for _, pipe := range pipes {
+		if formPipeline == pipe.GetSlug() {
+			execPipe = pipe
+			break
+		}
+	}
+
+	if execPipe == nil {
+		writeError(w, http.StatusUnprocessableEntity, "pipeline not found")
+		return
+	}
+
+	bucket := r.FormValue("bucket")
+	if bucket == "" {
+		writeError(w, http.StatusUnprocessableEntity, "bucket parameter missing")
+		return
+	}
+
+	// open file
+	file, err := handler.Open()
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "error reading uploaded file")
+		return
+	}
+
+	defer file.Close()
+
+	// read file to buffer
+	buf := bytes.NewBuffer(nil)
+	_, err = io.Copy(buf, file)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "error reading file from buffer")
+		return
+	}
+
+	// store uploaded file
+	_, err = storageProvider.StoreRaw(bucket, "source.jpg", buf.Bytes())
+	if err != nil {
+		return
+	}
+
+	// execute pipeline
+	output, err := execPipe.Run("source.jpg", bucket, storageProvider)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "error executing pipeline")
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+
+	err = json.NewEncoder(w).Encode(struct {
+		Message     string   `json:"message"`
+		OutputFiles []string `json:"output_files"`
+	}{"ok", []string{output}})
+
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
 	}
 }
 
+func RegisterRoutes(r *mux.Router, pipelines []pipelines.IPipeline, storageProvider storage.IStorageProvider) {
+	r.HandleFunc("/", IndexHandler).Methods("GET")
+	r.HandleFunc("/upload", func(w http.ResponseWriter, r *http.Request) {
+		UploadHandler(w, r, pipelines, storageProvider)
+	}).Methods("POST")
+	r.HandleFunc("/pipelines/{pipeline}", func(w http.ResponseWriter, r *http.Request) {
+		for _, pipeline := range pipelines {
+			if pipeline.GetSlug() == mux.Vars(r)["pipeline"] {
+				PipelineHandler(pipeline, storageProvider, w, r)
+				return
+			}
+		}
+
+		w.WriteHeader(404)
+	}).Methods("GET")
+}
+
 func main() {
 	appSettings, err := settings.LoadSettings(afero.NewOsFs())
 	if err != nil {
@@ -80,9 +176,7 @@ func main() {
 		r.Use(rateLimiterMiddleware.Middleware)
 	}
 
-	r.HandleFunc("/", IndexHandler)
-
-	RegisterPipelineRoutes(r, pipes, storageProvider)
+	RegisterRoutes(r, pipes, storageProvider)
 
 	err = http.ListenAndServe(appSettings.Endpoint, r)
 	if err != nil {

main_test.go

@@ -1,7 +1,9 @@
 package main
 
 import (
+	"encoding/base64"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -34,9 +36,9 @@ func TestEndpointRoute(t *testing.T) {
 		router := mux.NewRouter()
 		fs := storage.GetMemoryStorageProvider()
 
-		RegisterPipelineRoutes(router, []pipelines.IPipeline{data}, fs)
+		RegisterRoutes(router, []pipelines.IPipeline{data}, fs)
 
-		request, _ := http.NewRequest("GET", "/"+data.Slug, nil)
+		request, _ := http.NewRequest("GET", "/pipelines/"+data.Slug, nil)
 		responseRecorder := httptest.NewRecorder()
 
 		router.ServeHTTP(responseRecorder, request)
@@ -49,7 +51,7 @@ func TestEndpointRoute(t *testing.T) {
 	t.Run("Unregistered pipelines return 404", func(t *testing.T) {
 		router := mux.NewRouter()
 
-		request, _ := http.NewRequest("GET", "/"+data.Slug, nil)
+		request, _ := http.NewRequest("GET", "/pipelines/"+data.Slug, nil)
 		responseRecorder := httptest.NewRecorder()
 
 		router.ServeHTTP(responseRecorder, request)
@@ -57,3 +59,61 @@ func TestEndpointRoute(t *testing.T) {
 		assert.Equal(t, 404, responseRecorder.Code)
 	})
 }
+
+func TestUploadRoute(t *testing.T) {
+	t.Run("Test uploads missing multipart boundary", func(t *testing.T) {
+		router := mux.NewRouter()
+		fs := storage.GetMemoryStorageProvider()
+
+		RegisterRoutes(router, []pipelines.IPipeline{pipelines.Pipeline{
+			Name:           "",
+			Slug:           "",
+			Type:           0,
+			RemoveMetadata: false,
+			Steps:          []pipelines.Step{},
+			Output: struct {
+				Format  string `json:"format"`
+				Quality int    `json:"quality"`
+			}{"jpeg", 10},
+		}}, fs)
+
+		request, _ := http.NewRequest("POST", "/upload", nil)
+		request.Header["Content-Type"] = []string{"multipart/form-data"}
+		responseRecorder := httptest.NewRecorder()
+
+		router.ServeHTTP(responseRecorder, request)
+
+		assert.Equal(t, 0x1A6, responseRecorder.Code)
+		str, _ := base64.StdEncoding.DecodeString("eyJlcnJvciI6Im5" +
+			"vIG11bHRpcGFydCBib3VuZGFyeSBwYXJhbSBpbiBDb250ZW50LVR5cGUifQ==")
+		assert.JSONEq(t, string(str), responseRecorder.Body.String())
+	})
+
+	t.Run("Test uploads missing multipart boundary", func(t *testing.T) {
+		router := mux.NewRouter()
+		fs := storage.GetMemoryStorageProvider()
+
+		RegisterRoutes(router, []pipelines.IPipeline{pipelines.Pipeline{
+			Name:           "",
+			Slug:           "",
+			Type:           0,
+			RemoveMetadata: false,
+			Steps:          []pipelines.Step{},
+			Output: struct {
+				Format  string `json:"format"`
+				Quality int    `json:"quality"`
+			}{"jpeg", 10},
+		}}, fs)
+
+		request, _ := http.NewRequest("POST", "/upload", nil)
+		request.Header["Content-Type"] = []string{"multipart/form-data", "boundary=X-INSOMNIA-BOUNDARY"}
+		responseRecorder := httptest.NewRecorder()
+
+		router.ServeHTTP(responseRecorder, request)
+
+		assert.Equal(t, 0x1A6, responseRecorder.Code)
+		str, _ := base64.StdEncoding.DecodeString("eyJlcnJvciI6Im5vIG11bHRpcGFydCBib3VuZGFyeSBwYXJhbSBpbiBDb250ZW50LVR5cGUifQ==")
+		assert.JSONEq(t, string(str), responseRecorder.Body.String())
+		fmt.Println(responseRecorder.Body.String())
+	})
+}