1 """

     2 Tests the sync api

     3 """

     4 import datetime

     5 import logging

     6 

     7 from django.contrib.auth import get_user_model

     8 from django.contrib.contenttypes.models import ContentType

     9 from django.urls import reverse

    10 from django.utils import timezone

    11 from rest_framework import status

    12 from rest_framework.test import APITransactionTestCase

    13 

    14 from notes.models import Session, Note

    15 from auditlog.models import LogEntry

    16 

    17 logger = logging.getLogger(__name__)

    18 

    19 

    20 class SyncApiTests(APITransactionTestCase):

    21     '''

    22     Test Sync api

    23     '''

    24 

    25     def setUp(self):

    26         User = get_user_model()

    27         user1 = User.objects.create_user(

    28             username='test_user1',

    29             email='test_user@emial.com',

    30             password='top_secret'

    31         )

    32         user2 = User.objects.create_user(

    33             username='test_user2',

    34             email='test_user@emial.com',

    35             password='top_secret'

    36         )

    37         user3 = User.objects.create_user(

    38             username='test_user3',

    39             email='test_user@emial.com',

    40             password='top_secret'

    41         )

    42 

    43         self.session1 = Session.objects.create(

    44             title="a new session 1",

    45             description="Description 1",

    46             protocol="[]",

    47             owner=user1

    48         )

    49 

    50         self.session2 = Session.objects.create(

    51             title="a new session 2",

    52             description="Description 2",

    53             protocol="[]",

    54             owner=user2

    55         )

    56 

    57         Session.objects.create(

    58             title="a new session 3",

    59             description="Description 3",

    60             protocol="[]",

    61             owner=user3

    62         )

    63 

    64         self.note1 = Note.objects.create(

    65             tc_start=timezone.now(),

    66             tc_end=timezone.now(),

    67             session=self.session1,

    68             plain="example note 1",

    69             html="<i>example note 1</i>",

    70             raw="<i>example note 1</i>",

    71             margin_note="margin note 1",

    72             categorization="[]"

    73         )

    74 

    75         self.note2 = Note.objects.create(

    76             tc_start=timezone.now(),

    77             tc_end=timezone.now(),

    78             session=self.session1,

    79             plain="example note 1.1",

    80             html="<i>example note 1,1</i>",

    81             raw="<i>example note 1.1</i>",

    82             margin_note="margin note 1.1",

    83             categorization="[]"

    84         )

    85 

    86 

    87         Note.objects.create(

    88             tc_start=timezone.now(),

    89             tc_end=timezone.now(),

    90             session=self.session2,

    91             plain="example note 2",

    92             html="<i>example note</i>",

    93             raw="<i>example note</i>",

    94             margin_note="margin note",

    95             categorization="[]"

    96         )

    97 

    98     def test_not_authenticated(self):

    99         url = reverse('notes:sync-list')

   100         response = self.client.get(url)

   101         self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

   102 

   103     def test_simple(self):

   104         url = reverse('notes:sync-list')

   105         self.client.login(username='test_user1', password='top_secret')

   106         response = self.client.get(url)

   107         self.assertEqual(response.status_code, status.HTTP_200_OK)

   108 

   109     def test_bad_method(self):

   110         url = reverse('notes:sync-list')

   111         self.client.login(username='test_user1', password='top_secret')

   112         response = self.client.post(url)

   113         self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)

   114 

   115     def test_simple_output(self):

   116         url = reverse('notes:sync-list')

   117         self.client.login(username='test_user1', password='top_secret')

   118         response = self.client.get(url)

   119         self.assertEqual(response.status_code, status.HTTP_200_OK)

   120         json_resp = response.json()

   121         self.assertIn('sessions', json_resp)

   122         self.assertEqual(1, len(json_resp['sessions']))

   123         self.assertEqual('session', json_resp['sessions'][0]['type'])

   124         self.assertEqual(0, json_resp['sessions'][0]['action'])

   125         self.assertEqual(str(self.session1.ext_id), json_resp['sessions'][0]['ext_id'])

   126 

   127         self.assertIn('notes', json_resp)

   128         self.assertEqual(2, len(json_resp['notes']))

   129         for sync_def in json_resp['notes']:

   130             self.assertEqual('note', sync_def['type'])

   131             self.assertEqual(0, sync_def['action'])

   132             self.assertIn(sync_def['ext_id'],[str(self.note1.ext_id), str(self.note2.ext_id)])

   133 

   134     def test_modified_since_empty(self):

   135         url = reverse('notes:sync-list')

   136         self.client.login(username='test_user1', password='top_secret')

   137 

   138         nexthour = \

   139             datetime.datetime.utcnow().replace(tzinfo=timezone.utc) +\

   140             datetime.timedelta(hours=1)

   141 

   142         response = self.client.get(

   143             url,

   144             {'modified_since': nexthour.timestamp()},

   145             format='json'

   146         )

   147 

   148         self.assertEqual(response.status_code, status.HTTP_200_OK)

   149         json_resp = response.json()

   150         self.assertIn('sessions', json_resp)

   151         self.assertEqual(0, len(json_resp['sessions']))

   152         self.assertIn('notes', json_resp)

   153         self.assertEqual(0, len(json_resp['notes']))

   154 

   155     def test_modified_since_notes(self):

   156         url = reverse('notes:sync-list')

   157         self.client.login(username='test_user1', password='top_secret')

   158 

   159         nexthour = \

   160             datetime.datetime.utcnow().replace(tzinfo=timezone.utc) +\

   161             datetime.timedelta(hours=1)

   162 

   163         LogEntry.objects.filter(content_type=ContentType.objects.get_for_model(Note)).update(timestamp=nexthour)

   164 

   165         response = self.client.get(

   166             url,

   167             {'modified_since': (nexthour-datetime.timedelta(minutes=30)).timestamp()},

   168             format='json'

   169         )

   170 

   171         self.assertEqual(response.status_code, status.HTTP_200_OK)

   172         json_resp = response.json()

   173         self.assertIn('sessions', json_resp)

   174         self.assertEqual(0, len(json_resp['sessions']))

   175         self.assertIn('notes', json_resp)

   176         self.assertEqual(2, len(json_resp['notes']))

   177         for sync_def in json_resp['notes']:

   178             self.assertEqual('note', sync_def['type'])

   179             self.assertEqual(0, sync_def['action'])

   180             self.assertIn(sync_def['ext_id'],[str(self.note1.ext_id), str(self.note2.ext_id)])

   181 

   182 

   183     def test_modified_since_single_update(self):

   184         self.note2.plain = "plain text modified"

   185         self.note2.save()

   186 

   187         url = reverse('notes:sync-list')

   188         self.client.login(username='test_user1', password='top_secret')

   189 

   190         nexthour = \

   191             datetime.datetime.utcnow().replace(tzinfo=timezone.utc) +\

   192             datetime.timedelta(hours=1)

   193 

   194         LogEntry.objects.filter(

   195             content_type=ContentType.objects.get_for_model(Note),

   196             object_pk=self.note2.id,

   197             action=LogEntry.Action.UPDATE

   198         ).update(timestamp=nexthour)

   199 

   200         response = self.client.get(

   201             url,

   202             {'modified_since': (nexthour-datetime.timedelta(minutes=30)).timestamp()},

   203             format='json'

   204         )

   205 

   206         self.assertEqual(response.status_code, status.HTTP_200_OK)

   207         json_resp = response.json()

   208         self.assertIn('sessions', json_resp)

   209         self.assertEqual(0, len(json_resp['sessions']))

   210         self.assertIn('notes', json_resp)

   211         self.assertEqual(1, len(json_resp['notes']))

   212         sync_def = json_resp['notes'][0]

   213         self.assertEqual(LogEntry.Action.UPDATE, sync_def['action'])

   214 

   215     def test_note_delete(self):

   216         self.note2.delete()

   217         url = reverse('notes:sync-list')

   218         self.client.login(username='test_user1', password='top_secret')

   219         response = self.client.get(url)

   220         self.assertEqual(response.status_code, status.HTTP_200_OK)

   221         json_resp = response.json()

   222         self.assertIn('sessions', json_resp)

   223         self.assertEqual(1, len(json_resp['sessions']))

   224         self.assertEqual('session', json_resp['sessions'][0]['type'])

   225         self.assertEqual(0, json_resp['sessions'][0]['action'])

   226         self.assertEqual(str(self.session1.ext_id), json_resp['sessions'][0]['ext_id'])

   227 

   228         self.assertIn('notes', json_resp)

   229         self.assertEqual(1, len(json_resp['notes']))

   230         sync_def = json_resp['notes'][0]

   231         self.assertEqual('note', sync_def['type'])

   232         self.assertEqual(0, sync_def['action'])

   233         self.assertEqual(sync_def['ext_id'],str(self.note1.ext_id))