-
Notifications
You must be signed in to change notification settings - Fork 2k
/
test_dump.py
94 lines (82 loc) · 3.71 KB
/
test_dump.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
import json
import nose
from nose.tools import assert_equals
from pylons import config
import sqlalchemy.orm as orm
import paste.fixture
import ckan.config.middleware as middleware
import ckan.plugins as p
import ckan.lib.create_test_data as ctd
import ckan.model as model
import ckan.tests.legacy as tests
import ckanext.datastore.db as db
import ckanext.datastore.tests.helpers as helpers
class TestDatastoreDump(object):
sysadmin_user = None
normal_user = None
@classmethod
def setup_class(cls):
if helpers.should_skip_test_for_legacy():
raise nose.SkipTest("SQL tests are not supported in legacy mode")
wsgiapp = middleware.make_app(config['global_conf'], **config)
cls.app = paste.fixture.TestApp(wsgiapp)
if not tests.is_datastore_supported():
raise nose.SkipTest("Datastore not supported")
p.load('datastore')
ctd.CreateTestData.create()
cls.sysadmin_user = model.User.get('testsysadmin')
cls.normal_user = model.User.get('annafan')
resource = model.Package.get('annakarenina').resources[0]
cls.data = {
'resource_id': resource.id,
'force': True,
'aliases': 'books',
'fields': [{'id': u'b\xfck', 'type': 'text'},
{'id': 'author', 'type': 'text'},
{'id': 'published'},
{'id': u'characters', u'type': u'_text'}],
'records': [{u'b\xfck': 'annakarenina',
'author': 'tolstoy',
'published': '2005-03-01',
'nested': ['b', {'moo': 'moo'}],
u'characters': [u'Princess Anna', u'Sergius']},
{u'b\xfck': 'warandpeace', 'author': 'tolstoy',
'nested': {'a': 'b'}}]
}
postparams = '%s=1' % json.dumps(cls.data)
auth = {'Authorization': str(cls.sysadmin_user.apikey)}
res = cls.app.post('/api/action/datastore_create', params=postparams,
extra_environ=auth)
res_dict = json.loads(res.body)
assert res_dict['success'] is True
engine = db._get_engine({
'connection_url': config['ckan.datastore.write_url']})
cls.Session = orm.scoped_session(orm.sessionmaker(bind=engine))
@classmethod
def teardown_class(cls):
helpers.rebuild_all_dbs(cls.Session)
p.unload('datastore')
def test_dump_basic(self):
auth = {'Authorization': str(self.normal_user.apikey)}
res = self.app.get('/datastore/dump/{0}'.format(str(
self.data['resource_id'])), extra_environ=auth)
content = res.body.decode('utf-8')
expected = u'_id,b\xfck,author,published,characters,nested'
assert_equals(content[:len(expected)], expected)
assert 'warandpeace' in content
assert "[u'Princess Anna', u'Sergius']" in content
# get with alias instead of id
res = self.app.get('/datastore/dump/{0}'.format(str(
self.data['aliases'])), extra_environ=auth)
def test_dump_does_not_exist_raises_404(self):
auth = {'Authorization': str(self.normal_user.apikey)}
self.app.get('/datastore/dump/{0}'.format(str(
'does-not-exist')), extra_environ=auth, status=404)
def test_dump_limit(self):
auth = {'Authorization': str(self.normal_user.apikey)}
res = self.app.get('/datastore/dump/{0}?limit=1'.format(str(
self.data['resource_id'])), extra_environ=auth)
content = res.body.decode('utf-8')
expected = u'_id,b\xfck,author,published,characters,nested'
assert_equals(content[:len(expected)], expected)
assert_equals(len(content), 148)