improve testing and swap a couple of urls

api/drf_permissions.py

@@ -26,14 +26,9 @@ def has_permission(self, request, view):
         return request.user.profile.can_see_incoming
 
 
-class CanSeeContactNums(permissions.BasePermission):
-    def has_permission(self, request, view):
-        return request.user.profile.can_see_contact_nums
-
-
 class CanSeeKeyword(permissions.BasePermission):
     def has_object_permission(self, request, view, obj):
-        if not obj.is_locked:
+        if not obj.is_locked():
             return True
 
         return obj.can_user_access(request.user)

api/urls.py

@@ -35,6 +35,12 @@
             permission_classes=(IsAuthenticated, CanSeeIncoming)
         ),
         name='all_live_wall'),
+    url(r'^v1/sms/live_wall/only_live/in/$',
+        ApiCollectionAllWall.as_view(
+            permission_classes=(IsAuthenticated, CanSeeIncoming),
+            only_live=True
+        ),
+        name='all_live_wall_only_live'),
     url(r'^v1/sms/in/recpient/(?P<pk>\d+)/$',
         ApiCollectionRecentSms.as_view(
             permission_classes=(IsAuthenticated, CanSeeContactNames, CanSeeIncoming)),
@@ -50,6 +56,12 @@
             permission_classes=(IsAuthenticated, CanSeeKeywords, CanSeeKeyword, CanSeeIncoming),
         ),
         name='keyword_live_wall'),
+    url(r'^v1/sms/live_wall/only_live/in/keyword/(?P<pk>\d+)/$',
+        ApiCollectionKeywordWall.as_view(
+            permission_classes=(IsAuthenticated, CanSeeKeywords, CanSeeKeyword, CanSeeIncoming),
+            only_live=True,
+        ),
+        name='keyword_live_wall_only_live'),
     url(r'^v1/sms/in/keyword/(?P<pk>\d+)/archive/$',
         ApiCollectionKeywordSms.as_view(
             permission_classes=(IsAuthenticated, CanSeeKeywords, CanSeeKeyword, CanSeeIncoming),

api/views.py

@@ -81,7 +81,11 @@ class ApiCollectionKeywordSms(APIView):
     archive = False
 
     def get(self, request, format=None, **kwargs):
-        objs = SmsInbound.objects.filter(matched_keyword=str(Keyword.objects.get(pk=kwargs['pk'])))
+        keyword_obj = Keyword.objects.get(pk=kwargs['pk'])
+        self.check_object_permissions(request, keyword_obj)
+        objs = SmsInbound.objects.filter(
+            matched_keyword=str(keyword_obj)
+        )
         if self.archive:
             objs = objs.filter(is_archived=True)
         else:
@@ -95,18 +99,24 @@ class ApiCollectionKeywordWall(APIView):
     only_live = False
 
     def get(self, request, format=None, **kwargs):
+        keyword_obj = Keyword.objects.get(pk=kwargs['pk'])
+        self.check_object_permissions(request, keyword_obj)
         # check cache
         if self.only_live:
             objs = cache.get('keyword_{}_only_live'.format(kwargs['pk']))
             if objs is None:
-                objs = SmsInbound.objects.filter(matched_keyword=str(Keyword.objects.get(pk=kwargs['pk'])))
+                objs = SmsInbound.objects.filter(
+                    matched_keyword=str(keyword_obj)
+                )
                 objs = objs.filter(is_archived=False)
                 objs = objs.filter(display_on_wall=True)
                 cache.set('keyword_{}_only_live'.format(kwargs['pk']), objs, 120)
         else:
             objs = cache.get('keyword_{}_all'.format(kwargs['pk']))
             if objs is None:
-                objs = SmsInbound.objects.filter(matched_keyword=str(Keyword.objects.get(pk=kwargs['pk'])))
+                objs = SmsInbound.objects.filter(
+                    matched_keyword=str(keyword_obj)
+                )
                 objs = objs.filter(is_archived=False)
                 cache.set('keyword_{}_all'.format(kwargs['pk']), objs, 120)
 

apostello/tasks.py

@@ -212,7 +212,7 @@ def send_keyword_digest():
                     'Daily update for "{0}" responses'.format(
                         str(keyword)
                     ),
-                    "The following text messages have been received today:\n\n{0]".format(
+                    "The following text messages have been received today:\n\n{0}".format(
                         "\n".join([str(x) for x in new_responses])
                     ),
                     [subscriber.email]

apostello/templates/apostello/wall.html

@@ -21,15 +21,15 @@
 <script>
 React.render(
      React.createElement(ResponseWall,
-     {pollInterval: 500, curating:false, preview:false, url: "{%url 'api:keyword_live_wall' keyword.id %}"}),
+     {pollInterval: 500, curating:false, preview:false, url: "{%url 'api:keyword_live_wall_only_live' keyword.id %}"}),
      document.getElementById('content')
 );
 </script>
 {% else %}
 <script>
 React.render(
      React.createElement(ResponseWall,
-     {pollInterval: 500, curating:false, url: "{%url 'api:all_live_wall'%}", preview: false}),
+     {pollInterval: 500, curating:false, url: "{%url 'api:all_live_wall_only_live'%}", preview: false}),
      document.getElementById('content')
 );
 </script>

apostello/tests/conftest.py

@@ -1,21 +1,16 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime
 
+import pytest
+from allauth.account.models import EmailAddress
 from django.contrib.auth.models import User
 from django.test import Client
 from django.utils import timezone
 from django.utils.timezone import get_current_timezone
 
-import pytest
-from allauth.account.models import EmailAddress
 from apostello.models import *
 
 
-@pytest.fixture()
-def no_requests(monkeypatch):
-    monkeypatch.delattr("requests.sessions.Session.request")
-
-
 @pytest.fixture
 def recipients():
     calvin = Recipient.objects.create(first_name="John",

apostello/tests/test_management.py

@@ -1,14 +1,16 @@
 # -*- coding: utf-8 -*-
-from django.core.management import call_command
-
 import pytest
+from django.core.management import call_command
 
 
 @pytest.mark.django_db
 class TestManagementCommands():
 
-    def test_import(self, recipients, smsin):
+    def test_update_sms_name_fields(self, recipients, smsin):
         call_command('update_sms_name_fields')
 
-    def test_management(self):
+    def test_import_in(self):
         call_command('import_incoming_sms')
+
+    def test_import_out(self):
+        call_command('import_outgoing_sms')

apostello/tests/test_models.py

@@ -3,18 +3,18 @@
 from django.contrib.auth.models import User
 from django.utils import timezone
 
-from ..models import SmsOutbound
+from .. import models
 
 
 @pytest.mark.django_db
 class TestSms:
     def test_create_sms(self, recipients, groups):
-        sms = SmsOutbound(sid='dummy_sid',
-                          content='test',
-                          time_sent=timezone.now(),
-                          sent_by='test',
-                          recipient_group=groups['test_group'],
-                          recipient=recipients['calvin'])
+        sms = models.SmsOutbound(sid='dummy_sid',
+                                 content='test',
+                                 time_sent=timezone.now(),
+                                 sent_by='test',
+                                 recipient_group=groups['test_group'],
+                                 recipient=recipients['calvin'])
         assert 'test' == str(sms)
 
     def test_reimport_sms(self, smsin):
@@ -28,3 +28,18 @@ def test_display(self):
                                                    email='test3@example.com',
                                                    password='top_secret')
         assert "Profile: test_staff" == str(user_staff.profile)
+
+
+@pytest.mark.django_db
+class TestSiteConfig:
+    def test_display(self):
+        assert "Site Configuration" == str(models.SiteConfiguration.get_solo())
+
+
+@pytest.mark.django_db
+class TestDefaultResponses:
+    def test_fetch_default_reply_length(self):
+        assert 160 == models.fetch_default_resp_length()
+
+    def test_display(self):
+        assert "Default Responses" == str(models.DefaultResponses.get_solo())

apostello/tests/test_sms_freq.py

@@ -0,0 +1,14 @@
+import pytest
+
+from graphs.sms_freq import sms_graph_data
+
+
+@pytest.mark.django_db
+class TestSmsFreq:
+    def test_sms_freq_in(self, smsin):
+        graph_data = sms_graph_data(direction='in')
+        assert 3 in graph_data
+
+    def test_sms_freq_out(self, smsout):
+        graph_data = sms_graph_data(direction='out')
+        assert 1 in graph_data

apostello/tests/test_tasks.py

@@ -63,6 +63,8 @@ def test_send_keyword_digest(self, keywords, smsin, users):
                                         matched_keyword="test",
                                         sid='123fasdfdfaw45')
         sms.save()
+        keywords['test'].subscribed_to_digest.add(users['staff'])
+        keywords['test'].save()
         send_keyword_digest()
 
     def test_log_msg_in(self, recipients):
@@ -73,3 +75,6 @@ def test_log_msg_in(self, recipients):
         log_msg_in(p, datetime.now(), calvin.pk)
 
         assert SmsInbound.objects.filter(content="New test message").count() == 1
+
+    def test_warn_on_blacklist_receipt(self, recipients):
+        warn_on_blacklist_receipt(recipients['wesley'].pk, 'stop')

apostello/tests/test_views_twilio.py → apostello/tests/test_twilio_view.py

@@ -27,12 +27,16 @@ def _compute_signature(self, path, params):
             self.twilio_auth_token
         ).compute_signature(urljoin(self.base_url, path), params=params)
 
-    def get(self, path, data={}, **extra):
+    def get(self, path, data=None, **extra):
+        if data is None:
+            data = {}
         if 'HTTP_X_TWILIO_SIGNATURE' not in extra:
             extra.update({'HTTP_X_TWILIO_SIGNATURE': self._compute_signature(path, params=data)})
         return super(TwilioRequestFactory, self).get(path, data, **extra)
 
-    def post(self, path, data={}, content_type=None, **extra):
+    def post(self, path, data=None, content_type=None, **extra):
+        if data is None:
+            data = {}
         if 'HTTP_X_TWILIO_SIGNATURE' not in extra:
             extra.update({'HTTP_X_TWILIO_SIGNATURE': self._compute_signature(path, params=data)})
         if content_type is None: