File: test_sqla.py

package info (click to toggle)
python-bottle-cork 0.12.0-6
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, forky, sid, trixie
  • size: 936 kB
  • sloc: python: 6,862; makefile: 4
file content (153 lines) | stat: -rw-r--r-- 4,043 bytes parent folder | download | duplicates (4)
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
146
147
148
149
150
151
152
153


from sqlalchemy import create_engine, delete, select, \
    Column, ForeignKey, Integer, MetaData, String, Table, Unicode

from cork.backends import SqlAlchemyBackend
import pytest
import os
from cork import Cork

class MockedSession(object):
    """Mock Beaker session
    """
    def __init__(self, username=None):
        self.__username = username
        self.__saved = False

    def get(self, k, default):
        assert k in ('username')
        if self.__username is None:
            return default

        return self.__username

    def __getitem__(self, k):
        assert k in ('username')
        if self.__username is None:
            raise KeyError()

        return self.__username

    def __setitem__(self, k, v):
        assert k in ('username')
        self.__username = v
        self.__saved = False

    def delete(self):
        """Used during logout to delete the current session"""
        self.__username = None

    def save(self):
        self.__saved = True

#TODO: implement tests around MockedSession __saved

class MockedSessionCork(Cork):
    """Mocked Cork instance where the session is replaced with
    MockedSession
    """
    @property
    def _beaker_session(self):
        return self._mocked_beaker_session
def setup_postgresql_db(request):

    if os.environ.get('TRAVIS', False):
        # Using Travis-CI - https://travis-ci.org/
        db_name = 'myapp_test'
    else:
        db_name = 'cork_functional_test'

    uri = "postgresql+psycopg2://postgres:@/%s" % db_name
    mb = SqlAlchemyBackend(uri, initialize=True)

    # Purge DB
    mb._drop_all_tables()
    assert len(mb.roles) == 0
    assert len(mb.users) == 0

    # Create roles
    mb.roles.insert({'role': 'special', 'level': 200})
    mb.roles.insert({'role': 'admin', 'level': 100})
    mb.roles.insert({'role': 'editor', 'level': 60})
    mb.roles.insert({'role': 'user', 'level': 50})

    # Create admin
    mb.users.insert({
        "username": "admin",
        "email_addr": "admin@localhost.local",
        "desc": "admin test user",
        "role": "admin",
        "hash": "cLzRnzbEwehP6ZzTREh3A4MXJyNo+TV8Hs4//EEbPbiDoo+dmNg22f2RJC282aSwgyWv/O6s3h42qrA6iHx8yfw=",
        "creation_date": "2012-10-28 20:50:26.286723",
        "last_login": "2012-10-28 20:50:26.286723"
    })
    assert len(mb.roles) == 4
    assert len(mb.users) == 1

    def fin():
        mb._drop_all_tables()
        assert len(mb.roles) == 0
        assert len(mb.users) == 0

    request.addfinalizer(fin)
    return mb

@pytest.fixture
def backend(tmpdir, request):
    return setup_postgresql_db(request)

    raise Exception()

@pytest.fixture
def aaa_admin(templates_dir, backend):
    # Session with an admin user
    aaa = MockedSessionCork(
        templates_dir,
        backend=backend,
        email_sender='test@localhost',
        smtp_server='localhost',
    )
    aaa._mocked_beaker_session = MockedSession(username='admin')
    return aaa

def test_write_user_hash_bytes(aaa_admin):
    username = 'huh'
    h = b'1234'
    tstamp = "just a string"

    assert isinstance(h, type(b''))
    aaa_admin._store.users[username] = {
        'role': "user",
        'hash': h,
        'email_addr': "bar",
        'desc': "foo",
        'creation_date': tstamp,
        'last_login': tstamp
    }

    #eng = aaa_admin._store._engine
    #print(eng.execute("SELECT * FROM users").fetchall()[1])

    fetched_h = aaa_admin._store.users[username]['hash']
    assert isinstance(fetched_h, type(b''))
    assert fetched_h == b'1234'


def test_write_user_hash_unicode(aaa_admin):
    username = 'huh'
    h = u'1234'
    tstamp = "just a string"
    aaa_admin._store.users[username] = {
        'role': "user",
        'hash': h,
        'email_addr': "bar",
        'desc': "foo",
        'creation_date': tstamp,
        'last_login': tstamp
    }

    fetched_h = aaa_admin._store.users[username]['hash']
    eng = aaa_admin._store._engine
    print(eng.execute("SELECT * FROM users").fetchall()[1])
    assert fetched_h == u'1234'