| // Licensed to the Apache Software Foundation (ASF) under one or more |
| // contributor license agreements. See the NOTICE file distributed with |
| // this work for additional information regarding copyright ownership. |
| // The ASF licenses this file to You under the Apache License, Version 2.0 |
| // (the "License"); you may not use this file except in compliance with |
| // the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| package code_processing |
| |
| import ( |
| "context" |
| "encoding/json" |
| "fmt" |
| "github.com/google/go-cmp/cmp" |
| "io/fs" |
| "os" |
| "os/exec" |
| "path/filepath" |
| "reflect" |
| "strings" |
| "sync" |
| "testing" |
| "time" |
| |
| "github.com/go-redis/redismock/v8" |
| "github.com/google/uuid" |
| "go.uber.org/goleak" |
| |
| pb "beam.apache.org/playground/backend/internal/api/v1" |
| "beam.apache.org/playground/backend/internal/cache" |
| "beam.apache.org/playground/backend/internal/cache/local" |
| "beam.apache.org/playground/backend/internal/cache/redis" |
| "beam.apache.org/playground/backend/internal/db/entity" |
| "beam.apache.org/playground/backend/internal/environment" |
| "beam.apache.org/playground/backend/internal/executors" |
| "beam.apache.org/playground/backend/internal/fs_tool" |
| "beam.apache.org/playground/backend/internal/utils" |
| "beam.apache.org/playground/backend/internal/validators" |
| ) |
| |
| const ( |
| javaConfig = "{\n \"compile_cmd\": \"javac\",\n \"run_cmd\": \"java\",\n \"test_cmd\": \"java\",\n \"compile_args\": [\n \"-d\",\n \"bin\",\n \"-parameters\",\n \"-classpath\"\n ],\n \"run_args\": [\n \"-cp\",\n \"../bin:\"\n ],\n \"test_args\": [\n \"-cp\",\n \"bin:\",\n \"JUnit\"\n ]\n}" |
| pythonConfig = "{\n \"compile_cmd\": \"\",\n \"run_cmd\": \"python3\",\n \"compile_args\": [],\n \"run_args\": []\n}" |
| goConfig = "{\n \"compile_cmd\": \"go\",\n \"run_cmd\": \"\",\n \"compile_args\": [\n \"build\",\n \"-o\",\n \"bin\"\n ],\n \"run_args\": [\n ]\n}" |
| pipelinesFolder = "executable_files" |
| configFolder = "configs" |
| resourcesFolder = "resources" |
| helloWordPython = "if __name__ == \"__main__\":\n print(\"Hello world!\")\n" |
| helloWordGo = "package main\nimport \"fmt\"\nfunc main() {\n fmt.Println(\"hello world\")\n}\n" |
| helloWordJava = "class HelloWorld {\n public static void main(String[] args) {\n System.out.println(\"Hello world!\");\n }\n}" |
| graphFilePath = "resources/graph.dot" |
| jsonExtension = ".json" |
| ) |
| |
| var opt goleak.Option |
| var cacheService cache.Cache |
| |
| func TestMain(m *testing.M) { |
| setup() |
| opt = goleak.IgnoreCurrent() |
| exitValue := m.Run() |
| teardown() |
| os.Exit(exitValue) |
| } |
| |
| func setup() { |
| // create configs for java |
| err := os.MkdirAll(configFolder, fs.ModePerm) |
| if err != nil { |
| panic(err) |
| } |
| filePath := filepath.Join(configFolder, fmt.Sprintf("%s%s", pb.Sdk_SDK_JAVA.String(), jsonExtension)) |
| err = os.WriteFile(filePath, []byte(javaConfig), 0600) |
| if err != nil { |
| panic(err) |
| } |
| |
| // create dir with graph file |
| err = os.MkdirAll(resourcesFolder, fs.ModePerm) |
| if err != nil { |
| panic(err) |
| } |
| err = os.WriteFile(graphFilePath, []byte("graph"), 0600) |
| if err != nil { |
| panic(err) |
| } |
| |
| path, err := os.Getwd() |
| if err != nil { |
| panic(err) |
| } |
| err = os.Setenv("BEAM_SDK", pb.Sdk_SDK_JAVA.String()) |
| if err != nil { |
| panic(err) |
| } |
| err = os.Setenv("APP_WORK_DIR", path) |
| if err != nil { |
| panic(err) |
| } |
| |
| cacheService = local.New(context.Background()) |
| } |
| |
| func teardown() { |
| err := os.RemoveAll(configFolder) |
| if err != nil { |
| panic(fmt.Errorf("error during test teardown: %s", err.Error())) |
| } |
| err = os.RemoveAll(pipelinesFolder) |
| if err != nil { |
| panic(fmt.Errorf("error during test teardown: %s", err.Error())) |
| } |
| err = os.RemoveAll(resourcesFolder) |
| if err != nil { |
| panic(fmt.Errorf("error during test teardown: %s", err.Error())) |
| } |
| os.Clearenv() |
| } |
| |
| func Test_Process(t *testing.T) { |
| defer goleak.VerifyNone(t, opt) |
| appEnvs, err := environment.GetApplicationEnvsFromOsEnvs() |
| if err != nil { |
| panic(err) |
| } |
| sdkJavaEnv, err := environment.ConfigureBeamEnvs(appEnvs.WorkingDir()) |
| if err != nil { |
| panic(err) |
| } |
| sdkGoEnv := *sdkJavaEnv |
| sdkGoEnv.ApacheBeamSdk = pb.Sdk_SDK_GO |
| incorrectGoHelloWord := "package main\nimport \"fmt\"\nfunc main() {\n fmt.Println(\"hello world\").\n}\n" |
| type args struct { |
| ctx context.Context |
| appEnv *environment.ApplicationEnvs |
| sdkEnv *environment.BeamEnvs |
| pipelineId uuid.UUID |
| pipelineOptions string |
| } |
| tests := []struct { |
| name string |
| createExecFile bool |
| code string |
| cancelFunc bool |
| expectedStatus pb.Status |
| expectedRunOutput interface{} |
| expectedRunError interface{} |
| expectedCompileOutput interface{} |
| args args |
| }{ |
| { |
| // Test case with calling processCode method with small timeout. |
| // As a result status into cache should be set as Status_STATUS_RUN_TIMEOUT. |
| name: "Small pipeline execution timeout", |
| createExecFile: false, |
| code: "", |
| cancelFunc: false, |
| expectedStatus: pb.Status_STATUS_RUN_TIMEOUT, |
| expectedCompileOutput: nil, |
| expectedRunOutput: nil, |
| expectedRunError: nil, |
| args: args{ |
| ctx: context.Background(), |
| appEnv: &environment.ApplicationEnvs{}, |
| sdkEnv: sdkJavaEnv, |
| pipelineId: uuid.New(), |
| pipelineOptions: "", |
| }, |
| }, |
| { |
| // Test case with calling processCode method without preparing files with code. |
| // As a result status into cache should be set as Status_STATUS_VALIDATION_ERROR. |
| name: "Validation failed", |
| createExecFile: false, |
| code: "", |
| cancelFunc: false, |
| expectedStatus: pb.Status_STATUS_VALIDATION_ERROR, |
| expectedCompileOutput: nil, |
| expectedRunOutput: nil, |
| expectedRunError: nil, |
| args: args{ |
| ctx: context.Background(), |
| appEnv: appEnvs, |
| sdkEnv: sdkJavaEnv, |
| pipelineId: uuid.New(), |
| pipelineOptions: "", |
| }, |
| }, |
| { |
| // Test case with calling processCode method with incorrect code. |
| // As a result status into cache should be set as Status_STATUS_COMPILE_ERROR. |
| name: "Compilation failed", |
| createExecFile: true, |
| code: "MOCK_CODE", |
| cancelFunc: false, |
| expectedStatus: pb.Status_STATUS_COMPILE_ERROR, |
| expectedCompileOutput: "error: exit status 1\noutput: %s:1: error: reached end of file while parsing\nMOCK_CODE\n^\n1 error\n", |
| expectedRunOutput: nil, |
| expectedRunError: nil, |
| args: args{ |
| ctx: context.Background(), |
| appEnv: appEnvs, |
| sdkEnv: sdkJavaEnv, |
| pipelineId: uuid.New(), |
| pipelineOptions: "", |
| }, |
| }, |
| { |
| // Test case with calling processCode method with incorrect logic into code. |
| // As a result status into cache should be set as Status_STATUS_RUN_ERROR. |
| name: "Run failed", |
| createExecFile: true, |
| code: "class HelloWorld {\n public static void main(String[] args) {\n System.out.println(1/0);\n }\n}", |
| cancelFunc: false, |
| expectedStatus: pb.Status_STATUS_RUN_ERROR, |
| expectedCompileOutput: "", |
| expectedRunOutput: "", |
| expectedRunError: "error: exit status 1\noutput: Exception in thread \"main\" java.lang.ArithmeticException: / by zero\n\tat HelloWorld.main(%s.java:3)\n", |
| args: args{ |
| ctx: context.Background(), |
| appEnv: appEnvs, |
| sdkEnv: sdkJavaEnv, |
| pipelineId: uuid.New(), |
| pipelineOptions: "", |
| }, |
| }, |
| { |
| // Test case with calling processCode with canceling code processing. |
| // As a result status into cache should be set as Status_STATUS_CANCELED. |
| name: "Cancel", |
| createExecFile: true, |
| code: "class HelloWorld {\n public static void main(String[] args) {\n while(true){}\n }\n}", |
| cancelFunc: true, |
| expectedStatus: pb.Status_STATUS_CANCELED, |
| expectedCompileOutput: "", |
| expectedRunOutput: "", |
| expectedRunError: "", |
| args: args{ |
| ctx: context.Background(), |
| appEnv: appEnvs, |
| sdkEnv: sdkJavaEnv, |
| pipelineId: uuid.New(), |
| pipelineOptions: "", |
| }, |
| }, |
| { |
| // Test case with calling processCode without any error cases. |
| // As a result status into cache should be set as Status_STATUS_FINISHED. |
| name: "Processing complete successfully on java sdk", |
| createExecFile: true, |
| cancelFunc: false, |
| code: helloWordJava, |
| expectedStatus: pb.Status_STATUS_FINISHED, |
| expectedCompileOutput: "", |
| expectedRunOutput: "Hello world!\n", |
| expectedRunError: "", |
| args: args{ |
| ctx: context.Background(), |
| appEnv: appEnvs, |
| sdkEnv: sdkJavaEnv, |
| pipelineId: uuid.New(), |
| pipelineOptions: "", |
| }, |
| }, |
| { |
| // Test case with calling processCode method with incorrect go code. |
| // As a result status into cache should be set as Status_STATUS_PREPARATION_ERROR. |
| name: "Prepare step failed", |
| createExecFile: true, |
| code: incorrectGoHelloWord, |
| cancelFunc: false, |
| expectedStatus: pb.Status_STATUS_PREPARATION_ERROR, |
| expectedCompileOutput: nil, |
| expectedRunOutput: nil, |
| expectedRunError: nil, |
| args: args{ |
| ctx: context.Background(), |
| appEnv: appEnvs, |
| sdkEnv: &sdkGoEnv, |
| pipelineId: uuid.New(), |
| pipelineOptions: "", |
| }, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| lc, _ := fs_tool.NewLifeCycle(pb.Sdk_SDK_JAVA, tt.args.pipelineId, filepath.Join(os.Getenv("APP_WORK_DIR"), pipelinesFolder)) |
| err := lc.CreateFolders() |
| if err != nil { |
| t.Fatalf("error during prepare folders: %s", err.Error()) |
| } |
| |
| sources := []entity.FileEntity{{Name: "main.java", Content: tt.code, IsMain: true}} |
| if tt.createExecFile { |
| _ = lc.CreateSourceCodeFiles(sources) |
| } |
| if err = utils.SetToCache(cacheService, tt.args.pipelineId, cache.Canceled, false); err != nil { |
| t.Fatal("error during set cancel flag to cache") |
| } |
| if tt.cancelFunc { |
| go func(ctx context.Context, pipelineId uuid.UUID) { |
| // to imitate behavior of cancellation |
| time.Sleep(5 * time.Second) |
| _ = cacheService.SetValue(ctx, pipelineId, cache.Canceled, true) |
| }(tt.args.ctx, tt.args.pipelineId) |
| } |
| Process(tt.args.ctx, cacheService, lc, tt.args.pipelineId, tt.args.appEnv, tt.args.sdkEnv, tt.args.pipelineOptions) |
| |
| status, _ := cacheService.GetValue(tt.args.ctx, tt.args.pipelineId, cache.Status) |
| if !reflect.DeepEqual(status, tt.expectedStatus) { |
| t.Errorf("processCode() set status: %s, but expectes: %s", status, tt.expectedStatus) |
| } |
| |
| compileOutput, _ := cacheService.GetValue(tt.args.ctx, tt.args.pipelineId, cache.CompileOutput) |
| if tt.expectedCompileOutput != nil && strings.Contains(tt.expectedCompileOutput.(string), "%s") { |
| tt.expectedCompileOutput = fmt.Sprintf(tt.expectedCompileOutput.(string), lc.Paths.AbsoluteSourceFilePath) |
| } |
| if !reflect.DeepEqual(compileOutput, tt.expectedCompileOutput) { |
| t.Errorf("processCode() set compileOutput: %s, but expectes: %s", compileOutput, tt.expectedCompileOutput) |
| } |
| |
| runOutput, _ := cacheService.GetValue(tt.args.ctx, tt.args.pipelineId, cache.RunOutput) |
| if !reflect.DeepEqual(runOutput, tt.expectedRunOutput) { |
| t.Errorf("processCode() set runOutput: %s, but expectes: %s", runOutput, tt.expectedRunOutput) |
| } |
| |
| runError, _ := cacheService.GetValue(tt.args.ctx, tt.args.pipelineId, cache.RunError) |
| if tt.expectedRunError != nil && strings.Contains(tt.expectedRunError.(string), "%s") { |
| tt.expectedRunError = fmt.Sprintf(tt.expectedRunError.(string), tt.args.pipelineId) |
| } |
| if !reflect.DeepEqual(runError, tt.expectedRunError) { |
| t.Errorf("processCode() set runError: %s, but expectes: %s", runError, tt.expectedRunError) |
| } |
| }) |
| } |
| } |
| |
| func TestGetProcessingOutput(t *testing.T) { |
| defer goleak.VerifyNone(t, opt) |
| pipelineId := uuid.New() |
| incorrectConvertPipelineId := uuid.New() |
| err := cacheService.SetValue(context.Background(), pipelineId, cache.RunOutput, "MOCK_RUN_OUTPUT") |
| if err != nil { |
| panic(err) |
| } |
| err = cacheService.SetValue(context.Background(), incorrectConvertPipelineId, cache.RunOutput, cache.RunOutput) |
| if err != nil { |
| panic(err) |
| } |
| |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| key uuid.UUID |
| subKey cache.SubKey |
| errorTitle string |
| } |
| tests := []struct { |
| name string |
| args args |
| want string |
| wantErr bool |
| }{ |
| { |
| // Test case with calling GetProcessingOutput with pipelineId which doesn't contain run output. |
| // As a result, want to receive an error. |
| name: "Get run output with incorrect pipelineId", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: uuid.New(), |
| subKey: cache.RunOutput, |
| errorTitle: "", |
| }, |
| want: "", |
| wantErr: true, |
| }, |
| { |
| // Test case with calling GetProcessingOutput with pipelineId which contains incorrect run output. |
| // As a result, want to receive an error. |
| name: "Get run output with incorrect run output", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: incorrectConvertPipelineId, |
| subKey: cache.RunOutput, |
| errorTitle: "", |
| }, |
| want: "", |
| wantErr: true, |
| }, |
| { |
| // Test case with calling GetProcessingOutput with pipelineId which contains run output. |
| // As a result, want to receive an expected string. |
| name: "Get run output with correct pipelineId", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: pipelineId, |
| subKey: cache.RunOutput, |
| errorTitle: "", |
| }, |
| want: "MOCK_RUN_OUTPUT", |
| wantErr: false, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| got, err := GetProcessingOutput(tt.args.ctx, tt.args.cacheService, tt.args.key, tt.args.subKey, tt.args.errorTitle) |
| if (err != nil) != tt.wantErr { |
| t.Errorf("GetProcessingOutput() error = %v, wantErr %v", err, tt.wantErr) |
| return |
| } |
| if got != tt.want { |
| t.Errorf("GetProcessingOutput() got = %v, want %v", got, tt.want) |
| } |
| }) |
| } |
| } |
| |
| func TestGetProcessingStatus(t *testing.T) { |
| defer goleak.VerifyNone(t, opt) |
| pipelineId := uuid.New() |
| incorrectConvertPipelineId := uuid.New() |
| err := cacheService.SetValue(context.Background(), pipelineId, cache.Status, pb.Status_STATUS_FINISHED) |
| if err != nil { |
| panic(err) |
| } |
| err = cacheService.SetValue(context.Background(), incorrectConvertPipelineId, cache.Status, "MOCK_STATUS") |
| if err != nil { |
| panic(err) |
| } |
| |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| key uuid.UUID |
| errorTitle string |
| } |
| tests := []struct { |
| name string |
| args args |
| want pb.Status |
| wantErr bool |
| }{ |
| { |
| // Test case with calling GetProcessingStatus with pipelineId which doesn't contain status. |
| // As a result, want to receive an error. |
| name: "Get status with incorrect pipelineId", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: uuid.New(), |
| errorTitle: "", |
| }, |
| want: pb.Status_STATUS_UNSPECIFIED, |
| wantErr: true, |
| }, |
| { |
| // Test case with calling GetProcessingStatus with pipelineId which contains incorrect status value in cache. |
| // As a result, want to receive an error. |
| name: "Get status with incorrect cache value", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: incorrectConvertPipelineId, |
| errorTitle: "", |
| }, |
| want: pb.Status_STATUS_UNSPECIFIED, |
| wantErr: true, |
| }, |
| { |
| // Test case with calling GetProcessingStatus with pipelineId which contains status. |
| // As a result, want to receive an expected status. |
| name: "Get status with correct pipelineId", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: pipelineId, |
| errorTitle: "", |
| }, |
| want: pb.Status_STATUS_FINISHED, |
| wantErr: false, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| got, err := GetProcessingStatus(tt.args.ctx, tt.args.cacheService, tt.args.key, tt.args.errorTitle) |
| if (err != nil) != tt.wantErr { |
| t.Errorf("GetProcessingStatus() error = %v, wantErr %v", err, tt.wantErr) |
| return |
| } |
| if got != tt.want { |
| t.Errorf("GetProcessingStatus() got = %v, want %v", got, tt.want) |
| } |
| }) |
| } |
| } |
| |
| func TestGetLastIndex(t *testing.T) { |
| defer goleak.VerifyNone(t, opt) |
| pipelineId := uuid.New() |
| incorrectConvertPipelineId := uuid.New() |
| err := cacheService.SetValue(context.Background(), pipelineId, cache.RunOutputIndex, 2) |
| if err != nil { |
| panic(err) |
| } |
| |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| key uuid.UUID |
| subKey cache.SubKey |
| errorTitle string |
| } |
| tests := []struct { |
| name string |
| args args |
| want int |
| wantErr bool |
| }{ |
| { |
| // Test case with calling GetLastIndex with pipelineId which doesn't contain last index. |
| // As a result, want to receive an error. |
| name: "Get last index with incorrect pipelineId", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: uuid.New(), |
| subKey: cache.RunOutputIndex, |
| errorTitle: "", |
| }, |
| want: 0, |
| wantErr: true, |
| }, |
| { |
| // Test case with calling GetLastIndex with pipelineId which contains incorrect status value in cache. |
| // As a result, want to receive an error. |
| name: "Get last index with incorrect cache value", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: incorrectConvertPipelineId, |
| subKey: cache.RunOutputIndex, |
| errorTitle: "", |
| }, |
| want: 0, |
| wantErr: true, |
| }, |
| { |
| // Test case with calling GetLastIndex with pipelineId which contains last index. |
| // As a result, want to receive an expected last index. |
| name: "Get last index with correct pipelineId", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: pipelineId, |
| subKey: cache.RunOutputIndex, |
| errorTitle: "", |
| }, |
| want: 2, |
| wantErr: false, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| got, err := GetLastIndex(tt.args.ctx, tt.args.cacheService, tt.args.key, tt.args.subKey, tt.args.errorTitle) |
| if (err != nil) != tt.wantErr { |
| t.Errorf("GetLastIndex() error = %v, wantErr %v", err, tt.wantErr) |
| return |
| } |
| if got != tt.want { |
| t.Errorf("GetLastIndex() got = %v, want %v", got, tt.want) |
| } |
| }) |
| } |
| } |
| |
| func Test_getRunOrTestCmd(t *testing.T) { |
| runEx := executors.NewExecutorBuilder(). |
| WithRunner(). |
| WithCommand("runCommand"). |
| WithArgs([]string{"arg1"}). |
| WithPipelineOptions([]string{""}). |
| Build() |
| |
| testEx := executors.NewExecutorBuilder(). |
| WithTestRunner(). |
| WithCommand("testCommand"). |
| WithArgs([]string{"arg1"}). |
| Build() |
| |
| wantRunExec := exec.CommandContext(context.Background(), "runCommand", "arg1") |
| wantTestExec := exec.CommandContext(context.Background(), "testCommand", "arg1") |
| |
| type args struct { |
| isUnitTest bool |
| executor *executors.Executor |
| ctxWithTimeout context.Context |
| } |
| |
| tests := []struct { |
| name string |
| args args |
| want *exec.Cmd |
| }{ |
| { |
| //Get cmd objects with set run executor |
| name: "Get run cmd", |
| args: args{ |
| isUnitTest: false, |
| executor: &runEx, |
| ctxWithTimeout: context.Background(), |
| }, |
| want: wantRunExec, |
| }, |
| { |
| //Get cmd objects with set test executor |
| name: "Get test cmd", |
| args: args{ |
| isUnitTest: true, |
| executor: &testEx, |
| ctxWithTimeout: context.Background(), |
| }, |
| want: wantTestExec, |
| }, |
| } |
| |
| execComparer := cmp.Comparer(func(a exec.Cmd, b exec.Cmd) bool { |
| return a.Path == b.Path && |
| cmp.Equal(a.Args, b.Args) && |
| cmp.Equal(a.Env, b.Env) && |
| a.Dir == b.Dir |
| }) |
| |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| if got := getExecuteCmd(tt.args.isUnitTest, tt.args.executor, tt.args.ctxWithTimeout); !cmp.Equal(got, tt.want, execComparer) { |
| t.Errorf("getExecuteCmd() = '%v', want '%v', diff = %v", got, tt.want, cmp.Diff(got, tt.want, execComparer)) |
| } |
| }) |
| } |
| } |
| |
| func getSdkEnv(sdk pb.Sdk) (*environment.BeamEnvs, error) { |
| setupSDK(sdk) |
| appEnvs, err := environment.GetApplicationEnvsFromOsEnvs() |
| if err != nil { |
| return nil, err |
| } |
| sdkEnv, err := environment.ConfigureBeamEnvs(appEnvs.WorkingDir()) |
| if err != nil { |
| return nil, err |
| } |
| return sdkEnv, nil |
| } |
| |
| func setupSDK(sdk pb.Sdk) { |
| err := os.MkdirAll(configFolder, fs.ModePerm) |
| if err != nil { |
| panic(err) |
| } |
| filePath := filepath.Join(configFolder, fmt.Sprintf("%s%s", sdk.String(), jsonExtension)) |
| switch sdk { |
| case pb.Sdk_SDK_JAVA: |
| err = os.WriteFile(filePath, []byte(javaConfig), 0600) |
| case pb.Sdk_SDK_PYTHON: |
| err = os.WriteFile(filePath, []byte(pythonConfig), 0600) |
| case pb.Sdk_SDK_GO: |
| err = os.WriteFile(filePath, []byte(goConfig), 0600) |
| } |
| if err != nil { |
| panic(err) |
| } |
| |
| err = os.Setenv("BEAM_SDK", sdk.String()) |
| if err != nil { |
| panic(err) |
| } |
| err = os.Setenv("APP_WORK_DIR", "") |
| if err != nil { |
| panic(err) |
| } |
| err = os.Setenv("PREPARED_MOD_DIR", "") |
| if err != nil { |
| panic(err) |
| } |
| |
| cacheService = local.New(context.Background()) |
| } |
| |
| func teardownBenchmarks() { |
| err := os.RemoveAll(configFolder) |
| if err != nil { |
| panic(fmt.Errorf("error during test teardown: %s", err.Error())) |
| } |
| err = os.RemoveAll(pipelinesFolder) |
| if err != nil { |
| panic(fmt.Errorf("error during test teardown: %s", err.Error())) |
| } |
| } |
| |
| func prepareFiles(b *testing.B, pipelineId uuid.UUID, code string, sdk pb.Sdk) *fs_tool.LifeCycle { |
| lc, err := fs_tool.NewLifeCycle(sdk, pipelineId, pipelinesFolder) |
| if err != nil { |
| b.Fatalf("error during initializse lc: %s", err.Error()) |
| } |
| err = lc.CreateFolders() |
| if err != nil { |
| b.Fatalf("error during prepare folders: %s", err.Error()) |
| } |
| sources := []entity.FileEntity{{Name: "main.java", Content: code, IsMain: true}} |
| err = lc.CreateSourceCodeFiles(sources) |
| if err != nil { |
| b.Fatalf("error during prepare source code file: %s", err.Error()) |
| } |
| return lc |
| } |
| |
| func Benchmark_ProcessJava(b *testing.B) { |
| setupSDK(pb.Sdk_SDK_JAVA) |
| defer teardownBenchmarks() |
| |
| appEnv, err := environment.GetApplicationEnvsFromOsEnvs() |
| if err != nil { |
| b.Fatalf("error during preparing appEnv: %s", err) |
| } |
| sdkEnv, err := environment.ConfigureBeamEnvs(appEnv.WorkingDir()) |
| if err != nil { |
| b.Fatalf("error during preparing sdkEnv: %s", err) |
| } |
| |
| ctx := context.Background() |
| code := "class HelloWorld {\n public static void main(String[] args) {\n System.out.println(\"Hello world!\");\n }\n}" |
| |
| b.ResetTimer() |
| for i := 0; i < b.N; i++ { |
| b.StopTimer() |
| pipelineId := uuid.New() |
| lc := prepareFiles(b, pipelineId, code, pb.Sdk_SDK_JAVA) |
| if err = utils.SetToCache(cacheService, pipelineId, cache.Canceled, false); err != nil { |
| b.Fatal("error during set cancel flag to cache") |
| } |
| b.StartTimer() |
| |
| Process(ctx, cacheService, lc, pipelineId, appEnv, sdkEnv, "") |
| } |
| } |
| |
| func Benchmark_ProcessPython(b *testing.B) { |
| setupSDK(pb.Sdk_SDK_PYTHON) |
| defer teardownBenchmarks() |
| |
| appEnv, err := environment.GetApplicationEnvsFromOsEnvs() |
| if err != nil { |
| b.Fatalf("error during preparing appEnv: %s", err) |
| } |
| sdkEnv, err := environment.ConfigureBeamEnvs(appEnv.WorkingDir()) |
| if err != nil { |
| b.Fatalf("error during preparing sdkEnv: %s", err) |
| } |
| |
| ctx := context.Background() |
| wordCountCode := "import argparse\nimport logging\nimport re\n\nimport apache_beam as beam\nfrom apache_beam.io import ReadFromText\nfrom apache_beam.io import WriteToText\nfrom apache_beam.options.pipeline_options import PipelineOptions\nfrom apache_beam.options.pipeline_options import SetupOptions\n\n\nclass WordExtractingDoFn(beam.DoFn):\n \"\"\"Parse each line of input text into words.\"\"\"\n def process(self, element):\n \"\"\"Returns an iterator over the words of this element.\n\n The element is a line of text. If the line is blank, note that, too.\n\n Args:\n element: the element being processed\n\n Returns:\n The processed element.\n \"\"\"\n return re.findall(r'[\\w\\']+', element, re.UNICODE)\n\n\ndef run(argv=None, save_main_session=True):\n \"\"\"Main entry point; defines and runs the wordcount pipeline.\"\"\"\n parser = argparse.ArgumentParser()\n parser.add_argument(\n '--input',\n dest='input',\n default='gs://dataflow-samples/shakespeare/kinglear.txt',\n help='Input file to process.')\n parser.add_argument(\n '--output',\n dest='output',\n required=True,\n help='Output file to write results to.')\n known_args, pipeline_args = parser.parse_known_args(argv)\n\n # We use the save_main_session option because one or more DoFn's in this\n # workflow rely on global context (e.g., a module imported at module level).\n pipeline_options = PipelineOptions(pipeline_args)\n pipeline_options.view_as(SetupOptions).save_main_session = save_main_session\n\n # The pipeline will be run on exiting the with block.\n with beam.Pipeline(options=pipeline_options) as p:\n\n # Read the text file[pattern] into a PCollection.\n lines = p | 'Read' >> ReadFromText(known_args.input)\n\n counts = (\n lines\n | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))\n | 'PairWithOne' >> beam.Map(lambda x: (x, 1))\n | 'GroupAndSum' >> beam.CombinePerKey(sum))\n\n # Format the counts into a PCollection of strings.\n def format_result(word, count):\n return '%s: %d' % (word, count)\n\n output = counts | 'Format' >> beam.MapTuple(format_result)\n\n # Write the output using a \"Write\" transform that has side effects.\n # pylint: disable=expression-not-assigned\n output | 'Write' >> WriteToText(known_args.output)\n\n\nif __name__ == '__main__':\n logging.getLogger().setLevel(logging.INFO)\n run()" |
| pipelineOptions := "--output t.txt" |
| b.ResetTimer() |
| for i := 0; i < b.N; i++ { |
| b.StopTimer() |
| pipelineId := uuid.New() |
| lc := prepareFiles(b, pipelineId, wordCountCode, pb.Sdk_SDK_PYTHON) |
| if err = utils.SetToCache(cacheService, pipelineId, cache.Canceled, false); err != nil { |
| b.Fatal("error during set cancel flag to cache") |
| } |
| b.StartTimer() |
| |
| Process(ctx, cacheService, lc, pipelineId, appEnv, sdkEnv, pipelineOptions) |
| } |
| } |
| |
| func Benchmark_ProcessGo(b *testing.B) { |
| setupSDK(pb.Sdk_SDK_GO) |
| defer teardownBenchmarks() |
| |
| appEnv, err := environment.GetApplicationEnvsFromOsEnvs() |
| if err != nil { |
| b.Fatalf("error during preparing appEnv: %s", err) |
| } |
| sdkEnv, err := environment.ConfigureBeamEnvs(appEnv.WorkingDir()) |
| if err != nil { |
| b.Fatalf("error during preparing sdkEnv: %s", err) |
| } |
| |
| ctx := context.Background() |
| code := "package main\n\nimport \"fmt\"\n\nfunc main() {\n fmt.Println(\"Hello world!\")\n}" |
| |
| b.ResetTimer() |
| for i := 0; i < b.N; i++ { |
| b.StopTimer() |
| pipelineId := uuid.New() |
| lc := prepareFiles(b, pipelineId, code, pb.Sdk_SDK_GO) |
| if err = utils.SetToCache(cacheService, pipelineId, cache.Canceled, false); err != nil { |
| b.Fatal("error during set cancel flag to cache") |
| } |
| b.StartTimer() |
| |
| Process(ctx, cacheService, lc, pipelineId, appEnv, sdkEnv, "") |
| } |
| } |
| |
| func Benchmark_GetProcessingOutput(b *testing.B) { |
| pipelineId := uuid.New() |
| subKey := cache.RunOutput |
| ctx := context.Background() |
| |
| err := cacheService.SetValue(ctx, pipelineId, subKey, "MOCK_RUN_OUTPUT") |
| if err != nil { |
| b.Fatalf("error during prepare cache value: %s", err.Error()) |
| } |
| |
| b.ResetTimer() |
| for i := 0; i < b.N; i++ { |
| _, _ = GetProcessingOutput(ctx, cacheService, pipelineId, subKey, "") |
| } |
| } |
| |
| func Benchmark_GetProcessingStatus(b *testing.B) { |
| pipelineId := uuid.New() |
| subKey := cache.Status |
| ctx := context.Background() |
| |
| err := cacheService.SetValue(ctx, pipelineId, subKey, pb.Status_STATUS_FINISHED) |
| if err != nil { |
| b.Fatalf("error during prepare cache value: %s", err.Error()) |
| } |
| |
| b.ResetTimer() |
| for i := 0; i < b.N; i++ { |
| _, _ = GetProcessingStatus(ctx, cacheService, pipelineId, "") |
| } |
| } |
| |
| func Benchmark_GetLastIndex(b *testing.B) { |
| pipelineId := uuid.New() |
| subKey := cache.RunOutputIndex |
| ctx := context.Background() |
| |
| err := cacheService.SetValue(ctx, pipelineId, subKey, 5) |
| if err != nil { |
| b.Fatalf("error during prepare cache value: %s", err.Error()) |
| } |
| |
| b.ResetTimer() |
| for i := 0; i < b.N; i++ { |
| _, _ = GetLastIndex(ctx, cacheService, pipelineId, subKey, "") |
| } |
| } |
| |
| func Test_validateStep(t *testing.T) { |
| javaSdkEnv, err := getSdkEnv(pb.Sdk_SDK_JAVA) |
| if err != nil { |
| panic(err) |
| } |
| incorrectSdkEnv := &environment.BeamEnvs{ |
| ApacheBeamSdk: pb.Sdk_SDK_UNSPECIFIED, |
| ExecutorConfig: nil, |
| } |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| pipelineId uuid.UUID |
| sdkEnv *environment.BeamEnvs |
| pipelineLifeCycleCtx context.Context |
| validationResults *sync.Map |
| } |
| tests := []struct { |
| name string |
| args args |
| want int |
| code string |
| }{ |
| { |
| name: "Test validation step by checking number of validators", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: javaSdkEnv, |
| pipelineLifeCycleCtx: context.Background(), |
| validationResults: &sync.Map{}, |
| }, |
| want: 3, |
| code: helloWordJava, |
| }, |
| { |
| name: "Test validation step with incorrect sdkEnv", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: incorrectSdkEnv, |
| pipelineLifeCycleCtx: context.Background(), |
| validationResults: &sync.Map{}, |
| }, |
| want: 0, |
| code: helloWordJava, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| lc, _ := fs_tool.NewLifeCycle(pb.Sdk_SDK_JAVA, tt.args.pipelineId, filepath.Join(os.Getenv("APP_WORK_DIR"), pipelinesFolder)) |
| err := lc.CreateFolders() |
| if err != nil { |
| t.Fatalf("error during prepare folders: %s", err.Error()) |
| } |
| sources := []entity.FileEntity{{Name: "main.java", Content: tt.code, IsMain: true}} |
| _ = lc.CreateSourceCodeFiles(sources) |
| err = validateStep(tt.args.pipelineLifeCycleCtx, tt.args.cacheService, &lc.Paths, tt.args.pipelineId, tt.args.sdkEnv, tt.args.validationResults) |
| got := syncMapLen(tt.args.validationResults) |
| if err != nil && !reflect.DeepEqual(got, tt.want) { |
| t.Errorf("validateStep() = %d, want %d", got, tt.want) |
| } |
| }) |
| } |
| } |
| |
| func Test_prepareStep(t *testing.T) { |
| javaSdkEnv, err := getSdkEnv(pb.Sdk_SDK_JAVA) |
| if err != nil { |
| panic(err) |
| } |
| incorrectSdkEnv := &environment.BeamEnvs{ |
| ApacheBeamSdk: pb.Sdk_SDK_UNSPECIFIED, |
| ExecutorConfig: nil, |
| } |
| validationResults := sync.Map{} |
| validationResults.Store(validators.UnitTestValidatorName, false) |
| validationResults.Store(validators.KatasValidatorName, false) |
| pipelineLifeCycleCtx, cancel := context.WithTimeout(context.Background(), 1) |
| defer cancel() |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| pipelineId uuid.UUID |
| sdkEnv *environment.BeamEnvs |
| pipelineLifeCycleCtx context.Context |
| validationResults *sync.Map |
| } |
| tests := []struct { |
| name string |
| args args |
| code string |
| expectedStatus pb.Status |
| }{ |
| { |
| name: "Test preparer step working without an error", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: javaSdkEnv, |
| pipelineLifeCycleCtx: context.Background(), |
| validationResults: &validationResults, |
| }, |
| code: helloWordJava, |
| expectedStatus: pb.Status_STATUS_COMPILING, |
| }, |
| { |
| name: "Test preparer step working with incorrect sdkEnv", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: incorrectSdkEnv, |
| pipelineLifeCycleCtx: context.Background(), |
| validationResults: &validationResults, |
| }, |
| code: "", |
| expectedStatus: pb.Status_STATUS_ERROR, |
| }, |
| { |
| name: "Error during expired context of the example", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: javaSdkEnv, |
| pipelineLifeCycleCtx: pipelineLifeCycleCtx, |
| validationResults: &validationResults, |
| }, |
| code: "", |
| expectedStatus: pb.Status_STATUS_RUN_TIMEOUT, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| lc, _ := fs_tool.NewLifeCycle(pb.Sdk_SDK_JAVA, tt.args.pipelineId, filepath.Join(os.Getenv("APP_WORK_DIR"), pipelinesFolder)) |
| err := lc.CreateFolders() |
| if err != nil { |
| t.Fatalf("error during prepare folders: %s", err.Error()) |
| } |
| sources := []entity.FileEntity{{Name: "main.java", Content: tt.code, IsMain: true}} |
| _ = lc.CreateSourceCodeFiles(sources) |
| _ = prepareStep(tt.args.pipelineLifeCycleCtx, tt.args.cacheService, &lc.Paths, tt.args.pipelineId, tt.args.sdkEnv, tt.args.validationResults, nil) |
| status, _ := cacheService.GetValue(tt.args.ctx, tt.args.pipelineId, cache.Status) |
| if status != tt.expectedStatus { |
| t.Errorf("prepareStep: got status = %v, want %v", status, tt.expectedStatus) |
| } |
| }) |
| } |
| } |
| |
| func Test_compileStep(t *testing.T) { |
| sdkJavaEnv, err := getSdkEnv(pb.Sdk_SDK_JAVA) |
| if err != nil { |
| panic(err) |
| } |
| sdkPythonEnv, err := getSdkEnv(pb.Sdk_SDK_PYTHON) |
| if err != nil { |
| panic(err) |
| } |
| pipelineLifeCycleCtx, cancel := context.WithTimeout(context.Background(), 1) |
| defer cancel() |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| pipelineId uuid.UUID |
| sdkEnv *environment.BeamEnvs |
| isUnitTest bool |
| pipelineLifeCycleCtx context.Context |
| } |
| tests := []struct { |
| name string |
| args args |
| code string |
| expectedStatus pb.Status |
| }{ |
| { |
| name: "Test compilation step finishes successfully on java sdk", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: sdkJavaEnv, |
| isUnitTest: false, |
| pipelineLifeCycleCtx: context.Background(), |
| }, |
| code: helloWordJava, |
| expectedStatus: pb.Status_STATUS_EXECUTING, |
| }, |
| { |
| name: "Test compilation step finishes successfully on python sdk", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: sdkPythonEnv, |
| isUnitTest: false, |
| pipelineLifeCycleCtx: context.Background(), |
| }, |
| code: helloWordPython, |
| expectedStatus: pb.Status_STATUS_EXECUTING, |
| }, |
| { |
| name: "Error during expired context of the example", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| sdkEnv: sdkJavaEnv, |
| isUnitTest: false, |
| pipelineLifeCycleCtx: pipelineLifeCycleCtx, |
| }, |
| code: helloWordJava, |
| expectedStatus: pb.Status_STATUS_RUN_TIMEOUT, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| lc, _ := fs_tool.NewLifeCycle(tt.args.sdkEnv.ApacheBeamSdk, tt.args.pipelineId, filepath.Join(os.Getenv("APP_WORK_DIR"), pipelinesFolder)) |
| err := lc.CreateFolders() |
| if err != nil { |
| t.Fatalf("error during prepare folders: %s", err.Error()) |
| } |
| sources := []entity.FileEntity{{Name: "main.java", Content: tt.code, IsMain: true}} |
| _ = lc.CreateSourceCodeFiles(sources) |
| _ = compileStep(tt.args.pipelineLifeCycleCtx, tt.args.cacheService, &lc.Paths, tt.args.pipelineId, tt.args.sdkEnv, tt.args.isUnitTest) |
| status, _ := cacheService.GetValue(tt.args.ctx, tt.args.pipelineId, cache.Status) |
| if status != tt.expectedStatus { |
| t.Errorf("compileStep: got status = %v, want %v", status, tt.expectedStatus) |
| } |
| }) |
| } |
| } |
| |
| func Test_runStep(t *testing.T) { |
| sdkJavaEnv, err := getSdkEnv(pb.Sdk_SDK_JAVA) |
| if err != nil { |
| panic(err) |
| } |
| sdkPythonEnv, err := getSdkEnv(pb.Sdk_SDK_PYTHON) |
| if err != nil { |
| panic(err) |
| } |
| sdkGoEnv, err := getSdkEnv(pb.Sdk_SDK_GO) |
| if err != nil { |
| panic(err) |
| } |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| pipelineId uuid.UUID |
| isUnitTest bool |
| sdkEnv *environment.BeamEnvs |
| pipelineOptions string |
| pipelineLifeCycleCtx context.Context |
| createExecFile bool |
| } |
| tests := []struct { |
| name string |
| args args |
| code string |
| expectedStatus pb.Status |
| }{ |
| { |
| // Test case with calling runStep method on python sdk. |
| // cmd.Run return error during saving output. |
| // As a result, the pipeline status should be Status_STATUS_RUN_ERROR. |
| name: "Test run step working on python sdk", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| isUnitTest: false, |
| sdkEnv: sdkPythonEnv, |
| pipelineOptions: "", |
| pipelineLifeCycleCtx: context.Background(), |
| createExecFile: true, |
| }, |
| code: helloWordPython, |
| expectedStatus: pb.Status_STATUS_RUN_ERROR, |
| }, |
| { |
| // Test case with calling runStep method on go sdk. |
| // cmd.Run return error due to missing executable file. |
| // As a result, the pipeline status should be Status_STATUS_RUN_ERROR. |
| name: "Test run step working on go sdk", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.New(), |
| isUnitTest: true, |
| sdkEnv: sdkGoEnv, |
| pipelineOptions: "", |
| pipelineLifeCycleCtx: context.Background(), |
| createExecFile: true, |
| }, |
| code: helloWordGo, |
| expectedStatus: pb.Status_STATUS_RUN_ERROR, |
| }, |
| { |
| // Test case with calling runStep method without preparing files with code. |
| // As a result, the pipeline status should be Status_STATUS_ERROR. |
| name: "Test run step without preparing files with code", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| pipelineId: uuid.UUID{}, |
| isUnitTest: true, |
| sdkEnv: sdkJavaEnv, |
| pipelineOptions: "", |
| pipelineLifeCycleCtx: context.Background(), |
| createExecFile: false, |
| }, |
| code: helloWordJava, |
| expectedStatus: pb.Status_STATUS_ERROR, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| lc, _ := fs_tool.NewLifeCycle(tt.args.sdkEnv.ApacheBeamSdk, tt.args.pipelineId, filepath.Join(os.Getenv("APP_WORK_DIR"), pipelinesFolder)) |
| if tt.args.createExecFile { |
| err := lc.CreateFolders() |
| if err != nil { |
| t.Fatalf("error during prepare folders: %s", err.Error()) |
| } |
| sources := []entity.FileEntity{{Name: "main.java", Content: tt.code, IsMain: true}} |
| _ = lc.CreateSourceCodeFiles(sources) |
| } |
| _ = runStep(tt.args.pipelineLifeCycleCtx, tt.args.cacheService, &lc.Paths, tt.args.pipelineId, tt.args.isUnitTest, tt.args.sdkEnv, tt.args.pipelineOptions) |
| status, _ := cacheService.GetValue(tt.args.ctx, tt.args.pipelineId, cache.Status) |
| if status != tt.expectedStatus { |
| t.Errorf("runStep() got status = %v, want %v", status, tt.expectedStatus) |
| } |
| }) |
| } |
| } |
| |
| func syncMapLen(syncMap *sync.Map) int { |
| length := 0 |
| syncMap.Range(func(_, _ interface{}) bool { |
| length++ |
| return true |
| }) |
| return length |
| } |
| |
| func TestGetGraph(t *testing.T) { |
| ctx := context.Background() |
| pipelineId1 := uuid.New() |
| graph := "GRAPH" |
| err := cacheService.SetValue(ctx, pipelineId1, cache.Graph, graph) |
| if err != nil { |
| return |
| } |
| pipelineId2 := uuid.New() |
| err = cacheService.SetValue(ctx, pipelineId2, cache.Graph, 1) |
| if err != nil { |
| return |
| } |
| type args struct { |
| ctx context.Context |
| cacheService cache.Cache |
| key uuid.UUID |
| errorTitle string |
| } |
| tests := []struct { |
| name string |
| args args |
| want string |
| wantErr bool |
| }{ |
| { |
| name: "Get graph when key exist in cache", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: pipelineId1, |
| errorTitle: "error", |
| }, |
| want: graph, |
| wantErr: false, |
| }, |
| { |
| name: "Get graph when key doesn't exist in cache", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: uuid.New(), |
| errorTitle: "error", |
| }, |
| want: "", |
| wantErr: true, |
| }, |
| { |
| name: "Get graph when value from cache by key couldn't be converted to a string", |
| args: args{ |
| ctx: context.Background(), |
| cacheService: cacheService, |
| key: pipelineId2, |
| errorTitle: "error", |
| }, |
| want: "", |
| wantErr: true, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| got, err := GetGraph(tt.args.ctx, tt.args.cacheService, tt.args.key, tt.args.errorTitle) |
| if (err != nil) != tt.wantErr { |
| t.Errorf("GetGraph error = %v, wantErr %v", err, tt.wantErr) |
| return |
| } |
| if got != tt.want { |
| t.Errorf("GetGraph got = %v, want %v", got, tt.want) |
| } |
| }) |
| } |
| } |
| |
| func Test_processSetupError(t *testing.T) { |
| client, mock := redismock.NewClientMock() |
| pipelineId := uuid.New() |
| errorMessage := "MOCK_ERROR" |
| type args struct { |
| err error |
| pipelineId uuid.UUID |
| cacheService cache.Cache |
| ctxWithTimeout context.Context |
| } |
| tests := []struct { |
| name string |
| mocks func() |
| args args |
| wantErr bool |
| }{ |
| { |
| name: "Error during HSet operation", |
| mocks: func() { |
| mock.ExpectHSet(pipelineId.String(), "MOCK_VALUE").SetErr(fmt.Errorf(errorMessage)) |
| }, |
| args: args{ |
| err: fmt.Errorf(errorMessage), |
| pipelineId: pipelineId, |
| cacheService: &redis.Cache{ |
| Client: client, |
| }, |
| ctxWithTimeout: nil, |
| }, |
| wantErr: true, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| tt.mocks() |
| if err := processSetupError(tt.args.err, tt.args.pipelineId, tt.args.cacheService); (err != nil) != tt.wantErr { |
| t.Errorf("processSetupError() error = %v, wantErr %v", err, tt.wantErr) |
| } |
| }) |
| } |
| } |
| |
| func Test_processErrorWithSavingOutput(t *testing.T) { |
| client, mock := redismock.NewClientMock() |
| pipelineId := uuid.New() |
| errorMessage := "MOCK_ERROR" |
| subKey := cache.RunOutput |
| type args struct { |
| ctx context.Context |
| err error |
| errorOutput []byte |
| pipelineId uuid.UUID |
| subKey cache.SubKey |
| cacheService cache.Cache |
| errorTitle string |
| newStatus pb.Status |
| } |
| tests := []struct { |
| name string |
| mocks func() |
| args args |
| wantErr bool |
| }{ |
| { |
| name: "Error during HSet operation", |
| mocks: func() { |
| mock.ExpectHSet(pipelineId.String(), subKey).SetErr(fmt.Errorf(errorMessage)) |
| }, |
| args: args{ |
| ctx: context.Background(), |
| err: fmt.Errorf(errorMessage), |
| errorOutput: nil, |
| pipelineId: pipelineId, |
| subKey: subKey, |
| cacheService: &redis.Cache{Client: client}, |
| errorTitle: "", |
| newStatus: 0, |
| }, |
| wantErr: true, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| tt.mocks() |
| if err := processErrorWithSavingOutput(tt.args.err, tt.args.errorOutput, tt.args.pipelineId, tt.args.subKey, tt.args.cacheService, tt.args.errorTitle, tt.args.newStatus); (err != nil) != tt.wantErr { |
| t.Errorf("processErrorWithSavingOutput() error = %v, wantErr %v", err, tt.wantErr) |
| } |
| }) |
| } |
| } |
| |
| func Test_processRunError(t *testing.T) { |
| client, mock := redismock.NewClientMock() |
| pipelineId := uuid.New() |
| errorMessage := "MOCK_ERROR" |
| subKey := cache.RunError |
| errorChannel := make(chan error, 1) |
| errorChannel <- fmt.Errorf(errorMessage) |
| type args struct { |
| ctx context.Context |
| errorChannel chan error |
| errorOutput []byte |
| pipelineId uuid.UUID |
| cacheService cache.Cache |
| stopReadLogsChannel chan bool |
| finishReadLogsChannel chan bool |
| } |
| tests := []struct { |
| name string |
| mocks func() |
| args args |
| wantErr bool |
| }{ |
| { |
| name: "Error during HSet operation", |
| mocks: func() { |
| mock.ExpectHSet(pipelineId.String(), subKey).SetErr(fmt.Errorf(errorMessage)) |
| }, |
| args: args{ |
| ctx: context.Background(), |
| errorChannel: errorChannel, |
| errorOutput: nil, |
| pipelineId: pipelineId, |
| cacheService: &redis.Cache{Client: client}, |
| stopReadLogsChannel: nil, |
| finishReadLogsChannel: nil, |
| }, |
| wantErr: true, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| tt.mocks() |
| if err := processRunError(tt.args.errorChannel, tt.args.errorOutput, tt.args.pipelineId, tt.args.cacheService, tt.args.stopReadLogsChannel, tt.args.finishReadLogsChannel); (err != nil) != tt.wantErr { |
| t.Errorf("processRunError() error = %v, wantErr %v", err, tt.wantErr) |
| } |
| }) |
| } |
| } |
| |
| func Test_processCompileSuccess(t *testing.T) { |
| client, mock := redismock.NewClientMock() |
| pipelineId := uuid.New() |
| output := "output" |
| cacheMock := &redis.Cache{Client: client} |
| marshalLogs, _ := json.Marshal(cache.Logs) |
| marshalCompileOutput, _ := json.Marshal(cache.CompileOutput) |
| marshalRunOutput, _ := json.Marshal(cache.RunOutput) |
| marshalRunError, _ := json.Marshal(cache.RunError) |
| outputMarshal, _ := json.Marshal(output) |
| marshalEmptyString, _ := json.Marshal("") |
| type args struct { |
| ctx context.Context |
| output []byte |
| pipelineId uuid.UUID |
| cacheService cache.Cache |
| } |
| tests := []struct { |
| name string |
| mocks func() |
| args args |
| wantErr bool |
| }{ |
| { |
| name: "Error during set value to CompileOutput subKey", |
| mocks: func() { |
| }, |
| args: args{ |
| ctx: context.Background(), |
| output: []byte(output), |
| pipelineId: pipelineId, |
| cacheService: cacheMock, |
| }, |
| wantErr: true, |
| }, |
| { |
| name: "Error during set value to RunOutput subKey", |
| mocks: func() { |
| mock.ExpectHSet(pipelineId.String(), marshalCompileOutput, outputMarshal).SetVal(1) |
| }, |
| args: args{ |
| ctx: context.Background(), |
| output: []byte(output), |
| pipelineId: pipelineId, |
| cacheService: cacheMock, |
| }, |
| wantErr: true, |
| }, |
| { |
| name: "Error during set value to RunError subKey", |
| mocks: func() { |
| mock.ExpectHSet(pipelineId.String(), marshalCompileOutput, outputMarshal).SetVal(1) |
| mock.ExpectHSet(pipelineId.String(), marshalRunOutput, marshalEmptyString).SetVal(1) |
| }, |
| args: args{ |
| ctx: context.Background(), |
| output: []byte(output), |
| pipelineId: pipelineId, |
| cacheService: cacheMock, |
| }, |
| wantErr: true, |
| }, |
| { |
| name: "Error during set value to Logs subKey", |
| mocks: func() { |
| mock.ExpectHSet(pipelineId.String(), marshalCompileOutput, outputMarshal).SetVal(1) |
| mock.ExpectHSet(pipelineId.String(), marshalRunOutput, marshalEmptyString).SetVal(1) |
| mock.ExpectHSet(pipelineId.String(), marshalRunError, marshalEmptyString).SetVal(1) |
| }, |
| args: args{ |
| ctx: context.Background(), |
| output: []byte(output), |
| pipelineId: pipelineId, |
| cacheService: cacheMock, |
| }, |
| wantErr: true, |
| }, |
| { |
| name: "Error during set value to Graph subKey", |
| mocks: func() { |
| mock.ExpectHSet(pipelineId.String(), marshalCompileOutput, outputMarshal).SetVal(1) |
| mock.ExpectHSet(pipelineId.String(), marshalRunOutput, marshalEmptyString).SetVal(1) |
| mock.ExpectHSet(pipelineId.String(), marshalRunError, marshalEmptyString).SetVal(1) |
| mock.ExpectHSet(pipelineId.String(), marshalLogs, marshalEmptyString).SetVal(1) |
| }, |
| args: args{ |
| ctx: context.Background(), |
| output: []byte(output), |
| pipelineId: pipelineId, |
| cacheService: cacheMock, |
| }, |
| wantErr: true, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| tt.mocks() |
| if err := processCompileSuccess(tt.args.output, tt.args.pipelineId, tt.args.cacheService); (err != nil) != tt.wantErr { |
| t.Errorf("processCompileSuccess() error = %v, wantErr %v", err, tt.wantErr) |
| } |
| }) |
| } |
| } |
| |
| func Test_readGraphFile(t *testing.T) { |
| pipelineLifeCycleCtx, cancel := context.WithTimeout(context.Background(), 1*time.Second) |
| defer cancel() |
| type args struct { |
| pipelineLifeCycleCtx context.Context |
| backgroundCtx context.Context |
| cacheService cache.Cache |
| graphFilePath string |
| pipelineId uuid.UUID |
| } |
| tests := []struct { |
| name string |
| args args |
| }{ |
| { |
| name: "Successfully saving the prepared graph to the cache", |
| args: args{ |
| pipelineLifeCycleCtx: pipelineLifeCycleCtx, |
| backgroundCtx: context.Background(), |
| cacheService: cacheService, |
| graphFilePath: graphFilePath, |
| pipelineId: uuid.New(), |
| }, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| readGraphFile(tt.args.pipelineLifeCycleCtx, tt.args.cacheService, tt.args.graphFilePath, tt.args.pipelineId) |
| if v, _ := cacheService.GetValue(tt.args.backgroundCtx, tt.args.pipelineId, cache.Graph); v == nil { |
| t.Errorf("readGraphFile() error: the graph was not cached") |
| } |
| }) |
| } |
| } |
| |
| func Test_cancelCheck(t *testing.T) { |
| type args struct { |
| timeout time.Duration |
| cancelWaitTime time.Duration |
| setCancel bool |
| } |
| tests := []struct { |
| name string |
| args args |
| }{ |
| { |
| name: "Successfully canceling the pipeline immediately", |
| args: args{ |
| timeout: 5 * time.Minute, |
| cancelWaitTime: 0, |
| setCancel: true, |
| }, |
| }, |
| { |
| name: "Successfully canceling the pipeline after timeout", |
| args: args{ |
| timeout: 2 * pauseDuration, |
| cancelWaitTime: 3 * pauseDuration, |
| setCancel: true, |
| }, |
| }, |
| { |
| name: "Successfully canceling the pipeline after timeout (immediate timeout)", |
| args: args{ |
| timeout: 0, |
| cancelWaitTime: 3 * pauseDuration, |
| setCancel: true, |
| }, |
| }, |
| { |
| name: "Successfully timing out the pipeline without cancel", |
| args: args{ |
| timeout: 3 * pauseDuration, |
| cancelWaitTime: 0, |
| setCancel: false, |
| }, |
| }, |
| } |
| for _, tt := range tests { |
| t.Run(tt.name, func(t *testing.T) { |
| ctx, cancel := context.WithTimeout(context.Background(), tt.args.timeout) |
| defer cancel() |
| |
| id := uuid.New() |
| localCache := local.New(ctx) |
| |
| go cancelCheck(ctx, id, cancel, localCache) |
| |
| if tt.args.setCancel { |
| _ = localCache.SetValue(ctx, id, cache.Canceled, true) |
| } |
| |
| // Wait some time for the cancelCheck to be executed |
| time.Sleep(5 * pauseDuration) |
| |
| if err := ctx.Err(); err == nil { |
| t.Errorf("cancelCheck() error expected, err = %v", err) |
| } |
| }) |
| } |
| } |
