diff -r c653f49fabfb -r ba8bc0199464 src/notes/tests/api/sync.py --- /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="example note 1", + raw="example note 1", + 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="example note 1,1", + raw="example note 1.1", + 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="example note", + raw="example note", + 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))