<commit_before>from django.contrib.auth.models import User
from django.test import TestCase
from django.utils.timezone import utc
from django.db.models import Q

from zephyr.models import Zephyr, UserProfile, ZephyrClass, Recipient, Subscription, \
    filter_by_subscriptions, Realm, do_send_zephyr
from zephyr.views import get_updates
from zephyr.decorator import TornadoAsyncException

import datetime
import simplejson
import subprocess
subprocess.call("zephyr/tests/generate-fixtures");
from django.conf import settings

settings.ZEPHYR_LOG = "/tmp/test-zephyr-log"

class AuthedTestCase(TestCase):
    def login(self, username, password):
        return self.client.post('/accounts/login/',
                                {'username':username, 'password':password})

    def register(self, username, password):
        return self.client.post('/accounts/register/',
                                {'full_name':username, 'short_name':username,
                                 'email':username + "@humbughq.com",
                                 'username':username, 'password':password,
                                 'domain':'humbughq.com'})

    def get_userprofile(self, email):
        """
        Given an email address, return the UserProfile object for the
        User that has that email.
        """
        # Usernames are unique, even across Realms.
        return UserProfile.objects.get(user=User.objects.get(email=email))

    def send_zephyr(self, sender_name, recipient_name, zephyr_type):
        sender = self.get_userprofile(sender_name)
        if zephyr_type == Recipient.PERSONAL:
            recipient = self.get_userprofile(recipient_name)
        else:
            recipient = ZephyrClass.objects.get(name=recipient_name, realm=sender.realm)
        recipient = Recipient.objects.get(type_id=recipient.id, type=zephyr_type)
        pub_date = datetime.datetime.utcnow().replace(tzinfo=utc)
        do_send_zephyr(Zephyr(sender=sender, recipient=recipient, instance="test", pub_date=pub_date),
                       synced_from_mit=True)

    def users_subscribed_to_class(self, class_name, realm_domain):
        realm = Realm.objects.get(domain=realm_domain)
        zephyr_class = ZephyrClass.objects.get(name=class_name, realm=realm)
        recipient = Recipient.objects.get(type_id=zephyr_class.id, type=Recipient.CLASS)
        subscriptions = Subscription.objects.filter(recipient=recipient)

        return [subscription.userprofile.user for subscription in subscriptions]

    def zephyr_stream(self, user):
        return filter_by_subscriptions(Zephyr.objects.all(), user)

    def assert_json_success(self, result):
        """
        Successful POSTs return a 200 and JSON of the form {"result": "success",
        "msg": ""}.
        """
        self.assertEquals(result.status_code, 200)
        json = simplejson.loads(result.content)
        self.assertEquals(json.get("result"), "success")
        # We have a msg key for consistency with errors, but it typically has an
        # empty value.
        self.assertTrue("msg" in json)

    def assert_json_error(self, result, msg):
        """
        Invalid POSTs return a 400 and JSON of the form {"result": "error",
        "msg": "reason"}.
        """
        self.assertEquals(result.status_code, 400)
        json = simplejson.loads(result.content)
        self.assertEquals(json.get("result"), "error")
        self.assertEquals(json.get("msg"), msg)

class PublicURLTest(TestCase):
    """
    Account creation URLs are accessible even when not logged in. Authenticated
    URLs redirect to a page.
    """
    def fetch(self, urls, expected_status):
        for url in urls:
            response = self.client.get(url)
            self.assertEqual(response.status_code, expected_status,
                             msg="Expected %d, received %d for %s" % (
                    expected_status, response.status_code, url))

    def test_public_urls(self):
        """
        Pages that should return a 200 when not logged in.
        """
        urls = {200: ["/accounts/home/", "/accounts/login/", "/accounts/logout/",
                      "/accounts/register/"],
                302: ["/", "/zephyr/", "/json/subscriptions/list",
                      "/json/subscriptions/remove", "/json/subscriptions/add"],
                400: ["/accounts/register/"],
                }
        for status_code, url_set in urls.iteritems():
            self.fetch(url_set, status_code)


class LoginTest(AuthedTestCase):
    """
    Logging in, registration, and logging out.
    """
    fixtures = ['zephyrs.json']
<commit_msg>Fix registration test, now exercise the email sending and verification code<commit_after>from django.contrib.auth.models import User
from django.test import TestCase
from django.utils.timezone import utc
from django.db.models import Q

from zephyr.models import Zephyr, UserProfile, ZephyrClass, Recipient, Subscription, \
    filter_by_subscriptions, Realm, do_send_zephyr
from zephyr.views import get_updates
from zephyr.decorator import TornadoAsyncException

import datetime
import simplejson
import subprocess
subprocess.call("zephyr/tests/generate-fixtures");
from django.conf import settings
import re

settings.ZEPHYR_LOG = "/tmp/test-zephyr-log"
settings.EMAIL_BACKEND = 'django.core.mail.backends.locmem.EmailBackend'


def find_key_by_email(address):
    from django.core.mail import outbox
    key_regex = re.compile("accounts/do_confirm/([a-f0-9]{40})>")
    for message in reversed(outbox):
        if address in message.to:
            return key_regex.search(message.body).groups()[0]

class AuthedTestCase(TestCase):
    def login(self, username, password):
        return self.client.post('/accounts/login/',
                                {'username':username, 'password':password})

    def register(self, username, password):
        self.client.post('/accounts/home/',
                         {'email': username + '@humbughq.com'})
        return self.client.post('/accounts/register/',
                                {'full_name':username, 'short_name':username,
                                 'key': find_key_by_email(username + '@humbughq.com'),
                                 'username':username, 'password':password,
                                 'domain':'humbug<EMAIL>'})

    def get_userprofile(self, email):
        """
        Given an email address, return the UserProfile object for the
        User that has that email.
        """
        # Usernames are unique, even across Realms.
        return UserProfile.objects.get(user=User.objects.get(email=email))

    def send_zephyr(self, sender_name, recipient_name, zephyr_type):
        sender = self.get_userprofile(sender_name)
        if zephyr_type == Recipient.PERSONAL:
            recipient = self.get_userprofile(recipient_name)
        else:
            recipient = ZephyrClass.objects.get(name=recipient_name, realm=sender.realm)
        recipient = Recipient.objects.get(type_id=recipient.id, type=zephyr_type)
        pub_date = datetime.datetime.utcnow().replace(tzinfo=utc)
        do_send_zephyr(Zephyr(sender=sender, recipient=recipient, instance="test", pub_date=pub_date),
                       synced_from_mit=True)

    def users_subscribed_to_class(self, class_name, realm_domain):
        realm = Realm.objects.get(domain=realm_domain)
        zephyr_class = ZephyrClass.objects.get(name=class_name, realm=realm)
        recipient = Recipient.objects.get(type_id=zephyr_class.id, type=Recipient.CLASS)
        subscriptions = Subscription.objects.filter(recipient=recipient)

        return [subscription.userprofile.user for subscription in subscriptions]

    def zephyr_stream(self, user):
        return filter_by_subscriptions(Zephyr.objects.all(), user)

    def assert_json_success(self, result):
        """
        Successful POSTs return a 200 and JSON of the form {"result": "success",
        "msg": ""}.
        """
        self.assertEquals(result.status_code, 200)
        json = simplejson.loads(result.content)
        self.assertEquals(json.get("result"), "success")
        # We have a msg key for consistency with errors, but it typically has an
        # empty value.
        self.assertTrue("msg" in json)

    def assert_json_error(self, result, msg):
        """
        Invalid POSTs return a 400 and JSON of the form {"result": "error",
        "msg": "reason"}.
        """
        self.assertEquals(result.status_code, 400)
        json = simplejson.loads(result.content)
        self.assertEquals(json.get("result"), "error")
        self.assertEquals(json.get("msg"), msg)

class PublicURLTest(TestCase):
    """
    Account creation URLs are accessible even when not logged in. Authenticated
    URLs redirect to a page.
    """
    def fetch(self, urls, expected_status):
        for url in urls:
            response = self.client.get(url)
            self.assertEqual(response.status_code, expected_status,
                             msg="Expected %d, received %d for %s" % (
                    expected_status, response.status_code, url))

    def test_public_urls(self):
        """
        Pages that should return a 200 when not logged in.
        """
        urls = {200: ["/accounts/home/", "/accounts/login/", "/accounts/logout/"],
                302: ["/", "/zephyr/", "/json/subscriptions/list",
                      "/json/subscriptions/remove", "/json/subscriptions/add"],
                400: ["/accounts/register/"],
                }
        for status_code, url_set in urls.iteritems():
            self.fetch(url_set, status_code)


class LoginTest(AuthedTestCase):
    """
    Logging in, registration, and logging out.
    """
    fixtures = ['zephyrs.json']

{'fraction_non_alphanumeric': 0.11023541453428864, 'fraction_numerical': 0.0054247697031729785, 'mean_word_length': 2.6431767337807606, 'pattern_counts': {'":': 8, '<': 4, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3898573', 'n_tokens_mistral': 2860, 'n_tokens_neox': 2778, 'n_words': 630}