|
| 1 | +//go:build !windows |
| 2 | + |
| 3 | +package sqldb |
| 4 | + |
| 5 | +import ( |
| 6 | + "context" |
| 7 | + "strconv" |
| 8 | + "testing" |
| 9 | + "time" |
| 10 | + |
| 11 | + "github.com/stretchr/testify/assert" |
| 12 | + "github.com/stretchr/testify/require" |
| 13 | + testcontainers "github.com/testcontainers/testcontainers-go" |
| 14 | + testmysql "github.com/testcontainers/testcontainers-go/modules/mysql" |
| 15 | + "github.com/testcontainers/testcontainers-go/wait" |
| 16 | + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" |
| 17 | + "k8s.io/apimachinery/pkg/types" |
| 18 | + |
| 19 | + "github.com/argoproj/argo-workflows/v4/config" |
| 20 | + wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" |
| 21 | + sutils "github.com/argoproj/argo-workflows/v4/server/utils" |
| 22 | + "github.com/argoproj/argo-workflows/v4/util/instanceid" |
| 23 | + "github.com/argoproj/argo-workflows/v4/util/logging" |
| 24 | + usqldb "github.com/argoproj/argo-workflows/v4/util/sqldb" |
| 25 | +) |
| 26 | + |
| 27 | +// setupMySQLArchiveTest starts a MySQL or MariaDB container, runs migrations, and returns a WorkflowArchive. |
| 28 | +func setupMySQLArchiveTest(ctx context.Context, t *testing.T, v usqldb.MySQLVariant) WorkflowArchive { |
| 29 | + t.Helper() |
| 30 | + |
| 31 | + c, err := testmysql.Run(ctx, |
| 32 | + v.Image, |
| 33 | + testmysql.WithDatabase("argo"), |
| 34 | + testmysql.WithUsername("argo"), |
| 35 | + testmysql.WithPassword("argo"), |
| 36 | + testcontainers.WithWaitStrategy( |
| 37 | + wait.ForAll( |
| 38 | + wait.ForLog(v.WaitMessage).WithStartupTimeout(60*time.Second), |
| 39 | + wait.ForListeningPort("3306/tcp"), |
| 40 | + )), |
| 41 | + ) |
| 42 | + require.NoError(t, err) |
| 43 | + t.Cleanup(func() { |
| 44 | + if termErr := testcontainers.TerminateContainer(c); termErr != nil { |
| 45 | + t.Logf("failed to terminate container: %s", termErr) |
| 46 | + } |
| 47 | + }) |
| 48 | + |
| 49 | + host, err := c.Host(ctx) |
| 50 | + require.NoError(t, err) |
| 51 | + p, err := c.MappedPort(ctx, "3306/tcp") |
| 52 | + require.NoError(t, err) |
| 53 | + port, err := strconv.Atoi(p.Port()) |
| 54 | + require.NoError(t, err) |
| 55 | + |
| 56 | + proxy, err := usqldb.NewSessionProxy(ctx, usqldb.SessionProxyConfig{ |
| 57 | + DBConfig: config.DBConfig{ |
| 58 | + MySQL: &config.MySQLConfig{ |
| 59 | + DatabaseConfig: config.DatabaseConfig{ |
| 60 | + Database: "argo", |
| 61 | + Host: host, |
| 62 | + Port: port, |
| 63 | + }, |
| 64 | + }, |
| 65 | + }, |
| 66 | + Username: "argo", |
| 67 | + Password: "argo", |
| 68 | + }) |
| 69 | + require.NoError(t, err) |
| 70 | + |
| 71 | + err = Migrate(ctx, proxy.Session(), "test", "argo_workflows", proxy.DBType()) |
| 72 | + require.NoError(t, err) |
| 73 | + |
| 74 | + t.Cleanup(func() { proxy.Close() }) |
| 75 | + |
| 76 | + return NewWorkflowArchive(proxy, "test", "", instanceid.NewService("")) |
| 77 | +} |
| 78 | + |
| 79 | +// TestMySQLListWorkflows verifies that JSON_EXTRACT/JSON_UNQUOTE queries in |
| 80 | +// ListWorkflows execute correctly against both MySQL and MariaDB. |
| 81 | +func TestMySQLListWorkflows(t *testing.T) { |
| 82 | + for name, variant := range usqldb.MySQLVariants { |
| 83 | + t.Run(name, func(t *testing.T) { |
| 84 | + ctx := logging.TestContext(t.Context()) |
| 85 | + archive := setupMySQLArchiveTest(ctx, t, variant) |
| 86 | + |
| 87 | + now := metav1.Now() |
| 88 | + err := archive.ArchiveWorkflow(ctx, &wfv1.Workflow{ |
| 89 | + ObjectMeta: metav1.ObjectMeta{ |
| 90 | + Name: "test-wf", |
| 91 | + Namespace: "default", |
| 92 | + UID: types.UID("test-uid-001"), |
| 93 | + CreationTimestamp: now, |
| 94 | + Labels: map[string]string{"env": "test"}, |
| 95 | + Annotations: map[string]string{"note": "integration-test"}, |
| 96 | + }, |
| 97 | + Spec: wfv1.WorkflowSpec{ |
| 98 | + Suspend: new(true), |
| 99 | + Arguments: wfv1.Arguments{ |
| 100 | + Parameters: []wfv1.Parameter{ |
| 101 | + {Name: "msg", Value: wfv1.AnyStringPtr("hello")}, |
| 102 | + }, |
| 103 | + }, |
| 104 | + }, |
| 105 | + Status: wfv1.WorkflowStatus{ |
| 106 | + Phase: wfv1.WorkflowSucceeded, |
| 107 | + StartedAt: now, |
| 108 | + FinishedAt: now, |
| 109 | + Progress: "1/1", |
| 110 | + Message: "completed", |
| 111 | + EstimatedDuration: wfv1.EstimatedDuration(30), |
| 112 | + }, |
| 113 | + }) |
| 114 | + require.NoError(t, err) |
| 115 | + |
| 116 | + results, err := archive.ListWorkflows(ctx, sutils.ListOptions{Namespace: "default", Limit: 10}) |
| 117 | + require.NoError(t, err) |
| 118 | + require.Len(t, results, 1) |
| 119 | + |
| 120 | + wf := results[0] |
| 121 | + assert.Equal(t, "test-wf", wf.Name) |
| 122 | + assert.Equal(t, wfv1.WorkflowSucceeded, wf.Status.Phase) |
| 123 | + assert.Equal(t, wfv1.Progress("1/1"), wf.Status.Progress) |
| 124 | + assert.Equal(t, "completed", wf.Status.Message) |
| 125 | + assert.Equal(t, "test", wf.GetLabels()["env"]) |
| 126 | + assert.Equal(t, "integration-test", wf.GetAnnotations()["note"]) |
| 127 | + assert.Equal(t, new(true), wf.Spec.Suspend) |
| 128 | + assert.Equal(t, "hello", wf.Spec.Arguments.Parameters[0].Value.String()) |
| 129 | + assert.Equal(t, wfv1.EstimatedDuration(30), wf.Status.EstimatedDuration) |
| 130 | + }) |
| 131 | + } |
| 132 | +} |
0 commit comments