Add image flip step

config/example.json

@@ -14,11 +14,23 @@
             }
         },
         {
-            "name": "compress image",
+            "name": "rotate image",
             "type": 1,
             "options": {
-                "quality": 80
+                "angle": 90.0
             }
+        },
+        {
+            "name": "flip image",
+            "type": 2
+        },
+        {
+            "name": "grayscale",
+            "type": 3
         }
-    ]
+    ],
+    "output": {
+        "format": 0,
+        "quality": 90
+    }
 }

go.mod

@@ -4,6 +4,7 @@ go 1.17
 
 require (
 	github.com/bxcodec/faker/v3 v3.7.0
+	github.com/disintegration/imaging v1.6.2
 	github.com/gorilla/mux v1.8.0
 	github.com/spf13/afero v1.8.0
 	github.com/stretchr/testify v1.7.0
@@ -14,7 +15,8 @@ require (
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rogpeppe/go-internal v1.8.1-0.20211023094830-115ce09fd6b4 // indirect
-	golang.org/x/text v0.3.4 // indirect
+	golang.org/x/image v0.0.0-20211028202545-6944b10bf410 // indirect
+	golang.org/x/text v0.3.6 // indirect
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )

go.sum

@@ -51,6 +51,8 @@ github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnht
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/disintegration/imaging v1.6.2 h1:w1LecBlG2Lnp8B3jk5zSuNqd7b4DXhcjwek1ei82L+c=
+github.com/disintegration/imaging v1.6.2/go.mod h1:44/5580QXChDfwIclfc/PCwrr44amcmDAg8hxG0Ewe4=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -177,6 +179,9 @@ golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EH
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
+golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
+golang.org/x/image v0.0.0-20211028202545-6944b10bf410 h1:hTftEOvwiOq2+O8k2D5/Q7COC7k5Qcrgc2TFURJYnvQ=
+golang.org/x/image v0.0.0-20211028202545-6944b10bf410/go.mod h1:023OzeP/+EPmXeapQh35lcL3II3LrY8Ic+EFFKVhULM=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190301231843-5614ed5bae6f/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
@@ -288,8 +293,9 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.4 h1:0YWbFKbhXG/wIiuHDSKpS0Iy7FSA+u45VtBMfQcFTTc=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

main.go

@@ -2,9 +2,10 @@ package main
 
 import (
 	"encoding/json"
-	"github.com/geplauder/lithium/pipelines"
 	"net/http"
 
+	"github.com/geplauder/lithium/pipelines"
+	"github.com/geplauder/lithium/storage"
 	"github.com/gorilla/mux"
 )
 
@@ -16,7 +17,7 @@ type Metadata struct {
 	Version string
 }
 
-func PipelineHandler(pipeline pipelines.IPipeline, w http.ResponseWriter, r *http.Request) {
+func PipelineHandler(pipeline pipelines.IPipeline, storageProvider storage.IStorageProvider, w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
 	err := json.NewEncoder(w).Encode(pipeline)
 	if err != nil {
@@ -32,21 +33,24 @@ func IndexHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func RegisterPipelineRoutes(r *mux.Router, pipelines []pipelines.IPipeline) {
+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, w, r)
+			PipelineHandler(pipeline, storageProvider, w, r)
 		})
 	}
 }
 
 func main() {
+	storageProvider := storage.GetFileSystemStorageProvider("test", "")
+	storageProvider.StoreRaw("abc", "def.test", []byte{0x12, 0x10})
+
 	pipes := pipelines.LoadPipelines()
 
 	r := mux.NewRouter()
 	r.HandleFunc("/", IndexHandler)
 
-	RegisterPipelineRoutes(r, pipes)
+	RegisterPipelineRoutes(r, pipes, storageProvider)
 
 	err := http.ListenAndServe(":8000", r)
 	if err != nil {

main_test.go

@@ -3,12 +3,13 @@ package main
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/geplauder/lithium/pipelines"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
 	"github.com/bxcodec/faker/v3"
+	"github.com/geplauder/lithium/pipelines"
+	"github.com/geplauder/lithium/storage"
 	"github.com/gorilla/mux"
 	"github.com/stretchr/testify/assert"
 )
@@ -34,7 +35,9 @@ func TestEndpointRoute(t *testing.T) {
 
 	t.Run("Registered pipelines are valid routes", func(t *testing.T) {
 		router := mux.NewRouter()
-		RegisterPipelineRoutes(router, []pipelines.IPipeline{data})
+		fs := storage.GetMemoryStorageProvider()
+
+		RegisterPipelineRoutes(router, []pipelines.IPipeline{data}, fs)
 
 		request, _ := http.NewRequest("GET", "/"+data.Slug, nil)
 		responseRecorder := httptest.NewRecorder()

pipelines/executable_step.go

@@ -1,7 +1,14 @@
 package pipelines
 
+import (
+	"errors"
+	"fmt"
+	"github.com/disintegration/imaging"
+	"image"
+)
+
 type IExecutableStep interface {
-	Execute()
+	Execute(src image.Image) (image.Image, error)
 }
 
 // Resize image
@@ -15,19 +22,54 @@ type ResizeImageStep struct {
 	} `json:"options"`
 }
 
-func (s ResizeImageStep) Execute() {
-	// TODO
+func (s ResizeImageStep) Execute(src image.Image) (image.Image, error) {
+	src = imaging.Resize(src, s.Options.Width, s.Options.Height, imaging.Linear)
+	return src, nil
+}
+
+// Rotate image
+
+type RotateImageStep struct {
+	Step
+	Options struct {
+		Angle float64 `json:"angle"`
+	} `json:"options"`
+}
+
+func (s RotateImageStep) Execute(src image.Image) (image.Image, error) {
+	src = imaging.Rotate(src, s.Options.Angle, image.Black)
+	return src, nil
 }
 
-// Compress image
+// Flip image
 
-type CompressImageStep struct {
+type FlipImageStep struct {
 	Step
 	Options struct {
-		Quality int `json:"quality"`
+		Direction string `json:"direction"`
 	} `json:"options"`
 }
 
-func (s CompressImageStep) Execute() {
-	// TODO
+func (s FlipImageStep) Execute(src image.Image) (image.Image, error) {
+	switch s.Options.Direction {
+	case "h":
+		src = imaging.FlipH(src)
+	case "v":
+		src = imaging.FlipH(src)
+	default:
+		return src, errors.New(fmt.Sprintf("invalid flip direction: %s", s.Options.Direction))
+	}
+
+	return src, nil
+}
+
+// Grayscale image
+
+type GrayscaleImageStep struct {
+	Step
+}
+
+func (s GrayscaleImageStep) Execute(src image.Image) (image.Image, error) {
+	src = imaging.Grayscale(src)
+	return src, nil
 }

pipelines/executable_step_test.go

@@ -32,18 +32,15 @@ func TestDeserializeOptionsResizeImage(t *testing.T) {
 	})
 }
 
-func TestDeserializeOptionsCompressImage(t *testing.T) {
+func TestDeserializeMissingOptions(t *testing.T) {
 	const Payload string = `{
 		"name": "example pipeline",
 		"type": 0,
 		"removeMetadata": false,
 		"steps": [
 			{
-				"name": "compress image",
-				"type": 1,
-				"options": {
-					"quality": 80
-				}
+				"name": "resize image",
+				"type": 0
 			}
 		]
 	}`
@@ -53,11 +50,11 @@ func TestDeserializeOptionsCompressImage(t *testing.T) {
 
 		_, err := values[0].GetSteps()[0].GetExecutable()
 
-		assert.Equal(t, nil, err)
+		assert.EqualError(t, err, "unexpected end of JSON input")
 	})
 }
 
-func TestDeserializeMissingOptions(t *testing.T) {
+func TestLoadingImage(t *testing.T) {
 	const Payload string = `{
 		"name": "example pipeline",
 		"type": 0,
@@ -70,7 +67,7 @@ func TestDeserializeMissingOptions(t *testing.T) {
 		]
 	}`
 
-	t.Run("Image pipeline deserialization is successful", func(t *testing.T) {
+	t.Run("Loading image from filesystem to pipeline is successful", func(t *testing.T) {
 		values := DeserializePipelines([][]byte{[]byte(Payload)})
 
 		_, err := values[0].GetSteps()[0].GetExecutable()

pipelines/pipeline.go

@@ -1,8 +1,12 @@
 package pipelines
 
 import (
+	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"github.com/disintegration/imaging"
+	"github.com/geplauder/lithium/storage"
 	"io/fs"
 	"log"
 	"os"
@@ -23,6 +27,7 @@ type IPipeline interface {
 	GetSlug() string
 	GetType() PipelineType
 	GetSteps() []Step
+	Run(string, string, storage.IStorageProvider) (string, error)
 }
 
 type Pipeline struct {
@@ -31,6 +36,53 @@ type Pipeline struct {
 	Type           PipelineType `json:"type" faker:"-"`
 	RemoveMetadata bool         `json:"remove_metadata" faker:"-"`
 	Steps          []Step       `json:"steps" faker:"-"`
+	Output         struct {
+		Format  int `json:"format"`
+		Quality int `json:"quality"`
+	} `json:"output" faker:"-"`
+}
+
+func (p Pipeline) Run(srcPath, bucketName string, storageProvider storage.IStorageProvider) (string, error) {
+	fmt.Println("path: ", storageProvider.GetPath(bucketName, srcPath))
+
+	src, err := imaging.Open(storageProvider.GetPath(bucketName, srcPath))
+	if err != nil {
+		return "", errors.New(fmt.Sprintf("error opening file for processing: %s", err))
+	}
+
+	for _, step := range p.GetSteps() {
+		runner, err := step.GetExecutable()
+		if err != nil {
+			return "", err
+		}
+
+		src, err = runner.Execute(src)
+		if err != nil {
+			return "", err
+		}
+	}
+
+	format := imaging.Format(p.Output.Format)
+	var options []imaging.EncodeOption
+
+	if p.Output.Quality != 0 {
+		options = append(options, imaging.JPEGQuality(p.Output.Quality))
+	}
+
+	// encode image to io buffer
+	buffer := new(bytes.Buffer)
+	if err := imaging.Encode(buffer, src, format, options...); err != nil {
+		return "", err
+	}
+
+	const fileName = "output.jpg" // TODO make variable
+
+	_, err = storageProvider.StoreRaw(bucketName, fileName, buffer.Bytes())
+	if err != nil {
+		return "", err
+	}
+
+	return fileName, nil
 }
 
 func (p Pipeline) GetName() string {

pipelines/pipeline_test.go

@@ -1,38 +1,326 @@
 package pipelines
 
 import (
+	"github.com/geplauder/lithium/storage"
+	"image"
+	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
-func TestImagePipelineDeserialization(t *testing.T) {
-	const Payload string = `{
-		"name": "example pipeline",
-		"type": 0,
-		"removeMetadata": false,
-		"steps": [
-			{
-				"name": "resize image",
-				"type": 0
-			},
-			{
-				"name": "compress image",
-				"type": 1
-			}
-		]
-	}`
+// pipeline deserialization
 
+func TestPipelineDeserialization(t *testing.T) {
 	t.Run("Image pipeline deserialization is successful", func(t *testing.T) {
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 0,
+			"removeMetadata": false,
+			"steps": [
+				{
+					"name": "resize image",
+					"type": 0
+				},
+				{
+					"name": "compress image",
+					"type": 1
+				}
+			]
+		}`
+
 		values := DeserializePipelines([][]byte{[]byte(Payload)})
 
 		assert.Equal(t, 1, len(values), "Output should contain one element")
 		assert.Equal(t, "example pipeline", values[0].GetName())
 		assert.Equal(t, Image, values[0].GetType())
 	})
+
+	t.Run("Video pipelines deserialization is successful", func(t *testing.T) {
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 1,
+			"removeMetadata": false,
+			"steps": [
+				{
+					"name": "resize image",
+					"type": 0
+				},
+				{
+					"name": "compress image",
+					"type": 1
+				}
+			]
+		}`
+
+		values := DeserializePipelines([][]byte{[]byte(Payload)})
+
+		assert.Equal(t, 1, len(values), "Output should contain one element")
+		assert.Equal(t, "example pipeline", values[0].GetName())
+		assert.Equal(t, Video, values[0].GetType())
+	})
 }
 
-func TestVideoPipelineDeserialization(t *testing.T) {
+// image pipeline steps
+
+func TestExecuteSteps(t *testing.T) {
+	t.Run("Pipeline executes with no steps", func(t *testing.T) {
+		const Bucket string = "pipeline_test_01"
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 1,
+			"removeMetadata": false,
+			"steps": []
+		}`
+
+		wd, _ := os.Getwd()
+		pipe := DeserializePipelines([][]byte{[]byte(Payload)})[0]
+
+		storageProvider := storage.GetFileSystemStorageProvider("test", "..")
+
+		// copy test file to storage bucket
+		_, err := storageProvider.StoreExisting(Bucket, "source.jpg", filepath.Join(wd, "../tests/files/900x900.jpg"))
+		assert.Nil(t, err, "Test file should be readable")
+		assert.FileExists(t, storageProvider.GetPath(Bucket, "source.jpg"))
+
+		// run pipeline steps
+		dest, err := pipe.Run("source.jpg", Bucket, storageProvider)
+		assert.Nil(t, err)
+		assert.FileExists(t, storageProvider.GetPath(Bucket, dest))
+
+		// clean up
+		os.Remove(storageProvider.GetPath(Bucket, "source.jpg"))
+		os.Remove(storageProvider.GetPath(Bucket, dest))
+	})
+
+	t.Run("Image resizing is successful", func(t *testing.T) {
+		const Bucket string = "pipeline_test_02"
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 1,
+			"removeMetadata": false,
+			"steps": [
+				{
+					"name": "resize image",
+					"type": 0,
+					"options": {
+						"width": 1280,
+						"height": 720,
+						"upscale": false
+					}
+				}
+			]
+		}`
+
+		wd, _ := os.Getwd()
+		pipe := DeserializePipelines([][]byte{[]byte(Payload)})[0]
+
+		storageProvider := storage.GetFileSystemStorageProvider("test", "..")
+
+		// copy test file to storage bucket
+		_, err := storageProvider.StoreExisting(Bucket, "source.jpg", filepath.Join(wd, "../tests/files/900x900.jpg"))
+		assert.Nil(t, err, "Test file should be readable")
+		assert.FileExists(t, storageProvider.GetPath(Bucket, "source.jpg"))
+
+		// run pipeline steps
+		dest, err := pipe.Run("source.jpg", Bucket, storageProvider)
+		assert.Nil(t, err)
+		assert.FileExists(t, storageProvider.GetPath(Bucket, dest))
+
+		// read image config
+		file, err := storageProvider.OpenFile(Bucket, dest)
+		assert.Nil(t, err)
+
+		imgConf, _, err := image.DecodeConfig(file)
+		assert.Nil(t, err)
+
+		assert.Equal(t, 1280, imgConf.Width)
+		assert.Equal(t, 720, imgConf.Height)
+
+		// clean up
+		os.Remove(storageProvider.GetPath(Bucket, "source.jpg"))
+		os.Remove(storageProvider.GetPath(Bucket, dest))
+	})
+
+	t.Run("Image rotation step is successful", func(t *testing.T) {
+		const Bucket string = "pipeline_test_03"
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 1,
+			"removeMetadata": false,
+			"steps": [
+				{
+					"name": "rotate image",
+					"type": 1,
+					"options": {
+						"angle": 90.0
+					}
+				}
+			]
+		}`
+
+		wd, _ := os.Getwd()
+		pipe := DeserializePipelines([][]byte{[]byte(Payload)})[0]
+
+		storageProvider := storage.GetFileSystemStorageProvider("test", "..")
+
+		// copy test file to storage bucket
+		_, err := storageProvider.StoreExisting(Bucket, "source.jpg", filepath.Join(wd, "../tests/files/800x500.jpg"))
+		assert.Nil(t, err, "Test file should be readable")
+		assert.FileExists(t, storageProvider.GetPath(Bucket, "source.jpg"))
+
+		// run pipeline steps
+		dest, err := pipe.Run("source.jpg", Bucket, storageProvider)
+		assert.Nil(t, err)
+		assert.FileExists(t, storageProvider.GetPath(Bucket, dest))
+
+		// read image config
+		file, err := storageProvider.OpenFile(Bucket, dest)
+		assert.Nil(t, err)
+
+		imgConf, _, err := image.DecodeConfig(file)
+		assert.Nil(t, err)
+
+		assert.Equal(t, 500, imgConf.Width)
+		assert.Equal(t, 800, imgConf.Height)
+
+		// clean up
+		os.Remove(storageProvider.GetPath(Bucket, "source.jpg"))
+		os.Remove(storageProvider.GetPath(Bucket, dest))
+	})
+
+	t.Run("Image flip step is successful", func(t *testing.T) {
+		const Bucket string = "pipeline_test_06"
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 1,
+			"removeMetadata": false,
+			"steps": [
+				{
+					"name": "flip image",
+					"type": 2,
+					"options": {
+						"direction": "h"
+					}
+				}
+			]
+		}`
+
+		wd, _ := os.Getwd()
+		pipe := DeserializePipelines([][]byte{[]byte(Payload)})[0]
+
+		storageProvider := storage.GetFileSystemStorageProvider("test", "..")
+
+		// copy test file to storage bucket
+		_, err := storageProvider.StoreExisting(Bucket, "source.jpg", filepath.Join(wd, "../tests/files/800x500.jpg"))
+		assert.Nil(t, err, "Test file should be readable")
+		assert.FileExists(t, storageProvider.GetPath(Bucket, "source.jpg"))
+
+		// run pipeline steps
+		dest, err := pipe.Run("source.jpg", Bucket, storageProvider)
+		assert.Nil(t, err)
+		assert.FileExists(t, storageProvider.GetPath(Bucket, dest))
+
+		// read image config
+		file, err := storageProvider.OpenFile(Bucket, dest)
+		assert.Nil(t, err)
+
+		imgConf, _, err := image.DecodeConfig(file)
+		assert.Nil(t, err)
+
+		assert.Equal(t, 800, imgConf.Width)
+		assert.Equal(t, 500, imgConf.Height)
+
+		// clean up
+		os.Remove(storageProvider.GetPath(Bucket, "source.jpg"))
+		os.Remove(storageProvider.GetPath(Bucket, dest))
+	})
+
+	t.Run("Image flip step direction validation is successful", func(t *testing.T) {
+		const Bucket string = "pipeline_test_06"
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 1,
+			"removeMetadata": false,
+			"steps": [
+				{
+					"name": "flip image",
+					"type": 2,
+					"options": {
+						"direction": "f"
+					}
+				}
+			]
+		}`
+
+		wd, _ := os.Getwd()
+		pipe := DeserializePipelines([][]byte{[]byte(Payload)})[0]
+
+		storageProvider := storage.GetFileSystemStorageProvider("test", "..")
+
+		// copy test file to storage bucket
+		_, err := storageProvider.StoreExisting(Bucket, "source.jpg", filepath.Join(wd, "../tests/files/800x500.jpg"))
+		assert.Nil(t, err, "Test file should be readable")
+		assert.FileExists(t, storageProvider.GetPath(Bucket, "source.jpg"))
+
+		// run pipeline steps
+		_, err = pipe.Run("source.jpg", Bucket, storageProvider)
+		assert.EqualError(t, err, "invalid flip direction: f")
+
+		// clean up
+		os.Remove(storageProvider.GetPath(Bucket, "source.jpg"))
+	})
+
+	t.Run("Image grayscale step is successful", func(t *testing.T) {
+		const Bucket string = "pipeline_test_05"
+		const Payload string = `{
+			"name": "example pipeline",
+			"type": 1,
+			"removeMetadata": false,
+			"steps": [
+				{
+					"name": "grayscale",
+					"type": 3
+				}
+			]
+		}`
+
+		wd, _ := os.Getwd()
+		pipe := DeserializePipelines([][]byte{[]byte(Payload)})[0]
+
+		storageProvider := storage.GetFileSystemStorageProvider("test", "..")
+
+		// copy test file to storage bucket
+		_, err := storageProvider.StoreExisting(Bucket, "source.jpg", filepath.Join(wd, "../tests/files/900x900.jpg"))
+		assert.Nil(t, err, "Test file should be readable")
+		assert.FileExists(t, storageProvider.GetPath(Bucket, "source.jpg"))
+
+		// run pipeline steps
+		dest, err := pipe.Run("source.jpg", Bucket, storageProvider)
+		assert.Nil(t, err)
+		assert.FileExists(t, storageProvider.GetPath(Bucket, dest))
+
+		// read image config
+		file, err := storageProvider.OpenFile(Bucket, dest)
+		assert.Nil(t, err)
+
+		imgConf, _, err := image.DecodeConfig(file)
+		assert.Nil(t, err)
+
+		assert.Equal(t, 900, imgConf.Width)
+		assert.Equal(t, 900, imgConf.Height)
+
+		// clean up
+		os.Remove(storageProvider.GetPath(Bucket, "source.jpg"))
+		os.Remove(storageProvider.GetPath(Bucket, dest))
+	})
+}
+
+// output options
+
+func TestEncoding(t *testing.T) {
+	const Bucket string = "pipeline_test_04"
 	const Payload string = `{
 		"name": "example pipeline",
 		"type": 1,
@@ -40,20 +328,37 @@ func TestVideoPipelineDeserialization(t *testing.T) {
 		"steps": [
 			{
 				"name": "resize image",
-				"type": 0
-			},
-			{
-				"name": "compress image",
-				"type": 1
+				"type": 0,
+				"options": {
+					"width": 1280,
+					"height": 720,
+					"upscale": false
+				}
 			}
-		]
+		],
+		"output": {
+			"quality": 50
+		}
 	}`
 
-	t.Run("Video pipelines deserialization is successful", func(t *testing.T) {
-		values := DeserializePipelines([][]byte{[]byte(Payload)})
+	t.Run("Image encoding with jpeg quality is successful", func(t *testing.T) {
+		wd, _ := os.Getwd()
+		pipe := DeserializePipelines([][]byte{[]byte(Payload)})[0]
 
-		assert.Equal(t, 1, len(values), "Output should contain one element")
-		assert.Equal(t, "example pipeline", values[0].GetName())
-		assert.Equal(t, Video, values[0].GetType())
+		storageProvider := storage.GetFileSystemStorageProvider("test", "..")
+
+		// copy test file to storage bucket
+		_, err := storageProvider.StoreExisting(Bucket, "source.jpg", filepath.Join(wd, "../tests/files/900x900.jpg"))
+		assert.Nil(t, err, "Test file should be readable")
+		assert.FileExists(t, storageProvider.GetPath(Bucket, "source.jpg"))
+
+		// run pipeline steps
+		dest, err := pipe.Run("source.jpg", Bucket, storageProvider)
+		assert.Nil(t, err)
+		assert.FileExists(t, storageProvider.GetPath(Bucket, dest))
+
+		// clean up
+		os.Remove(storageProvider.GetPath(Bucket, "source.jpg"))
+		os.Remove(storageProvider.GetPath(Bucket, dest))
 	})
 }

pipelines/step.go

@@ -9,7 +9,9 @@ type StepType int
 
 const (
 	TypeResizeImageStep StepType = iota
-	TypeCompressImageStep
+	TypeRotateImageStep
+	TypeFlipImageStep
+	TypeGrayscaleImageStep
 )
 
 type Step struct {
@@ -20,6 +22,7 @@ type Step struct {
 
 func (s Step) GetExecutable() (IExecutableStep, error) {
 	switch s.GetType() {
+
 	case TypeResizeImageStep:
 		step := ResizeImageStep{}
 		if err := json.Unmarshal(s.Options, &step.Options); err != nil {
@@ -27,12 +30,22 @@ func (s Step) GetExecutable() (IExecutableStep, error) {
 		}
 		return step, nil
 
-	case TypeCompressImageStep:
-		step := CompressImageStep{}
+	case TypeRotateImageStep:
+		step := RotateImageStep{}
 		if err := json.Unmarshal(s.Options, &step.Options); err != nil {
 			return nil, err
 		}
 		return step, nil
+
+	case TypeFlipImageStep:
+		step := FlipImageStep{}
+		if err := json.Unmarshal(s.Options, &step.Options); err != nil {
+			return nil, err
+		}
+		return step, nil
+
+	case TypeGrayscaleImageStep:
+		return GrayscaleImageStep{}, nil
 	}
 
 	return nil, errors.New("invalid type")

storage/storage.go

@@ -7,30 +7,69 @@ import (
 	"github.com/spf13/afero"
 )
 
+const StorageFolderName = "files"
+
 type IStorageProvider interface {
-	storeRaw(bucketName string, objectName string, data []byte) string
-	storeExisting(bucketName string, objectName string, existingFilePath string) string
+	StoreRaw(bucketName string, objectName string, data []byte) (string, error)
+	StoreExisting(bucketName string, objectName string, existingFilePath string) (string, error)
+	GetPath(bucketName string, objectName string) string
 }
 
 type FileSystemStorageProvider struct {
 	fileSystem afero.Fs
 	basePath   string
+	wd         string
 }
 
-func (sp FileSystemStorageProvider) storeRaw(bucketName string, objectName string, data []byte) string {
+func (sp FileSystemStorageProvider) StoreRaw(bucketName string, objectName string, data []byte) (string, error) {
 	directoryPath := filepath.Join(sp.basePath, bucketName)
 
-	sp.fileSystem.MkdirAll(directoryPath, os.ModePerm)
+	if err := sp.fileSystem.MkdirAll(directoryPath, os.ModePerm); err != nil {
+		return "", err
+	}
 
 	filePath := filepath.Join(directoryPath, objectName)
 
-	afero.WriteFile(sp.fileSystem, filePath, data, os.ModePerm)
+	if err := afero.WriteFile(sp.fileSystem, filePath, data, os.ModePerm); err != nil {
+		return "", err
+	}
+
+	return filePath, nil
+}
+
+func (sp FileSystemStorageProvider) StoreExisting(bucketName string, objectName string, existingFilePath string) (string, error) {
+	bytesRead, err := os.ReadFile(existingFilePath)
+	if err != nil {
+		return "", err
+	}
+
+	return sp.StoreRaw(bucketName, objectName, bytesRead)
+}
+
+func (sp FileSystemStorageProvider) GetPath(bucketName string, objectName string) string {
+	return filepath.Join(sp.wd, StorageFolderName, sp.basePath, bucketName, objectName)
+}
 
-	return filePath
+func (sp FileSystemStorageProvider) OpenFile(bucketName string, objectName string) (*os.File, error) {
+	return os.Open(sp.GetPath(bucketName, objectName))
 }
 
-func (sp FileSystemStorageProvider) storeExisting(bucketName string, objectName string, existingFilePath string) string {
-	bytesRead, _ := afero.ReadFile(sp.fileSystem, existingFilePath)
+func GetFileSystemStorageProvider(basePath string, wd string) FileSystemStorageProvider {
+	if wd == "" {
+		wd, _ = os.Getwd()
+	}
+
+	return FileSystemStorageProvider{
+		fileSystem: afero.NewBasePathFs(afero.NewOsFs(), filepath.Join(wd, StorageFolderName)),
+		basePath:   basePath,
+		wd:         wd,
+	}
+}
 
-	return sp.storeRaw(bucketName, objectName, bytesRead)
+// TODO: Move this out of this file
+func GetMemoryStorageProvider() FileSystemStorageProvider {
+	return FileSystemStorageProvider{
+		fileSystem: afero.NewBasePathFs(afero.NewMemMapFs(), "/"),
+		basePath:   "/tmp/foo/bar",
+	}
 }

storage/storage_test.go

@@ -19,7 +19,8 @@ func TestFileSystemStorageProvider(t *testing.T) {
 			basePath:   "/tmp/foo/bar",
 		}
 
-		finalPath := provider.storeRaw("test", "test.bin", dummyData)
+		finalPath, err := provider.StoreRaw("test", "test.bin", dummyData)
+		assert.Nil(t, err)
 		assert.Equal(t, "/tmp/foo/bar/test/test.bin", finalPath)
 
 		exists, _ := afero.Exists(fileSystem, "/tmp/foo/bar/test/test.bin")
@@ -32,14 +33,16 @@ func TestFileSystemStorageProvider(t *testing.T) {
 	t.Run("storeExisting method stores files in filesystem", func(t *testing.T) {
 		fileSystem := afero.NewMemMapFs()
 
-		afero.WriteFile(fileSystem, "/tmp/existing.bin", dummyData, os.ModePerm)
+		err := os.WriteFile("/tmp/existing.bin", dummyData, os.ModePerm)
+		assert.Nil(t, err)
 
 		provider := FileSystemStorageProvider{
 			fileSystem: fileSystem,
 			basePath:   "/tmp/foo/bar",
 		}
 
-		finalPath := provider.storeExisting("test", "test.bin", "/tmp/existing.bin")
+		finalPath, err := provider.StoreExisting("test", "test.bin", "/tmp/existing.bin")
+		assert.Nil(t, err)
 		assert.Equal(t, "/tmp/foo/bar/test/test.bin", finalPath)
 
 		exists, _ := afero.Exists(fileSystem, "/tmp/foo/bar/test/test.bin")
@@ -48,4 +51,18 @@ func TestFileSystemStorageProvider(t *testing.T) {
 		content, _ := afero.ReadFile(fileSystem, "/tmp/foo/bar/test/test.bin")
 		assert.Equal(t, dummyData, content)
 	})
+
+	t.Run("getPath method returns correct path", func(t *testing.T) {
+		fileSystem := afero.NewMemMapFs()
+
+		provider := FileSystemStorageProvider{
+			fileSystem: fileSystem,
+			basePath:   "/tmp/foo/bar",
+		}
+
+		_, err := provider.StoreRaw("test", "test.bin", dummyData)
+		assert.Nil(t, err)
+
+		assert.Equal(t, "files/tmp/foo/bar/test/test.bin", provider.GetPath("test", "test.bin"))
+	})
 }