-
Notifications
You must be signed in to change notification settings - Fork 2k
/
test_cli.py
111 lines (85 loc) · 3.5 KB
/
test_cli.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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import os
import csv
from nose.tools import assert_equal
from ckan import model
from ckan.lib.cli import ManageDb,SearchIndexCommand
from ckan.lib.create_test_data import CreateTestData
from ckan.common import json
from ckan.lib.search import index_for,query_for
class TestDb:
@classmethod
def setup_class(cls):
cls.db = ManageDb('db')
CreateTestData.create()
# delete warandpeace
rev = model.repo.new_revision()
model.Package.by_name(u'warandpeace').delete()
model.repo.commit_and_remove()
@classmethod
def teardown_class(cls):
model.repo.rebuild_db()
def test_simple_dump_csv(self):
csv_filepath = '/tmp/dump.tmp'
self.db.args = ('simple-dump-csv %s' % csv_filepath).split()
self.db.simple_dump_csv()
assert os.path.exists(csv_filepath), csv_filepath
f_obj = open(csv_filepath, "r")
reader = csv.reader(f_obj)
rows = [row for row in reader]
assert_equal(rows[0][:3], ['id', 'name', 'title'])
pkg_names = set(row[1] for row in rows[1:])
assert 'annakarenina' in pkg_names, pkg_names
assert 'warandpeace' not in pkg_names, pkg_names
def test_simple_dump_json(self):
json_filepath = '/tmp/dump.tmp'
self.db.args = ('simple-dump-json %s' % json_filepath).split()
self.db.simple_dump_json()
assert os.path.exists(json_filepath), json_filepath
f_obj = open(json_filepath, "r")
rows = json.loads(f_obj.read())
assert set(rows[0].keys()) > set(('id', 'name', 'title')), rows[0].keys()
pkg_names = set(row['name'] for row in rows)
assert 'annakarenina' in pkg_names, pkg_names
assert 'warandpeace' not in pkg_names, pkg_names
class FakeOptions():
def __init__(self,**kwargs):
for key in kwargs:
setattr(self,key,kwargs[key])
class TestSearch:
@classmethod
def setup_class(cls):
cls.search = SearchIndexCommand('search-index')
cls.index = index_for(model.Package)
cls.query = query_for(model.Package)
CreateTestData.create()
@classmethod
def teardown_class(cls):
model.repo.rebuild_db()
def test_clear_and_rebuild_index(self):
# Clear index
self.search.args = ()
self.search.options = FakeOptions()
self.search.clear()
self.query.run({'q':'*:*'})
assert self.query.count == 0
# Rebuild index
self.search.args = ()
self.search.options = FakeOptions(only_missing=False, force=False, refresh=False, commit_each=False, quiet=False)
self.search.rebuild()
pkg_count = model.Session.query(model.Package).filter(model.Package.state==u'active').count()
self.query.run({'q':'*:*'})
assert self.query.count == pkg_count
def test_clear_and_rebuild_only_one(self):
pkg_count = model.Session.query(model.Package).filter(model.Package.state==u'active').count()
# Clear index for annakarenina
self.search.args = ('clear annakarenina').split()
self.search.options = FakeOptions()
self.search.clear()
self.query.run({'q':'*:*'})
assert self.query.count == pkg_count - 1
# Rebuild index for annakarenina
self.search.args = ('rebuild annakarenina').split()
self.search.options = FakeOptions(only_missing=False,force=False,refresh=False,commit_each=False)
self.search.rebuild()
self.query.run({'q':'*:*'})
assert self.query.count == pkg_count