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
|
# Copyright (C) 2015 Custodia Project Contributors - see LICENSE file
from __future__ import absolute_import
from tests.client import LocalConnection
import json
import os
import signal
import subprocess
import time
import unittest
class CustodiaTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
env = os.environ.copy()
env['PYTHONPATH'] = './'
pexec = env.get('CUSTODIAPYTHON', 'python')
devnull = open(os.devnull, "w")
p = subprocess.Popen([pexec, 'custodia/custodia'], env=env,
stdout=devnull, stderr=devnull)
cls.custodia_process = p
time.sleep(1)
cls.dfl_headers = {'REMOTE_USER': 'test',
'Content-Type': 'application/json'}
@classmethod
def tearDownClass(self):
self.custodia_process.kill()
self.custodia_process.wait()
for fname in ['server_socket', 'secrets.db']:
try:
os.unlink(fname)
except OSError:
pass
def _make_request(self, cmd, path, headers=None, body=None):
conn = LocalConnection('./server_socket')
conn.connect()
conn.request(cmd, path, body=body, headers=self.dfl_headers)
return conn.getresponse()
def test_connect(self):
r = self._make_request('GET', '/', {'REMOTE_USER': 'tests'})
self.assertEqual(r.status, 200)
def test_simple_0_set_key(self):
data = {'type': 'simple', 'value': 'VmVycnlTZWNyZXQK'}
r = self._make_request('PUT', '/secrets/test/key',
self.dfl_headers, json.dumps(data))
self.assertEqual(r.status, 201)
def test_simple_1_get_key(self):
r = self._make_request('GET', '/secrets/test/key', self.dfl_headers)
self.assertEqual(r.status, 200)
body = r.read().decode('utf-8')
data = {'type': 'simple', 'value': 'VmVycnlTZWNyZXQK'}
self.assertEqual(json.loads(body), data)
def test_simple_2_del_key(self):
r = self._make_request('DELETE', '/secrets/test/key', self.dfl_headers)
self.assertEqual(r.status, 204)
|
