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
|
# -*- encoding: utf-8 -*-
try:
import unittest2 as unittest
except ImportError:
import unittest
import os
import sqlite3
import tempfile
from isso import config
from isso.db import SQLite3
from isso.compat import iteritems
class TestDBMigration(unittest.TestCase):
def setUp(self):
fd, self.path = tempfile.mkstemp()
def tearDown(self):
os.unlink(self.path)
def test_defaults(self):
conf = config.new({
"general": {
"dbpath": "/dev/null",
"max-age": "1h"
}
})
db = SQLite3(self.path, conf)
self.assertEqual(db.version, SQLite3.MAX_VERSION)
self.assertTrue(db.preferences.get("session-key", "").isalnum())
def test_session_key_migration(self):
conf = config.new({
"general": {
"dbpath": "/dev/null",
"max-age": "1h"
}
})
conf.set("general", "session-key", "supersecretkey")
with sqlite3.connect(self.path) as con:
con.execute("PRAGMA user_version = 1")
con.execute("CREATE TABLE threads (id INTEGER PRIMARY KEY)")
db = SQLite3(self.path, conf)
self.assertEqual(db.version, SQLite3.MAX_VERSION)
self.assertEqual(db.preferences.get("session-key"),
conf.get("general", "session-key"))
# try again, now with the session-key removed from our conf
conf.remove_option("general", "session-key")
db = SQLite3(self.path, conf)
self.assertEqual(db.version, SQLite3.MAX_VERSION)
self.assertEqual(db.preferences.get("session-key"),
"supersecretkey")
def test_limit_nested_comments(self):
tree = {
1: None,
2: None,
3: 2,
4: 3,
7: 3,
5: 2,
6: None
}
with sqlite3.connect(self.path) as con:
con.execute("PRAGMA user_version = 2")
con.execute("CREATE TABLE threads ("
" id INTEGER PRIMARY KEY,"
" uri VARCHAR UNIQUE,"
" title VARCHAR)")
con.execute("CREATE TABLE comments ("
" tid REFERENCES threads(id),"
" id INTEGER PRIMARY KEY,"
" parent INTEGER,"
" created FLOAT NOT NULL, modified FLOAT,"
" text VARCHAR, email VARCHAR, website VARCHAR,"
" mode INTEGER,"
" remote_addr VARCHAR,"
" likes INTEGER DEFAULT 0,"
" dislikes INTEGER DEFAULT 0,"
" voters BLOB)")
con.execute("INSERT INTO threads (uri, title) VALUES (?, ?)", ("/", "Test"))
for (id, parent) in iteritems(tree):
con.execute("INSERT INTO comments ("
" tid, parent, created)"
"VALUEs (?, ?, ?)", (id, parent, id))
conf = config.new({
"general": {
"dbpath": "/dev/null",
"max-age": "1h"
}
})
SQLite3(self.path, conf)
flattened = [
(1, None),
(2, None),
(3, 2),
(4, 2),
(5, 2),
(6, None),
(7, 2)
]
with sqlite3.connect(self.path) as con:
rv = con.execute("SELECT id, parent FROM comments ORDER BY created").fetchall()
self.assertEqual(flattened, rv)
|
