--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/notes/tests/api/sync.py Mon Jul 24 16:58:34 2017 +0200
@@ -0,0 +1,233 @@
+"""
+Tests the sync api
+"""
+import datetime
+import logging
+
+from django.contrib.auth import get_user_model
+from django.contrib.contenttypes.models import ContentType
+from django.urls import reverse
+from django.utils import timezone
+from rest_framework import status
+from rest_framework.test import APITransactionTestCase
+
+from notes.models import Session, Note
+from auditlog.models import LogEntry
+
+logger = logging.getLogger(__name__)
+
+
+class SyncApiTests(APITransactionTestCase):
+ '''
+ Test Sync api
+ '''
+
+ def setUp(self):
+ User = get_user_model()
+ user1 = User.objects.create_user(
+ username='test_user1',
+ email='test_user@emial.com',
+ password='top_secret'
+ )
+ user2 = User.objects.create_user(
+ username='test_user2',
+ email='test_user@emial.com',
+ password='top_secret'
+ )
+ user3 = User.objects.create_user(
+ username='test_user3',
+ email='test_user@emial.com',
+ password='top_secret'
+ )
+
+ self.session1 = Session.objects.create(
+ title="a new session 1",
+ description="Description 1",
+ protocol="[]",
+ owner=user1
+ )
+
+ self.session2 = Session.objects.create(
+ title="a new session 2",
+ description="Description 2",
+ protocol="[]",
+ owner=user2
+ )
+
+ Session.objects.create(
+ title="a new session 3",
+ description="Description 3",
+ protocol="[]",
+ owner=user3
+ )
+
+ self.note1 = Note.objects.create(
+ tc_start=timezone.now(),
+ tc_end=timezone.now(),
+ session=self.session1,
+ plain="example note 1",
+ html="<i>example note 1</i>",
+ raw="<i>example note 1</i>",
+ margin_note="margin note 1",
+ categorization="[]"
+ )
+
+ self.note2 = Note.objects.create(
+ tc_start=timezone.now(),
+ tc_end=timezone.now(),
+ session=self.session1,
+ plain="example note 1.1",
+ html="<i>example note 1,1</i>",
+ raw="<i>example note 1.1</i>",
+ margin_note="margin note 1.1",
+ categorization="[]"
+ )
+
+
+ Note.objects.create(
+ tc_start=timezone.now(),
+ tc_end=timezone.now(),
+ session=self.session2,
+ plain="example note 2",
+ html="<i>example note</i>",
+ raw="<i>example note</i>",
+ margin_note="margin note",
+ categorization="[]"
+ )
+
+ def test_not_authenticated(self):
+ url = reverse('notes:sync-list')
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+ def test_simple(self):
+ url = reverse('notes:sync-list')
+ self.client.login(username='test_user1', password='top_secret')
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+ def test_bad_method(self):
+ url = reverse('notes:sync-list')
+ self.client.login(username='test_user1', password='top_secret')
+ response = self.client.post(url)
+ self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
+
+ def test_simple_output(self):
+ url = reverse('notes:sync-list')
+ self.client.login(username='test_user1', password='top_secret')
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ json_resp = response.json()
+ self.assertIn('sessions', json_resp)
+ self.assertEqual(1, len(json_resp['sessions']))
+ self.assertEqual('session', json_resp['sessions'][0]['type'])
+ self.assertEqual(0, json_resp['sessions'][0]['action'])
+ self.assertEqual(str(self.session1.ext_id), json_resp['sessions'][0]['ext_id'])
+
+ self.assertIn('notes', json_resp)
+ self.assertEqual(2, len(json_resp['notes']))
+ for sync_def in json_resp['notes']:
+ self.assertEqual('note', sync_def['type'])
+ self.assertEqual(0, sync_def['action'])
+ self.assertIn(sync_def['ext_id'],[str(self.note1.ext_id), str(self.note2.ext_id)])
+
+ def test_modified_since_empty(self):
+ url = reverse('notes:sync-list')
+ self.client.login(username='test_user1', password='top_secret')
+
+ nexthour = \
+ datetime.datetime.utcnow().replace(tzinfo=timezone.utc) +\
+ datetime.timedelta(hours=1)
+
+ response = self.client.get(
+ url,
+ {'modified_since': nexthour.timestamp()},
+ format='json'
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ json_resp = response.json()
+ self.assertIn('sessions', json_resp)
+ self.assertEqual(0, len(json_resp['sessions']))
+ self.assertIn('notes', json_resp)
+ self.assertEqual(0, len(json_resp['notes']))
+
+ def test_modified_since_notes(self):
+ url = reverse('notes:sync-list')
+ self.client.login(username='test_user1', password='top_secret')
+
+ nexthour = \
+ datetime.datetime.utcnow().replace(tzinfo=timezone.utc) +\
+ datetime.timedelta(hours=1)
+
+ LogEntry.objects.filter(content_type=ContentType.objects.get_for_model(Note)).update(timestamp=nexthour)
+
+ response = self.client.get(
+ url,
+ {'modified_since': (nexthour-datetime.timedelta(minutes=30)).timestamp()},
+ format='json'
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ json_resp = response.json()
+ self.assertIn('sessions', json_resp)
+ self.assertEqual(0, len(json_resp['sessions']))
+ self.assertIn('notes', json_resp)
+ self.assertEqual(2, len(json_resp['notes']))
+ for sync_def in json_resp['notes']:
+ self.assertEqual('note', sync_def['type'])
+ self.assertEqual(0, sync_def['action'])
+ self.assertIn(sync_def['ext_id'],[str(self.note1.ext_id), str(self.note2.ext_id)])
+
+
+ def test_modified_since_single_update(self):
+ self.note2.plain = "plain text modified"
+ self.note2.save()
+
+ url = reverse('notes:sync-list')
+ self.client.login(username='test_user1', password='top_secret')
+
+ nexthour = \
+ datetime.datetime.utcnow().replace(tzinfo=timezone.utc) +\
+ datetime.timedelta(hours=1)
+
+ LogEntry.objects.filter(
+ content_type=ContentType.objects.get_for_model(Note),
+ object_pk=self.note2.id,
+ action=LogEntry.Action.UPDATE
+ ).update(timestamp=nexthour)
+
+ response = self.client.get(
+ url,
+ {'modified_since': (nexthour-datetime.timedelta(minutes=30)).timestamp()},
+ format='json'
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ json_resp = response.json()
+ self.assertIn('sessions', json_resp)
+ self.assertEqual(0, len(json_resp['sessions']))
+ self.assertIn('notes', json_resp)
+ self.assertEqual(1, len(json_resp['notes']))
+ sync_def = json_resp['notes'][0]
+ self.assertEqual(LogEntry.Action.UPDATE, sync_def['action'])
+
+ def test_note_delete(self):
+ self.note2.delete()
+ url = reverse('notes:sync-list')
+ self.client.login(username='test_user1', password='top_secret')
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ json_resp = response.json()
+ self.assertIn('sessions', json_resp)
+ self.assertEqual(1, len(json_resp['sessions']))
+ self.assertEqual('session', json_resp['sessions'][0]['type'])
+ self.assertEqual(0, json_resp['sessions'][0]['action'])
+ self.assertEqual(str(self.session1.ext_id), json_resp['sessions'][0]['ext_id'])
+
+ self.assertIn('notes', json_resp)
+ self.assertEqual(1, len(json_resp['notes']))
+ sync_def = json_resp['notes'][0]
+ self.assertEqual('note', sync_def['type'])
+ self.assertEqual(0, sync_def['action'])
+ self.assertEqual(sync_def['ext_id'],str(self.note1.ext_id))