automated ci for oauth

twitter_v2/cli/main.py

+import os
 import socketserver
 import webbrowser
 from enum import Enum
+from pathlib import Path
 from queue import Queue
+from threading import Thread
+from time import sleep
 from typing import List, Optional
 
 import typer
 from loguru import logger
 from pymemri.pod.client import PodClient
 from pymemri.pod.utils import read_pod_key
+from selenium import webdriver
+from selenium.webdriver.common.by import By
 
 from twitter_v2.cli import get_request_handler
 
@@ -87,6 +93,7 @@ def simulate_oauth2_flow(
     pod: PodType = PodType.local,
     database_key: Optional[str] = None,
     owner_key: Optional[str] = None,
+    automated: bool = False,
 ):
     pod_full_address = POD_ADDRESS[pod]
     if database_key is None:
@@ -104,7 +111,14 @@ def simulate_oauth2_flow(
         owner_key=owner_key,
     )
 
-    token = run_oauth2_flow(client=client)
+    if automated:
+        token = run_oauth2_flow_automated(client=client)
+        dir = Path.home() / ".memri"
+        dir.mkdir(exist_ok=True, parents=True)
+        with open(str(dir / "twitter_access_token"), "w") as f:
+            f.write(str(token))
+    else:
+        token = run_oauth2_flow(client=client)
     logger.info(f"Access token: {token}")
 
 
@@ -116,10 +130,75 @@ def run_oauth2_flow(client: PodClient):
         port=3667,
     )
     url = handler.get_url()
+    print(url)
     webbrowser.open(url)
     token = handler.get_token()
     return token
 
 
+def emulate_oauth_user_flow_selenium(url, user, password):
+    """run selenium to click through the twitter oauth flow
+
+    Args:
+        url (str): twitter url to visit
+        user (str): user, can be username, or email
+        password (str): pw corresponding to the user
+    """
+    op = webdriver.ChromeOptions()
+    # these options are required to make sure twitter doesnt block us
+    user_agent = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"
+    "(KHTML, like Gecko) Chrome/60.0.3112.50 Safari/537.36"
+    op.add_argument("headless")
+    op.add_argument(f"user-agent={user_agent}")
+    driver = webdriver.Chrome(options=op)
+    driver.get(url)
+    sleep(3)
+    print("clicking first button")
+    button1 = driver.find_element(
+        By.XPATH, "//a[@data-testid='OAuth_Consent_Log_In_Button']"
+    )
+    button1.click()
+    sleep(2)
+    username_field = driver.find_elements("name", "text")[0]
+    username_field.send_keys(user)
+    print("clicking second button")
+    button2 = driver.find_elements(By.XPATH, "//div[@role='button']")[2]
+    button2.click()
+    sleep(2)
+    pw_field = driver.find_elements("name", "password")[0]
+    pw_field.send_keys(password)
+    print("clicking third button")
+    button3 = driver.find_elements(By.XPATH, "//div[@role='button']")[2]
+    button3.click()
+    sleep(5)
+    print("clicking last (4th) button")
+    button4 = driver.find_element(
+        By.XPATH, "//div[@data-testid='OAuth_Consent_Button']"
+    )
+    driver.execute_script("arguments[0].click();", button4)
+    print("twitter oauth selenium flow done")
+    # dont remove this last sleep, its very important as it gives the server
+    # time to handle the request
+    sleep(3)
+
+
+def run_oauth2_flow_automated(client: PodClient):
+    handler = TokenCallbackHandler(
+        client=client,
+        scheme="http",
+        host="localhost",
+        port=3667,
+    )
+    url = handler.get_url()
+    # webbrowser.open(url)
+    user = os.environ.get("TWITTER_TEST_USER_CI")
+    pw = os.environ.get("TWITTER_TEST_USER_PASSWORD_CI")
+    assert user is not None and pw is not None
+    thread = Thread(target=emulate_oauth_user_flow_selenium, args=(url, user, pw))
+    thread.start()
+    token = handler.get_token()
+    return token
+
+
 if __name__ == "__main__":
     app()