# This file is registered as a pytest plugin, meaning it will automatically loaded. # All fixtures in this file will be available without manual loading. import os import re # from datetime import datetime, timedelta from pathlib import Path from typing import Generator, Protocol, TypedDict import pytest from dotenv import dotenv_values from icecream import ic # type: ignore from imbox import Imbox # type: ignore from playwright.sync_api import BrowserContext, expect from pytest import Parser from pytest_abra import BaseUrl, DirManager, EnvFile def pytest_addoption(parser: Parser): parser.addoption("--runner_index", action="store", type=int) parser.addoption("--output_dir", action="store", type=Path) parser.addoption("--session_id", action="store", type=str) parser.addoption("--timeout", action="store", type=int, default=30_000) @pytest.fixture(autouse=True) def set_expect_timeout(request): TIMEOUT = request.config.getoption("--timeout") expect.set_options(timeout=TIMEOUT) @pytest.fixture def context(context: BrowserContext, request) -> BrowserContext: # note: because this has the existing context fixture as an argument, it is ensured # that the original fixture is called first and then overwritten by this custom one. TIMEOUT = request.config.getoption("--timeout") LOCALE = {"Accept-Language": "de_DE"} context.set_default_timeout(TIMEOUT) context.set_extra_http_headers(LOCALE) return context @pytest.fixture(scope="session") def DIR(request) -> DirManager: """Fixture holding test directories DIR.OUTPUT DIR.SESSION DIR.STATES DIR.RESULTS DIR.STATUS""" output_dir = request.config.getoption("--output_dir") assert output_dir, "pytest argument --output_dir not set" session_id = request.config.getoption("--session_id") assert session_id, "pytest argument --session_id not set" dirmanager = DirManager(output_dir=output_dir, session_id=session_id) dirmanager.create_all_dirs() return dirmanager @pytest.fixture(scope="session") def env_files(DIR: DirManager) -> list[EnvFile]: """list of EnvFile objects created from the given env files""" env_files_dict: dict[int, EnvFile] = dict() for env_path in DIR.ENV_FILES.glob("*.env"): config: dict[str, str] = dotenv_values(env_path) # type: ignore env_type = config["TYPE"] result = re.search(r"(\d+)-*", env_path.name) assert result runner_index = int(result[1]) env_files_dict[runner_index] = EnvFile(env_path=env_path, env_config=config, env_type=env_type) keys = list(env_files_dict.keys()) keys.sort() return [env_files_dict[key] for key in keys] @pytest.fixture(scope="session") def env_config(request, env_files: list[EnvFile]) -> dict[str, str]: """Current env_config""" runner_index = request.config.getoption("--runner_index") return env_files[runner_index].env_config @pytest.fixture(scope="session") def URL(env_config: dict[str, str]) -> BaseUrl: """BaseUrl object based on current DOMAIN""" return BaseUrl(netloc=env_config["DOMAIN"]) @pytest.fixture(scope="session") def imap_client() -> Generator[Imbox, None, None]: """imap email client using credentials from environment variables""" assert os.environ["IMAP_HOST"], "required environment variable is undefined" assert os.environ["IMAP_PORT"], "required environment variable is undefined" assert os.environ["IMAP_USER"], "required environment variable is undefined" assert os.environ["IMAP_PASS"], "required environment variable is undefined" imbox = Imbox( hostname=os.environ["IMAP_HOST"], port=os.environ["IMAP_PORT"], username=os.environ["IMAP_USER"], password=os.environ["IMAP_PASS"], ssl=True, ssl_context=None, starttls=False, ) yield imbox imbox.logout() class Body(TypedDict): plain: list html: list class Message(Protocol): sent_from: list sent_to: list subject: str headers: list date: str body: Body @pytest.fixture def imap_recent_messages(imap_client: Imbox) -> list[Message]: """Get all messages from [n_minutes] ago till now. # iterate with for uid, message in messages: print(uid, message.subject, message.date)""" # N_MINUTES = 30 # n_minutes_ago = datetime.now() - timedelta(minutes=N_MINUTES) uids: list[bytes] = [] messages: list[Message] = [] # for uid, message in imap_client.messages(date__gt=n_minutes_ago): for uid, message in imap_client.messages(): ic("one time") uids.append(uid) messages.append(message) return messages