|
|
from django.test import TestCase
|
|
|
|
|
|
from boards.models import KeyPair, Post, Tag
|
|
|
from boards.models.post.sync import SyncManager
|
|
|
from boards.tests.mocks import MockRequest
|
|
|
from boards.views.sync import response_get, response_list
|
|
|
|
|
|
__author__ = 'neko259'
|
|
|
|
|
|
|
|
|
class SyncTest(TestCase):
|
|
|
def test_get(self):
|
|
|
"""
|
|
|
Forms a GET request of a post and checks the response.
|
|
|
"""
|
|
|
|
|
|
key = KeyPair.objects.generate_key(primary=True)
|
|
|
tag, created = Tag.objects.get_or_create_with_alias(name='tag1')
|
|
|
post = Post.objects.create_post(title='test_title',
|
|
|
text='test_text\rline two',
|
|
|
tags=[tag])
|
|
|
|
|
|
request = MockRequest()
|
|
|
request.body = (
|
|
|
'<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)
|
|
|
)
|
|
|
|
|
|
response = response_get(request).content.decode()
|
|
|
self.assertTrue(
|
|
|
'<status>success</status>'
|
|
|
'<models>'
|
|
|
'<model name="post">'
|
|
|
'<content>'
|
|
|
'<id key="%s" local-id="%d" type="%s" />'
|
|
|
'<title>%s</title>'
|
|
|
'<text>%s</text>'
|
|
|
'<tags><tag>%s</tag></tags>'
|
|
|
'<pub-time>%s</pub-time>'
|
|
|
'<version>%s</version>'
|
|
|
'</content>' % (
|
|
|
post.global_id.key,
|
|
|
post.global_id.local_id,
|
|
|
post.global_id.key_type,
|
|
|
post.title,
|
|
|
post.get_sync_text(),
|
|
|
post.get_thread().get_tags().first().get_name(),
|
|
|
post.get_pub_time_str(),
|
|
|
post.version,
|
|
|
) in response,
|
|
|
'Wrong response generated for the GET request.')
|
|
|
|
|
|
post.delete()
|
|
|
key.delete()
|
|
|
|
|
|
KeyPair.objects.generate_key(primary=True)
|
|
|
|
|
|
SyncManager.parse_response_get(response, None)
|
|
|
self.assertEqual(1, Post.objects.count(),
|
|
|
'Post was not created from XML response.')
|
|
|
|
|
|
parsed_post = Post.objects.first()
|
|
|
self.assertEqual('tag1',
|
|
|
parsed_post.get_thread().get_tags().first().get_name(),
|
|
|
'Invalid tag was parsed.')
|
|
|
|
|
|
SyncManager.parse_response_get(response, None)
|
|
|
self.assertEqual(1, Post.objects.count(),
|
|
|
'The same post was imported twice.')
|
|
|
|
|
|
self.assertEqual(1, parsed_post.global_id.signature_set.count(),
|
|
|
'Signature was not saved.')
|
|
|
|
|
|
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>'
|
|
|
'<id key="%s" local-id="%d" type="%s" />'
|
|
|
'<title>%s</title>'
|
|
|
'<text>%s</text>'
|
|
|
'<tags><tag>%s</tag></tags>'
|
|
|
'<pub-time>%s</pub-time>'
|
|
|
'<version>%s</version>'
|
|
|
'</content>' % (
|
|
|
post.global_id.key,
|
|
|
post.global_id.local_id,
|
|
|
post.global_id.key_type,
|
|
|
post.title,
|
|
|
post.get_sync_text(),
|
|
|
post.get_thread().get_tags().first().get_name(),
|
|
|
post.get_pub_time_str(),
|
|
|
post.version,
|
|
|
) in response,
|
|
|
'Wrong response generated for the GET request.')
|
|
|
|
|
|
def test_list_all(self):
|
|
|
key = KeyPair.objects.generate_key(primary=True)
|
|
|
tag, created = Tag.objects.get_or_create_with_alias(name='tag1')
|
|
|
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="{}" />'
|
|
|
'<version>{}</version>'
|
|
|
'</model>'
|
|
|
'<model>'
|
|
|
'<id key="{}" local-id="{}" type="{}" />'
|
|
|
'<version>{}</version>'
|
|
|
'</model>'
|
|
|
'</models>'.format(
|
|
|
post.global_id.key,
|
|
|
post.global_id.local_id,
|
|
|
post.global_id.key_type,
|
|
|
post.version,
|
|
|
post2.global_id.key,
|
|
|
post2.global_id.local_id,
|
|
|
post2.global_id.key_type,
|
|
|
post2.version,
|
|
|
) 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)
|
|
|
tag, created = Tag.objects.get_or_create_with_alias(name='tag1')
|
|
|
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="{}" />'
|
|
|
'<version>{}</version>'
|
|
|
'</model>'
|
|
|
'</models>'.format(
|
|
|
post.global_id.key,
|
|
|
post.global_id.local_id,
|
|
|
post.global_id.key_type,
|
|
|
post.version,
|
|
|
) 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)
|
|
|
tag, created = Tag.objects.get_or_create_with_alias(name='tag1')
|
|
|
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.')
|
|
|
|
|
|
def test_list_pub_time(self):
|
|
|
key = KeyPair.objects.generate_key(primary=True)
|
|
|
tag, created = Tag.objects.get_or_create_with_alias(name='tag1')
|
|
|
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="{}" />'
|
|
|
'<version>{}</version>'
|
|
|
'</model>'
|
|
|
'</models>'.format(
|
|
|
post2.global_id.key,
|
|
|
post2.global_id.local_id,
|
|
|
post2.global_id.key_type,
|
|
|
post2.version,
|
|
|
) in response_thread,
|
|
|
'Wrong response generated for the LIST request for posts of '
|
|
|
'existing thread.')
|
|
|
|
|
|
|
