changed base URL to use mock birdsite and small fixes for e2e cases

extensions/twitter_v2_facade/content_source.py

@@ -101,7 +101,7 @@ def cache_tweets_response (response_tweets, query_type, auth_user_id, user_id =
     media = includes and includes.media or []
     ref_tweets = includes and includes.tweets or []
     
-    if response_tweets.meta and 'next_token' in response_tweets.meta:
+    if response_tweets.meta and response_tweets.meta.next_token:
         next_token = response_tweets.meta.next_token
     else:
         next_token = None

extensions/twitter_v2_facade/oauth2_login.py

@@ -104,7 +104,8 @@ def get_loggedin_html ():
     authorization_response = redirect_uri + '?code={}&state={}'.format(code, state)
     
     # Fetch your access token
-    token_url = "https://api.twitter.com/2/oauth2/token"
+    #token_url = "https://api.twitter.com/2/oauth2/token"
+    token_url = "http://localhost:5900/2/oauth2/token"
     
     
 
@@ -140,7 +141,8 @@ def get_loggedin_html ():
     # Make a request to the users/me endpoint to get your user ID
     user_me = requests.request(
         "GET",
-        "https://api.twitter.com/2/users/me",
+        #"https://api.twitter.com/2/users/me",
+        "http://localhost:5900/2/users/me",
         headers={"Authorization": "Bearer {}".format(access)},
     ).json()
     user_id = user_me["data"]["id"]
@@ -194,7 +196,8 @@ def get_login_html ():
     oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)
 
     # Create an authorize URL
-    auth_url = "https://twitter.com/i/oauth2/authorize"
+    auth_url = "http://localhost:5900/2/oauth2/authorize"
+    #auth_url = "https://twitter.com/i/oauth2/authorize"
     authorization_url, state = oauth.authorization_url(
         auth_url, code_challenge=code_challenge, code_challenge_method="S256"
     )
@@ -233,7 +236,8 @@ def refresh_token ():
     }
     
     
-    response = requests.post('https://api.twitter.com/2/oauth2/token', data=data, headers=headers)
+    response = requests.post('http://localhost:5900/2/oauth2/token', data=data, headers=headers)
+    #response = requests.post('https://api.twitter.com/2/oauth2/token', data=data, headers=headers)
     
     result = json.loads(response.text)
     
@@ -269,7 +273,8 @@ def get_twitter_app_refresh_token ():
         'grant_type': 'client_credentials'
     }
     
-    response = requests.post('https://api.twitter.com/oauth2/token', data=data, headers=headers)
+    #response = requests.post('https://api.twitter.com/oauth2/token', data=data, headers=headers)
+    response = requests.post('http://localhost:5900/2/oauth2/token', data=data, headers=headers)
     
     result = json.loads(response.text)
     

lib/twitter_v2/api.py

@@ -8,6 +8,9 @@ import sqlite3
 
 from twitter_v2.types import TweetSearchResponse, DMEventsResponse, UserSearchResponse
 
+#BASE_URL = 'https://api.twitter.com'
+BASE_URL = 'http://localhost:5900'
+
 # https://developer.twitter.com/en/docs/twitter-api/v1/tweets/curate-a-collection/api-reference/get-collections-entries
 # we can perhaps steal a token from the TweetDeck Console, otherwise we need to apply for Standard v1.1 / Elevated
 class ApiV11TweetCollectionSource:
@@ -27,8 +30,9 @@ class ApiV11TweetCollectionSource:
         return
 
 class TwitterApiV2SocialGraph:
-    def __init__ (self, token):
+    def __init__ (self, token, base_url = BASE_URL):
         self.token = token
+        self.base_url = base_url
         
     def get_user (self, user_id, is_username=False, return_dataclass=False):
         # GET /2/users/:id
@@ -39,7 +43,7 @@ class TwitterApiV2SocialGraph:
         # GET /2/users/by?usernames=
         # GET /2/users?ids=
         
-        
+        print('get_users: ' + json.dumps(user_ids) )
         
         user_fields = ["id", "created_at", "name", "username", "location", "profile_image_url", "verified", "description", "public_metrics", "protected", "pinned_tweet_id", "url"]
         
@@ -50,10 +54,10 @@ class TwitterApiV2SocialGraph:
         }
         
         if are_usernames:
-            url = "https://api.twitter.com/2/users/by"
+            url = f"{self.base_url}/2/users/by"
             params['usernames'] = user_ids
         else:
-            url = "https://api.twitter.com/2/users"
+            url = f"{self.base_url}/2/users"
             params['ids'] = user_ids
 
         headers = {
@@ -63,13 +67,16 @@ class TwitterApiV2SocialGraph:
         response = requests.get(url, params=params, headers=headers)
         result = json.loads(response.text)
         
+        print("get_users, response=")
+        print(json.dumps(result, indent=2))
+        
         return self._parse_user_search_response(result, return_dataclass=return_dataclass)
         
     def get_following (self, user_id, 
                         max_results = 50, pagination_token = None, return_dataclass=False):
         # GET /2/users/:id/following
         
-        url = "https://api.twitter.com/2/users/{}/following".format(user_id)
+        url = f"{self.base_url}/2/users/{user_id}/following"
         
         
         user_fields = ["id", "created_at", "name", "username", "location", "profile_image_url", "verified"]
@@ -108,7 +115,7 @@ class TwitterApiV2SocialGraph:
                         max_results = 50, pagination_token = None, return_dataclass=False):
         # GET /2/users/:id/followers
         
-        url = "https://api.twitter.com/2/users/{}/followers".format(user_id)
+        url = f"{self.base_url}/2/users/{user_id}/followers"
         
         
         user_fields = ["id", "created_at", "name", "username", "location", "profile_image_url", "verified", "description", "public_metrics", "protected", "pinned_tweet_id", "url"]
@@ -143,13 +150,14 @@ class TwitterApiV2SocialGraph:
         return
 
 class ApiV2ConversationSource:
-    def __init__ (self, token):
+    def __init__ (self, token, base_url = BASE_URL):
         self.token = token
+        self.base_url = base_url
     
     def get_recent_events (self, max_results = None, pagination_token = None):
 
         # https://developer.twitter.com/en/docs/twitter-api/direct-messages/lookup/api-reference/get-dm_events
-        url = "https://api.twitter.com/2/dm_events"
+        url = f"{self.base_url}/2/dm_events"
         
         tweet_fields = ["created_at", "conversation_id",  "referenced_tweets", "text", "public_metrics", "entities", "attachments"]
         media_fields = ["alt_text", "type", "preview_image_url", "public_metrics", "url", "media_key", "duration_ms", "width", "height", "variants"]
@@ -195,7 +203,7 @@ class ApiV2ConversationSource:
         return
     
     def send_message (self, dm_conversation_id, text, attachments = None):
-        url = f'/2/dm_conversations/{dm_conversation_id}/messages'
+        url = f'{self.base_url}/2/dm_conversations/{dm_conversation_id}/messages'
         
         body = {
             'text': text
@@ -220,14 +228,15 @@ class ApiV2ConversationSource:
         return result 
         
 class ApiV2TweetSource:
-    def __init__ (self, token):
+    def __init__ (self, token, base_url = BASE_URL):
         self.token = token
+        self.base_url = base_url
         
         
     def create_tweet (self, text, 
         reply_to_tweet_id = None, quote_tweet_id = None):
         
-        url = "https://api.twitter.com/2/tweets"
+        url = f"{self.base_url}/2/tweets"
         
         tweet = {
             'text': text
@@ -255,7 +264,7 @@ class ApiV2TweetSource:
         
     def retweet (self, tweet_id, user_id):
         
-        url = "https://api.twitter.com/2/users/{}/retweets".format(user_id)
+        url = f"{self.base_url}/2/users/{user_id}/retweets"
         
         retweet = {
             'tweet_id': tweet_id
@@ -275,7 +284,7 @@ class ApiV2TweetSource:
     
     def delete_retweet (self, tweet_id, user_id):
         
-        url = "https://api.twitter.com/2/users/{}/retweets/{}".format(user_id, tweet_id)
+        url = f"{self.base_url}/2/users/{user_id}/retweets/{tweet_id}"
    
         headers = {
             'Authorization': 'Bearer {}'.format(self.token)
@@ -288,7 +297,7 @@ class ApiV2TweetSource:
     
     def bookmark (self, tweet_id, user_id):
         
-        url = "https://api.twitter.com/2/users/{}/bookmarks".format(user_id)
+        url = f"{self.base_url}/2/users/{user_id}/bookmarks"
         
         bookmark = {
             'tweet_id': tweet_id
@@ -308,7 +317,7 @@ class ApiV2TweetSource:
         
     def delete_bookmark (self, tweet_id, user_id):
         
-        url = "https://api.twitter.com/2/users/{}/bookmarks/{}".format(user_id, tweet_id)
+        url = f"{self.base_url}/2/users/{user_id}/bookmarks/{tweet_id}"
    
         headers = {
             'Authorization': 'Bearer {}'.format(self.token)
@@ -328,7 +337,7 @@ class ApiV2TweetSource:
         Get a user's timeline as viewed by the user themselves.
         """
         
-        path = 'users/{}/timelines/{}'.format(user_id, variant)
+        path = f'users/{user_id}/timelines/{variant}'
         
         return self.get_timeline(path, 
             max_results=max_results, pagination_token=pagination_token, since_id=since_id, until_id=until_id, end_time=end_time, start_time=start_time, return_dataclass=True) 
@@ -350,7 +359,7 @@ class ApiV2TweetSource:
         
         token = self.token
         
-        url = "https://api.twitter.com/2/{}".format(path)
+        url = f"{self.base_url}/2/{path}"
         
         tweet_fields = ["created_at", "conversation_id",  "referenced_tweets", "text", "public_metrics", "entities", "attachments"]
         media_fields = ["alt_text", "type", "preview_image_url", "public_metrics", "url", "media_key", "duration_ms", "width", "height", "variants"]
@@ -428,7 +437,7 @@ class ApiV2TweetSource:
             if not return_dataclass:
                 return response_json
             
-            raise 'error converting response to dataclass'
+            raise Exception('error converting response to dataclass')
         
         if return_dataclass:
             return typed_resp
@@ -483,7 +492,7 @@ class ApiV2TweetSource:
                     
         token = self.token
         
-        url = "https://api.twitter.com/2/tweets"
+        url = f"{self.base_url}/2/tweets"
         
         tweet_fields = ["created_at", "conversation_id",  "referenced_tweets", "text", "public_metrics", "entities", "attachments"]
         media_fields = ["alt_text", "type", "preview_image_url", "public_metrics", "url", "media_key", "duration_ms", "width", "height", "variants"]
@@ -532,7 +541,7 @@ class ApiV2TweetSource:
         
         token = self.token
         
-        url = "https://api.twitter.com/2/tweets/search/recent"
+        url = f"{self.base_url}/2/tweets/search/recent"
         
         tweet_fields = ["created_at", "conversation_id",  "referenced_tweets", "text", "public_metrics", "entities", "attachments"]
         media_fields = ["alt_text", "type", "preview_image_url", "public_metrics", "url", "media_key", "duration_ms", "width", "height", "variants"]
@@ -596,7 +605,7 @@ class ApiV2TweetSource:
         """
         token = self.token
         
-        url = "https://api.twitter.com/2/tweets/counts/recent"
+        url = f"{self.base_url}/2/tweets/counts/recent"
         
         
         params = {
@@ -736,7 +745,7 @@ class ApiV2TweetSource:
         #  GET /2/tweets/:id/liking_users
         # User rate limit (User context): 75 requests per 15-minute window per each authenticated user
 
-        url = f"https://api.twitter.com/2/tweets/{tweet_id}/liking_users"
+        url = f"{self.base_url}/2/tweets/{tweet_id}/liking_users"
         
         user_fields = ["id", "created_at", "name", "username", "location", "profile_image_url", "verified", "description", "public_metrics", "protected", "pinned_tweet_id", "url"]
         
@@ -766,7 +775,7 @@ class ApiV2TweetSource:
         
     def delete_like (self, tweet_id, user_id):
         
-        url = "https://api.twitter.com/2/users/{}/likes/{}".format(user_id, tweet_id)
+        url = f"{self.base_url}/2/users/{user_id}/likes/{tweet_id}"
    
         headers = {
             'Authorization': 'Bearer {}'.format(self.token)