~homeworkprod/byceps: tests/helpers.py - sourcehut git


                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
"""
tests.helpers
~~~~~~~~~~~~~

:Copyright: 2006-2019 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""

from contextlib import contextmanager

from flask import appcontext_pushed, g

from byceps.application import create_app
from byceps.database import db
from byceps.services.authentication.session import service as session_service
from byceps.services.authorization import service as authorization_service
from byceps.services.email import service as email_service
from byceps.services.party import service as party_service
from byceps.services.site import service as site_service

from testfixtures.brand import create_brand as _create_brand
from testfixtures.party import create_party as _create_party
from testfixtures.user import create_user as _create_user, \
    create_user_with_detail as _create_user_with_detail

from .base import CONFIG_FILENAME_TEST_PARTY


@contextmanager
def app_context(*, config_filename=CONFIG_FILENAME_TEST_PARTY):
    app = create_app(config_filename)

    with app.app_context():
        yield app


@contextmanager
def current_party_set(app, party):
    def handler(sender, **kwargs):
        g.party_id = party.id
        g.brand_id = party.brand_id

    with appcontext_pushed.connected_to(handler, app):
        yield


@contextmanager
def current_user_set(app, user):
    def handler(sender, **kwargs):
        g.current_user = user

    with appcontext_pushed.connected_to(handler, app):
        yield


def create_user(*args, **kwargs):
    user = _create_user(*args, **kwargs)

    db.session.add(user)
    db.session.commit()

    return user


def create_user_with_detail(*args, **kwargs):
    user = _create_user_with_detail(*args, **kwargs)

    db.session.add(user)
    db.session.commit()

    return user


def assign_permissions_to_user(user_id, role_id, permission_ids,
                               *, initiator_id=None):
    """Create the role and permissions, assign the permissions to the
    role, and assign the role to the user.
    """
    role = authorization_service.create_role(role_id, role_id)

    for permission_id in permission_ids:
        permission = authorization_service.create_permission(permission_id,
                                                             permission_id)
        authorization_service.assign_permission_to_role(permission.id, role.id)

    authorization_service.assign_role_to_user(role.id, user_id,
                                              initiator_id=initiator_id)


def create_brand(brand_id='acmecon', title='ACME Entertainment Convention'):
    brand = _create_brand(id=brand_id, title=title)

    db.session.add(brand)
    db.session.commit()

    return brand


def create_party(brand_id, party_id='acmecon-2014', title='ACMECon 2014'):
    party = _create_party(id=party_id, title=title, brand_id=brand_id)

    db.session.add(party)
    db.session.commit()

    return party_service._db_entity_to_party(party)


def create_site(party_id, *, site_id='acmecon-2014-website', title='Website',
                server_name='www.example.com', email_config_id='acmecon'):
    return site_service.create_site(site_id, party_id, title, server_name,
                                    email_config_id)


def create_email_config(config_id='acmecon', sender_address='info@example.com',
                        sender_name='ACMECon', contact_address=None):
    email_service.set_config(config_id, sender_address, sender_name=sender_name,
                             contact_address=contact_address)


@contextmanager
def http_client(app, *, user_id=None):
    """Provide an HTTP client.

    If a user ID is given, the client authenticates with the user's
    credentials.
    """
    client = app.test_client()

    if user_id is not None:
        _add_user_credentials_to_session(client, user_id)

    yield client


def _add_user_credentials_to_session(client, user_id):
    session_token = session_service.find_session_token_for_user(user_id)

    with client.session_transaction() as session:
        session['user_id'] = str(user_id)
        session['user_auth_token'] = str(session_token.token)


def login_user(user_id):
    """Authenticate the user to create a session."""
    session_service.get_session_token(user_id)