agilentics / boiler
"""Test fixtures.

The suite runs against an in-memory SQLite database, so it needs no Postgres and
CI can run it before the Cloud SQL Auth Proxy is up (the `integration` marker is
reserved for tests that genuinely need a live database).

Two ways to reach the database:

  `session`     a plain Session for core-unit tests, which take their session as
                an argument and never touch Flask.
  `app_client`  a Flask test client whose requests hit the same in-memory DB, by
                reconfiguring the app's scoped_session to the test engine.

StaticPool keeps a single connection, which is what makes an in-memory SQLite
database persist across the many sessions one test opens.
"""
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from db.db_declarations import Base
from utils.db_wrapper import db_session


@pytest.fixture
def engine():
    eng = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(eng)
    return eng


@pytest.fixture
def session(engine):
    Session = sessionmaker(bind=engine)
    s = Session()
    try:
        yield s
    finally:
        s.close()


@pytest.fixture
def app_client(engine):
    # Point the app's scoped_session at the test engine. Endpoints imported the
    # `db_session` proxy by reference, so reconfiguring its bind makes every
    # request query SQLite without touching the endpoint code.
    db_session.remove()
    db_session.configure(bind=engine)

    import main
    main.app.config.update(TESTING=True)
    with main.app.test_client() as client:
        yield client

    db_session.remove()


PASSWORD = "correcthorsebatterystaple"


@pytest.fixture
def account(app_client):
    """A seeded organisation with one `member` in it, and a client signed into
    it. Returns (client, user, org)."""
    from app.core import auth as auth_core
    from db.db_declarations import Organization

    org = Organization(name="Acme", slug="acme")
    db_session.add(org)
    db_session.commit()

    user = auth_core.create_user(db_session, "dev@example.com", PASSWORD, "Dev")
    auth_core.add_membership(db_session, user, org, "member")

    r = app_client.post("/api/login",
                        json={"email": "dev@example.com", "password": PASSWORD})
    assert r.status_code == 200, r.get_json()
    return app_client, user, org


def csrf_for(client) -> str:
    """The token the browser would have read from the page's meta tag.

    Any rendered page mints it into the session. Following redirects is what
    makes this work in both states: signed out, "/" redirects to the sign-in
    page; signed in, the sign-in page redirects back to "/". Either way the
    response that finally renders carries the tag.
    """
    import re

    html = client.get("/", follow_redirects=True).get_data(as_text=True)
    match = re.search(r'name="csrf-token" content="([^"]+)"', html)
    assert match, "no csrf-token meta tag on the rendered page"
    return match.group(1)