conftest.py

from pathlib import Path
import os
import shutil
import allure
import pytest
from playwright.sync_api import (
    Browser,
    BrowserContext,
    Page,
    expect,
    sync_playwright,
)
import time
from config.settings import Settings
from pages.accountsPage import AccountPage
from pages.homePage import HomePage
from pages.loginPage import LoginPage
from pages.productPage import ProductPage
from pages.cartPage import CartPage
from components.navBar import NavBar
from utils.product_catalog import get_happy_path_product_ids, get_boundary_product_ids, get_eco_product_id


_AUTH_STATE: dict[str, float] = {}


def ensure_fresh_auth(browser: Browser, auth_file: str, credentials: dict, worker_id: str):
    """Re-authenticates and rewrites auth_file if the JWT is at or past
    the refresh threshold. Call before creating any authenticated context."""
    age = time.time() - _AUTH_STATE.get(worker_id, 0)
    if age >= Settings.TOKEN_REFRESH_THRESHOLD_SECONDS:
        print(f"\n[auth:{worker_id}] Token age {age:.0f}s >= {Settings.TOKEN_REFRESH_THRESHOLD_SECONDS}s -- refreshing...")
        create_storage_state(browser, auth_file, credentials)
        _AUTH_STATE[worker_id] = time.time()


@pytest.fixture(scope="session", autouse=True)
def cleanup_auth_files():
    """
    Remove any existing authentication state files before the test session starts.
    Fresh auth files will be generated by create_auth_state().
    """

    auth_dir = Path.cwd()

    for auth_file in auth_dir.glob("auth_*.json"):
        try:
            auth_file.unlink()
            print(f"Deleted auth file: {auth_file.name}")
        except OSError as e:
            print(f"Failed to delete {auth_file.name}: {e}")

    yield


@pytest.fixture(scope="session")
def worker_id():
    worker = os.environ.get("PYTEST_XDIST_WORKER", "master")
    return "gw0" if worker == "master" else worker


def pytest_addoption(parser):
    """
    Register framework-specific command line options.
    """

    parser.addoption("--browser", action="store", default="chromium", 
                     help="Comma-separated browsers (e.g. chromium,firefox,webkit)")


def pytest_generate_tests(metafunc):
    if "browser_name" in metafunc.fixturenames:
        browsers = [b.strip() for b in metafunc.config.getoption("--browser").split(",")]
        valid = {"chromium", "firefox", "webkit"}
        invalid = set(browsers) - valid

        if invalid:
            raise pytest.UsageError(f"Unsupported browser(s): {', '.join(invalid)}")

        metafunc.parametrize("browser_name", browsers, scope="session")

    if "product_page" in metafunc.fixturenames:
        if "happy_path" in metafunc.definition.keywords:
            ids = get_happy_path_product_ids(count_per_category=2)
            metafunc.parametrize("product_page", ids, indirect=True)
        elif "boundary" in metafunc.definition.keywords:
            boundary_map = get_boundary_product_ids()
            ids = [v for v in boundary_map.values() if v]
            metafunc.parametrize("product_page", ids, indirect=True)
        elif "eco" in metafunc.definition.keywords:
            eco_id = get_eco_product_id()
            ids = [eco_id] if eco_id else []
            metafunc.parametrize("product_page", ids, indirect=True)


@pytest.fixture(scope="session", autouse=True)
def prepare_artifacts():
    """
    Prepare artifact folders before execution.
    """

    directories = [
        Settings.SCREENSHOT_DIR,
        Settings.TRACE_DIR,
        Settings.VIDEO_DIR
    ]

    for directory in directories:
        shutil.rmtree(directory, ignore_errors=True)
        Path(directory).mkdir(parents=True, exist_ok=True)

    yield

@pytest.fixture(scope="session")
def playwright_instance():
    with sync_playwright() as p:
        p.selectors.set_test_id_attribute("data-test")  # site convention differs from Playwright default
        yield p


@pytest.fixture(scope="session")
def browser(playwright_instance, browser_name):
    """
    Launch the requested browser.
    """
    browser_type = getattr(
        playwright_instance,
        browser_name
    )

    browser = browser_type.launch(
        headless=Settings.HEADLESS,
        slow_mo=Settings.SLOW_MO
    )

    yield browser

    browser.close()


@pytest.fixture
def context(browser: Browser):
    context = browser.new_context()
    context.tracing.start(screenshots=True, snapshots=True, sources=True)
    yield context
    context.close()


def collect_artifacts(page: Page, context: BrowserContext, request, browser_name: str, worker_id: str):
    failed = (hasattr(request.node, "rep_call") and request.node.rep_call.failed)

    if failed:
        screenshot = Settings.get_screenshot_path(
            browser_name=browser_name,
            worker_id=worker_id,
            test_name=request.node.name,
        )
        try:
            page.screenshot(path=str(screenshot), full_page=True)
            allure.attach.file(str(screenshot), name="Failure Screenshot",
                               attachment_type=allure.attachment_type.PNG)
        except Exception as e:
            print(e)

        trace = Settings.get_trace_path(browser_name=browser_name, worker_id=worker_id, test_name=request.node.name)
        try:
            context.tracing.stop(path=str(trace))
        except Exception as e:
            print(e)
    else:
        context.tracing.stop()


@pytest.fixture
def page(request, context: BrowserContext, browser_name, worker_id):
    page = context.new_page()
    yield page
    collect_artifacts(page, context, request, browser_name, worker_id)
    page.close()


@pytest.fixture
def login_page(page: Page):
    login_page = LoginPage(page)
    login_page.navigate(Settings.BASE_URL)
    expect(login_page.get_by_role("link", name="Practice Software Testing - Toolshop")).to_be_visible()
    sign_in = login_page.get_by_role("link", name="Sign in")
    expect(sign_in).to_be_visible()
    sign_in.click()
    return login_page


@pytest.fixture(scope="session")
def auth_file(worker_id, browser_name):
    return f"auth_{worker_id}_{browser_name}.json"


@pytest.fixture(scope="session")
def credentials(worker_id):
    return Settings.get_worker_credentials(worker_id)


def create_storage_state(browser: Browser, auth_file: str, credentials):
    try:
        context = browser.new_context()
        page = context.new_page()
        login_page = LoginPage(page)
        login_page.navigate(Settings.BASE_URL)
        login_page.get_by_role("link", name="Sign in").click()
        login_page.login_and_wait_for_profile(credentials["username"], credentials["password"])
        login_page.to_be_visible(login_page.user_button(credentials["name"]), timeout=30000)
        context.storage_state(path=auth_file)
        if not Path(auth_file).is_file():
            raise RuntimeError(
                f"Storage state file '{auth_file}' was not created."
            )
    except Exception:
        try:
            alert = page.get_by_role("alert")
            if alert.is_visible(timeout=1000):
                print(alert.first.text_content())
        except Exception:
            pass
        raise
    finally:
        page.close()
        context.close()


@pytest.fixture(scope="session", autouse=True)
def cleanup_all_favorites(browser: Browser, auth_file, worker_id):
    """
    Session-scoped teardown-only sweep. Runs once per xdist worker, after
    all tests in that worker finish. Deletes every favorite currently on
    this worker's account -- catches orphans from skipped/crashed tests
    that per-instance tracking would miss.
    """
    yield  # no setup -- pure teardown after this point
    ensure_fresh_auth(browser, auth_file, credentials, worker_id)
    context = browser.new_context(storage_state=auth_file)
    page = context.new_page()
    try:
        # localStorage from storage_state only attaches once a page
        # navigates to the matching origin -- must load BASE_URL first.
        page.goto(Settings.BASE_URL)
        headers = HomePage(page).get_auth_header()  # any BasePage subclass works

        resp = context.request.get(f"{Settings.API_BASE_URL}/favorites", headers=headers)
        favorites = resp.json() if resp.ok else []
        print(f"\n[cleanup:{worker_id}] Found {len(favorites)} favorite(s) to sweep")

        for fav in favorites:
            fav_id = fav["id"]
            del_resp = context.request.delete(
                f"{Settings.API_BASE_URL}/favorites/{fav_id}", headers=headers
            )
            print(f"[cleanup:{worker_id}] DELETE favorite {fav_id} -> {del_resp.status}")
    except Exception as e:
        print(f"[cleanup:{worker_id}] Favorites sweep failed: {e}")
    finally:
        page.close()
        context.close()


@pytest.fixture(scope="session", autouse=True)
def create_auth_state(browser: Browser, auth_file: str, credentials, worker_id):
    for attempt in range(1, Settings.AUTH_RETRY_COUNT + 1):
        try:
            print(f"\nCreating auth state ({attempt}/{Settings.AUTH_RETRY_COUNT})...")
            create_storage_state(browser, auth_file, credentials)
            _AUTH_STATE[worker_id] = time.time()   # <-- new line
            print("Auth state created successfully.")
            return
        except Exception as e:
            print(f"Attempt {attempt} failed: {e}")
            if attempt == Settings.AUTH_RETRY_COUNT:
                raise
            print("Retrying authentication...\n")


@pytest.fixture
def auth_context(browser: Browser, auth_file, credentials, worker_id):
    ensure_fresh_auth(browser, auth_file, credentials, worker_id)
    context = browser.new_context(storage_state=auth_file)
    context.tracing.start(screenshots=True, snapshots=True, sources=True)
    yield context
    context.close()

@pytest.fixture
def auth_page(request, auth_context: BrowserContext, browser_name, worker_id):
    page = auth_context.new_page()

    yield page

    collect_artifacts(page, auth_context, request, browser_name, worker_id)

    page.close()

@pytest.fixture
def account_page(auth_page: Page):
    account_page = AccountPage(auth_page)
    # account_page.page.pause()
    account_page.navigate(Settings.ACCOUNTS_URL)

    expect(auth_page.get_by_role("heading", name="My account")).to_be_visible()

    return account_page


@pytest.fixture
def home_page(auth_page: Page):
    home_page = HomePage(auth_page)

    with auth_page.expect_response(lambda r: "/products" in r.url and r.request.method == "QUERY" and r.ok):
        home_page.navigate(Settings.BASE_URL)

    expect(auth_page.get_by_alt_text("Banner")).to_be_visible()

    return home_page


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    outcome = yield
    report = outcome.get_result()

    if report.when != "call":
        return

    setattr(item, f"rep_{report.when}", report)


@pytest.fixture
def product_page(auth_page: Page, request):
    product_id = request.param
    prod_page = ProductPage(auth_page, product_id=product_id)
    # prod_page.page.pause()
    prod_page.navigate(f"{Settings.BASE_URL}/product/{product_id}")
    try:
        expect(prod_page.product_name).to_be_visible(timeout=5000)
    except AssertionError:
        pytest.skip(f"Product {product_id} no longer resolves -- stale catalog reference (list/detail data drift)")

    yield prod_page

@pytest.fixture
def cart_page(auth_page: Page):
    """
    Factory fixture -- returns a callable, not an eager page object,
    because cart state must exist BEFORE navigating (add via ProductPage
    first). Instantiates NavBar directly rather than depending on
    HomePage, since the cart icon is a global layout element reachable
    from any authenticated page (e.g. straight from ProductPage after
    add_to_cart()).
    """
    def _open() -> CartPage:
        NavBar(auth_page).cart_link.click()
        return CartPage(auth_page)
    return _open