Merge pull request #635 from ericchiang/dev-transaction-tests

storage/conformance: add tests for transactional guarantees

storage/conformance/conformance.go

@@ -20,22 +20,12 @@ import (
 // ensure that values being tested on never expire.
 var neverExpire = time.Now().UTC().Add(time.Hour * 24 * 365 * 100)
 
-// RunTests runs a set of conformance tests against a storage. newStorage should
-// return an initialized but empty storage. The storage will be closed at the
-// end of each test run.
-func RunTests(t *testing.T, newStorage func() storage.Storage) {
-	tests := []struct {
+type subTest struct {
 	name string
 	run  func(t *testing.T, s storage.Storage)
-	}{
-		{"AuthCodeCRUD", testAuthCodeCRUD},
-		{"AuthRequestCRUD", testAuthRequestCRUD},
-		{"ClientCRUD", testClientCRUD},
-		{"RefreshTokenCRUD", testRefreshTokenCRUD},
-		{"PasswordCRUD", testPasswordCRUD},
-		{"KeysCRUD", testKeysCRUD},
-		{"GarbageCollection", testGC},
-	}
+}
+
+func runTests(t *testing.T, newStorage func() storage.Storage, tests []subTest) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			s := newStorage()
@@ -45,6 +35,21 @@ func RunTests(t *testing.T, newStorage func() storage.Storage) {
 	}
 }
 
+// RunTests runs a set of conformance tests against a storage. newStorage should
+// return an initialized but empty storage. The storage will be closed at the
+// end of each test run.
+func RunTests(t *testing.T, newStorage func() storage.Storage) {
+	runTests(t, newStorage, []subTest{
+		{"AuthCodeCRUD", testAuthCodeCRUD},
+		{"AuthRequestCRUD", testAuthRequestCRUD},
+		{"ClientCRUD", testClientCRUD},
+		{"RefreshTokenCRUD", testRefreshTokenCRUD},
+		{"PasswordCRUD", testPasswordCRUD},
+		{"KeysCRUD", testKeysCRUD},
+		{"GarbageCollection", testGC},
+	})
+}
+
 func mustLoadJWK(b string) *jose.JSONWebKey {
 	var jwt jose.JSONWebKey
 	if err := jwt.UnmarshalJSON([]byte(b)); err != nil {

storage/conformance/transactions.go

@@ -0,0 +1,54 @@
+// +build go1.7
+
+package conformance
+
+import (
+	"testing"
+
+	"github.com/coreos/dex/storage"
+)
+
+// RunTransactionTests runs a test suite aimed a verifying the transaction
+// guarantees of the storage interface. Atomic updates, deletes, etc. The
+// storage returned by newStorage will be closed at the end of each test run.
+//
+// This call is separate from RunTests because some storage perform extremely
+// poorly under deadlocks, such as SQLite3, while others may be working towards
+// conformance.
+func RunTransactionTests(t *testing.T, newStorage func() storage.Storage) {
+	runTests(t, newStorage, []subTest{
+		{"ClientConcurrentUpdate", testClientConcurrentUpdate},
+	})
+}
+
+func testClientConcurrentUpdate(t *testing.T, s storage.Storage) {
+	c := storage.Client{
+		ID:           storage.NewID(),
+		Secret:       "foobar",
+		RedirectURIs: []string{"foo://bar.com/", "https://auth.example.com"},
+		Name:         "dex client",
+		LogoURL:      "https://goo.gl/JIyzIC",
+	}
+
+	if err := s.CreateClient(c); err != nil {
+		t.Fatalf("create client: %v", err)
+	}
+
+	var err1, err2 error
+
+	err1 = s.UpdateClient(c.ID, func(old storage.Client) (storage.Client, error) {
+		old.Secret = "new secret 1"
+		err2 = s.UpdateClient(c.ID, func(old storage.Client) (storage.Client, error) {
+			old.Secret = "new secret 2"
+			return old, nil
+		})
+		return old, nil
+	})
+
+	t.Logf("update1: %v", err1)
+	t.Logf("update2: %v", err2)
+
+	if err1 == nil && err2 == nil {
+		t.Errorf("update client: concurrent updates both returned no error")
+	}
+}

storage/kubernetes/storage.go

@@ -75,10 +75,13 @@ func (c *Config) open() (*client, error) {
 		return nil, fmt.Errorf("create client: %v", err)
 	}
 
-	// Don't try to synchronize this because creating third party resources is not
-	// a synchronous event. Even after the API server returns a 200, it can still
-	// take several seconds for them to actually appear.
 	ctx, cancel := context.WithCancel(context.Background())
+
+	// Try to synchronously create the third party resources once. This doesn't mean
+	// they'll immediately be available, but ensures that the client will actually try
+	// once.
+	if err := cli.createThirdPartyResources(); err != nil {
+		log.Printf("failed creating third party resources: %v", err)
 		go func() {
 			for {
 				if err := cli.createThirdPartyResources(); err != nil {
@@ -94,12 +97,20 @@ func (c *Config) open() (*client, error) {
 				}
 			}
 		}()
+	}
 
 	// If the client is closed, stop trying to create third party resources.
 	cli.cancel = cancel
 	return cli, nil
 }
 
+// createThirdPartyResources attempts to create the third party resources dex
+// requires or identifies that they're already enabled.
+//
+// Creating a third party resource does not mean that they'll be immediately available.
+//
+// TODO(ericchiang): Provide an option to wait for the third party resources
+// to actually be available.
 func (cli *client) createThirdPartyResources() error {
 	for _, r := range thirdPartyResources {
 		err := cli.postResource("extensions/v1beta1", "", "thirdpartyresources", r)

storage/kubernetes/storage_test.go

@@ -1,6 +1,7 @@
 package kubernetes
 
 import (
+	"fmt"
 	"os"
 	"testing"
 
@@ -78,7 +79,7 @@ func TestURLFor(t *testing.T) {
 
 func TestStorage(t *testing.T) {
 	client := loadClient(t)
-	conformance.RunTests(t, func() storage.Storage {
+	newStorage := func() storage.Storage {
 		for _, resource := range []string{
 			resourceAuthCode,
 			resourceAuthRequest,
@@ -88,9 +89,14 @@ func TestStorage(t *testing.T) {
 			resourcePassword,
 		} {
 			if err := client.deleteAll(resource); err != nil {
+				// Fatalf sometimes doesn't print the error message.
+				fmt.Fprintf(os.Stderr, "delete all %q failed: %v\n", resource, err)
 				t.Fatalf("delete all %q failed: %v", resource, err)
 			}
 		}
 		return client
-	})
+	}
+
+	conformance.RunTests(t, newStorage)
+	conformance.RunTransactionTests(t, newStorage)
 }