Repository: sivchari/gotwtr
Pull request #51 (2021-10-15 04:33:52)

Impl filtered stream
relate to #29 

# TODO 
- [x] impl retrieve stream rules
- [x] add tests for add or delete rules to your stream
- [x] impl  add or delete rules to your stream
- [x] add tests for add or delete rules to your stream
- [x] impl https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/get-tweets-search-stream
- [x] add tests for the one above


Patch:
diff --git a/client.go b/client.go
index a10bb13..ec3fdd8 100644
--- a/client.go
+++ b/client.go
@@ -11,10 +11,11 @@ type client struct {
 }
 
 const (
-	spaceLookUpMaxIDs         = 100
-	tweetLookUpMaxIDs         = 100
-	userLookUpMaxIDs          = 100
-	tweetSearchMaxQueryLength = 512
+	spaceLookUpMaxIDs           = 100
+	tweetLookUpMaxIDs           = 100
+	userLookUpMaxIDs            = 100
+	tweetSearchMaxQueryLength   = 512
+	filteredStreamRuleMaxLength = 512
 )
 
 // TODO: Add HideReplies interface
@@ -22,13 +23,16 @@ const (
 
 type Client interface {
 	// CountsFullArchiveTweet(ctx context.Context, query string, opt ...*TweetCountsOption) (*TweetCountsResponse, error)
+	AddOrDeleteRules(ctx context.Context, body *AddOrDeleteJSONBody, opt ...*AddOrDeleteRulesOption) (*AddOrDeleteRulesResponse, error)
 	CountsRecentTweet(ctx context.Context, query string, opt ...*TweetCountsOption) (*TweetCountsResponse, error)
+	ConnectToStream(ctx context.Context, ch chan<- ConnectToStreamResponse, errCh chan<- error, opt ...*ConnectToStreamOption) *ConnectToStream
 	Followers(ctx context.Context, id string, opt ...*FollowOption) (*FollowersResponse, error)
 	Following(ctx context.Context, id string, opt ...*FollowOption) (*FollowingResponse, error)
 	LookUpSpaces(ctx context.Context, ids []string, opt ...*SpaceLookUpOption) (*SpaceLookUpResponse, error)
 	LookUpSpaceByID(ctx context.Context, id string, opt ...*SpaceLookUpOption) (*SpaceLookUpByIDResponse, error)
 	LookUpTweets(ctx context.Context, ids []string, opt ...*TweetLookUpOption) (*TweetLookUpResponse, error)
 	LookUpTweetByID(ctx context.Context, id string, opt ...*TweetLookUpOption) (*TweetLookUpByIDResponse, error)
+	RetrieveStreamRules(ctx context.Context, opt ...*RetrieveStreamRulesOption) (*RetrieveStreamRulesResponse, error)
 	LookUpUsers(ctx context.Context, ids []string, opt ...*UserLookUpOption) (*UserLookUpResponse, error)
 	LookUpUserByID(ctx context.Context, id string, opt ...*UserLookUpOption) (*UserLookUpByIDResponse, error)
 	LookUpUserByUserName(ctx context.Context, name string, opt ...*UserLookUpOption) (*UserLookUpByUserNameResponse, error)
@@ -73,10 +77,18 @@ func WithHTTPClient(httpClient *http.Client) ClientOption {
 // 	return countFullArchiveTweet(ctx, c, query, opt...)
 // }
 
+func (c *client) AddOrDeleteRules(ctx context.Context, body *AddOrDeleteJSONBody, opt ...*AddOrDeleteRulesOption) (*AddOrDeleteRulesResponse, error) {
+	return addOrDeleteRules(ctx, c, body, opt...)
+}
+
 func (c *client) CountsRecentTweet(ctx context.Context, query string, opt ...*TweetCountsOption) (*TweetCountsResponse, error) {
 	return countsRecentTweet(ctx, c, query, opt...)
 }
 
+func (c *client) ConnectToStream(ctx context.Context, ch chan<- ConnectToStreamResponse, errCh chan<- error, opt ...*ConnectToStreamOption) *ConnectToStream {
+	return connectToStream(ctx, c, ch, errCh, opt...)
+}
+
 func (c *client) Followers(ctx context.Context, id string, opt ...*FollowOption) (*FollowersResponse, error) {
 	return followers(ctx, c, id, opt...)
 }
@@ -160,3 +172,7 @@ func (c *client) UserMentionTimeline(ctx context.Context, id string, opt ...*Use
 func (c *client) UserTweetTimeline(ctx context.Context, id string, opt ...*UserTweetTimelineOpts) (*UserTweetTimelineResponse, error) {
 	return userTweetTimeline(ctx, c, id, opt...)
 }
+
+func (c *client) RetrieveStreamRules(ctx context.Context, opt ...*RetrieveStreamRulesOption) (*RetrieveStreamRulesResponse, error) {
+	return retrieveStreamRules(ctx, c, opt...)
+}
diff --git a/endpoint.go b/endpoint.go
index bd7e246..a6854c9 100644
--- a/endpoint.go
+++ b/endpoint.go
@@ -2,6 +2,7 @@ package gotwtr
 
 // EndpointURL is the base URL for the Twitter V2 API.
 var (
+	filteredStream    = "https://api.twitter.com/2/tweets/search/stream"
 	baseUserPath      = "https://api.twitter.com/2/users"
 	baseTweetPath     = "https://api.twitter.com/2/tweets"
 	recentTweetCounts = "https://api.twitter.com/2/tweets/counts/recent"
diff --git a/error.go b/error.go
index 71e3398..5348bd4 100644
--- a/error.go
+++ b/error.go
@@ -11,15 +11,20 @@ func (e *HTTPError) Error() string {
 }
 
 type APIResponseError struct {
-	Title        string      `json:"title"`
-	Detail       string      `json:"detail"`
-	Type         string      `json:"type"`
-	ResourceType string      `json:"resource_type"`
-	ResourceID   string      `json:"resource_id"`
-	Parameter    string      `json:"parameter"`
-	Parameters   Parameter   `json:"parameters"`
-	Message      string      `json:"message"`
-	Value        interface{} `json:"value"`
+	Title              string      `json:"title"`
+	Detail             string      `json:"detail"`
+	Type               string      `json:"type"`
+	ResourceType       string      `json:"resource_type"`
+	ResourceID         string      `json:"resource_id"`
+	Parameter          string      `json:"parameter"`
+	Parameters         Parameter   `json:"parameters"`
+	Message            string      `json:"message"`
+	Value              interface{} `json:"value"`
+	Reason             string      `json:"reason,omitempty"`
+	ClientID           string      `json:"client_id,omitempty"`
+	RequiredEnrollment string      `json:"required_enrollment,omitempty"`
+	RegistrationURL    string      `json:"registration_url,omitempty"`
+	ConnectionIssue    string      `json:"connection_issue,omitempty"`
 }
 
 type Parameter struct {
diff --git a/example/tweet/filteredstream/main.go b/example/tweet/filteredstream/main.go
new file mode 100644
index 0000000..a922f62
--- /dev/null
+++ b/example/tweet/filteredstream/main.go
@@ -0,0 +1,71 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/sivchari/gotwtr"
+)
+
+func main() {
+	client := gotwtr.New("key")
+	_, err := client.AddOrDeleteRules(context.Background(), &gotwtr.AddOrDeleteJSONBody{
+		Add: []*gotwtr.AddRule{
+			{
+				Value: "puppy has:media",
+				Tag:   "puppies with media",
+			},
+			{
+				Value: "meme has:images",
+			},
+		},
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	// connect to  stream
+	ch := make(chan gotwtr.ConnectToStreamResponse, 5)
+	errCh := make(chan error)
+	stream := client.ConnectToStream(context.Background(), ch, errCh)
+	fmt.Println("streaming...")
+	ctx, cancel := context.WithCancel(context.Background())
+	go func(ctx context.Context) {
+		for {
+			select {
+			case data := <-ch:
+				fmt.Println(data.Tweet)
+			case err := <-errCh:
+				fmt.Println(err)
+			case <-ctx.Done():
+				break
+			}
+		}
+	}(ctx)
+	time.Sleep(time.Second * 10)
+	cancel()
+	stream.Stop()
+	fmt.Println("done")
+
+	// retrieve Stream rules
+	ts, err := client.RetrieveStreamRules(context.Background())
+	if err != nil {
+		panic(err)
+	}
+	var ids []string
+	for _, t := range ts.Rules {
+		fmt.Println(t)
+		ids = append(ids, t.ID)
+	}
+
+	// delete Stream rules
+	_, err = client.AddOrDeleteRules(context.Background(), &gotwtr.AddOrDeleteJSONBody{
+		Delete: &gotwtr.DeleteRule{
+			IDs: ids,
+		},
+	})
+	if err != nil {
+		panic(err)
+	}
+}
diff --git a/filtered_stream.go b/filtered_stream.go
new file mode 100644
index 0000000..5226a11
--- /dev/null
+++ b/filtered_stream.go
@@ -0,0 +1,183 @@
+package gotwtr
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+)
+
+func retrieveStreamRules(ctx context.Context, c *client, opt ...*RetrieveStreamRulesOption) (*RetrieveStreamRulesResponse, error) {
+	filteredStreamRules := filteredStream + "/rules"
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, filteredStreamRules, nil)
+	if err != nil {
+		return nil, fmt.Errorf("retrieve stream rules new request with ctx: %w", err)
+	}
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.bearerToken))
+
+	var topt RetrieveStreamRulesOption
+	switch len(opt) {
+	case 0:
+		// do nothing
+	case 1:
+		topt = *opt[0]
+	default:
+		return nil, errors.New("retrieve stream rules: only one option is allowed")
+	}
+	topt.addQuery(req)
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("retrieve stream rules: %w", err)
+	}
+	defer resp.Body.Close()
+
+	var tweet RetrieveStreamRulesResponse
+	if err := json.NewDecoder(resp.Body).Decode(&tweet); err != nil {
+		return nil, fmt.Errorf("retrieve stream rules decode: %w", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		return &tweet, &HTTPError{
+			APIName: "retrieve stream rules",
+			Status:  resp.Status,
+			URL:     req.URL.String(),
+		}
+	}
+
+	return &tweet, nil
+}
+
+func addOrDeleteRules(ctx context.Context, c *client, body *AddOrDeleteJSONBody, opt ...*AddOrDeleteRulesOption) (*AddOrDeleteRulesResponse, error) {
+	filteredStreamRules := filteredStream + "/rules"
+
+	switch {
+	case len(body.Add) == 0 && len(body.Delete.IDs) == 0:
+		return nil, errors.New("add or delete rules: add or delete.ids are required")
+	default:
+	}
+
+	for _, rule := range body.Add {
+		switch {
+		case len(rule.Value) == 0:
+			return nil, errors.New("add or delete rules: add is required")
+		case filteredStreamRuleMaxLength < len(rule.Value):
+			return nil, errors.New("add or delete rules: delete is required")
+		default:
+		}
+	}
+
+	jsonStr, err := json.Marshal(body)
+	if err != nil {
+		return nil, errors.New("add or delete rules : can not marshal")
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, filteredStreamRules, bytes.NewBuffer(jsonStr))
+	if err != nil {
+		return nil, fmt.Errorf("add or delete rules new request with ctx: %w", err)
+	}
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.bearerToken))
+	req.Header.Set("Content-Type", "application/json")
+
+	var topt AddOrDeleteRulesOption
+	switch len(opt) {
+	case 0:
+		// do nothing
+	case 1:
+		topt = *opt[0]
+	default:
+		return nil, errors.New("add or delete rules: only one option is allowed")
+	}
+	topt.addQuery(req)
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("add or delete rules: %w", err)
+	}
+	defer resp.Body.Close()
+
+	var addOrDelete AddOrDeleteRulesResponse
+	if err := json.NewDecoder(resp.Body).Decode(&addOrDelete); err != nil {
+		return nil, fmt.Errorf("add or delete rules decode: %w", err)
+	}
+	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK {
+		return &addOrDelete, &HTTPError{
+			APIName: "add or delete",
+			Status:  resp.Status,
+			URL:     req.URL.String(),
+		}
+	}
+
+	return &addOrDelete, nil
+}
+
+func (s *ConnectToStream) Stop() {
+	close(s.done)
+	s.wg.Wait()
+}
+
+func (s *ConnectToStream) retry(req *http.Request) {
+	defer s.wg.Done()
+	resp, err := s.client.Do(req)
+	if err != nil {
+		s.errCh <- err
+		return
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		s.errCh <- &HTTPError{
+			APIName: "connect to stream",
+			Status:  resp.Status,
+			URL:     req.URL.String(),
+		}
+		return
+	}
+	dec := json.NewDecoder(resp.Body)
+
+	for !stopped(s.done) {
+		var connectToStream ConnectToStreamResponse
+		err := dec.Decode(&connectToStream)
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			s.errCh <- err
+		}
+		s.ch <- connectToStream
+	}
+}
+
+func connectToStream(ctx context.Context, c *client, ch chan<- ConnectToStreamResponse, errCh chan<- error, opt ...*ConnectToStreamOption) *ConnectToStream {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, filteredStream, nil)
+	if err != nil {
+		errCh <- fmt.Errorf("connect to stream new request with ctx: %w", err)
+	}
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.bearerToken))
+
+	var copt ConnectToStreamOption
+	switch len(opt) {
+	case 0:
+		// do nothing
+	case 1:
+		copt = *opt[0]
+	default:
+		errCh <- errors.New("connect to stream: only one option is allowed")
+	}
+	copt.addQuery(req)
+
+	s := &ConnectToStream{
+		client: c.client,
+		errCh:  errCh,
+		ch:     ch,
+		done:   make(chan struct{}),
+		wg:     &sync.WaitGroup{},
+	}
+
+	s.wg.Add(1)
+	go s.retry(req)
+	return s
+}
diff --git a/tweet.go b/tweet.go
index 5525753..fc8a57c 100644
--- a/tweet.go
+++ b/tweet.go
@@ -240,6 +240,75 @@ type RetweetsLookupMeta struct {
 	ResultCount int `json:"result_count"`
 }
 
+type RetrieveStreamRulesResponse struct {
+	Rules  []*FilteredRule          `json:"data"`
+	Meta   *RetrieveStreamRulesMeta `json:"meta"`
+	Errors []*APIResponseError      `json:"errors,omitempty"`
+}
+
+type FilteredRule struct {
+	ID    string `json:"id"`
+	Value string `json:"value"`
+	Tag   string `json:"tag,omitempty"`
+}
+
+type RetrieveStreamRulesMeta struct {
+	Sent string // TODO: Is it number ?
+}
+
+type AddOrDeleteJSONBody struct {
+	Add    []*AddRule  `json:"add,omitempty"`
+	Delete *DeleteRule `json:"delete,omitempty"`
+}
+
+type AddRule struct {
+	Value string `json:"value"`
+	Tag   string `json:"tag,omitempty"`
+}
+
+type DeleteRule struct {
+	IDs []string `json:"ids"`
+}
+
+type AddOrDeleteRulesResponse struct {
+	Rules  []*FilteredRule       `json:"data"`
+	Meta   *AddOrDeleteRulesMeta `json:"meta"`
+	Errors []*APIResponseError   `json:"errors,omitempty"`
+}
+
+type AddOrDeleteRulesMeta struct {
+	Sent    string                  `json:"sent"` // TODO: Is it number ?
+	Summary *AddOrDeleteMetaSummary `json:"summary"`
+}
+
+type AddOrDeleteMetaSummary struct {
+	Created    int `json:"created"`
+	NotCreated int `json:"not_created"`
+	Deleted    int `json:"deleted"`
+	NotDeleted int `json:"not_deleted"`
+	Valid      int `json:"valid"`
+	Invalid    int `json:"invalid"`
+}
+
+type ConnectToStreamResponse struct {
+	Tweet         *Tweet          `json:"data"`
+	Includes      *TweetIncludes  `json:"includes,omitempty"`
+	MatchingRules []*MatchingRule `json:"matching_rules"`
+}
+
+type MatchingRule struct {
+	ID  string `json:"id"`
+	Tag string `json:"tag"`
+}
+
+type ConnectToStream struct {
+	client *http.Client
+	errCh  chan<- error
+	ch     chan<- ConnectToStreamResponse
+	done   chan struct{}
+	wg     *sync.WaitGroup
+}
+
 type PostRetweetResponse struct {
 	Retweeted *Retweeted          `json:"data"`
 	Errors    []*APIResponseError `json:"errors,omitempty"`
diff --git a/tweet_option.go b/tweet_option.go
index a480c07..fafb6e7 100644
--- a/tweet_option.go
+++ b/tweet_option.go
@@ -329,3 +329,66 @@ func (t RetweetsLookupOpts) addQuery(req *http.Request) {
 		req.URL.RawQuery = q.Encode()
 	}
 }
+
+type RetrieveStreamRulesOption struct {
+	IDs []string
+}
+
+func (t *RetrieveStreamRulesOption) addQuery(req *http.Request) {
+	q := req.URL.Query()
+	if len(t.IDs) > 0 {
+		q.Add("ids", strings.Join(t.IDs, ","))
+	}
+	if len(q) > 0 {
+		req.URL.RawQuery = q.Encode()
+	}
+}
+
+type AddOrDeleteRulesOption struct {
+	DryRun bool // If it is true, test a the syntax of your rule without submitting it
+}
+
+func (t *AddOrDeleteRulesOption) addQuery(req *http.Request) {
+	q := req.URL.Query()
+	if t.DryRun {
+		q.Add("dry_run", strconv.FormatBool(t.DryRun))
+	}
+	if len(q) > 0 {
+		req.URL.RawQuery = q.Encode()
+	}
+}
+
+type ConnectToStreamOption struct {
+	// BackfillMinutes int `json:"backfill_minutes"` // This feature is currently only available to the Academic Research product track.
+	Expansions  []Expansion
+	MediaFields []MediaField
+	PlaceFields []PlaceField
+	PollFields  []PollField
+	TweetFields []TweetField
+	UserFields  []UserField
+}
+
+func (t *ConnectToStreamOption) addQuery(req *http.Request) {
+	q := req.URL.Query()
+	if len(t.Expansions) > 0 {
+		q.Add("expansions", strings.Join(expansionsToString(t.Expansions), ","))
+	}
+	if len(t.MediaFields) > 0 {
+		q.Add("media.fields", strings.Join(mediaFieldsToString(t.MediaFields), ","))
+	}
+	if len(t.PlaceFields) > 0 {
+		q.Add("place.fields", strings.Join(placeFieldsToString(t.PlaceFields), ","))
+	}
+	if len(t.PollFields) > 0 {
+		q.Add("poll.fields", strings.Join(pollFieldsToString(t.PollFields), ","))
+	}
+	if len(t.TweetFields) > 0 {
+		q.Add("tweet.fields", strings.Join(tweetFieldsToString(t.TweetFields), ","))
+	}
+	if len(t.UserFields) > 0 {
+		q.Add("user.fields", strings.Join(userFieldsToString(t.UserFields), ","))
+	}
+	if len(q) > 0 {
+		req.URL.RawQuery = q.Encode()
+	}
+}


Test patch:
diff --git a/filtered_stream_test.go b/filtered_stream_test.go
new file mode 100644
index 0000000..9fa5414
--- /dev/null
+++ b/filtered_stream_test.go
@@ -0,0 +1,507 @@
+package gotwtr_test
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/sivchari/gotwtr"
+)
+
+func Test_retrieveStreamRules(t *testing.T) {
+	type args struct {
+		ctx    context.Context
+		client *http.Client
+		opt    []*gotwtr.RetrieveStreamRulesOption
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    *gotwtr.RetrieveStreamRulesResponse
+		wantErr bool
+	}{
+		{
+			name: "Success 200 Retrieve Rules",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{
+					    "data": [
+					        {
+					            "id": "1273636687768285186",
+					            "value": "meme has:images"
+					        },
+					        {
+					            "id": "1273636687768285187",
+					            "value": "puppy has:media",
+					            "tag": "puppies with media"
+					        }
+					    ],
+					    "meta": {
+					        "sent": "2020-06-18T15:21:58.638Z"
+					    }
+					}`
+					return &http.Response{
+						StatusCode: http.StatusOK,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.RetrieveStreamRulesOption{},
+			},
+			want: &gotwtr.RetrieveStreamRulesResponse{
+				Rules: []*gotwtr.FilteredRule{
+					{
+						ID:    "1273636687768285186",
+						Value: "meme has:images",
+					},
+					{
+						ID:    "1273636687768285187",
+						Value: "puppy has:media",
+						Tag:   "puppies with media",
+					},
+				},
+				Meta: &gotwtr.RetrieveStreamRulesMeta{
+					Sent: "2020-06-18T15:21:58.638Z",
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "Success 200 Retrieve Rules with Option",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{
+					    "data": [
+					        {
+					            "id": "1273636687768285186",
+					            "value": "meme has:images"
+					        }
+					    ],
+					    "meta": {
+					        "sent": "2020-06-18T15:21:58.638Z"
+					    }
+					}`
+					return &http.Response{
+						StatusCode: http.StatusOK,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.RetrieveStreamRulesOption{
+					{
+						[]string{"1273636687768285186"},
+					},
+				},
+			},
+			want: &gotwtr.RetrieveStreamRulesResponse{
+				Rules: []*gotwtr.FilteredRule{
+					{
+						ID:    "1273636687768285186",
+						Value: "meme has:images",
+					},
+				},
+				Meta: &gotwtr.RetrieveStreamRulesMeta{
+					Sent: "2020-06-18T15:21:58.638Z",
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "403 Forbidden Client Not Enrolled",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{
+						"errors": [
+							{
+							    "client_id": "16340226",
+							    "required_enrollment": "Standard Basic",
+							    "registration_url": "https://developer.twitter.com/en/account",
+							    "title": "Client Forbidden",
+							    "detail": "This request must be made using an approved developer account that is enrolled in the requested endpoint. Learn more by visiting our documentation.",
+							    "reason": "client-not-enrolled",
+							    "type": "https://api.twitter.com/2/problems/client-forbidden"
+							}
+						]
+					}`
+					return &http.Response{
+						StatusCode: http.StatusForbidden,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.RetrieveStreamRulesOption{},
+			},
+			want: &gotwtr.RetrieveStreamRulesResponse{
+				Errors: []*gotwtr.APIResponseError{
+					{
+						ClientID:           "16340226",
+						RequiredEnrollment: "Standard Basic",
+						RegistrationURL:    "https://developer.twitter.com/en/account",
+						Title:              "Client Forbidden",
+						Detail:             "This request must be made using an approved developer account that is enrolled in the requested endpoint. Learn more by visiting our documentation.",
+						Reason:             "client-not-enrolled",
+						Type:               "https://api.twitter.com/2/problems/client-forbidden",
+					},
+				},
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			c := gotwtr.New("test-key", gotwtr.WithHTTPClient(tt.args.client))
+			got, err := c.RetrieveStreamRules(tt.args.ctx, tt.args.opt...)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("searchRecentTweets() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if diff := cmp.Diff(got, tt.want); diff != "" {
+				t.Errorf("searchRecentTweets() mismatch (-want +got):\n%s", diff)
+				return
+			}
+		})
+	}
+}
+
+func Test_AddOrDeleteRules(t *testing.T) {
+	type args struct {
+		ctx    context.Context
+		client *http.Client
+		opt    []*gotwtr.AddOrDeleteRulesOption
+		body   *gotwtr.AddOrDeleteJSONBody
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    *gotwtr.AddOrDeleteRulesResponse
+		wantErr bool
+	}{
+		{
+			name: "Success 201 Created List of Rules",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{
+					    "data": [
+					        {
+					            "value": "meme has:images",
+					            "id": "1273636687768285186"
+					        },
+					        {
+					            "value": "puppy has:media",
+					            "tag": "puppies with media",
+					            "id": "1273636687768285187"
+					        }
+					    ],
+					    "meta": {
+					        "sent": "2020-06-18T15:20:24.063Z",
+					        "summary": {
+					            "created": 2,
+					            "not_created": 0,
+					            "valid": 2,
+					            "invalid": 0
+					        }
+					    }
+					}}`
+					return &http.Response{
+						StatusCode: http.StatusCreated,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.AddOrDeleteRulesOption{},
+				body: &gotwtr.AddOrDeleteJSONBody{
+					Add: []*gotwtr.AddRule{
+						{
+							Value: "puppy has:media",
+							Tag:   "puppies with media",
+						},
+						{
+							Value: "meme has:images",
+						},
+					},
+					Delete: &gotwtr.DeleteRule{},
+				},
+			},
+			want: &gotwtr.AddOrDeleteRulesResponse{
+				Rules: []*gotwtr.FilteredRule{
+					{
+						ID:    "1273636687768285186",
+						Value: "meme has:images",
+					},
+					{
+						ID:    "1273636687768285187",
+						Value: "puppy has:media",
+						Tag:   "puppies with media",
+					},
+				},
+				Meta: &gotwtr.AddOrDeleteRulesMeta{
+					Sent: "2020-06-18T15:20:24.063Z",
+					Summary: &gotwtr.AddOrDeleteMetaSummary{
+						Created:    2,
+						NotCreated: 0,
+						Valid:      2,
+						Invalid:    0,
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "Success 201 Created List of Rules with Dry Run",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{
+					    "data": [
+					        {
+					            "value": "tostones recipe",
+					            "id": "1273646795642421249"
+					        }
+					    ],
+					    "meta": {
+					        "sent": "2020-06-18T16:00:33.972Z",
+					        "summary": {
+					            "created": 1,
+					            "not_created": 0,
+					            "valid": 1,
+					            "invalid": 0
+					        }
+					    }
+					}`
+					return &http.Response{
+						StatusCode: http.StatusCreated,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.AddOrDeleteRulesOption{
+					{
+						DryRun: true,
+					},
+				},
+				body: &gotwtr.AddOrDeleteJSONBody{
+					Add: []*gotwtr.AddRule{
+						{
+							Value: "tostones recipe",
+						},
+					},
+					Delete: &gotwtr.DeleteRule{},
+				},
+			},
+			want: &gotwtr.AddOrDeleteRulesResponse{
+				Rules: []*gotwtr.FilteredRule{
+					{
+						Value: "tostones recipe",
+						ID:    "1273646795642421249",
+					},
+				},
+				Meta: &gotwtr.AddOrDeleteRulesMeta{
+					Sent: "2020-06-18T16:00:33.972Z",
+					Summary: &gotwtr.AddOrDeleteMetaSummary{
+						Created:    1,
+						NotCreated: 0,
+						Valid:      1,
+						Invalid:    0,
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "Success 200 DeleteRule Rule",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{
+					    "meta": {
+					        "sent": "2020-07-09T21:13:18.284Z",
+					        "summary": {
+					            "deleted": 1,
+					            "not_deleted": 0
+					        }
+					    }
+					}`
+					return &http.Response{
+						StatusCode: http.StatusOK,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.AddOrDeleteRulesOption{},
+				body: &gotwtr.AddOrDeleteJSONBody{
+					Add: []*gotwtr.AddRule{},
+					Delete: &gotwtr.DeleteRule{
+						IDs: []string{"1273636687768285186"},
+					},
+				},
+			},
+			want: &gotwtr.AddOrDeleteRulesResponse{
+				Meta: &gotwtr.AddOrDeleteRulesMeta{
+					Sent: "2020-07-09T21:13:18.284Z",
+					Summary: &gotwtr.AddOrDeleteMetaSummary{
+						Deleted:    1,
+						NotDeleted: 0,
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "403 Forbidden Client Not Enrolled",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{
+						"errors": [
+							{
+							    "client_id": "16340226",
+							    "required_enrollment": "Standard Basic",
+							    "registration_url": "https://developer.twitter.com/en/account",
+							    "title": "Client Forbidden",
+							    "detail": "This request must be made using an approved developer account that is enrolled in the requested endpoint. Learn more by visiting our documentation.",
+							    "reason": "client-not-enrolled",
+							    "type": "https://api.twitter.com/2/problems/client-forbidden"
+							}
+						]
+					}`
+					return &http.Response{
+						StatusCode: http.StatusForbidden,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.AddOrDeleteRulesOption{},
+				body: &gotwtr.AddOrDeleteJSONBody{
+					Add: []*gotwtr.AddRule{
+						{
+							Value: "tostones recipe",
+						},
+					},
+					Delete: &gotwtr.DeleteRule{},
+				},
+			},
+			want: &gotwtr.AddOrDeleteRulesResponse{
+				Errors: []*gotwtr.APIResponseError{
+					{
+						ClientID:           "16340226",
+						RequiredEnrollment: "Standard Basic",
+						RegistrationURL:    "https://developer.twitter.com/en/account",
+						Title:              "Client Forbidden",
+						Detail:             "This request must be made using an approved developer account that is enrolled in the requested endpoint. Learn more by visiting our documentation.",
+						Reason:             "client-not-enrolled",
+						Type:               "https://api.twitter.com/2/problems/client-forbidden",
+					},
+				},
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			c := gotwtr.New("test-key", gotwtr.WithHTTPClient(tt.args.client))
+			got, err := c.AddOrDeleteRules(tt.args.ctx, tt.args.body, tt.args.opt...)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("addOrDelteRules() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if diff := cmp.Diff(got, tt.want); diff != "" {
+				t.Errorf("addOrDelteRules() mismatch (-want +got):\n%s", diff)
+				return
+			}
+		})
+	}
+}
+
+func Test_ConnectToStream(t *testing.T) {
+	type args struct {
+		ctx    context.Context
+		client *http.Client
+		opt    []*gotwtr.ConnectToStreamOption
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    *gotwtr.ConnectToStreamResponse
+		wantErr bool
+	}{
+		{
+			name: "200 ok",
+			args: args{
+				ctx: context.Background(),
+				client: mockHTTPClient(func(request *http.Request) *http.Response {
+					body := `{"data":{"id":"1228393702244134912","text":"What did the developer write in their Valentine’s card?\n  \nwhile(true) {\n    I = Love(You);  \n}"},"matching_rules":[{"id":"1452189330902970370","tag":"has:media puppies with media"}]}`
+					return &http.Response{
+						StatusCode: http.StatusOK,
+						Body:       io.NopCloser(strings.NewReader(body)),
+					}
+				}),
+				opt: []*gotwtr.ConnectToStreamOption{},
+			},
+			want: &gotwtr.ConnectToStreamResponse{
+				Tweet: &gotwtr.Tweet{
+					ID:   "1228393702244134912",
+					Text: "What did the developer write in their Valentine’s card?\n  \nwhile(true) {\n    I = Love(You);  \n}",
+				},
+				MatchingRules: []*gotwtr.MatchingRule{
+					{
+						ID:  "1452189330902970370",
+						Tag: "has:media puppies with media",
+					},
+				},
+			},
+			wantErr: false,
+		},
+		/*
+			TODO: must consider way of processing APIError such as kind of late limit
+			{
+				name: "429 - Too Many Requests",
+				args: args{
+					ctx: context.Background(),
+					client: mockHTTPClient(func(request *http.Request) *http.Response {
+						body := `{ "title": "ConnectionException", "detail": "This stream is currently at the maximum allowed connection limit.", "connection_issue": "TooManyConnections","type": "https://api.twitter.com/2/problems/streaming-connection"}`
+						return &http.Response{
+							StatusCode: http.StatusTooManyRequests,
+							Body:       io.NopCloser(strings.NewReader(body)),
+						}
+					}),
+					opt: []*gotwtr.ConnectToStreamOption{},
+				},
+				want: &gotwtr.ConnectToStreamResponse{
+					Error: &gotwtr.APIResponseError{
+						Title:           "ConnectionException",
+						Detail:          "This stream is currently at the maximum allowed connection limit.",
+						ConnectionIssue: "TooManyConnections",
+						Type:            "https://api.twitter.com/2/problems/streaming-connection",
+					},
+				},
+				wantErr: true,
+			},
+		*/
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			ch := make(chan gotwtr.ConnectToStreamResponse)
+			errCh := make(chan error)
+			c := gotwtr.New("test-key", gotwtr.WithHTTPClient(tt.args.client))
+			c.ConnectToStream(tt.args.ctx, ch, errCh, tt.args.opt...)
+			select {
+			case got := <-ch:
+				if diff := cmp.Diff(&got, tt.want); diff != "" {
+					t.Errorf("ConnectToStream() mismatch (-want +got):\n%s", diff)
+					return
+				}
+			case err := <-errCh:
+				if (err != nil) != tt.wantErr {
+					t.Errorf("ConnectToStream() error = %v, wantErr %v", err, tt.wantErr)
+					return
+				}
+			}
+		})
+	}
+}