Coverage for src/postorius/views/list.py : 77%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
# -*- coding: utf-8 -*- # Copyright (C) 1998-2019 by the Free Software Foundation, Inc. # # This file is part of Postorius. # # Postorius is free software: you can redistribute it and/or modify it under # the terms of the GNU General Public License as published by the Free # Software Foundation, either version 3 of the License, or (at your option) # any later version. # # Postorius is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or # FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for # more details. # # You should have received a copy of the GNU General Public License along with # Postorius. If not, see <http://www.gnu.org/licenses/>.
list_moderator_required, list_owner_required, superuser_required) AlterMessagesForm, ArchiveSettingsForm, DigestSettingsForm, DMARCMitigationsForm, ListAddBanForm, ListAnonymousSubscribe, ListAutomaticResponsesForm, ListHeaderMatchForm, ListHeaderMatchFormset, ListIdentityForm, ListMassRemoval, ListMassSubscription, ListNew, ListSubscribe, ListSubscriptionPolicyForm, MemberForm, MemberModeration, MessageAcceptanceForm, MultipleChoiceForm, UserPreferences)
"""Who 'owns' the token returned from the registrar?"""
# List of allowed roles for the memberships. The string value matches the # exact value Core's REST API expects.
"""Prepare regex based query to search partial email addresses.
Core's `members/find` API allows searching for memberships based on regex. This methods prepares a valid regex to pass on the the REST API. """ else:
"""Handle GET for Member view.
This includes all the membership roles (self.allowed_roles). """ # If the role is misspelled, redirect to the default subscribers. return redirect('list_members', list_id, 'member')
context['query'], role=role, count=count, page=page)
find_method, request.GET.get('page', 1), request.GET.get('count', 25), paginator_class=MailmanPaginator) context['members'].paginator.count) 'No {}s were found matching the search.'.format(role)) else:
"""Handle POST for members. Unsubscribe all the members selected.""" form = MultipleChoiceForm(request.POST) if form.is_valid(): members = form.cleaned_data['choices'] for member in members: self.mailing_list.unsubscribe(member) messages.success( request, _('The selected members have been unsubscribed')) return redirect('list_members', self.mailing_list.list_id, role)
"""Handle POST for membership roles owner, moderator and non-member.
Add memberships if the form is valid otherwise redirect to list_members page with an error message. """ role=role, address=member_form.cleaned_data['email'], display_name=member_form.cleaned_data['display_name']) request, _('{0}s has been added with the role {1}s'.format( member_form.cleaned_data['email'], role))) except HTTPError as e: messages.error(request, e.msg) else:
"""Handle POST for list members page.
List members page have more than one forms, depending on the membership type.
- Regular subscribers have a MultipleChoiceForm, which returns a list of emails that needs to be unsubscribed from the MailingList. This is handled by :func:`self._member_post` method.
- Owners, moderators and non-members have a MemberForm which allows adding new members with the given roles. This is handled by :func:`self._non_member_post` method.
""" return redirect('list_members', list_id, 'member')
return self._member_post(request, role) else:
def list_member_options(request, list_id, email): except Mailman404Error: return render(request, template_name, {'nolists': 'true'}) (key, getattr(mm_member, key)) for key in MemberModeration.base_fields ]) preferences_form = UserPreferences( request.POST, preferences=member_prefs) if preferences_form.is_valid(): try: preferences_form.save() except HTTPError as e: messages.error(request, e.msg) else: messages.success(request, _("The member's preferences" " have been updated.")) return redirect('list_member_options', list_id, email) request.POST, initial=initial_moderation) messages.info( request, _("No change to the member's moderation.")) return redirect('list_member_options', list_id, email) # In general, it would be a very bad idea to loop over the # fields and try to set them one by one, However, # moderation form has only one field. except HTTPError as e: messages.error(request, e.msg) else: "settings have been updated.")) else: 'mm_member': mm_member, 'list': mm_list, 'preferences_form': preferences_form, 'moderation_form': moderation_form, })
"""Shows common list metrics. """
'user_subscribed': False, 'subscribed_address': None, 'public_archive': False, 'hyperkitty_enabled': False} 'hyperkitty' not in settings.INSTALLED_APPS): # avoid systematic test failure when HyperKitty is installed # (missing VCR request, see below). 'hyperkitty' in self.mailing_list.archivers and # noqa: W504 self.mailing_list.archivers['hyperkitty']): data['hyperkitty_enabled'] = True user=request.user, verified=True).order_by( "email").values_list("email", flat=True) else: else:
"""Change mailing list subscription """
def post(self, request, list_id): user=request.user, verified=True).order_by( "email").values_list("email", flat=True) # Find the currently subscribed email else: messages.error(request, _('You are already subscribed')) else: # Since the action is done via the web UI, no email # confirmation is needed. email, pre_confirmed=True) response.get('token_owner') == TokenOwner.moderator): # noqa: E501 messages.success( request, _('Your request to change the email for' ' this subscription was submitted and' ' is waiting for moderator approval.')) else: _('Subscription changed to %s') % email) else: messages.error(request, _('Something went wrong. Please try again.')) except HTTPError as e: messages.error(request, e.msg)
""" view name: `list_subscribe` """
def post(self, request, list_id): """ Subscribes an email address to a mailing list via POST and redirects to the `list_summary` view. """ user=request.user, verified=True).order_by( "email").values_list("email", flat=True) email, pre_verified=True, pre_confirmed=True) response.get('token_owner') == TokenOwner.moderator): request, _('Your subscription request has been' ' submitted and is waiting for moderator' ' approval.')) else: _('You are subscribed to %s.') % self.mailing_list.fqdn_listname) else: _('Something went wrong. Please try again.'))
""" view name: `list_anonymous_subscribe` """
""" Subscribes an email address to a mailing list via POST and redirects to the `list_summary` view. This view is used for unauthenticated users and asks Mailman core to verify the supplied email address. """ pre_confirmed=False) 'further instructions')) else: messages.error(request, _('Something went wrong. Please try again.')) except HTTPError as e: messages.error(request, e.msg)
"""Unsubscribe from a mailing list."""
def post(self, request, *args, **kwargs): email = request.POST['email'] try: self.mailing_list.unsubscribe(email) messages.success(request, _('%s has been unsubscribed' ' from this list.') % email) except ValueError as e: messages.error(request, e) return redirect('list_summary', self.mailing_list.list_id)
def list_mass_subscribe(request, list_id): # Parse the data to get the address and the display name display_name=display_name, pre_verified=True, pre_confirmed=True, pre_approved=True) request, _('The address %(address)s has been' ' subscribed to %(list)s.') % {'address': address, 'list': mailing_list.fqdn_listname}) except HTTPError as e: messages.error(request, e) except ValidationError: messages.error(request, _('The email address %s' ' is not valid.') % address) else: form = ListMassSubscription() {'form': form, 'list': mailing_list})
"""Class For Mass Removal"""
def get(self, request, *args, **kwargs): form = ListMassRemoval() return render(request, 'postorius/lists/mass_removal.html', {'form': form, 'list': self.mailing_list})
def post(self, request, *args, **kwargs): form = ListMassRemoval(request.POST) if not form.is_valid(): messages.error(request, _('Please fill out the form correctly.')) else: for address in form.cleaned_data['emails']: try: validate_email(address) self.mailing_list.unsubscribe(address.lower()) messages.success( request, _('The address %(address)s has been' ' unsubscribed from %(list)s.') % {'address': address, 'list': self.mailing_list.fqdn_listname}) except (HTTPError, ValueError) as e: messages.error(request, e) except ValidationError: messages.error(request, _('The email address %s' ' is not valid.') % address) return redirect('mass_removal', self.mailing_list.list_id)
for message_id in message_ids: action(message_id)
mailing_list = List.objects.get_or_404(fqdn_listname=list_id) if request.method == 'POST': form = MultipleChoiceForm(request.POST) if form.is_valid(): message_ids = form.cleaned_data['choices'] try: if 'accept' in request.POST: _perform_action(message_ids, mailing_list.accept_message) messages.success(request, _('The selected messages were accepted')) elif 'reject' in request.POST: _perform_action(message_ids, mailing_list.reject_message) messages.success(request, _('The selected messages were rejected')) elif 'discard' in request.POST: _perform_action(message_ids, mailing_list.discard_message) messages.success(request, _('The selected messages were discarded')) except HTTPError: messages.error(request, _('Message could not be found')) else: form = MultipleChoiceForm() held_messages = paginate( mailing_list.get_held_page, request.GET.get('page'), request.GET.get('count'), paginator_class=MailmanPaginator) context = { 'list': mailing_list, 'held_messages': held_messages, 'form': form, 'ACTION_CHOICES': ACTION_CHOICES, } return render(request, 'postorius/lists/held_messages.html', context)
def moderate_held_message(request, list_id): """Moderate one held message""" elif 'reject' in request.POST: mailing_list.reject_message(msg.request_id) messages.success(request, _('The message was rejected')) elif 'discard' in request.POST: mailing_list.discard_message(msg.request_id) messages.success(request, _('The message was discarded')) request, _('Moderation action for {} set to {}'.format( member, moderation_choices[moderation_choice]))) except ValueError as e: messages.error( request, _('Failed to set moderation action: {}'.format(e)))
def csv_view(request, list_id): """Export all the subscriber in csv """ mm_lists = List.objects.get_or_404(fqdn_listname=list_id)
response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = ( 'attachment; filename="Subscribers.csv"')
writer = csv.writer(response) if mm_lists: for i in mm_lists.members: writer.writerow([i.email])
return response
for style in styles['styles']] # Reorder to put the default at the beginning
""" Add a new mailing list. If the request to the function is a GET request an empty form for creating a new list will be displayed. If the request method is POST the form will be evaluated. If the form is considered correct the list gets created and otherwise the form with the data filled in before the last POST request is returned. The user must be logged in to create a new list. """ # grab domain mail_host=form.cleaned_data['mail_host']) # creating the list form.cleaned_data['listname'], style_name=form.cleaned_data['list_style']) form.cleaned_data['description'] list_id=mailing_list.list_id) # TODO catch correct Error class: # Right now, there is no good way to detect that this is a # duplicate mailing list request other than checking the # reason for 400 error. 'listname', _('Mailing List already exists.')) # Otherwise just render the generic error page. return render(request, 'postorius/errors/generic.html', {'error': e.msg}) else: else: form = ListNew(choosable_domains, choosable_styles, initial={ 'list_owner': request.user.email, 'advertised': True, })
"""Return unique lists from a list of mailing lists."""
def list_index_authenticated(request): """Index page for authenticated users.
Index page for authenticated users is slightly different than un-authenticated ones. Authenticated users will see all their memberships in the index page.
This view is not paginated and will show all the lists.
"""
# Get all the verified addresses of the user. user=request.user, verified=True).order_by( "email").values_list("email", flat=True)
# Get all the mailing lists for the current user. getattr(settings, 'FILTER_VHOST', False)) else None client.find_lists(user_email, role=role, mail_host=mail_host)) # No lists exist with the given role for the given user. # If the user has no list that they are subscriber/owner/moderator of, we # just redirect them to the index page with all lists. # Render the list index page. 'lists': _unique_lists(all_lists), 'domain_count': len(choosable_domains), 'role': role } request, 'postorius/index.html', context )
"""Show a table of all public mailing lists.""" # TODO maxking: Figure out why does this view accept POST request and why # can't it be just a GET with list parameter. # If the user is logged-in, show them only related lists in the index, # except role is present in requests.GET.
getattr(settings, 'FILTER_VHOST', False)) else None advertised=advertised, mail_host=mail_host, count=count, page=page)
_get_list_page, request.GET.get('page'), request.GET.get('count'), paginator_class=MailmanPaginator)
{'lists': lists, 'all_lists': True, 'domain_count': len(choosable_domains)})
def list_delete(request, list_id): """Deletes a list but asks for confirmation first. """ else: kwargs={'list_id': list_id}) {'submit_url': submit_url, 'cancel_url': cancel_url, 'list': the_list})
def list_pending_confirmations(request, list_id): """Shows a list of subscription requests. """ return _list_subscriptions( request=request, list_id=list_id, token_owner=TokenOwner.subscriber, template='postorius/lists/pending_confirmations.html', page_title=_('Subscriptions pending user confirmation'), )
def list_subscription_requests(request, list_id): """Shows a list of subscription requests.""" request=request, list_id=list_id, token_owner=TokenOwner.moderator, template='postorius/lists/subscription_requests.html', page_title=_('Subscriptions pending approval'), )
for req in m_list.requests if req['token_owner'] == token_owner] requests, request.GET.get('page'), request.GET.get('count', 25)) {'list': m_list, 'paginated_requests': paginated_requests, 'page_title': page_title, 'page_subtitle': page_subtitle})
def handle_subscription_request(request, list_id, request_id, action): """ Handle a subscription request. Possible actions: - accept - defer - reject - discard """ 'accept': _('The request has been accepted.'), 'reject': _('The request has been rejected.'), 'discard': _('The request has been discarded.'), 'defer': _('The request has been defered.'), } # Moderate request and add feedback message to session. except HTTPError as e: if e.code == 409: messages.success(request, _('The request was already moderated: %s') % e.reason) else: messages.error(request, _('The request could not be moderated: %s') % e.reason)
('list_identity', _('List Identity')), ('automatic_responses', _('Automatic Responses')), ('alter_messages', _('Alter Messages')), ('dmarc_mitigations', _('DMARC Mitigations')), ('digest', _('Digest')), ('message_acceptance', _('Message Acceptance')), ('archiving', _('Archiving')), ('subscription_policy', _('Subscription Policy')), )
'list_identity': ListIdentityForm, 'automatic_responses': ListAutomaticResponsesForm, 'alter_messages': AlterMessagesForm, 'dmarc_mitigations': DMARCMitigationsForm, 'digest': DigestSettingsForm, 'message_acceptance': MessageAcceptanceForm, 'archiving': ArchiveSettingsForm, 'subscription_policy': ListSubscriptionPolicyForm, }
template='postorius/lists/settings.html'): """ View and edit the settings of a list. The function requires the user to be logged in and have the permissions necessary to perform the action.
Use /<NAMEOFTHESECTION>/<NAMEOFTHEOPTION> to show only parts of the settings <param> is optional / is used to differ in between section and option might result in using //option """ visible_section = 'list_identity' except KeyError: raise Http404('No such settings section') # List settings are grouped an processed in different forms. else: except HTTPError as e: messages.error( request, _('An error occured: ') + e.reason) else:
'form': form, 'section_names': SETTINGS_SECTION_NAMES, 'list': m_list, 'visible_section': visible_section, })
template='postorius/lists/confirm_remove_role.html'): """Removes a list moderator or owner.""" messages.error(request, _('The user %(email)s is not in the %(role)s group') % {'email': address, 'role': role}) return redirect('list_members', the_list.list_id, role)
user=request.user, verified=True).order_by( "email").values_list("email", flat=True) # The user is removing themselves, redirect to the list info page # because they won't have access to the members page anyway.
except HTTPError as e: messages.error(request, _('The user could not be removed: %(msg)s') % {'msg': e.msg}) return redirect('list_members', the_list.list_id, role) ' from the %(role)s group.') % {'address': address, 'role': role}) return render(request, template, {'role': role, 'address': address, 'list_id': the_list.list_id})
def remove_all_subscribers(request, list_id):
"""Empty the list by unsubscribing all members."""
messages.error(request, _('No member is subscribed to the list currently.')) return redirect('mass_removal', mlist.list_id) # TODO maxking: This doesn't scale. Either the Core should provide # an endpoint to remove all subscribers or there should be some # better way to do this. Maybe, Core can take a list of email # addresses in batches of 50 and unsubscribe all of them. ' unsubscribed from the list.')) except Exception as e: messages.error(request, e) 'postorius/lists/confirm_removeall_subscribers.html', {'list': mlist})
def list_bans(request, list_id): """ Ban or unban email addresses. """ # Get the list and cache the archivers property.
# Process form submission. 'The email {} has been banned.'.format( addban_form.cleaned_data['email']))) request, _('An error occured: %s') % e.reason) except ValueError as e: messages.error(request, _('Invalid data: %s') % e) 'The email {} has been un-banned'.format( request.POST['email']))) messages.error(request, _('An error occured: %s') % e.reason) else: list(ban_list), request.GET.get('page'), request.GET.get('count')) {'list': m_list, 'addban_form': addban_form, 'banned_addresses': banned_addresses, })
def list_header_matches(request, list_id): """ View and edit the list's header matches. """ ListHeaderMatchForm, extra=1, can_delete=True, can_order=True, formset=ListHeaderMatchFormset) dict([ (key, getattr(hm, key)) for key in ListHeaderMatchForm.base_fields ]) for hm in header_matches]
# Process form submission. # Purge the existing header_matches # Add the ones in the form
# If ORDER is None (new header match), add it last. # The new header match form was not filled header=form.cleaned_data['header'], pattern=form.cleaned_data['pattern'], action=form.cleaned_data['action'], ) request, _('An error occured: %s') % e.reason) ' successfully modified.')) else: # Adapt the last form to create new matches
'list': m_list, 'formset': formset, }) |