forked from splunk/splunk-sdk-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_kvstore_conf.py
More file actions
executable file
·99 lines (86 loc) · 3.83 KB
/
test_kvstore_conf.py
File metadata and controls
executable file
·99 lines (86 loc) · 3.83 KB
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
#!/usr/bin/env python
#
# Copyright 2011-2014 Splunk, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"): you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import testlib
try:
import unittest
except ImportError:
import unittest2 as unittest
import splunklib.client as client
class KVStoreConfTestCase(testlib.SDKTestCase):
def setUp(self):
super(KVStoreConfTestCase, self).setUp()
self.service.namespace['owner'] = 'nobody'
self.service.namespace['app'] = 'search'
self.confs = self.service.kvstore
if ('test' in self.confs):
self.confs['test'].delete()
def test_owner_restriction(self):
self.service.namespace['owner'] = 'admin'
self.assertRaises(client.HTTPError, lambda: self.confs.list())
self.service.namespace['owner'] = 'nobody'
def test_create_delete_collection(self):
self.confs.create('test')
self.assertTrue('test' in self.confs)
self.confs['test'].delete()
self.assertTrue(not 'test' in self.confs)
def test_update_collection(self):
self.confs.create('test')
self.confs['test'].post(**{'accelerated_fields.ind1': '{"a": 1}', 'field.a': 'number'})
self.assertEqual(self.confs['test']['field.a'], 'number')
self.assertEqual(self.confs['test']['accelerated_fields.ind1'], '{"a": 1}')
self.confs['test'].delete()
def test_update_fields(self):
self.confs.create('test')
self.confs['test'].post(**{'field.a': 'number'})
self.assertEqual(self.confs['test']['field.a'], 'number')
self.confs['test'].update_field('a', 'string')
self.assertEqual(self.confs['test']['field.a'], 'string')
self.confs['test'].delete()
def test_create_unique_collection(self):
self.confs.create('test')
self.assertTrue('test' in self.confs)
self.assertRaises(client.HTTPError, lambda: self.confs.create('test'))
self.confs['test'].delete()
def test_overlapping_collections(self):
self.service.namespace['app'] = 'system'
self.confs.create('test')
self.service.namespace['app'] = 'search'
self.confs.create('test')
self.assertEqual(self.confs['test']['eai:appName'], 'search')
self.service.namespace['app'] = 'system'
self.assertEqual(self.confs['test']['eai:appName'], 'system')
self.service.namespace['app'] = 'search'
self.confs['test'].delete()
self.confs['test'].delete()
"""
def test_create_accelerated_fields_fields(self):
self.confs.create('test', indexes={'foo': '{"foo": 1}', 'bar': {'bar': -1}}, **{'field.foo': 'string'})
self.assertEqual(self.confs['test']['accelerated_fields.foo'], '{"foo": 1}')
self.assertEqual(self.confs['test']['field.foo'], 'string')
self.assertRaises(client.HTTPError, lambda: self.confs['test'].post(**{'accelerated_fields.foo': 'THIS IS INVALID'}))
self.assertEqual(self.confs['test']['accelerated_fields.foo'], '{"foo": 1}')
self.confs['test'].update_accelerated_fields('foo', '')
self.assertEqual(self.confs['test']['accelerated_fields.foo'], None)
"""
def tearDown(self):
if ('test' in self.confs):
self.confs['test'].delete()
if __name__ == "__main__":
try:
import unittest2 as unittest
except ImportError:
import unittest
unittest.main()