-
Notifications
You must be signed in to change notification settings - Fork 683
/
migration_c5a02eb52f2d.py
89 lines (74 loc) · 2.94 KB
/
migration_c5a02eb52f2d.py
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
import random
import uuid
from db import db
from journalist_app import create_app
from sqlalchemy import text
from sqlalchemy.exc import OperationalError
from .helpers import random_chars
class Helper:
def __init__(self):
self.journalist_id = None
class UpgradeTester(Helper):
def __init__(self, config):
Helper.__init__(self)
self.config = config
self.app = create_app(config)
def create_journalist(self):
params = {
"uuid": str(uuid.uuid4()),
"username": random_chars(50),
"nonce": random.randint(20, 100),
}
sql = """INSERT INTO journalists (uuid, username, session_nonce)
VALUES (:uuid, :username, :nonce)"""
return db.engine.execute(text(sql), **params).lastrowid
def add_revoked_token(self):
params = {
"journalist_id": self.journalist_id,
"token": "abc123",
}
sql = """INSERT INTO revoked_tokens (journalist_id, token)
VALUES (:journalist_id, :token)
"""
db.engine.execute(text(sql), **params)
def load_data(self):
with self.app.app_context():
self.journalist_id = self.create_journalist()
self.add_revoked_token()
def check_upgrade(self):
with self.app.app_context():
sql = "SELECT session_nonce FROM journalists WHERE id = :id"
params = {"id": self.journalist_id}
try:
db.engine.execute(text(sql), **params).fetchall()
except OperationalError:
pass
sql = "SELECT * FROM revoked_tokens WHERE id = :id"
try:
db.engine.execute(text(sql), **params).fetchall()
except OperationalError:
pass
class DowngradeTester(Helper):
def __init__(self, config):
Helper.__init__(self)
self.config = config
self.app = create_app(config)
def create_journalist(self):
params = {"uuid": str(uuid.uuid4()), "username": random_chars(50)}
sql = """INSERT INTO journalists (uuid, username)
VALUES (:uuid, :username)"""
return db.engine.execute(text(sql), **params).lastrowid
def load_data(self):
with self.app.app_context():
self.journalist_id = self.create_journalist()
def check_downgrade(self):
with self.app.app_context():
sql = "SELECT session_nonce FROM journalists WHERE id = :id"
params = {"id": self.journalist_id}
res = db.engine.execute(text(sql), **params).fetchone()
assert isinstance(res["session_nonce"], int)
sql = """INSERT INTO revoked_tokens (journalist_id, token)
VALUES (:journalist_id, :token)"""
params = {"journalist_id": self.journalist_id, "token": "abc789"}
res = db.engine.execute(text(sql), **params).lastrowid
assert isinstance(res, int)