Made SyncManager's methods static

boards/management/commands/sync_with_server.py

@@ -1,59 +1,59 b''
 import re
 import urllib.parse
 import httplib2
 import xml.etree.ElementTree as ET
 
 from django.core.management import BaseCommand
 from boards.models import GlobalId
 from boards.models.post.sync import SyncManager
 
 __author__ = 'neko259'
 
 
 REGEX_GLOBAL_ID = re.compile(r'(\w+)::([\w\+/]+)::(\d+)')
 
 
 class Command(BaseCommand):
     help = 'Send a sync or get request to the server.' + \
            'sync_with_server <server_url> [post_global_id]'
 
     def add_arguments(self, parser):
         parser.add_argument('url', type=str)
         parser.add_argument('global_id', type=str)
 
     def handle(self, *args, **options):
         url = options.get('url')
         global_id_str = options.get('global_id')
         if global_id_str:
             match = REGEX_GLOBAL_ID.match(global_id_str)
             if match:
                 key_type = match.group(1)
                 key = match.group(2)
                 local_id = match.group(3)
 
                 global_id = GlobalId(key_type=key_type, key=key,
                                      local_id=local_id)
 
                 xml = GlobalId.objects.generate_request_get([global_id])
                 # body = urllib.parse.urlencode(data)
                 h = httplib2.Http()
                 response, content = h.request(url, method="POST", body=xml)
 
-                SyncManager().parse_response_get(content)
+                SyncManager.parse_response_get(content)
             else:
                 raise Exception('Invalid global ID')
         else:
             h = httplib2.Http()
             response, content = h.request(url, method="POST")
 
             print(content)
 
             root = ET.fromstring(content)
             status = root.findall('status')[0].text
             if status == 'success':
                 models = root.findall('models')[0]
                 for model in models:
                     model_content = model[0]
                     print(model_content.findall('text')[0].text)
             else:
                 raise Exception('Invalid response status')

boards/models/post/sync.py

@@ -1,128 +1,129 b''
 import xml.etree.ElementTree as et
 from django.db import transaction
 from boards.models import KeyPair, GlobalId, Signature, Post
 
 ENCODING_UNICODE = 'unicode'
 
 TAG_MODEL = 'model'
 TAG_REQUEST = 'request'
 TAG_RESPONSE = 'response'
 TAG_ID = 'id'
 TAG_STATUS = 'status'
 TAG_MODELS = 'models'
 TAG_TITLE = 'title'
 TAG_TEXT = 'text'
 TAG_THREAD = 'thread'
 TAG_PUB_TIME = 'pub-time'
 TAG_SIGNATURES = 'signatures'
 TAG_SIGNATURE = 'signature'
 TAG_CONTENT = 'content'
 TAG_ATTACHMENTS = 'attachments'
 TAG_ATTACHMENT = 'attachment'
 
 TYPE_GET = 'get'
 
 ATTR_VERSION = 'version'
 ATTR_TYPE = 'type'
 ATTR_NAME = 'name'
 ATTR_VALUE = 'value'
 ATTR_MIMETYPE = 'mimetype'
 
 STATUS_SUCCESS = 'success'
 
 
-# TODO Make this fully static
 class SyncManager:
-    def generate_response_get(self, model_list: list):
+    @staticmethod
+    def generate_response_get(model_list: list):
         response = et.Element(TAG_RESPONSE)
 
         status = et.SubElement(response, TAG_STATUS)
         status.text = STATUS_SUCCESS
 
         models = et.SubElement(response, TAG_MODELS)
 
         for post in model_list:
             model = et.SubElement(models, TAG_MODEL)
             model.set(ATTR_NAME, 'post')
 
             content_tag = et.SubElement(model, TAG_CONTENT)
 
             tag_id = et.SubElement(content_tag, TAG_ID)
             post.global_id.to_xml_element(tag_id)
 
             title = et.SubElement(content_tag, TAG_TITLE)
             title.text = post.title
 
             text = et.SubElement(content_tag, TAG_TEXT)
             text.text = post.get_sync_text()
 
             if not post.is_opening():
                 thread = et.SubElement(content_tag, TAG_THREAD)
                 thread_id = et.SubElement(thread, TAG_ID)
                 post.get_thread().get_opening_post().global_id.to_xml_element(thread_id)
             else:
                 # TODO Output tags here
                 pass
 
             pub_time = et.SubElement(content_tag, TAG_PUB_TIME)
             pub_time.text = str(post.get_pub_time_str())
 
             signatures_tag = et.SubElement(model, TAG_SIGNATURES)
             post_signatures = post.signature.all()
             if post_signatures:
                 signatures = post.signatures
             else:
                 # TODO Maybe the signature can be computed only once after
                 # the post is added? Need to add some on_save signal queue
                 # and add this there.
                 key = KeyPair.objects.get(public_key=post.global_id.key)
                 signatures = [Signature(
                     key_type=key.key_type,
                     key=key.public_key,
                     signature=key.sign(et.tostring(model, ENCODING_UNICODE)),
                 )]
             for signature in signatures:
                 signature_tag = et.SubElement(signatures_tag, TAG_SIGNATURE)
                 signature_tag.set(ATTR_TYPE, signature.key_type)
                 signature_tag.set(ATTR_VALUE, signature.signature)
 
         return et.tostring(response, ENCODING_UNICODE)
 
+    @staticmethod
     @transaction.atomic
-    def parse_response_get(self, response_xml):
+    def parse_response_get(response_xml):
         tag_root = et.fromstring(response_xml)
         tag_status = tag_root.find(TAG_STATUS)
         if STATUS_SUCCESS == tag_status.text:
             tag_models = tag_root.find(TAG_MODELS)
             for tag_model in tag_models:
                 tag_content = tag_model.find(TAG_CONTENT)
                 tag_id = tag_content.find(TAG_ID)
                 global_id, exists = GlobalId.from_xml_element(tag_id)
 
                 if exists:
                     print('Post with same ID already exists')
                 else:
                     global_id.save()
 
                     title = tag_content.find(TAG_TITLE).text
                     text = tag_content.find(TAG_TEXT).text
                     pub_time = tag_content.find(TAG_PUB_TIME).text
 
                     thread = tag_content.find(TAG_THREAD)
                     if thread:
                         opening_post = Post.objects.get(
                             id=thread.find(TAG_ID).text)
                     else:
                         opening_post = None
                         # TODO Get tags here
 
                     # TODO Check that the replied posts are already present
                     # before adding new ones
 
                     post = Post.objects.import_post(
                         title=title, text=text, pub_time=pub_time,
                         opening_post=opening_post)
                     post.global_id = global_id
         else:
             # TODO Throw an exception?
             pass

boards/tests/test_keys.py

@@ -1,87 +1,87 b''
 from base64 import b64encode
 import logging
 
 from django.test import TestCase
 from boards.models import KeyPair, GlobalId, Post
 from boards.models.post.sync import SyncManager
 
 logger = logging.getLogger(__name__)
 
 
 class KeyTest(TestCase):
     def test_create_key(self):
         key = KeyPair.objects.generate_key('ecdsa')
 
         self.assertIsNotNone(key, 'The key was not created.')
 
     def test_validation(self):
         key = KeyPair.objects.generate_key(key_type='ecdsa')
         message = 'msg'
         signature = key.sign(message)
         valid = KeyPair.objects.verify(key.public_key, message, signature,
                                        key_type='ecdsa')
 
         self.assertTrue(valid, 'Message verification failed.')
 
     def test_primary_constraint(self):
         KeyPair.objects.generate_key(key_type='ecdsa', primary=True)
 
         with self.assertRaises(Exception):
             KeyPair.objects.generate_key(key_type='ecdsa', primary=True)
 
     def test_model_id_save(self):
         model_id = GlobalId(key_type='test', key='test key', local_id='1')
         model_id.save()
 
     def test_request_get(self):
         post = self._create_post_with_key()
 
         request = GlobalId.objects.generate_request_get([post.global_id])
         logger.debug(request)
 
         key = KeyPair.objects.get(primary=True)
         self.assertTrue('<request type="get" version="1.0">'
                         '<model name="post" version="1.0">'
                         '<id key="%s" local-id="1" type="%s" />'
                         '</model>'
                         '</request>' % (
                             key.public_key,
                             key.key_type,
                         ) in request,
                         'Wrong XML generated for the GET request.')
 
     def test_response_get(self):
         post = self._create_post_with_key()
         reply_post = Post.objects.create_post(title='test_title',
                                               text='[post]%d[/post]' % post.id,
                                               thread=post.get_thread())
 
-        response = SyncManager().generate_response_get([reply_post])
+        response = SyncManager.generate_response_get([reply_post])
         logger.debug(response)
 
         key = KeyPair.objects.get(primary=True)
         self.assertTrue('<status>success</status>'
                         '<models>'
                         '<model name="post">'
                         '<content>'
                         '<id key="%s" local-id="%d" type="%s" />'
                         '<title>test_title</title>'
                         '<text>[post]%s[/post]</text>'
                         '<thread><id key="%s" local-id="%d" type="%s" /></thread>'
                         '<pub-time>%s</pub-time>'
                         '</content>' % (
                             key.public_key,
                             reply_post.id,
                             key.key_type,
                             str(post.global_id),
                             key.public_key,
                             post.id,
                             key.key_type,
                             str(reply_post.get_pub_time_str()),
                         ) in response,
                         'Wrong XML generated for the GET response.')
 
     def _create_post_with_key(self):
         KeyPair.objects.generate_key(primary=True)
 
         return Post.objects.create_post(title='test_title', text='test_text')

boards/tests/test_sync.py

@@ -1,60 +1,60 b''
 from boards.models import KeyPair, Post
 from boards.models.post.sync import SyncManager
 from boards.tests.mocks import MockRequest
 from boards.views.sync import response_get
 
 __author__ = 'neko259'
 
 
 from django.test import TestCase
 
 
 class SyncTest(TestCase):
     def test_get(self):
         """
         Forms a GET request of a post and checks the response.
         """
 
         KeyPair.objects.generate_key(primary=True)
         post = Post.objects.create_post(title='test_title', text='test_text')
 
         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>'
             '<pub-time>%s</pub-time>'
             '</content>' % (
                 post.global_id.key,
                 post.id,
                 post.global_id.key_type,
                 post.title,
                 post.get_raw_text(),
                 post.get_pub_time_str(),
             ) in response_get(request).content.decode(),
             'Wrong response generated for the GET request.')
 
         post.delete()
 
-        SyncManager().parse_response_get(response)
+        SyncManager.parse_response_get(response)
         self.assertEqual(1, Post.objects.count(),
                          'Post was not created from XML response.')
 
-        SyncManager().parse_response_get(response)
+        SyncManager.parse_response_get(response)
         self.assertEqual(1, Post.objects.count(),
                          'The same post was imported twice.')

boards/views/api.py

@@ -1,258 +1,258 b''
 import json
 import logging
 
 import xml.etree.ElementTree as ET
 
 from django.db import transaction
 from django.http import HttpResponse
 from django.shortcuts import get_object_or_404
 from django.core import serializers
 
 from boards.forms import PostForm, PlainErrorList
 from boards.models import Post, Thread, Tag, GlobalId
 from boards.models.post.sync import SyncManager
 from boards.utils import datetime_to_epoch
 from boards.views.thread import ThreadView
 from boards.models.user import Notification
 from boards.mdx_neboard import Parser
 
 
 __author__ = 'neko259'
 
 PARAMETER_TRUNCATED = 'truncated'
 PARAMETER_TAG = 'tag'
 PARAMETER_OFFSET = 'offset'
 PARAMETER_DIFF_TYPE = 'type'
 PARAMETER_POST = 'post'
 PARAMETER_UPDATED = 'updated'
 PARAMETER_LAST_UPDATE = 'last_update'
 PARAMETER_THREAD = 'thread'
 PARAMETER_UIDS = 'uids'
 
 DIFF_TYPE_HTML = 'html'
 DIFF_TYPE_JSON = 'json'
 
 STATUS_OK = 'ok'
 STATUS_ERROR = 'error'
 
 logger = logging.getLogger(__name__)
 
 
 @transaction.atomic
 def api_get_threaddiff(request):
     """
     Gets posts that were changed or added since time
     """
 
     thread_id = request.POST.get(PARAMETER_THREAD)
     uids_str = request.POST.get(PARAMETER_UIDS).strip()
     uids = uids_str.split(' ')
 
     thread = get_object_or_404(Post, id=thread_id).get_thread()
 
     json_data = {
         PARAMETER_UPDATED: [],
         PARAMETER_LAST_UPDATE: None,  # TODO Maybe this can be removed already?
     }
     posts = Post.objects.filter(threads__in=[thread]).exclude(uid__in=uids)
 
     diff_type = request.GET.get(PARAMETER_DIFF_TYPE, DIFF_TYPE_HTML)
 
     for post in posts:
         json_data[PARAMETER_UPDATED].append(get_post_data(post.id, diff_type,
             request))
     json_data[PARAMETER_LAST_UPDATE] = str(thread.last_edit_time)
 
     return HttpResponse(content=json.dumps(json_data))
 
 
 def api_add_post(request, opening_post_id):
     """
     Adds a post and return the JSON response for it
     """
 
     opening_post = get_object_or_404(Post, id=opening_post_id)
 
     logger.info('Adding post via api...')
 
     status = STATUS_OK
     errors = []
 
     if request.method == 'POST':
         form = PostForm(request.POST, request.FILES, error_class=PlainErrorList)
         form.session = request.session
 
         if form.need_to_ban:
             # Ban user because he is suspected to be a bot
             # _ban_current_user(request)
             status = STATUS_ERROR
         if form.is_valid():
             post = ThreadView().new_post(request, form, opening_post,
                                          html_response=False)
             if not post:
                 status = STATUS_ERROR
             else:
                 logger.info('Added post #%d via api.' % post.id)
         else:
             status = STATUS_ERROR
             errors = form.as_json_errors()
 
     response = {
         'status': status,
         'errors': errors,
     }
 
     return HttpResponse(content=json.dumps(response))
 
 
 def get_post(request, post_id):
     """
     Gets the html of a post. Used for popups. Post can be truncated if used
     in threads list with 'truncated' get parameter.
     """
 
     post = get_object_or_404(Post, id=post_id)
     truncated = PARAMETER_TRUNCATED in request.GET
 
     return HttpResponse(content=post.get_view(truncated=truncated))
 
 
 def api_get_threads(request, count):
     """
     Gets the JSON thread opening posts list.
     Parameters that can be used for filtering:
     tag, offset (from which thread to get results)
     """
 
     if PARAMETER_TAG in request.GET:
         tag_name = request.GET[PARAMETER_TAG]
         if tag_name is not None:
             tag = get_object_or_404(Tag, name=tag_name)
             threads = tag.get_threads().filter(archived=False)
     else: 
         threads = Thread.objects.filter(archived=False)
 
     if PARAMETER_OFFSET in request.GET:
         offset = request.GET[PARAMETER_OFFSET]
         offset = int(offset) if offset is not None else 0
     else:
         offset = 0
 
     threads = threads.order_by('-bump_time')
     threads = threads[offset:offset + int(count)]
 
     opening_posts = []
     for thread in threads:
         opening_post = thread.get_opening_post()
 
         # TODO Add tags, replies and images count
         post_data = get_post_data(opening_post.id, include_last_update=True)
         post_data['bumpable'] = thread.can_bump()
         post_data['archived'] = thread.archived
 
         opening_posts.append(post_data)
 
     return HttpResponse(content=json.dumps(opening_posts))
 
 
 # TODO Test this
 def api_get_tags(request):
     """
     Gets all tags or user tags.
     """
 
     # TODO Get favorite tags for the given user ID
 
     tags = Tag.objects.get_not_empty_tags()
 
     term = request.GET.get('term')
     if term is not None:
         tags = tags.filter(name__contains=term)
 
     tag_names = [tag.name for tag in tags]
 
     return HttpResponse(content=json.dumps(tag_names))
 
 
 # TODO The result can be cached by the thread last update time
 # TODO Test this
 def api_get_thread_posts(request, opening_post_id):
     """
     Gets the JSON array of thread posts
     """
 
     opening_post = get_object_or_404(Post, id=opening_post_id)
     thread = opening_post.get_thread()
     posts = thread.get_replies()
 
     json_data = {
         'posts': [],
         'last_update': None,
     }
     json_post_list = []
 
     for post in posts:
         json_post_list.append(get_post_data(post.id))
     json_data['last_update'] = datetime_to_epoch(thread.last_edit_time)
     json_data['posts'] = json_post_list
 
     return HttpResponse(content=json.dumps(json_data))
 
 
 def api_get_notifications(request, username):
     last_notification_id_str = request.GET.get('last', None)
     last_id = int(last_notification_id_str) if last_notification_id_str is not None else None
 
     posts = Notification.objects.get_notification_posts(username=username,
                                                         last=last_id)
 
     json_post_list = []
     for post in posts:
         json_post_list.append(get_post_data(post.id))
     return HttpResponse(content=json.dumps(json_post_list))
 
 
 def api_get_post(request, post_id):
     """
     Gets the JSON of a post. This can be
     used as and API for external clients.
     """
 
     post = get_object_or_404(Post, id=post_id)
 
     json = serializers.serialize("json", [post], fields=(
         "pub_time", "_text_rendered", "title", "text", "image",
         "image_width", "image_height", "replies", "tags"
     ))
 
     return HttpResponse(content=json)
 
 
 # TODO Remove this method and use post method directly
 def get_post_data(post_id, format_type=DIFF_TYPE_JSON, request=None,
                   include_last_update=False):
     post = get_object_or_404(Post, id=post_id)
     return post.get_post_data(format_type=format_type, request=request,
                               include_last_update=include_last_update)
 
 
 def api_get_preview(request):
     raw_text = request.POST['raw_text']
 
     parser = Parser()
     return HttpResponse(content=parser.parse(parser.preparse(raw_text)))
 
 
 # TODO Make a separate module for sync API methods
 def sync_pull(request):
     """
     Return 'pull' request response for all posts.
     """
     request_xml = request.get('xml')
     if request_xml is None:
         posts = Post.objects.all()
     else:
         pass  # TODO Parse the XML and get filters from it
 
-    xml = SyncManager().generate_response_get(posts)
+    xml = SyncManager.generate_response_get(posts)
     return HttpResponse(content=xml)

boards/views/sync.py

@@ -1,49 +1,49 b''
 import xml.etree.ElementTree as et
 from django.http import HttpResponse, Http404
 from boards.models import GlobalId, Post
 from boards.models.post.sync import SyncManager
 
 
 def response_pull(request):
     pass
 
 
 def response_get(request):
     """
     Processes a GET request with post ID list and returns the posts XML list.
     Request should contain an 'xml' post attribute with the actual request XML.
     """
 
     request_xml = request.body
 
     if request_xml is None:
         return HttpResponse(content='Use the API')
 
     posts = []
 
     root_tag = et.fromstring(request_xml)
     model_tag = root_tag[0]
     for id_tag in model_tag:
         global_id, exists = GlobalId.from_xml_element(id_tag)
         if exists:
             posts.append(Post.objects.get(global_id=global_id))
 
-    response_xml = SyncManager().generate_response_get(posts)
+    response_xml = SyncManager.generate_response_get(posts)
 
     return HttpResponse(content=response_xml)
 
 
 def get_post_sync_data(request, post_id):
     try:
         post = Post.objects.get(id=post_id)
     except Post.DoesNotExist:
         raise Http404()
 
     content = 'Global ID: %s\n\nXML: %s' \
-              % (post.global_id, SyncManager().generate_response_get([post]))
+              % (post.global_id, SyncManager.generate_response_get([post]))
 
 
     return HttpResponse(
         content_type='text/plain',
         content=content,
     )
  No newline at end of file