feat: api ci test; auto otp

Author: proffapt

.dockerignore

@@ -11,4 +11,8 @@ README.md
 credentials.json
 token.json
 erpcreds.py
-erpcreds.py.template
+erpcreds.py.template
+
+api-integration-test.py
+.Dockerfile-dev
+docker-compose-dev.yaml

api-integration-test.py

@@ -1,3 +1,4 @@
+import argparse
 import requests
 import erpcreds
 
@@ -15,20 +16,19 @@ def get_secret_question():
     if res_json["status"] == "success":
         secret_question = res_json["SECRET_QUESTION"]
         session_token = res_json["SESSION_TOKEN"]
-        print(f"✅ Secret Question: {secret_question}")
         print(f"✅ Session Token: {session_token}")
         return secret_question, session_token
     else:
         print(f"❌ Error: {res_json['message']}")
         exit(1)
 
-def request_otp(session_token, secret_question):
+def request_otp(session_token, secret_answer):
     url = f"{BASE_URL}/request-otp"
     headers = {"Session-Token": session_token}
     data = {
         "roll_number": erpcreds.ROLL_NUMBER,
         "password": erpcreds.PASSWORD,
-        "secret_answer": erpcreds.SECURITY_QUESTIONS_ANSWERS[secret_question],
+        "secret_answer": secret_answer,
     }
 
     response = session.post(url, headers=headers, data=data)
@@ -40,13 +40,13 @@ def request_otp(session_token, secret_question):
         print(f"❌ Error: {res_json['message']}")
         exit(1)
 
-def login(session_token, secret_question, otp):
+def login(session_token, secret_answer, otp):
     url = f"{BASE_URL}/login"
     headers = {"Session-Token": session_token}
     data = {
         "roll_number": erpcreds.ROLL_NUMBER,
         "password": erpcreds.PASSWORD,
-        "secret_answer": erpcreds.SECURITY_QUESTIONS_ANSWERS[secret_question],
+        "secret_answer": secret_answer,
         "otp": otp,
     }
 
@@ -76,11 +76,96 @@ def download_timetable(sso_token):
         print(f"❌ Error: {response.json().get('message', 'Unknown error')}")
         exit(1)
 
+def _getOTP(OTP_CHECK_INTERVAL: float, session_token: str, log: bool = False):
+    import time
+    import base64
+    from googleapiclient.discovery import build
+
+
+    def getMailID(service):
+        subject = "OTP for Sign In in ERP Portal of IIT Kharagpur"
+
+        query = f"subject:{subject}"
+        results = service.users().messages().list(userId="me", q=query, maxResults=1).execute()
+        messages = results.get("messages", [])
+        
+        if messages:
+            message = service.users().messages().get(userId="me", id=messages[0]["id"]).execute()
+            return message["id"]
+        
+        return None
+
+    def generate_token():
+        """Generates token.json from credentials.json file with readonly access to mails."""
+        import os
+        from google.oauth2.credentials import Credentials
+        from google.auth.transport.requests import Request
+        from google_auth_oauthlib.flow import InstalledAppFlow
+
+        token_path = os.path.join(os.getcwd(), "token.json")
+        credentials_path = os.path.join(os.getcwd(), "credentials.json")
+
+        scopes = ["https://www.googleapis.com/auth/gmail.readonly"]
+
+        creds = None
+        if os.path.exists(token_path):
+            creds = Credentials.from_authorized_user_file(token_path, scopes)
+        if not creds or not creds.valid:
+            if creds and creds.expired and creds.refresh_token:
+                creds.refresh(Request())
+            else:
+                flow = InstalledAppFlow.from_client_secrets_file(credentials_path, scopes)
+                creds = flow.run_local_server(port=0)
+
+            if not os.path.exists(token_path):
+                with open(token_path, "w") as token:
+                    token.write(creds.to_json())
+
+        return creds
+
+    creds = generate_token()
+    service = build("gmail", "v1", credentials=creds)
+    
+    latest_mail_id = getMailID(service)
+    request_otp(session_token, secret_answer)
+    if log: 
+        print("⏱︎ Waiting for OTP ...")
+    
+    while True:
+        if (mail_id := getMailID(service)) != latest_mail_id:
+            break
+        time.sleep(OTP_CHECK_INTERVAL)
+    
+    mail = service.users().messages().get(userId="me", id=mail_id).execute()
+    if "body" in mail["payload"]:
+        body_data = mail["payload"]["body"]["data"]
+        decoded_body_data = base64.urlsafe_b64decode(body_data).decode("utf-8")
+        otp = [part for part in decoded_body_data.split() if part.isdigit()][-1]
+
+        return otp
+
+def _parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "-o",
+        "--otp",
+        action="store_true",
+        help="Automate OTP fetching",
+    )
+    args = parser.parse_args()
+    return args
+
 if __name__ == "__main__":
     secret_question, session_token = get_secret_question()
-    request_otp(session_token, secret_question)
-    
-    otp = input("Enter OTP received via email: ").strip()
-    sso_token = login(session_token, secret_question, otp)
+    secret_answer = erpcreds.SECURITY_QUESTIONS_ANSWERS[secret_question]
+
+    args = _parse_args()
+    if args.otp:
+        otp = _getOTP(OTP_CHECK_INTERVAL=2, session_token=session_token, log=True)
+    else:
+        request_otp(session_token, secret_answer)
+        otp = input("Enter OTP received via email: ").strip()
+
+    sso_token = login(session_token, secret_answer, otp)
 
     download_timetable(sso_token)

requirements.ci.txt

@@ -0,0 +1,6 @@
+argparse
+requests==2.32.3
+google-api-python-client==2.160.0
+google-auth==2.38.0
+google-auth-httplib2==0.2.0
+google-auth-oauthlib==1.2.1