arno/isso
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add comment.hash to recognize user by email or ip fallback

Browse Source 
also: fixed test_comment json.dumps(json.loads(json.dumps(...)))
madness.
This commit is contained in:
Martin Zimmermann 2013-09-05 23:09:56 +02:00
parent 35926037a6
commit 59706815e7
4 changed files with 60 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
isso/db.py
View File

@ -88,7 +88,7 @@ class SQLite(Abstract):
sql = ('main.comments (path VARCHAR(255) NOT NULL, id INTEGER NOT NULL,'
'created FLOAT NOT NULL, modified FLOAT, text VARCHAR,'
'author VARCHAR(64), email VARCHAR(64), website VARCHAR(64),'
'parent INTEGER, mode INTEGER, PRIMARY KEY (id, path))')
'parent INTEGER, mode INTEGER, hash CHAR(32), PRIMARY KEY (id, path))')
con.execute("CREATE TABLE IF NOT EXISTS %s;" % sql)
# increment id if (id, path) is no longer unique
@ -105,8 +105,8 @@ class SQLite(Abstract):
return None
return Comment(
text=query[4], author=query[5], email=query[6], website=query[7], parent=query[8],
path=query[0], id=query[1], created=query[2], modified=query[3], mode=query[9],
text=query[4], author=query[5], hash=query[6], website=query[7], parent=query[8],
path=query[0], id=query[1], created=query[2], modified=query[3], mode=query[9]
)
def add(self, path, c):
@ -114,7 +114,7 @@ class SQLite(Abstract):
keys = ','.join(self.fields)
values = ','.join('?' * len(self.fields))
con.execute('INSERT INTO comments (%s) VALUES (%s);' % (keys, values), (
path, 0, c.created, c.modified, c.text, c.author, c.email, c.website,
path, 0, c.created, c.modified, c.text, c.author, c.hash, c.website,
c.parent, self.mode)
)

18
isso/models.py
View File

@ -7,6 +7,9 @@ import json
import time
import hashlib
aluhut = lambda ip: hashlib.sha1(ip + '\x082@t9*\x17\xad\xc1\x1c\xa5\x98').hexdigest()
class Comment(object):
"""This class represents a regular comment. It needs at least a text
field, all other fields are optional (or automatically set by the
@ -22,13 +25,13 @@ class Comment(object):
normal and queued using MODE=3.
"""
protected = ['path', 'id', 'mode', 'created', 'modified']
fields = ['text', 'author', 'email', 'website', 'parent']
protected = ['path', 'id', 'mode', 'created', 'modified', 'hash']
fields = ['text', 'author', 'website', 'parent']
def __init__(self, **kw):
for field in self.protected + self.fields:
self.__dict__[field] = kw.get(field)
setattr(self, field, kw.get(field))
def iteritems(self, protected=True):
for field in self.fields:
@ -38,10 +41,15 @@ class Comment(object):
yield field, getattr(self, field)
@classmethod
def fromjson(self, data):
def fromjson(self, data, ip='127.0.0.1'):
if '.' in ip:
ip = ip.rsplit('.', 1)[0] + '.0'
data = json.loads(data)
comment = Comment(created=time.time())
comment = Comment(
created=time.time(),
hash=hashlib.md5(data.get('email', aluhut(ip))).hexdigest())
for field in self.fields:
if field == 'text' and field not in data:

4
isso/views/comment.py
View File

@ -43,11 +43,11 @@ def create(app, environ, request, uri):
return Response('URI does not exist', 400)
try:
comment = models.Comment.fromjson(request.data)
comment = models.Comment.fromjson(request.data, ip=request.remote_addr)
except ValueError as e:
return Response(unicode(e), 400)
for attr in 'author', 'email', 'website':
for attr in 'author', 'website':
if getattr(comment, attr) is not None:
try:
setattr(comment, attr, cgi.escape(getattr(comment, attr)))

59
specs/test_comment.py
View File

@ -11,25 +11,33 @@ from isso import Isso
from isso.models import Comment
def comment(**kw):
return Comment.fromjson(Isso.dumps(kw))
class FakeIP(object):
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
environ['REMOTE_ADDR'] = '192.168.1.1'
return self.app(environ, start_response)
class TestComments(unittest.TestCase):
def setUp(self):
fd, self.path = tempfile.mkstemp()
self.app = Isso(self.path, '...', '...', 15*60, "...")
self.app.wsgi_app = FakeIP(self.app.wsgi_app)
self.client = Client(self.app, Response)
self.get = lambda *x, **z: self.client.get(*x, **z)
self.put = lambda *x, **z: self.client.put(*x, **z)
self.post = lambda *x, **z: self.client.post(*x, **z)
self.delete = lambda *x, **z: self.client.delete(*x, **z)
self.get = self.client.get
self.put = self.client.put
self.post = self.client.post
self.delete = self.client.delete
def testGet(self):
self.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='Lorem ipsum ...')))
self.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'Lorem ipsum ...'}))
r = self.get('/?uri=%2Fpath%2F&id=1')
assert r.status_code == 200
@ -40,7 +48,7 @@ class TestComments(unittest.TestCase):
def testCreate(self):
rv = self.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='Lorem ipsum ...')))
rv = self.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'Lorem ipsum ...'}))
assert rv.status_code == 201
assert len(filter(lambda header: header[0] == 'Set-Cookie', rv.headers)) == 1
@ -54,7 +62,7 @@ class TestComments(unittest.TestCase):
def testCreateAndGetMultiple(self):
for i in range(20):
self.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='Spam')))
self.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'Spam'}))
r = self.get('/?uri=%2Fpath%2F')
assert r.status_code == 200
@ -70,9 +78,9 @@ class TestComments(unittest.TestCase):
def testUpdate(self):
self.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='Lorem ipsum ...')))
self.put('/?uri=%2Fpath%2F&id=1', data=Isso.dumps(comment(
text='Hello World', author='me', website='http://example.com/')))
self.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'Lorem ipsum ...'}))
self.put('/?uri=%2Fpath%2F&id=1', data=json.dumps({
'text': 'Hello World', 'author': 'me', 'website': 'http://example.com/'}))
r = self.get('/?uri=%2Fpath%2F&id=1&plain=1')
assert r.status_code == 200
@ -85,7 +93,7 @@ class TestComments(unittest.TestCase):
def testDelete(self):
self.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='Lorem ipsum ...')))
self.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'Lorem ipsum ...'}))
r = self.delete('/?uri=%2Fpath%2F&id=1')
assert r.status_code == 200
assert json.loads(r.data) == None
@ -94,8 +102,8 @@ class TestComments(unittest.TestCase):
def testDeleteWithReference(self):
client = Client(self.app, Response)
client.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='First')))
client.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='First', parent=1)))
client.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'First'}))
client.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'First', 'parent': 1}))
r = client.delete('/?uri=%2Fpath%2F&id=1')
assert r.status_code == 200
@ -110,7 +118,7 @@ class TestComments(unittest.TestCase):
for path in paths:
assert self.post('/new?' + urllib.urlencode({'uri': path}),
data=Isso.dumps(comment(text='...'))).status_code == 201
data=json.dumps({'text': '...'})).status_code == 201
for path in paths:
assert self.get('/?' + urllib.urlencode({'uri': path})).status_code == 200
@ -119,11 +127,26 @@ class TestComments(unittest.TestCase):
def testDeleteAndCreateByDifferentUsersButSamePostId(self):
mallory = Client(self.app, Response)
mallory.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='Foo')))
mallory.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'Foo'}))
mallory.delete('/?uri=%2Fpath%2F&id=1')
bob = Client(self.app, Response)
bob.post('/new?uri=%2Fpath%2F', data=Isso.dumps(comment(text='Bar')))
bob.post('/new?uri=%2Fpath%2F', data=json.dumps({'text': 'Bar'}))
assert mallory.delete('/?uri=%2Fpath%2F&id=1').status_code == 403
assert bob.delete('/?uri=%2Fpath%2F&id=1').status_code == 200
def testHash(self):
a = self.post('/new?uri=%2Fpath%2F', data=json.dumps({"text": "Aaa"}))
b = self.post('/new?uri=%2Fpath%2F', data=json.dumps({"text": "Bbb"}))
c = self.post('/new?uri=%2Fpath%2F', data=json.dumps({"text": "Ccc", "email": "..."}))
assert a.status_code == b.status_code == c.status_code == 201
a = json.loads(a.data)
b = json.loads(b.data)
c = json.loads(c.data)
assert a['hash'] != '192.168.1.1'
assert a['hash'] == b['hash']
assert a['hash'] != c['hash']