Refactor views

src/colab_superarchives/views.py

@@ -31,7 +31,6 @@ class ThreadView(View):
     http_method_names = [u'get', u'post']
 
     def get(self, request, mailinglist, thread_token):
-
         thread = get_object_or_404(Thread, subject_token=thread_token,
                                    mailinglist__name=mailinglist)
 
@@ -43,16 +42,10 @@ def get(self, request, mailinglist, thread_token):
         )
 
         if all_privates.count(thread.mailinglist.name):
-            if not request.user.is_authenticated():
+            if not request.user.is_authenticated() or \
+                not self.mailing_list_in_user_list(request.user,
+                                                   thread.mailinglist.name):
                 raise PermissionDenied
-            else:
-                user = User.objects.get(username=request.user)
-                emails = user.emails.values_list('address', flat=True)
-                lists_for_user = mailman.get_user_mailinglists(user)
-                listnames_for_user = mailman.extract_listname_from_list(
-                    lists_for_user)
-                if thread.mailinglist.name not in listnames_for_user:
-                    raise PermissionDenied
 
         thread.hit(request)
 
@@ -88,6 +81,16 @@ def get(self, request, mailinglist, thread_token):
 
         return render(request, 'message-thread.html', context)
 
+    def mailing_list_in_user_list(self, user_name, name_mailing_list):
+        user = User.objects.get(username=user_name)
+        lists_for_user = mailman.get_user_mailinglists(user)
+        listnames_for_user = mailman.extract_listname_from_list(
+            lists_for_user)
+        if name_mailing_list not in listnames_for_user:
+            return False
+        else:
+            return True
+
     def post(self, request, mailinglist, thread_token):
         try:
             thread = Thread.objects.get(subject_token=thread_token,
@@ -106,15 +109,10 @@ def post(self, request, mailinglist, thread_token):
         url = urlparse.urljoin(settings.MAILMAN_API_URL,
                                'sendmail/' + mailinglist)
 
-        error_msg = None
-        try:
-            resp = requests.post(url, data=data, timeout=2)
-        except requests.exceptions.ConnectionError:
-            resp = None
-            error_msg = _('Error trying to connect to Mailman API')
-        except requests.exceptions.Timeout:
-            resp = None
-            error_msg = _('Timeout trying to connect to Mailman API')
+        response_data = self.get_response_message_sent(url, data)
+
+        resp = response_data[0]
+        error_msg = response_data[1]
 
         if resp and resp.status_code == 200:
             messages.success(request, _(
@@ -125,19 +123,47 @@ def post(self, request, mailinglist, thread_token):
             ))
         else:
             if not error_msg:
-                if resp is not None:
-                    if resp.status_code == 400:
-                        error_msg = _('You cannot send an empty email')
-                    elif resp.status_code == 404:
-                        error_msg = _('Mailing list does not exist')
-                else:
-                    error_msg = \
-                        _('Unknown error trying to connect to Mailman API')
+                error_msg = self.return_error_message(resp)
 
             messages.error(request, error_msg)
 
         return self.get(request, mailinglist, thread_token)
 
+    def get_response_message_sent(self, url, data):
+        error_msg = None
+
+        try:
+            resp = requests.post(url, data=data, timeout=2)
+        except requests.exceptions.ConnectionError:
+            resp = None
+            error_msg = _('Error trying to connect to Mailman API')
+        except requests.exceptions.Timeout:
+            resp = None
+            error_msg = _('Timeout trying to connect to Mailman API')
+
+        response_data = []
+        response_data.append(resp)
+        response_data.append(error_msg)
+
+        return response_data
+
+    def return_error_message(self, response):
+        return_message = ""
+
+        if response is not None:
+            if response.status_code == 400:
+                return_message = _('You cannot send an empty email')
+            elif response.status_code == 404:
+                return_message = _('Mailing list does not exist')
+            else:
+                return_message = \
+                    _('Unknown error trying to connect to Mailman API')
+        else:
+            return_message = \
+                _('Unknown error trying to connect to Mailman API')
+
+        return return_message
+
 
 class ThreadDashboardView(ListView):
     http_method_names = ['get']
@@ -163,8 +189,7 @@ class EmailView(View):
     http_method_names = [u'head', u'get', u'post', u'delete', u'update']
 
     def get(self, request, key):
-        """Validate an email with the given key"""
-
+        """Validate an email with the given key."""
         try:
             email_val = EmailAddressValidation.objects.get(validation_key=key)
         except EmailAddressValidation.DoesNotExist:
@@ -198,8 +223,7 @@ def get(self, request, key):
 
     @method_decorator(login_required)
     def post(self, request, key):
-        """Create new email address that will wait for validation"""
-
+        """Create new email address that will wait for validation."""
         email = request.POST.get('email')
         user_id = request.POST.get('user')
         if not email:
@@ -219,7 +243,6 @@ def post(self, request, key):
     @method_decorator(login_required)
     def delete(self, request, key):
         """Remove an email address, validated or not."""
-
         request.DELETE = http.QueryDict(request.body)
         email_addr = request.DELETE.get('email')
         user_id = request.DELETE.get('user')
@@ -249,7 +272,6 @@ def delete(self, request, key):
     @method_decorator(login_required)
     def update(self, request, key):
         """Set an email address as primary address."""
-
         request.UPDATE = http.QueryDict(request.body)
 
         email_addr = request.UPDATE.get('email')

tests/test_mailinglist_view.py

@@ -109,9 +109,7 @@ def test_private_list_access_with_user_without_permission(self, mock):
            return_value=[{'listname': 'privatelist'}])
     def test_private_list_access_with_user_permission(self, mock):
         self.authenticate_user()
-        response = self.client.get(
-            '/archives/mailinglist/privatelist'
-        )
+        response = self.client.get('/archives/mailinglist/privatelist')
 
         self.assertEqual(200, response.status_code)
         self.assertEqual(len(response.context['thread_list']), 1)

tests/test_privatelist.py

@@ -1,10 +1,12 @@
 # -*- coding:utf-8 -*-
-import mock
-
+from mock import patch
+from mock import Mock
+import requests
+from requests.exceptions import Timeout
 from colab_superarchives.utils import mailman
 from django.test import TestCase, Client
 from colab_superarchives.widgets.dashboard_latest_threads import DashboardLatestThreadsWidget
-
+from colab_superarchives.views import ThreadView
 
 class ArchivesViewTest(TestCase):
 
@@ -17,11 +19,11 @@ def authenticate_user(self):
         self.client.login(username='johndoe', password='1234')
 
     def test_see_only_private_list_if_member(self):
-        mailman.get_user_mailinglists = mock.Mock(
+        mailman.get_user_mailinglists = Mock(
             return_value=[{'listname': 'privatelist'}])
-        mailman.extract_listname_from_list = mock.Mock(
+        mailman.extract_listname_from_list = Mock(
             return_value=['privatelist'])
-        mailman.list_users = mock.Mock(return_value="['johndoe@example.com']")
+        mailman.list_users = Mock(return_value="['johndoe@example.com']")
 
         self.authenticate_user()
         request = self.client.get('/archives/thread/')
@@ -41,9 +43,9 @@ def test_see_only_public_if_not_logged_in(self):
         self.assertEqual(1, len(list_data))
 
     def test_see_private_thread_in_dashboard_if_member(self):
-        mailman.get_user_mailinglists = mock.Mock(
+        mailman.get_user_mailinglists = Mock(
             return_value="[{'listname': 'privatelist'}]")
-        mailman.extract_listname_from_list = mock.Mock(
+        mailman.extract_listname_from_list = Mock(
             return_value="['privatelist']")
 
         self.authenticate_user()
@@ -72,3 +74,78 @@ def test_dont_see_private_threads_in_profile_if_logged_out(self):
         emails = request.context['emails']
 
         self.assertEqual(1, len(emails))
+
+class ThreadViewTest(TestCase):
+
+    fixtures = ['mailinglistdata.json']
+
+    def setUp(self):
+        self.client = Client()
+        self.thread_view = ThreadView()
+        self.data = {
+            'in_reply_to': 1,
+            'email_from': "test@test.com",
+            'name_from': "John Doe",
+            'subject': "Subject Test",
+            'body': "Email body test",
+        }
+        url = "http://localhost/"
+
+        self.response = requests.post(url, data=self.data, timeout=2)
+
+    def authenticate_user(self):
+        self.client.login(username='johndoe', password='1234')
+
+    def test_mailing_list_in_user_list(self):
+        self.authenticate_user()
+
+        user_name = "johndoe"
+        list_name = "privatelist"
+
+        list_in_user = self.thread_view.mailing_list_in_user_list(user_name,
+                                                                  list_name)
+        self.assertTrue(list_in_user)
+
+    def test_error_message_empty_email(self):
+        self.response.status_code = 400
+        error_message = self.thread_view.return_error_message(self.response)
+        self.assertEqual(error_message, 'You cannot send an empty email')
+
+    def test_error_message_mailing_list_not_exist(self):
+        self.response.status_code = 404
+        error_message = self.thread_view.return_error_message(self.response)
+        self.assertEqual(error_message, 'Mailing list does not exist')
+
+    def test_error_message_unknown_error(self):
+        self.response.status_code = 999999
+        error_message = self.thread_view.return_error_message(self.response)
+        self.assertEqual(error_message,
+                         'Unknown error trying to connect to Mailman API')
+
+    def test_error_message_with_response_none(self):
+        response = None
+        error_message = self.thread_view.return_error_message(response)
+        self.assertEqual(error_message,
+                         'Unknown error trying to connect to Mailman API')
+
+    def test_get_response_connection_error(self):
+        wrong_url = "http://error.url.not.exist"
+        return_message = self.thread_view.get_response_message_sent(
+            wrong_url, self.data)[1]
+        self.assertEqual(return_message,
+                         'Error trying to connect to Mailman API')
+
+    @patch('requests.post', side_effect=Timeout())
+    def test_get_response_timeout(self, mock):
+        any_url = "http://anything.com"
+        return_message = self.thread_view.get_response_message_sent(
+            any_url, self.data)[1]
+        self.assertEqual(return_message,
+                         'Timeout trying to connect to Mailman API')
+
+    @patch('requests.post', return_value=None)
+    def test_get_response_successfully(self, mock):
+        any_url = "http://anything.com"
+        return_message = self.thread_view.get_response_message_sent(
+            any_url, self.data)[1]
+        self.assertEqual(return_message, None)