Browse Source

don't run some tests twice due to bad subclassing... add support

to save/restore the state of the server..
main
John-Mark Gurney 5 years ago
parent
commit
bf544391fa
1 changed files with 51 additions and 17 deletions
  1. +51
    -17
      ui/server.py

+ 51
- 17
ui/server.py View File

@@ -14,7 +14,7 @@ from klein import Klein
from kleintest import * from kleintest import *
from twisted.trial import unittest from twisted.trial import unittest
from twisted.web.iweb import IRequest from twisted.web.iweb import IRequest
from cli import Persona, MDBase
from cli import _asn1coder, Persona, MDBase


import hashlib import hashlib
import mock import mock
@@ -26,8 +26,21 @@ import uuid
defaultfile = 'mediaserver.store.pasn1' defaultfile = 'mediaserver.store.pasn1'
class MEDAServer: class MEDAServer:
def __init__(self, fname): def __init__(self, fname):
self._trustedkeys = {}
self._objstore = {}
self._fname = fname
try:
data = _asn1coder.loads(open(fname).read())
self._trustedkeys = {}
for i in data['trustedkeys']:
self.addpubkey(i)
if data['objstore']:
self._objstore = { uuid.UUID(bytes=k):
map(MDBase.create_obj, v) for k, v in
data['objstore'].iteritems() }
else:
self._objstore = {}
except IOError:
self._trustedkeys = {}
self._objstore = {}


app = Klein() app = Klein()


@@ -37,7 +50,13 @@ class MEDAServer:
self._trustedkeys[persona.uuid] = persona self._trustedkeys[persona.uuid] = persona


def store(self): def store(self):
pass
obj = {
'trustedkeys': [ i.get_pubkey() for i in self._trustedkeys.itervalues() ],
'objstore': self._objstore,
}

with open(self._fname, 'w') as fp:
fp.write(_asn1coder.dumps(obj))


@app.route('/obj/<id>') @app.route('/obj/<id>')
def obj_lookup(self, request, id): def obj_lookup(self, request, id):
@@ -97,7 +116,7 @@ def main():
if __name__ == '__main__': # pragma: no cover if __name__ == '__main__': # pragma: no cover
main() main()


class _TestCases(unittest.TestCase):
class _BaseServerTest(unittest.TestCase):
def setUp(self): def setUp(self):
d = os.path.realpath(tempfile.mkdtemp()) d = os.path.realpath(tempfile.mkdtemp())
self.basetempdir = d self.basetempdir = d
@@ -109,11 +128,7 @@ class _TestCases(unittest.TestCase):
shutil.rmtree(self.basetempdir) shutil.rmtree(self.basetempdir)
self.basetempdir = None self.basetempdir = None


def test_404(self):
h = hashlib.sha256(open('fixtures/testfiles/test.txt').read()).hexdigest()
r = self.requests.get('/chash/%s' % h)
self.assertEqual(r.status_code, 404)

class _TestCases(_BaseServerTest):
def test_objlookup(self): def test_objlookup(self):
# that when fetching an non-existant object # that when fetching an non-existant object
r = self.requests.get('/obj/%s' % str(uuid.uuid4())) r = self.requests.get('/obj/%s' % str(uuid.uuid4()))
@@ -128,10 +143,8 @@ class _TestCases(unittest.TestCase):
self.assertEqual(r.status_code, 400) self.assertEqual(r.status_code, 400)


def test_pubkeystorage(self): def test_pubkeystorage(self):
import cli

# that an identity # that an identity
persona = cli.Persona()
persona = Persona()
persona.generate_key() persona.generate_key()


# that by default, put's # that by default, put's
@@ -171,13 +184,19 @@ class _TestCases(unittest.TestCase):
# that when stored # that when stored
self.medaserver.store() self.medaserver.store()


tmpmedaserver = MEDAServer(self.medaserverfile)
# and restarted
self.medaserver = MEDAServer(self.medaserverfile)
self.requests = FakeRequests(self.medaserver.app)

# that fetching a previously successful object
r = self.requests.get('/obj/%s' % str(persona.uuid))

# it is successful
self.assertEqual(r.status_code, 200)


@mock.patch('klein.Klein.run') @mock.patch('klein.Klein.run')
def test_addpubkey(self, apprun): def test_addpubkey(self, apprun):
import cli

persona = cli.Persona()
persona = Persona()
persona.generate_key() persona.generate_key()


with nested(mock.patch('server.MEDAServer.addpubkey'), with nested(mock.patch('server.MEDAServer.addpubkey'),
@@ -205,3 +224,18 @@ class _TestCases(unittest.TestCase):


kleinrun.assert_called() kleinrun.assert_called()
storefun.assert_called() storefun.assert_called()

class _TestPostConfig(_BaseServerTest):
def setUp(self):
_BaseServerTest.setUp(self)

persona = Persona()
persona.generate_key()

self.persona = persona
self.medaserver.addpubkey(persona.get_pubkey())

def test_hashlookup(self):
h = hashlib.sha256(open('fixtures/testfiles/test.txt').read()).hexdigest()
r = self.requests.get('/lookup/%s' % h)
self.assertEqual(r.status_code, 404)

Loading…
Cancel
Save