test_sync.py
253 lines
| 9.2 KiB
| text/x-python
|
PythonLexer
neko259
|
r1586 | from django.test import TestCase | ||
neko259
|
r1239 | from boards.models import KeyPair, Post, Tag | ||
neko259
|
r1229 | from boards.models.post.sync import SyncManager | ||
neko259
|
r836 | from boards.tests.mocks import MockRequest | ||
neko259
|
r1834 | from boards.views.sync import response_get, response_list | ||
neko259
|
r836 | |||
__author__ = 'neko259' | ||||
class SyncTest(TestCase): | ||||
def test_get(self): | ||||
""" | ||||
Forms a GET request of a post and checks the response. | ||||
""" | ||||
neko259
|
r1503 | key = KeyPair.objects.generate_key(primary=True) | ||
neko259
|
r1895 | tag, created = Tag.objects.get_or_create_with_alias(name='tag1') | ||
neko259
|
r1504 | post = Post.objects.create_post(title='test_title', | ||
text='test_text\rline two', | ||||
neko259
|
r1239 | tags=[tag]) | ||
neko259
|
r836 | |||
request = MockRequest() | ||||
neko259
|
r1227 | request.body = ( | ||
neko259
|
r836 | '<request type="get" version="1.0">' | ||
'<model name="post" version="1.0">' | ||||
'<id key="%s" local-id="%d" type="%s" />' | ||||
'</model>' | ||||
'</request>' % (post.global_id.key, | ||||
post.id, | ||||
post.global_id.key_type) | ||||
) | ||||
neko259
|
r1229 | response = response_get(request).content.decode() | ||
neko259
|
r836 | self.assertTrue( | ||
'<status>success</status>' | ||||
'<models>' | ||||
neko259
|
r838 | '<model name="post">' | ||
'<content>' | ||||
neko259
|
r1928 | '<id key="{}" local-id="{}" type="{}" />' | ||
'<title>{}</title>' | ||||
'<text>{}</text>' | ||||
'<tags><tag>{}</tag></tags>' | ||||
'<pub-time>{}</pub-time>' | ||||
'<update-time>{}</update-time>' | ||||
'</content>'.format( | ||||
neko259
|
r836 | post.global_id.key, | ||
neko259
|
r1503 | post.global_id.local_id, | ||
neko259
|
r836 | post.global_id.key_type, | ||
post.title, | ||||
neko259
|
r1504 | post.get_sync_text(), | ||
neko259
|
r1895 | post.get_thread().get_tags().first().get_name(), | ||
neko259
|
r1229 | post.get_pub_time_str(), | ||
neko259
|
r1928 | post.last_edit_time, | ||
neko259
|
r1503 | ) in response, | ||
neko259
|
r935 | 'Wrong response generated for the GET request.') | ||
neko259
|
r1229 | |||
post.delete() | ||||
neko259
|
r1503 | key.delete() | ||
KeyPair.objects.generate_key(primary=True) | ||||
neko259
|
r1229 | |||
neko259
|
r1511 | SyncManager.parse_response_get(response, None) | ||
neko259
|
r1229 | self.assertEqual(1, Post.objects.count(), | ||
'Post was not created from XML response.') | ||||
neko259
|
r1232 | |||
neko259
|
r1239 | parsed_post = Post.objects.first() | ||
self.assertEqual('tag1', | ||||
neko259
|
r1895 | parsed_post.get_thread().get_tags().first().get_name(), | ||
neko259
|
r1239 | 'Invalid tag was parsed.') | ||
neko259
|
r1511 | SyncManager.parse_response_get(response, None) | ||
neko259
|
r1232 | self.assertEqual(1, Post.objects.count(), | ||
'The same post was imported twice.') | ||||
neko259
|
r1244 | |||
self.assertEqual(1, parsed_post.global_id.signature_set.count(), | ||||
'Signature was not saved.') | ||||
neko259
|
r1503 | |||
post = parsed_post | ||||
# Trying to sync the same once more | ||||
response = response_get(request).content.decode() | ||||
self.assertTrue( | ||||
'<status>success</status>' | ||||
'<models>' | ||||
'<model name="post">' | ||||
'<content>' | ||||
neko259
|
r1928 | '<id key="{}" local-id="{}" type="{}" />' | ||
'<title>{}</title>' | ||||
'<text>{}</text>' | ||||
'<tags><tag>{}</tag></tags>' | ||||
'<pub-time>{}</pub-time>' | ||||
'<update-time>{}</update-time>' | ||||
'</content>'.format( | ||||
neko259
|
r1503 | post.global_id.key, | ||
post.global_id.local_id, | ||||
post.global_id.key_type, | ||||
post.title, | ||||
neko259
|
r1504 | post.get_sync_text(), | ||
neko259
|
r1895 | post.get_thread().get_tags().first().get_name(), | ||
neko259
|
r1503 | post.get_pub_time_str(), | ||
neko259
|
r2104 | post.last_edit_time, | ||
neko259
|
r1503 | ) in response, | ||
'Wrong response generated for the GET request.') | ||||
neko259
|
r1834 | |||
def test_list_all(self): | ||||
key = KeyPair.objects.generate_key(primary=True) | ||||
neko259
|
r1895 | tag, created = Tag.objects.get_or_create_with_alias(name='tag1') | ||
neko259
|
r1834 | post = Post.objects.create_post(title='test_title', | ||
text='test_text\rline two', | ||||
tags=[tag]) | ||||
post2 = Post.objects.create_post(title='test title 2', | ||||
text='test text 2', | ||||
tags=[tag]) | ||||
request_all = MockRequest() | ||||
request_all.body = ( | ||||
'<request type="list" version="1.0">' | ||||
'<model name="post" version="1.0">' | ||||
'</model>' | ||||
'</request>' | ||||
) | ||||
response_all = response_list(request_all).content.decode() | ||||
self.assertTrue( | ||||
'<status>success</status>' | ||||
'<models>' | ||||
'<model>' | ||||
'<id key="{}" local-id="{}" type="{}" />' | ||||
neko259
|
r1928 | '<update-time>{}</update-time>' | ||
neko259
|
r1834 | '</model>' | ||
'<model>' | ||||
'<id key="{}" local-id="{}" type="{}" />' | ||||
neko259
|
r1928 | '<update-time>{}</update-time>' | ||
neko259
|
r1834 | '</model>' | ||
'</models>'.format( | ||||
post.global_id.key, | ||||
post.global_id.local_id, | ||||
post.global_id.key_type, | ||||
neko259
|
r1928 | post.last_edit_time, | ||
neko259
|
r1834 | post2.global_id.key, | ||
post2.global_id.local_id, | ||||
post2.global_id.key_type, | ||||
neko259
|
r1928 | post2.last_edit_time, | ||
neko259
|
r1834 | ) in response_all, | ||
'Wrong response generated for the LIST request for all posts.') | ||||
def test_list_existing_thread(self): | ||||
key = KeyPair.objects.generate_key(primary=True) | ||||
neko259
|
r1895 | tag, created = Tag.objects.get_or_create_with_alias(name='tag1') | ||
neko259
|
r1834 | post = Post.objects.create_post(title='test_title', | ||
text='test_text\rline two', | ||||
tags=[tag]) | ||||
post2 = Post.objects.create_post(title='test title 2', | ||||
text='test text 2', | ||||
tags=[tag]) | ||||
request_thread = MockRequest() | ||||
request_thread.body = ( | ||||
'<request type="list" version="1.0">' | ||||
'<model name="post" version="1.0">' | ||||
'<thread>{}</thread>' | ||||
'</model>' | ||||
'</request>'.format( | ||||
post.id, | ||||
) | ||||
) | ||||
response_thread = response_list(request_thread).content.decode() | ||||
self.assertTrue( | ||||
'<status>success</status>' | ||||
'<models>' | ||||
'<model>' | ||||
'<id key="{}" local-id="{}" type="{}" />' | ||||
neko259
|
r1928 | '<update-time>{}</update-time>' | ||
neko259
|
r1834 | '</model>' | ||
'</models>'.format( | ||||
post.global_id.key, | ||||
post.global_id.local_id, | ||||
post.global_id.key_type, | ||||
neko259
|
r1928 | post.last_edit_time, | ||
neko259
|
r1834 | ) in response_thread, | ||
'Wrong response generated for the LIST request for posts of ' | ||||
'existing thread.') | ||||
def test_list_non_existing_thread(self): | ||||
key = KeyPair.objects.generate_key(primary=True) | ||||
neko259
|
r1895 | tag, created = Tag.objects.get_or_create_with_alias(name='tag1') | ||
neko259
|
r1834 | post = Post.objects.create_post(title='test_title', | ||
text='test_text\rline two', | ||||
tags=[tag]) | ||||
post2 = Post.objects.create_post(title='test title 2', | ||||
text='test text 2', | ||||
tags=[tag]) | ||||
request_thread = MockRequest() | ||||
request_thread.body = ( | ||||
'<request type="list" version="1.0">' | ||||
'<model name="post" version="1.0">' | ||||
'<thread>{}</thread>' | ||||
'</model>' | ||||
'</request>'.format( | ||||
0, | ||||
) | ||||
) | ||||
response_thread = response_list(request_thread).content.decode() | ||||
self.assertTrue( | ||||
'<status>success</status>' | ||||
'<models />' | ||||
in response_thread, | ||||
'Wrong response generated for the LIST request for posts of ' | ||||
'non-existing thread.') | ||||
neko259
|
r1846 | |||
def test_list_pub_time(self): | ||||
key = KeyPair.objects.generate_key(primary=True) | ||||
neko259
|
r1895 | tag, created = Tag.objects.get_or_create_with_alias(name='tag1') | ||
neko259
|
r1846 | post = Post.objects.create_post(title='test_title', | ||
text='test_text\rline two', | ||||
tags=[tag]) | ||||
post2 = Post.objects.create_post(title='test title 2', | ||||
text='test text 2', | ||||
tags=[tag]) | ||||
request_thread = MockRequest() | ||||
request_thread.body = ( | ||||
'<request type="list" version="1.0">' | ||||
'<model name="post" version="1.0">' | ||||
'<timestamp_from>{}</timestamp_from>' | ||||
'</model>' | ||||
'</request>'.format( | ||||
post.pub_time, | ||||
) | ||||
) | ||||
response_thread = response_list(request_thread).content.decode() | ||||
self.assertTrue( | ||||
'<status>success</status>' | ||||
'<models>' | ||||
'<model>' | ||||
'<id key="{}" local-id="{}" type="{}" />' | ||||
neko259
|
r1928 | '<update-time>{}</update_time>' | ||
neko259
|
r1846 | '</model>' | ||
'</models>'.format( | ||||
post2.global_id.key, | ||||
post2.global_id.local_id, | ||||
post2.global_id.key_type, | ||||
neko259
|
r1928 | post2.last_edit_time, | ||
neko259
|
r1846 | ) in response_thread, | ||
'Wrong response generated for the LIST request for posts of ' | ||||
'existing thread.') | ||||