« back to all changes in this revision
Viewing changes to dmedia/tests/test_startup.py
- browse files at revision 459
- compare with another revision
- download diff
- download tarball
- view history from revision 459
- Committer: Jason Gerard DeRose
- Date: 2012-10-01 07:11:00 UTC
- mfrom: (457.1.69 peering)
- Revision ID: jderose@novacut.com-20121001071100-iachj5q1o1pbrtic
Unified web server for file transfer & CouchDB proxy [Part 1]
- files added:
- .bzrignore
- doc
- doc/_static
- doc/_templates
added
removed
dmedia/tests/test_startup.py
1
# dmedia: distributed media library
2
# Copyright (C) 2012 Novacut Inc
3
#
4
# This file is part of `dmedia`.
5
#
6
# `dmedia` is free software: you can redistribute it and/or modify it under
7
# the terms of the GNU Affero General Public License as published by the Free
8
# Software Foundation, either version 3 of the License, or (at your option) any
9
# later version.
10
#
11
# `dmedia` is distributed in the hope that it will be useful, but WITHOUT ANY
12
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
13
# A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
14
# details.
15
#
16
# You should have received a copy of the GNU Affero General Public License along
17
# with `dmedia`. If not, see <http://www.gnu.org/licenses/>.
18
#
19
# Authors:
20
# Jason Gerard DeRose <jderose@novacut.com>
21
22
"""
23
Unit tests for `dmedia.startup`.
24
"""
25
26
from unittest import TestCase
27
import os
28
from os import path
29
import json
30
from copy import deepcopy
31
32
import usercouch
33
from microfiber import random_id
34
35
from .base import TempDir
36
from dmedia.peering import PKI
37
from dmedia import startup
38
39
40
class TestFunctions(TestCase):
41
def test_load_config(self):
42
tmp = TempDir()
43
filename = tmp.join('foo.json')
44
45
with self.assertRaises(IOError) as cm:
46
startup.load_config(filename)
47
48
config = {'junk': random_id()}
49
json.dump(config, open(filename, 'w'))
50
self.assertEqual(startup.load_config(filename), config)
51
52
open(filename, 'w').write('bad json, not treat for you')
53
with self.assertRaises(ValueError) as cm:
54
startup.load_config(filename)
55
56
def test_save_config(self):
57
tmp = TempDir()
58
config = {'stuff': random_id()}
59
filename = tmp.join('foo.json')
60
61
with self.assertRaises(TypeError) as cm:
62
startup.save_config(filename, object)
63
self.assertFalse(path.exists(filename))
64
self.assertTrue(path.isfile(filename + '.tmp'))
65
66
startup.save_config(filename, config)
67
self.assertTrue(path.isfile(filename))
68
self.assertFalse(path.exists(filename + '.tmp'))
69
self.assertEqual(json.load(open(filename, 'r')), config)
70
71
def test_get_usercouch(self):
72
tmp = TempDir()
73
couch = startup.get_usercouch(tmp.dir)
74
self.assertIsInstance(couch, usercouch.UserCouch)
75
self.assertEqual(couch.basedir, tmp.dir)
76
self.assertIsNone(couch.couchdb)
77
self.assertIsInstance(couch.pki, PKI)
78
self.assertIs(couch.pki.ssldir, couch.paths.ssl)
79
80
# Test lockfile
81
with self.assertRaises(usercouch.LockError) as cm:
82
couch2 = startup.get_usercouch(tmp.dir)
83
self.assertEqual(cm.exception.lockfile, tmp.join('lockfile'))
84
85
def test_machine_filename(self):
86
tmp = TempDir()
87
couch = startup.get_usercouch(tmp.dir)
88
self.assertEqual(
89
startup.machine_filename(couch),
90
path.join(tmp.dir, 'machine.json')
91
)
92
93
def test_user_filename(self):
94
tmp = TempDir()
95
couch = startup.get_usercouch(tmp.dir)
96
self.assertEqual(
97
startup.user_filename(couch),
98
path.join(tmp.dir, 'user.json')
99
)
100
101
def test_has_machine(self):
102
tmp = TempDir()
103
couch = startup.get_usercouch(tmp.dir)
104
self.assertFalse(startup.has_machine(couch))
105
tmp.touch('machine.json')
106
self.assertTrue(startup.has_machine(couch))
107
108
def test_has_user(self):
109
tmp = TempDir()
110
couch = startup.get_usercouch(tmp.dir)
111
self.assertFalse(startup.has_user(couch))
112
tmp.touch('user.json')
113
self.assertTrue(startup.has_user(couch))
114
115
def test_init_machine(self):
116
tmp = TempDir()
117
couch = startup.get_usercouch(tmp.dir)
118
self.assertIsNone(startup.init_machine(couch))
119
doc = json.load(open(tmp.join('machine.json'), 'r'))
120
self.assertIsInstance(doc, dict)
121
122
def test_init_user(self):
123
tmp = TempDir()
124
couch = startup.get_usercouch(tmp.dir)
125
machine_id = random_id(25)
126
self.assertIsNone(startup.init_user(couch, machine_id))
127
doc = json.load(open(tmp.join('user.json'), 'r'))
128
self.assertIsInstance(doc, dict)
129
130
def test_bootstrap_args(self):
131
tmp = TempDir()
132
couch = startup.get_usercouch(tmp.dir)
133
machine_id = random_id(25)
134
user_id = random_id(25)
135
cert_id = random_id(25)
136
oauth = usercouch.random_oauth()
137
user = {
138
'_id': user_id,
139
'oauth': deepcopy(oauth),
140
'certs': {machine_id: cert_id},
141
}
142
(auth, config) = startup.bootstrap_args(couch, machine_id, user)
143
self.assertEqual(auth, 'oauth')
144
self.assertEqual(config,
145
{
146
'username': 'admin',
147
'bind_address': '0.0.0.0',
148
'oauth': oauth,
149
'ssl': {
150
'key_file': tmp.join('ssl', cert_id + '.key'),
151
'cert_file': tmp.join('ssl', cert_id + '.cert'),
152
},
153
'replicator': {
154
'ca_file': tmp.join('ssl', user_id + '.cert'),
155
},
156
}
157
)
158
self.assertEqual(
159
startup.bootstrap_args(couch, machine_id, None),
160
('basic', {'username': 'admin'})
161
)
162
Loggerhead is a web-based interface for Breezy
Version: 2.0.1