Function TestRedshiftSubscriptionBatchedPollingUser

tests/redshift_surface_test.go:51–101

(t *testing.T)

}

func TestRedshiftSubscriptionBatchedPollingUser(t *testing.T) {
	if dbType != "redshift" {
		t.Skip("redshift-only test")
	}

	if _, err := db.Exec(`UPDATE users SET phone = '555-redshift-initial' WHERE id = 1`); err != nil {
		t.Fatal(err)
	}

	gj, err := core.NewGraphJin(newConfig(&core.Config{
		DBType:           "redshift",
		DisableAllowList: true,
		SubsPollDuration: 200 * time.Millisecond,
	}), db)
	if err != nil {
		t.Fatal(err)
	}
	defer gj.Close()

	m, err := gj.Subscribe(context.Background(), `
	subscription redshift_user {
		users(id: $id) {
			id
			email
			phone
		}
	}`, json.RawMessage(`{"id":1}`), nil)
	if err != nil {
		t.Fatal(err)
	}
	defer m.Unsubscribe()

	initial := waitRedshiftSubResult(t, m, 5*time.Second)
	if !strings.Contains(initial.SQL(), `JSON_PARSE(?)`) || !strings.Contains(initial.SQL(), `UNNEST(_gj_sub_input._gj_params) WITH OFFSET`) {
		t.Fatalf("redshift subscription did not use batched wrapper:\n%s", initial.SQL())
	}
	if !bytes.Contains(initial.Data, []byte(`"email":"ada@example.com"`)) ||
		!bytes.Contains(initial.Data, []byte(`"phone":"555-redshift-initial"`)) {
		t.Fatalf("unexpected initial subscription result: %s", initial.Data)
	}

	assertNoRedshiftSubResult(t, m, 700*time.Millisecond)

	if _, err := db.Exec(`UPDATE users SET phone = '555-redshift-updated' WHERE id = 1`); err != nil {
		t.Fatal(err)
	}
	changed := waitRedshiftSubResult(t, m, 5*time.Second)
	if !bytes.Contains(changed.Data, []byte(`"phone":"555-redshift-updated"`)) {
		t.Fatalf("unexpected changed subscription result: %s", changed.Data)
	}
}

func TestRedshiftSubscriptionCursorBatchedPolling(t *testing.T) {
	if dbType != "redshift" {