import sys, os base_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) sys.path.insert(0, base_path) from utils.trace_cleaner import trace_cleaner # <--- Added import for trace_cleaner import email import imaplib import time from playwright.sync_api import TimeoutError, expect, sync_playwright EMAIL_USERNAME = "expensifytesting1@gmail.com" EMAIL_PASSWORD = "" EMAIL_USERNAME_POSTFIX = "46528" def generate_user_email(user_id=None): """ Generate an email address for a user login. """ temp = EMAIL_USERNAME.rsplit("@", 1) username = temp[0].strip() domain = temp[1].strip() return f"{username}+{EMAIL_USERNAME_POSTFIX}{user_id or ''}@{domain}" def clear_inbox(username, password): """ Delete all the messages from the Inbox. """ return with imaplib.IMAP4_SSL(host="imap.gmail.com") as imap: imap.login(username, password) imap.select("inbox") imap.store("1:*", "+FLAGS", "\\Deleted") imap.expunge() imap.close() def get_otp_from_email(username, password, retries=12, delay=5): """ Read the OTP email and return the OTP code. """ return "123456" with imaplib.IMAP4_SSL(host="imap.gmail.com") as imap: imap.login(username, password) for i in range(1, retries + 1): imap.select("inbox") status, messages = imap.search(None, "ALL") if status == "OK": for message_id in reversed(messages[0].split()): status, data = imap.fetch(message_id, "(RFC822)") if status == "OK": email_message = email.message_from_bytes(data[0][1]) subject, encoding = email.header.decode_header(email_message["Subject"])[0] if isinstance(subject, bytes): subject = subject.decode(encoding) if subject.startswith("Expensify magic sign-in code:"): otp_code = subject.split(":")[-1].strip() return otp_code time.sleep(delay) imap.close() raise AssertionError("Failed to read the OTP from the email") def login_member(p, member_first_name: str, member_last_name: str) -> tuple: browser = p.chromium.launch( headless=True, args=["--disable-web-security", "--disable-features=IsolateOrigins,site-per-process"], ) context = browser.new_context() page = context.new_page() user_email = generate_user_email() page.goto("https://dev.new.expensify.com:8082/") clear_inbox(EMAIL_USERNAME, EMAIL_PASSWORD) page.locator('input[type="email"]').fill(user_email) page.locator('button[tabindex="0"]').click() try: page.locator('button[tabindex="0"]').click() except Exception: pass try: expect(page.get_by_test_id("SignInPage").get_by_test_id("validateCode")).to_be_visible(timeout=5000) except AssertionError: page.get_by_test_id("SignInPage").get_by_role("button", name="Join").click() else: otp_code = get_otp_from_email(EMAIL_USERNAME, EMAIL_PASSWORD) page.get_by_test_id("SignInPage").get_by_test_id("validateCode").fill(otp_code) try: page.get_by_test_id("SignInPage").get_by_role("button", name="Sign in").click(timeout=2000) except (AssertionError, TimeoutError): pass try: expect(page.get_by_text("What do you want to do today?")).to_be_visible(timeout=5000) except AssertionError: pass else: page.get_by_label("Track and budget expenses").click() page.locator('input[name="fwork"]').fill(f"Work_{int(time.time())}") page.get_by_role("button", name="Continue").last.click() page.get_by_role("textbox", name="First name").fill(member_first_name) page.get_by_role("textbox", name="Last name").fill(member_last_name) page.get_by_role("button", name="Continue").click() expect(page.get_by_test_id("BaseSidebarScreen")).to_be_visible(timeout=10000) return browser, page, user_email def add_member(page): page.get_by_label("My settings").click() page.get_by_test_id("InitialSettingsPage").get_by_label("Workspaces").get_by_text("Workspaces").click() test_workspace = "WS Test" existing_workspaces = [ item.split("\n")[0].strip() for item in page.get_by_test_id("WorkspacesListPage").get_by_label("row", exact=True).all_inner_texts() ] if test_workspace not in existing_workspaces: page.get_by_test_id("WorkspacesListPage").get_by_role("button", name="New workspace").first.click() page.wait_for_timeout(2000) page.get_by_test_id("WorkspacePageWithSections").get_by_text("Name", exact=True).click() page.get_by_test_id("WorkspaceNamePage").get_by_role("textbox").fill(test_workspace) page.get_by_test_id("WorkspaceNamePage").get_by_role("button", name="Save").click() page.wait_for_timeout(2000) page.get_by_test_id("WorkspaceInitialPage").get_by_role("button", name="Back").click() page.get_by_test_id("WorkspacesListPage").get_by_label("row").get_by_text(test_workspace).first.click() page.get_by_text("More features").click() try: expect(page.get_by_label("Set up custom fields for")).to_be_checked(timeout=1000) except (AssertionError, TimeoutError): page.get_by_label("Set up custom fields for").click() page.get_by_role("button", name="Upgrade").click() page.get_by_role("button", name="Got it, thanks").click() page.get_by_test_id("WorkspaceInitialPage").get_by_text("Report fields").click() field_name = f"Filed_{int(time.time())}" page.get_by_role("button", name="Add field").click() page.get_by_test_id("CreateReportFieldsPage").get_by_text("Name").click() page.get_by_role("textbox", name="Name").fill(field_name) page.get_by_test_id("TextSelectorModal").get_by_role("button", name="Save").click() page.get_by_test_id("CreateReportFieldsPage").get_by_text("Type").click() page.get_by_test_id("selection-list").get_by_label("List").click() page.get_by_role("button", name="Save").click() page.locator(f"#field_id_{field_name.upper()}").click() page.get_by_text("List values").click() page.get_by_role("button", name="Add value").click() page.get_by_role("textbox", name="Value").fill("c479") page.get_by_role("button", name="Save").click() page.get_by_role("button", name="Add value").click() page.get_by_role("textbox", name="Value").fill("b789") page.get_by_role("button", name="Save").click() page.get_by_role("button", name="Add value").click() page.get_by_role("textbox", name="Value").fill("t963") page.get_by_role("button", name="Save").click() first_list_value = page.locator('div[data-testid="selection-list"]').nth(1) assert ( first_list_value.inner_text() == "b789\nEnabled\nc479\nEnabled\nt963\nEnabled" ), "The first list value is not 'b789'" def test_run(admin_first_name="Yagna", admin_last_name="Sai", member_first_name="Sai", member_last_name="Yagna"): with sync_playwright() as p: browser_member = None page_member = None try: browser_member, page_member, user_email = login_member(p, member_first_name, member_last_name) page_member.context.tracing.start(screenshots=True, snapshots=True, sources=True) add_member(page_member) except Exception as e: print("Test failed with exception:", e) raise finally: trace_path = "/app/expensify/user_tool/output_browser1.zip" if page_member is not None: try: page_member.context.tracing.stop(path=trace_path) except Exception as stop_err: print("Error while stopping trace:", stop_err) try: trace_cleaner(trace_path) except Exception as cleaner_err: print("Error while cleaning trace:", cleaner_err) if browser_member is not None: browser_member.close() if __name__ == "__main__": test_run()