text:
import csv
import os
import tempfile
from datetime import datetime
from os import path
from StringIO import StringIO
from zipfile import ZipFile

from go.vumitools.tests.helpers import djangotest_imports

with djangotest_imports(globals()):
    from django.conf import settings
    from django.core import mail
    from django.core.files.storage import default_storage
    from django.core.urlresolvers import reverse
    from django.utils.html import escape
    from django.templatetags.l10n import localize

    from go.contacts.parsers.base import FieldNormalizer
    from go.base.tests.helpers import GoDjangoTestCase, DjangoVumiApiHelper

TEST_GROUP_NAME = u"Test Group"
TEST_CONTACT_NAME = u"Name"
TEST_CONTACT_SURNAME = u"Surname"
TEST_CONTACT_MSISDN = u"+27761234567"

def newest(models):
    return max(models, key=lambda m: m.created_at)

def person_url(person_key):
    return reverse('contacts:person', kwargs={'person_key': person_key})

def group_url(group_key):
    return reverse('contacts:group', kwargs={'group_key': group_key})

def get_all_contact_keys_for_group(contact_store, group):
    """
    Walk the index pages returned from get_contact_keys_for_group() and collect
    all the keys.
    """
    keys = []
    index_page = contact_store.get_contact_keys_for_group(group)
    while index_page is not None:
        keys.extend(index_page)
        index_page = index_page.next_page()
    return keys

class BaseContactsTestCase(GoDjangoTestCase):
    def setUp(self):
        self.vumi_helper = self.add_helper(DjangoVumiApiHelper())
        self.user_helper = self.vumi_helper.make_django_user()
        self.user_email = self.user_helper.get_django_user().email
        self.contact_store = self.user_helper.user_api.contact_store
        self.client = self.vumi_helper.get_client()
        self.clear_tmp_storage()

    def clear_tmp_storage(self):
        try:
            _folders, files = default_storage.listdir("tmp")
        except (NotImplementedError, OSError):
            # exceptions indicated that listdir is not supported by
            # default storage or tmp does not yet exist.
            files = []
        for filename in files:
            default_storage.delete(path.join("tmp", filename))

    def mkcontact(self, name=None, surname=None, msisdn=u'+1234567890',
                  **kwargs):
        return self.contact_store.new_contact(
            name=unicode(name or TEST_CONTACT_NAME),
            surname=unicode(surname or TEST_CONTACT_SURNAME),
            msisdn=unicode(msisdn), **kwargs)

    def get_group_contact_keys(self, group):
        contact_keys = []
        contact_keys_page = group.backlinks.contact_keys()
        while contact_keys_page is not None:
            contact_keys.extend(contact_keys_page)
            contact_keys_page = contact_keys_page.next_page()
        return contact_keys

    def assert_group_contacts_count(self, group, count):
        contact_keys = self.get_group_contact_keys(group)
        self.assertEqual(len(contact_keys), count)

class TestContacts(BaseContactsTestCase):
    def test_redirect_index(self):
        response = self.client.get(reverse('contacts:index'))
        self.assertRedirects(response, reverse('contacts:groups'))

    def get_all_contacts(self, keys=None):
        if keys is None:
            keys = self.contact_store.list_contacts()
        contacts = []
        for batch in self.contact_store.contacts.load_all_bunches(keys):
            contacts.extend(batch)
        return contacts

    def get_latest_contact(self):
        return max(self.get_all_contacts(), key=lambda c: c.created_at)

    def test_contact_details(self):
        group_1 = self.contact_store.new_group(u'lerp')
        group_2 = self.contact_store.new_group(u'larp')

        contact = self.mkcontact(
            name='Foo',
            surname='Bar',
            groups=[group_1, group_2],
            extra={'black': 'moth', 'super': 'rainbow'})

        response = self.client.get(person_url(contact.key))

        self.assertContains(
            response,
            "Contact record created at %s." % localize(contact.created_at))

        self.assertContains(response, 'Foo')
        self.assertContains(response, 'Bar')

        self.assertContains(response, 'lerp')
        self.assertContains(response, 'larp')

        self.assertContains(response, 'Extra details')
        self.assertContains(response, 'black')
        self.assertContains(response, 'moth')
        self.assertContains(response, 'super')
        self.assertContains(response, 'rainbow')

        self.assertNotContains(response, 'Subscriptions')

    def test_subscriptions(self):
        contact = self.mkcontact(
            name='Foo',
            surname='Bar',
            subscription={'app_sub_1': 1, 'app_sub_2': 2})

        response = self.client.get(person_url(contact.key))

        self.assertContains(response, 'Subscriptions')
        self.assertContains(response, 'app_sub_1')
        self.assertContains(response, '1')
        self.assertContains(response, 'app_sub_2')
        self.assertContains(response, '2')

        self.assertNotContains(response, 'Extra details')

    def test_contact_creation_form(self):
        response = self.client.get(reverse('contacts:new_person'))
        for label in ["Name", "Surname", "Email address", "Groups"]:
            self.assertContains(response, label)
        self.assertNotContains(response, "Contact record created at")

    def test_contact_creation(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        response = self.client.post(reverse('contacts:new_person'), {
            'name': 'New',
            'surname': 'Person',
            'msisdn': '27761234567',
            'groups': [group.key],
        })
        contact = self.get_latest_contact()
        self.assertRedirects(response, person_url(contact.key))

    def test_contact_deleting(self):
        contact = self.mkcontact()
        person_url = reverse('contacts:person', kwargs={
            'person_key': contact.key,
        })
        response = self.client.post(person_url, {
            '_delete': True,
        })
        self.assertEqual(response.status_code, 302)
        self.assertTrue(response['Location'].endswith(
            reverse('contacts:people')))

        # After deleting the person should return a 404 page
        response = self.client.get(person_url)
        self.assertEqual(response.status_code, 404)

    def test_contact_update(self):
        contact = self.mkcontact()
        response = self.client.post(person_url(contact.key), {
            'name': 'changed name',
            'surname': 'changed surname',
            'msisdn': '112',
            'groups': [g.key for g in self.contact_store.list_groups()],
        })
        self.assertRedirects(response, person_url(contact.key))
        # reload to check
        contact = self.contact_store.get_contact_by_key(contact.key)
        self.assertEqual(contact.name, 'changed name')
        self.assertEqual(contact.surname, 'changed surname')
        self.assertEqual(contact.msisdn, '112')
        self.assertEqual(
            set(contact.groups.keys()),
            set([g.key for g in self.contact_store.list_groups()]))

    def test_list_contacts(self):
        """
        When we have fewer contacts than the default limit, the response
        contains all of them.
        """
        contacts = [self.mkcontact(), self.mkcontact()]
        ckeys = sorted(c.key for c in contacts)

        response = self.client.get(reverse('contacts:people'), {})
        response_contacts = response.context['selected_contacts']
        self.assertEqual(ckeys, sorted(c.key for c in response_contacts))
        self.assertContains(response, "Showing 2 of 2 contact(s)")
        self.assertContains(response, ckeys[0])
        self.assertContains(response, ckeys[1])

    def test_list_contacts_within_limit(self):
        """
        When we have fewer contacts than the specified limit, the response
        contains all of them.
        """
        contacts = [self.mkcontact(), self.mkcontact(), self.mkcontact()]
        ckeys = sorted(c.key for c in contacts)

        response = self.client.get(reverse('contacts:people'), {'limit': 5})
        response_contacts = response.context['selected_contacts']
        self.assertEqual(ckeys, sorted(c.key for c in response_contacts))
        self.assertContains(response, "Showing 3 of 3 contact(s)")
        self.assertContains(response, ckeys[0])
        self.assertContains(response, ckeys[1])
        self.assertContains(response, ckeys[2])

    def test_list_contacts_limit(self):
        """
        When we have more contacts than the limit, the response only contains
        a subset of the contacts.
        """
        contacts = [self.mkcontact(), self.mkcontact(), self.mkcontact()]
        ckeys = sorted(c.key for c in contacts)

        response = self.client.get(reverse('contacts:people'), {'limit': 2})
        response_contacts = response.context['selected_contacts']
        # This assumes we're using raw index pagination (which sorts by key)
        # under the hood.
        self.assertEqual(ckeys[:2], sorted(c.key for c in response_contacts))
        self.assertContains(response, "Showing 2 of 3 contact(s)")
        self.assertContains(response, ckeys[0])
        self.assertContains(response, ckeys[1])
        self.assertNotContains(response, ckeys[2])

    def test_list_contacts_fetch_limit(self):
        """
        When we have more contacts than the fetch limit, the response only
        contains a subset of the contacts and doesn't display the total number
        of contacts.
        """
        contacts = [self.mkcontact(), self.mkcontact(), self.mkcontact()]
        ckeys = sorted(c.key for c in contacts)

        response = self.client.get(
            reverse('contacts:people'), {'limit': 1, '_fetch_limit': 2})
        response_contacts = response.context['selected_contacts']
        # This assumes we're using raw index pagination (which sorts by key)
        # under the hood.
        self.assertEqual(ckeys[:1], sorted(c.key for c in response_contacts))
        self.assertContains(response, "Showing 1 of 2+ contact(s)")
        self.assertContains(response, ckeys[0])
        self.assertNotContains(response, ckeys[1])
        self.assertNotContains(response, ckeys[2])

    def test_contact_exporting(self):
        c1 = self.mkcontact()
        c1.extra['foo'] = u'bar'
        c1.extra['bar'] = u'baz'
        c1.save()

        c2 = self.mkcontact()
        c2.extra['foo'] = u'lorem'
        c2.extra['bar'] = u'ipsum'
        c2.save()

        response = self.client.post(reverse('contacts:people'), {
            '_export': True,
            'contact': [c1.key, c2.key],
        })

        self.assertContains(
            response,
            "The export is scheduled and should complete within a few"
            " minutes.")

        self.assertEqual(len(mail.outbox), 1)
        [email] = mail.outbox
        [(file_name, contents, mime_type)] = email.attachments

        self.assertEqual(email.recipients(), [self.user_email])
        self.assertTrue('Contacts export' in email.subject)
        self.assertTrue('2 contact(s)' in email.body)
        self.assertEqual(file_name, 'contacts-export.zip')

        zipfile = ZipFile(StringIO(contents), 'r')
        csv_contents = zipfile.open('contacts-export.csv', 'r').read()

        [header, c1_data, c2_data, _] = csv_contents.split('\r\n')

        self.assertEqual(
            header,
            ','.join([
                'key', 'name', 'surname', 'email_address', 'msisdn', 'dob',
                'twitter_handle', 'facebook_id', 'bbm_pin', 'gtalk_id',
                'mxit_id', 'wechat_id', 'created_at', 'bar',
                'foo']))

        self.assertTrue(c1_data.startswith(c1.key))
        self.assertTrue(c1_data.endswith('baz,bar'))
        self.assertTrue(c2_data.startswith(c2.key))
        self.assertTrue(c2_data.endswith('ipsum,lorem'))
        self.assertTrue(contents)
        self.assertEqual(mime_type, 'application/zip')

    def test_contact_exporting_too_many(self):
        """
        If we have too many contacts to export, we only export a subset.
        """
        self.vumi_helper.patch_settings(CONTACT_EXPORT_TASK_LIMIT=2)

        c1 = self.mkcontact()
        c1.extra['foo'] = u'bar'
        c1.save()

        c2 = self.mkcontact()
        c2.extra['foo'] = u'baz'
        c2.save()

        c3 = self.mkcontact()
        c3.extra['foo'] = u'quux'
        c3.save()

        response = self.client.post(reverse('contacts:people'), {
            '_export': True,
            'contact': [c1.key, c2.key, c3.key],
        })

        self.assertContains(
            response,
            "The export is scheduled and should complete within a few"
            " minutes.")

        self.assertEqual(len(mail.outbox), 1)
        [email] = mail.outbox
        [(file_name, contents, mime_type)] = email.attachments

        self.assertEqual(email.recipients(), [self.user_email])
        self.assertTrue('Contacts export' in email.subject)
        self.assertTrue(
            'NOTE: There are too many contacts to export.' in email.body)
        self.assertTrue('2 (out of 3) contacts' in email.body)
        self.assertEqual(file_name, 'contacts-export.zip')

        zipfile = ZipFile(StringIO(contents), 'r')
        csv_contents = zipfile.open('contacts-export.csv', 'r').read()

        [header, c1_data, c2_data, _] = csv_contents.split('\r\n')

        self.assertEqual(
            header,
            ','.join([
                'key', 'name', 'surname', 'email_address', 'msisdn', 'dob',
                'twitter_handle', 'facebook_id', 'bbm_pin', 'gtalk_id',
                'mxit_id', 'wechat_id', 'created_at', 'foo']))

        self.assertTrue(c1_data.startswith(c1.key))
        self.assertTrue(c1_data.endswith(',bar'))
        self.assertTrue(c2_data.startswith(c2.key))
        self.assertTrue(c2_data.endswith(',baz'))
        self.assertTrue(contents)
        self.assertEqual(mime_type, 'application/zip')

    def test_exporting_all_contacts(self):
        c1 = self.mkcontact()
        c1.extra['foo'] = u'bar'
        c1.extra['bar'] = u'baz'
        c1.save()

        c2 = self.mkcontact()
        c2.extra['foo'] = u'lorem'
        c2.extra['bar'] = u'ipsum'
        c2.save()

        response = self.client.post(reverse('contacts:people'), {
            '_export_all': True,
        })

        self.assertContains(
            response,
            "The export is scheduled and should complete within a few"
            " minutes.")

        self.assertEqual(len(mail.outbox), 1)
        [email] = mail.outbox
        [(file_name, contents, mime_type)] = email.attachments

        self.assertEqual(email.recipients(), [self.user_email])
        self.assertTrue('Contacts export' in email.subject)
        self.assertTrue('2 contact(s)' in email.body)
        self.assertEqual(file_name, 'contacts-export.zip')

        zipfile = ZipFile(StringIO(contents), 'r')
        csv_contents = zipfile.open('contacts-export.csv', 'r').read()

        [header, c1_data, c2_data, _] = csv_contents.split('\r\n')

        self.assertEqual(
            header,
            ','.join([
                'key', 'name', 'surname', 'email_address', 'msisdn', 'dob',
                'twitter_handle', 'facebook_id', 'bbm_pin', 'gtalk_id',
                'mxit_id', 'wechat_id', 'created_at', 'bar',
                'foo']))

        self.assertTrue(c1_data.startswith(c1.key))
        self.assertTrue(c1_data.endswith('baz,bar'))
        self.assertTrue(c2_data.startswith(c2.key))
        self.assertTrue(c2_data.endswith('ipsum,lorem'))
        self.assertTrue(contents)
        self.assertEqual(mime_type, 'application/zip')

    def specify_columns(self, group_key, columns=None, import_rule=None,
                        follow=False):
        group_url = reverse('contacts:group', kwargs={
            'group_key': group_key,
        })
        defaults = {
            'column-0': 'name',
            'column-1': 'surname',
            'column-2': 'msisdn',
            'normalize-0': '',
            'normalize-1': '',
            'normalize-2': '',
            '_complete_contact_upload': '1',
        }
        if columns:
            defaults.update(columns)
        if import_rule is not None:
            defaults['import_rule'] = import_rule
        return self.client.post(group_url, defaults, follow=follow)

    def test_contact_upload_into_new_group(self):
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                        'fixtures', 'sample-contacts.csv'))

        response = self.client.post(reverse('contacts:people'), {
            'file': csv_file,
            'name': 'a new group',
        })

        group = newest(self.contact_store.list_groups())
        self.assertEqual(group.name, u"a new group")
        self.assertRedirects(response, group_url(group.key))
        self.assert_group_contacts_count(group, 0)

        response = self.specify_columns(group.key)
        self.assertRedirects(response, group_url(group.key))
        self.assert_group_contacts_count(group, 3)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_contact_upload_into_existing_group(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                        'fixtures', 'sample-contacts.csv'), 'r')
        response = self.client.post(reverse('contacts:people'), {
            'file': csv_file,
            'contact_group': group.key
        })

        self.assertRedirects(response, group_url(group.key))
        group = self.contact_store.get_group(group.key)
        self.assert_group_contacts_count(group, 0)
        response = self.specify_columns(group.key)
        self.assertRedirects(response, group_url(group.key))
        self.assert_group_contacts_count(group, 3)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_uploading_unicode_chars_in_csv(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                                  'fixtures', 'sample-unicode-contacts.csv'))

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group.key,
            'file': csv_file,
        })
        self.assertRedirects(response, group_url(group.key))

        response = self.specify_columns(group.key)
        self.assertRedirects(response, group_url(group.key))
        group = self.contact_store.get_group(group.key)
        self.assert_group_contacts_count(group, 3)
        self.assertEqual(len(mail.outbox), 1)
        self.assertTrue('successfully' in mail.outbox[0].subject)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_uploading_windows_linebreaks_in_csv(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                                  'fixtures',
                                  'sample-windows-linebreaks-contacts.csv'))

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group.key,
            'file': csv_file,
        })
        self.assertRedirects(response, group_url(group.key))

        response = self.specify_columns(group.key, columns={
            'column-0': 'msisdn',
            'column-1': 'area',
            'column-2': 'nairobi_1',
            'column-3': 'baba dogo',
            'column-4': 'age',
            'column-5': 'gender',
            'column-6': 'language',
            'column-7': 'occupation',
            'normalize-0': '',
            'normalize-1': '',
            'normalize-2': '',
            'normalize-3': '',
            'normalize-4': '',
            'normalize-5': '',
            'normalize-6': '',
            'normalize-7': '',
        })
        self.assertRedirects(response, group_url(group.key))
        group = self.contact_store.get_group(group.key)
        self.assert_group_contacts_count(group, 2)
        self.assertEqual(len(mail.outbox), 1)
        self.assertTrue('successfully' in mail.outbox[0].subject)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_uploading_single_colum(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                                  'fixtures',
                                  'sample-single-column-contacts.csv'))

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group.key,
            'file': csv_file,
        })
        self.assertRedirects(response, group_url(group.key))

        response = self.specify_columns(group.key, columns={
            'column-0': 'msisdn',
            'normalize-0': '',
        })
        self.assertRedirects(response, group_url(group.key))

        group = self.contact_store.get_group(group.key)
        self.assert_group_contacts_count(group, 2)
        self.assertEqual(len(mail.outbox), 1)
        self.assertTrue('successfully' in mail.outbox[0].subject)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_upload_with_contact_uuid(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                                  'fixtures',
                                  'sample-contacts-with-uuid-headers.csv'))

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group.key,
            'file': csv_file,
        })
        self.assertRedirects(response, group_url(group.key))
        preview_response = self.client.get(group_url(group.key))
        self.assertContains(
            preview_response, 'The file includes contact UUIDs.')

    def test_upload_without_contact_uuid(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                                  'fixtures',
                                  'sample-contacts-with-headers.csv'))

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group.key,
            'file': csv_file,
        })
        self.assertRedirects(response, group_url(group.key))
        preview_response = self.client.get(group_url(group.key))
        self.assertContains(
            preview_response, 'The file does not include contact UUIDs.')

    def create_temp_csv_file(self):
        return tempfile.NamedTemporaryFile(delete=False, suffix='.csv')

    def create_csv(self, fieldnames, data):
        fp = self.create_temp_csv_file()
        csv_writer = csv.DictWriter(fp, fieldnames=fieldnames)
        csv_writer.writerow(dict(zip(fieldnames, fieldnames)))
        for row in data:
            csv_writer.writerow(row)
        fp.seek(0)
        return fp

    def test_import_upload_is_truth(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)

        # create existing contacts that'll be updated.
        contact_data = []
        for i in range(3):
            # the original contact
            contact = self.mkcontact(name='', surname='', msisdn='270000000')
            # Litmus to ensure we don't butcher stuff
            contact.extra['litmus_stay'] = u'red'
            contact.extra['litmus_overwrite'] = u'blue'
            contact.subscription['sub'] = u'the-subscription'
            contact.dob = datetime(2014, 1, 2)
            contact.save()
            # what we're going to update
            contact_data.append({
                u'created_at': '2014-01-0%dT00:00:00' % (i + 1,),
                u'key': contact.key,
                u'name': u'name %s' % (i,),
                u'surname': u'surname %s' % (i,),
                u'msisdn': u'271111111%s' % (i,),
                u'litmus_new': u'green',
                u'litmus_overwrite': u'purple',
            })

        csv = self.create_csv(
            ['key', 'created_at', 'name', 'surname', 'msisdn',
             'litmus_overwrite', 'litmus_new'],
            contact_data)

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group.key,
            'file': csv,
        })

        self.assertRedirects(response, group_url(group.key))
        response = self.specify_columns(group.key, columns={
            'column-0': 'key',
            'column-1': 'created_at',
            'column-2': 'name',
            'column-3': 'surname',
            'column-4': 'msisdn',
            'column-5': 'litmus_overwrite',
            'column-6': 'litmus_new',
            'normalize-0': '',
            'normalize-1': '',
            'normalize-2': '',
            'normalize-3': '',
            'normalize-4': 'msisdn_za',
            'normalize-5': '',
            'normalize-6': '',
        }, import_rule='upload_is_truth')
        self.assertRedirects(response, group_url(group.key))

        group = self.contact_store.get_group(group.key)
        self.assert_group_contacts_count(group, 3)
        [email] = mail.outbox

        self.assertEqual('Contact import completed.', email.subject)
        self.assertTrue(
            "We've successfully imported 3 of your contact(s)" in email.body)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

        updated_contacts = [
            self.contact_store.get_contact_by_key(contact['key'])
            for contact in contact_data]
        self.assertEqual(
            set([contact.created_at for contact in updated_contacts]),
            set([datetime(2014, 1, 1),
                 datetime(2014, 1, 2),
                 datetime(2014, 1, 3)]))
        self.assertEqual(
            set([contact.name for contact in updated_contacts]),
            set(['name 0', 'name 1', 'name 2']))
        self.assertEqual(
            set([contact.surname for contact in updated_contacts]),
            set(['surname 0', 'surname 1', 'surname 2']))
        # these are normalized for ZA
        self.assertEqual(
            set([contact.msisdn for contact in updated_contacts]),
            set(['+2711111110', '+2711111111', '+2711111112']))

        # check the litmus
        self.assertEqual(
            set([contact.extra['litmus_stay']
                 for contact in updated_contacts]),
            set(['red']))
        self.assertEqual(
            set([contact.extra['litmus_new']
                 for contact in updated_contacts]),
            set(['green']))
        self.assertEqual(
            set([contact.extra['litmus_overwrite']
                 for contact in updated_contacts]),
            set(['purple']))
        self.assertEqual(
            set([contact.dob for contact in updated_contacts]),
            set([datetime(2014, 1, 2)]))
        self.assertEqual(
            set([contact.subscription['sub']
                 for contact in updated_contacts]),
            set(['the-subscription']))

        for contact in updated_contacts:
            self.assertEqual(
                set(contact.extra.keys()),
                set(['litmus_stay', 'litmus_new', 'litmus_overwrite']))

        os.unlink(csv.name)

    def test_import_existing_is_truth(self):
        group1 = self.contact_store.new_group(TEST_GROUP_NAME)
        group2 = self.contact_store.new_group(TEST_GROUP_NAME + ' 2')

        # create existing contacts that'll be updated.
        contact_data = []
        for i in range(3):
            # the original contact
            contact = self.mkcontact(
                name='foo', surname='bar', msisdn='270000000')
            # Litmus to ensure we don't butcher stuff
            contact.extra['litmus_stay'] = u'red'
            contact.subscription['sub'] = u'the-subscription'
            contact.created_at = datetime(2014, 1, i + 1)
            contact.dob = datetime(2014, 1, 2)
            contact.add_to_group(group2)
            contact.save()
            # what we're going to update
            contact_data.append({
                u'created_at': datetime(1970, 1, 1),
                u'key': contact.key,
                u'name': u'name %s' % (i,),
                u'surname': u'surname %s' % (i,),
                u'msisdn': u'271111111%s' % (i,),
                u'litmus_stay': u'green',
                u'litmus_new': u'blue',
            })

        csv = self.create_csv(
            ['key', 'created_at', 'name', 'surname', 'msisdn',
             'litmus_stay', 'litmus_new'],
            contact_data)

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group1.key,
            'file': csv,
        })

        self.assertRedirects(response, group_url(group1.key))
        response = self.specify_columns(group1.key, columns={
            'column-0': 'key',
            'column-1': 'created_at',
            'column-2': 'name',
            'column-3': 'surname',
            'column-4': 'msisdn',
            'column-5': 'litmus_stay',
            'column-6': 'litmus_new',
            'normalize-0': '',
            'normalize-1': '',
            'normalize-2': '',
            'normalize-3': '',
            'normalize-4': 'msisdn_za',
            'normalize-5': '',
            'normalize-6': '',
        }, import_rule='existing_is_truth')
        self.assertRedirects(response, group_url(group1.key))

        group = self.contact_store.get_group(group1.key)
        self.assert_group_contacts_count(group, 3)
        [email] = mail.outbox

        self.assertEqual('Contact import completed.', email.subject)
        self.assertTrue(
            "We've successfully imported 3 of your contact(s)" in email.body)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

        updated_contacts = [
            self.contact_store.get_contact_by_key(contact['key'])
            for contact in contact_data]
        self.assertEqual(
            set([contact.created_at for contact in updated_contacts]),
            set([datetime(2014, 1, 1),
                 datetime(2014, 1, 2),
                 datetime(2014, 1, 3)]))
        self.assertEqual(
            set([contact.name for contact in updated_contacts]),
            set(['foo']))
        self.assertEqual(
            set([contact.surname for contact in updated_contacts]),
            set(['bar']))
        # these are normalized for ZA
        self.assertEqual(
            set([contact.msisdn for contact in updated_contacts]),
            set(['270000000']))
        # check the litmus
        self.assertEqual(
            set([contact.extra['litmus_stay']
                 for contact in updated_contacts]),
            set(['red']))
        self.assertEqual(
            set([contact.extra['litmus_new']
                 for contact in updated_contacts]),
            set(['blue']))
        self.assertEqual(
            set([contact.dob for contact in updated_contacts]),
            set([datetime(2014, 1, 2)]))
        self.assertEqual(
            set([contact.dob for contact in updated_contacts]),
            set([datetime(2014, 1, 2)]))
        self.assertEqual(
            set([contact.subscription['sub']
                 for contact in updated_contacts]),
            set(['the-subscription']))

        for contact in updated_contacts:
            self.assertEqual(
                set(contact.extra.keys()),
                set(['litmus_stay', 'litmus_new']))

        groups = []
        for contact in updated_contacts:
            groups.extend(contact.groups.keys())

        self.assertEqual(
            set(groups),
            set([group1.key, group2.key]))

        os.unlink(csv.name)

    def test_import_existing_is_truth_fail_on_empty_key(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)

        # create existing contacts that'll be updated.
        contact_data = []
        contacts = []
        for i in range(3):
            # the original contact
            contact = self.mkcontact(
                name='foo', surname='bar', msisdn='270000000')
            # what we're going to update
            contacts.append(contact)
            contact_data.append({
                u'key': contact.key,
                u'twitter_handle': u'tweetor%d' % (i,),
            })

        # add a bad key
        contact_data[1]['key'] = ''

        csv = self.create_csv(['key', 'twitter_handle'], contact_data)

        response = self.client.post(reverse('contacts:people'), {
            'contact_group': group.key,
            'file': csv,
        })

        self.assertRedirects(response, group_url(group.key))
        response = self.specify_columns(group.key, columns={
            'column-0': 'key',
            'column-1': 'twitter_handle',
            'normalize-0': '',
            'normalize-1': '',
        }, import_rule='existing_is_truth')
        self.assertRedirects(response, group_url(group.key))

        [email] = mail.outbox

        self.assertEqual('Contact import completed.', email.subject)
        self.assertTrue(
            "We've successfully imported 2 of your contact(s)" in email.body)
        self.assertTrue(
            "Unfortunately there were also 1 errors. These are listed below:"
            in email.body)
        self.assertTrue(
            "row 2: No key provided" in email.body)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

        new_contact = self.contact_store.get_contact_by_key(contacts[0].key)
        self.assertEqual(new_contact.twitter_handle, 'tweetor0')
        self.assertEqual(new_contact.groups.keys(), [group.key])

        new_contact = self.contact_store.get_contact_by_key(contacts[1].key)
        self.assertEqual(new_contact.twitter_handle, None)
        self.assertEqual(new_contact.groups.keys(), [])

        new_contact = self.contact_store.get_contact_by_key(contacts[2].key)
        self.assertEqual(new_contact.twitter_handle, 'tweetor2')
        self.assertEqual(new_contact.groups.keys(), [group.key])

        os.unlink(csv.name)

    def test_uploading_unicode_chars_in_csv_into_new_group(self):
        new_group_name = u'Testing a ünicode grøüp'
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                                  'fixtures', 'sample-unicode-contacts.csv'))

        response = self.client.post(reverse('contacts:people'), {
            'name': new_group_name,
            'file': csv_file,
        })

        group = newest(self.contact_store.list_groups())
        self.assertEqual(group.name, new_group_name)
        self.assertRedirects(response, group_url(group.key))
        response = self.specify_columns(group_key=group.key)
        self.assertRedirects(response, group_url(group.key))
        self.assert_group_contacts_count(group, 3)
        self.assertEqual(len(mail.outbox), 1)
        self.assertTrue('successfully' in mail.outbox[0].subject)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_contact_upload_from_group_page(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)

        group_url = reverse('contacts:group', kwargs={
            'group_key': group.key
        })

        csv_file = open(
            path.join(settings.PROJECT_ROOT, 'base',
                      'fixtures', 'sample-contacts.csv'), 'r')
        response = self.client.post(group_url, {
            'file': csv_file,
        })

        # It should redirect to the group page
        self.assertRedirects(response, group_url)

        # Wich should show the column-matching dialogue
        response = self.client.get(group_url)
        self.assertContains(response,
                            'Please match the sample to the fields provided')

        # The path of the uploaded file should have been set
        self.assertTrue('uploaded_contacts_file_name' in self.client.session)
        self.assertTrue('uploaded_contacts_file_path' in self.client.session)

        file_name = self.client.session['uploaded_contacts_file_name']
        self.assertEqual(file_name, 'sample-contacts.csv')

        # Nothing should have been written to the db by now.
        self.assert_group_contacts_count(group, 0)

        # Now submit the column names and check that things have been written
        # to the db
        response = self.specify_columns(group.key)
        # Check the redirect
        self.assertRedirects(response, group_url)
        # 3 records should have been written to the db.
        self.assert_group_contacts_count(group, 3)
        self.assertEqual(len(mail.outbox), 1)
        self.assertTrue('successfully' in mail.outbox[0].subject)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_graceful_error_handling_on_upload_failure(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        group_url = reverse('contacts:group', kwargs={
            'group_key': group.key
        })

        # Carefully crafted but bad CSV data
        wrong_file = StringIO(',,\na,b,c\n"')
        wrong_file.name = 'fubar.csv'

        response = self.client.post(group_url, {
            'file': wrong_file
        })

        response = self.client.get(group_url)
        self.assertContains(response, 'Something is wrong with the file')
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_contact_upload_failure(self):
        self.assertEqual(len(self.contact_store.list_groups()), 0)
        response = self.client.post(reverse('contacts:people'), {
            'name': 'a new group',
            'file': None,
        })
        self.assertContains(response, 'Something went wrong with the upload')
        self.assertEqual(len(self.contact_store.list_groups()), 0)
        self.assertEqual(len(mail.outbox), 0)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_contact_parsing_failure(self):
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                        'fixtures', 'sample-broken-contacts.csv'))
        response = self.client.post(reverse('contacts:people'), {
            'name': 'broken contacts group',
            'file': csv_file,
        }, follow=True)
        self.assertContains(response, "Something is wrong with the file")
        self.assertNotContains(response, "contact_data_headers")

    def test_contact_parsing_failure_bad_delimiter(self):
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                        'fixtures', 'sample-contacts-bad-delimiter.csv'))
        response = self.client.post(reverse('contacts:people'), {
            'name': 'broken contacts group',
            'file': csv_file,
        }, follow=True)
        group = newest(self.contact_store.list_groups())
        self.assertRedirects(response, group_url(group.key))
        # We only get one column here.
        self.assertContains(response, 'column-0')
        self.assertNotContains(response, 'column-1')
        # Now we follow the response and check that we fail sensibly by
        # reporting that we don't have the expected fields.
        response = self.specify_columns(group_key=group.key, columns={
            'column-0': 'name',
            'normalize-0': '',
        }, follow=True)
        self.assertRedirects(response, group_url(group.key))

        messages = [(m.tags, m.message) for m in response.context['messages']]
        self.assertEqual(messages, [
            ("error", "Please specify a Contact Number field."),
        ])

        group = newest(self.contact_store.list_groups())
        self.assert_group_contacts_count(group, 0)
        self.assertEqual(len(mail.outbox), 0)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_contact_parsing_failure_no_msisdn_field(self):
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                        'fixtures', 'sample-contacts.csv'))
        response = self.client.post(reverse('contacts:people'), {
            'name': 'broken contacts group',
            'file': csv_file,
        }, follow=True)
        group = newest(self.contact_store.list_groups())
        self.assertRedirects(response, group_url(group.key))

        # Here we don't specify an msisdn field, so that we explode in
        # the next bit.
        response = self.specify_columns(group_key=group.key, columns={
            'column-0': 'name',
            'column-1': 'surname',
            'column-2': 'accidental_extra',
            'normalize-0': '',
            'normalize-1': '',
            'normalize-2': '',
        }, follow=True)
        self.assertRedirects(response, group_url(group.key))

        messages = [(m.tags, m.message) for m in response.context['messages']]
        self.assertEqual(messages, [
            ("error", "Please specify a Contact Number field."),
        ])

        group = newest(self.contact_store.list_groups())
        self.assert_group_contacts_count(group, 0)
        self.assertEqual(len(mail.outbox), 0)
        self.assertEqual(default_storage.listdir("tmp"), ([], []))

    def test_normalization(self):
        csv_file = open(path.join(settings.PROJECT_ROOT, 'base',
                        'fixtures', 'sample-non-normalized-contacts.csv'))
        response = self.client.post(reverse('contacts:people'), {
            'name': 'non-normalized-contacts',
            'file': csv_file,
        })
        group = newest(self.contact_store.list_groups())
        self.assertRedirects(response, group_url(group.key))
        response = self.specify_columns(group_key=group.key, columns={
            'column-0': 'name',
            'column-1': 'surname',
            'column-2': 'integer',
            'column-3': 'float',
            'column-4': 'msisdn',
            'normalize-0': 'string',
            'normalize-1': 'string',
            'normalize-2': 'integer',
            'normalize-3': 'float',
            'normalize-4': 'msisdn_za',
        })
        contacts = self.get_all_contacts(self.get_group_contact_keys(group))

        self.assertTrue(all([contact.msisdn == '+27761234561' for contact in
                        contacts]))
        self.assertTrue(all([contact.extra['integer'] == '2' for contact in
                        contacts]))
        self.assertTrue(all([contact.extra['float'] == '2.0' for contact in
                        contacts]))

    def test_contact_querying(self):
        contact = self.mkcontact()
        people_url = reverse('contacts:people')

        # test no-match
        response = self.client.get(people_url, {
            'q': 'this should not match',
        })
        self.assertContains(response, 'No contacts match')

        # test match
        response = self.client.get(people_url, {
            'q': TEST_CONTACT_NAME,
        })
        self.assertContains(response, person_url(contact.key))

    def test_contact_key_value_query(self):
        contact = self.mkcontact()
        people_url = reverse('contacts:people')
        self.client.get(people_url, {
            'q': 'name:%s' % (contact.name,)
        })

class TestGroups(BaseContactsTestCase):
    def get_all_contacts(self, keys=None):
        if keys is None:
            keys = self.contact_store.list_contacts()
        contacts = []
        for batch in self.contact_store.contacts.load_all_bunches(keys):
            contacts.extend(batch)
        return contacts

    def get_latest_contact(self):
        return max(self.get_all_contacts(), key=lambda c: c.created_at)

    def list_group_keys(self):
        return [group.key for group in self.contact_store.list_groups()]

    def test_groups_creation(self):
        response = self.client.post(reverse('contacts:groups'), {
            'name': 'a new group',
            '_new_group': '1',
        })
        group = newest(self.contact_store.list_groups())
        self.assertNotEqual(group, None)
        self.assertEqual(u'a new group', group.name)
        self.assertRedirects(response, group_url(group.key))

    def test_group_updating(self):
        group = self.contact_store.new_group(u'old name')
        response = self.client.post(
            reverse('contacts:group', kwargs={'group_key': group.key}),
            {'name': 'new name', '_save_group': '1'})
        updated_group = self.contact_store.get_group(group.key)
        self.assertEqual('new name', updated_group.name)
        self.assertRedirects(response, group_url(group.key))

    def test_groups_creation_with_funny_chars(self):
        response = self.client.post(reverse('contacts:groups'), {
            'name': "a new group! with cüte chars's",
            '_new_group': '1',
        })
        group = newest(self.contact_store.list_groups())
        self.assertNotEqual(group, None)
        self.assertEqual(u"a new group! with cüte chars's", group.name)
        self.assertRedirects(response, group_url(group.key))

    def test_group_contact_querying(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        contact = self.mkcontact(groups=[group])
        # test no-match
        response = self.client.get(group_url(group.key), {
            'q': 'this should not match',
        })
        self.assertContains(response, 'No contacts match')

        # test match name
        response = self.client.get(group_url(group.key), {
            'q': TEST_CONTACT_NAME,
        })
        self.assertContains(response, person_url(contact.key))

    def test_group_contact_query_limits(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)

        for i in range(10):
            self.mkcontact(groups=[group])

        default_limit = self.client.get(group_url(group.key), {
            'q': TEST_CONTACT_NAME,
        })

        self.assertContains(
            default_limit,
            escape("Showing 10 of the group's 10 contact(s)"))

        custom_limit = self.client.get(group_url(group.key), {
            'q': TEST_CONTACT_NAME,
            'limit': 5,
        })
        self.assertContains(
            custom_limit,
            escape("Showing 5 of the group's 10 contact(s)"))

    def test_multiple_group_deletion(self):
        group_1 = self.contact_store.new_group(TEST_GROUP_NAME)
        group_2 = self.contact_store.new_group(TEST_GROUP_NAME)

        # Delete the groups
        groups_url = reverse('contacts:groups')
        response = self.client.post(groups_url, {
            'group': [group_1.key, group_2.key],
            '_delete': True,
        })
        self.assertRedirects(response, groups_url)
        self.assertEqual(self.contact_store.list_groups(), [])

    def test_removing_contacts_from_group(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        c1 = self.mkcontact(groups=[group])
        c2 = self.mkcontact(groups=[group])

        group_url = reverse('contacts:group', kwargs={'group_key': group.key})
        response = self.client.post(group_url, {
            '_remove': True,
            'contact': [c1.key]
        })
        self.assertRedirects(response, group_url)

        self.assertEqual(
            [c2.key],
            get_all_contact_keys_for_group(self.contact_store, group))

    def test_group_empty_post(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)

        self.assertEqual(self.list_group_keys(), [group.key])
        group_url = reverse('contacts:group', kwargs={'group_key': group.key})
        response = self.client.post(group_url)
        self.assertRedirects(response, group_url)

        self.assertEqual(self.list_group_keys(), [group.key])

    def test_group_deletion(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)

        # Create a contact in the group
        response = self.client.post(reverse('contacts:new_person'), {
            'name': 'New',
            'surname': 'Person',
            'msisdn': '27761234567',
            'groups': [group.key],
        })

        contact = self.get_latest_contact()
        self.assertRedirects(response, person_url(contact.key))

        # Delete the group
        group_url = reverse('contacts:group', kwargs={
            'group_key': group.key,
        })
        response = self.client.post(group_url, {
            '_delete_group': True,
        })
        self.assertEqual(response.status_code, 302)
        self.assertTrue(response['Location'].endswith(
            reverse('contacts:index')))

        reloaded_contact = self.get_latest_contact()
        self.assertEqual(reloaded_contact.key, contact.key)
        self.assertEqual(reloaded_contact.groups.keys(), [])

    def test_group_clearing(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        # Create a contact in the group
        response = self.client.post(reverse('contacts:new_person'), {
            'name': 'New',
            'surname': 'Person',
            'msisdn': '27761234567',
            'groups': [group.key],
        })

        contact = self.get_latest_contact()
        self.assertRedirects(response, person_url(contact.key))

        # Clear the group
        group_url = reverse('contacts:group', kwargs={
            'group_key': group.key,
        })
        response = self.client.post(group_url, {
            '_delete_group_contacts': True,
        })
        self.assertRedirects(response, group_url)

        self.assertEqual(
            [], get_all_contact_keys_for_group(self.contact_store, group))
        self.assertFalse(contact in self.contact_store.list_contacts())

    def test_group_contact_export(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        contact = self.mkcontact(groups=[group])
        # Clear the group
        group_url = reverse('contacts:group', kwargs={
            'group_key': group.key,
        })

        # add some extra info to ensure it gets exported properly
        contact.extra['foo'] = u'bar'
        contact.extra['bar'] = u'baz'
        contact.save()

        response = self.client.post(group_url, {'_export': True})

        self.assertRedirects(response, group_url)
        self.assertEqual(len(mail.outbox), 1)
        [email] = mail.outbox
        [(file_name, contents, mime_type)] = email.attachments

        self.assertEqual(email.recipients(), [self.user_email])
        self.assertTrue(
            '%s contacts export' % (group.name,) in email.subject)
        self.assertTrue(
            '1 contact(s) from group "%s" attached' % (group.name,)
            in email.body)
        self.assertEqual(file_name, 'contacts-export.zip')

        zipfile = ZipFile(StringIO(contents), 'r')
        csv_contents = zipfile.open('contacts-export.csv', 'r').read()

        [header, csv_contact, _] = csv_contents.split('\r\n')

        self.assertEqual(
            header,
            ','.join([
                'key', 'name', 'surname', 'email_address', 'msisdn', 'dob',
                'twitter_handle', 'facebook_id', 'bbm_pin', 'gtalk_id',
                'mxit_id', 'wechat_id', 'created_at', 'bar',
                'foo']))

        self.assertTrue(csv_contact.startswith(contact.key))
        self.assertTrue(csv_contact.endswith('baz,bar'))
        self.assertTrue(contents)
        self.assertEqual(mime_type, 'application/zip')

    def test_group_contact_export_with_prefix(self):
        group = self.contact_store.new_group(TEST_GROUP_NAME)
        contact = self.mkcontact(groups=[group])
        # Clear the group
        group_url = reverse('contacts:group', kwargs={
            'group_key': group.key,
        })

        # add some extra info to ensure it gets exported properly
        contact.extra['msisdn'] = u'bar'
        contact.extra['name'] = u'baz'
        contact.save()

        response = self.client.post(group_url, {'_export': True})

        self.assertRedirects(response, group_url)
        self.assertEqual(len(mail.outbox), 1)
        [email] = mail.outbox
        [(file_name, contents, mime_type)] = email.attachments

        self.assertEqual(email.recipients(), [self.user_email])
        self.assertTrue(
            '%s contacts export' % (group.name,) in email.subject)
        self.assertTrue(
            '1 contact(s) from group "%s" attached' % (group.name,)
            in email.body)
        self.assertEqual(file_name, 'contacts-export.zip')

        zipfile = ZipFile(StringIO(contents), 'r')
        csv_contents = zipfile.open('contacts-export.csv', 'r').read()

        [header, csv_contact, _] = csv_contents.split('\r\n')
        self.assertEqual(
            header,
            ','.join([
                'key', 'name', 'surname', 'email_address', 'msisdn', 'dob',
                'twitter_handle', 'facebook_id', 'bbm_pin', 'gtalk_id',
                'mxit_id', 'wechat_id', 'created_at', 'extras-msisdn',
                'extras-name']))

        self.assertTrue(csv_contact.startswith(contact.key))
        self.assertTrue(csv_contact.endswith('bar,baz'))
        self.assertTrue(contents)
        self.assertEqual(mime_type, 'application/zip')

    def test_multiple_group_exportation(self):
        group_1 = self.contact_store.new_group(u'Test Group 1')
        contact_1 = self.mkcontact(groups=[group_1])
        contact_1.extra['foo'] = u'bar'
        contact_1.extra['bar'] = u'baz'
        contact_1.save()

        group_2 = self.contact_store.new_group(u'Test Group 2')
        contact_2 = self.mkcontact(groups=[group_2])
        contact_2.extra['foo'] = u'lorem'
        contact_2.extra['bar'] = u'ipsum'
        contact_2.save()

        groups_url = reverse('contacts:groups')
        self.client.post(groups_url, {
            'group': [group_1.key, group_2.key],
            '_export': True,
        })

        self.assertEqual(len(mail.outbox), 1)
        [email] = mail.outbox
        [(file_name, contents, mime_type)] = email.attachments

        self.assertEqual(email.recipients(), [self.user_email])
        self.assertTrue('Contacts export' in email.subject)
        self.assertTrue(
            '2 contact(s) from the following groups:',
            '\n  - Test Group 1'
            '\n  - Test Group 2'
            in email.body)
        self.assertEqual(file_name, 'contacts-export.zip')

        zipfile = ZipFile(StringIO(contents), 'r')
        csv_contents = zipfile.open('contacts-export.csv', 'r').read()

        [header, c1_data, c2_data, _] = csv_contents.split('\r\n')

        self.assertEqual(
            header,
            ','.join([
                'key', 'name', 'surname', 'email_address', 'msisdn', 'dob',
                'twitter_handle', 'facebook_id', 'bbm_pin', 'gtalk_id',
                'mxit_id', 'wechat_id', 'created_at', 'bar',
                'foo']))

        self.assertTrue(c1_data.startswith(contact_1.key))
        self.assertTrue(c1_data.endswith('baz,bar'))
        self.assertTrue(c2_data.startswith(contact_2.key))
        self.assertTrue(c2_data.endswith('ipsum,lorem'))
        self.assertTrue(contents)
        self.assertEqual(mime_type, 'application/zip')

class TestSmartGroups(BaseContactsTestCase):
    def mksmart_group(self, query, name='a smart group'):
        response = self.client.post(reverse('contacts:groups'), {
            'name': name,
            'query': query,
            '_new_smart_group': '1',
        })
        group = newest(self.contact_store.list_groups())
        self.assertRedirects(response, group_url(group.key))
        return group

    def list_group_keys(self):
        return [group.key for group in self.contact_store.list_groups()]

    def add_to_group(self, contact, group):
        contact.add_to_group(group)
        contact.save()
        return contact

    def test_smart_groups_creation(self):
        group = self.mksmart_group('msisdn:\+12*')
        self.assertEqual(u'a smart group', group.name)
        self.assertEqual(u'msisdn:\+12*', group.query)

    def test_smart_group_empty_post(self):
        group = self.mksmart_group('msisdn:\+12*')
        group_url = reverse('contacts:group', kwargs={'group_key': group.key})
        response = self.client.post(group_url)
        self.assertRedirects(response, group_url)
        self.assertEqual(self.list_group_keys(), [group.key])

    def test_smart_group_deletion(self):
        group = self.mksmart_group('msisdn:\+12*')
        self.assertEqual(self.list_group_keys(), [group.key])
        group_url = reverse('contacts:group', kwargs={'group_key': group.key})
        response = self.client.post(group_url, {'_delete_group': 1})
        self.assertRedirects(response, reverse('contacts:index'),
                             target_status_code=302)
        self.assertEqual(self.list_group_keys(), [])

    def test_smart_group_clearing(self):
        contact = self.mkcontact()
        group = self.mksmart_group('msisdn:\+12*')
        self.assertEqual(
            [contact.key],
            get_all_contact_keys_for_group(self.contact_store, group))
        response = self.client.post(
            reverse('contacts:group', kwargs={'group_key': group.key}),
            {'_delete_group_contacts': 1})
        self.assertRedirects(response, reverse('contacts:group', kwargs={
            'group_key': group.key}))
        self.assertEqual(
            [], get_all_contact_keys_for_group(self.contact_store, group))

    def test_smart_group_updating(self):
        group = self.mksmart_group('msisdn:\+12*')
        response = self.client.post(
            reverse('contacts:group', kwargs={'group_key': group.key}),
            {'name': 'foo', 'query': 'name:bar', '_save_group': 1})
        self.assertRedirects(response, reverse('contacts:group', kwargs={
            'group_key': group.key}))
        saved_group = self.contact_store.get_group(group.key)
        self.assertEqual(saved_group.name, 'foo')
        self.assertEqual(saved_group.query, 'name:bar')

    def test_smart_groups_no_matches_results(self):
        response = self.client.post(reverse('contacts:groups'), {
            'name': 'a smart group',
            'query': 'msisdn:\+12*',
            '_new_smart_group': '1',
        })
        group = newest(self.contact_store.list_groups())
        conversation = self.user_helper.create_conversation(u'bulk_message')
        conversation.groups.add(group)
        conversation.save()

        self.assertRedirects(response, group_url(group.key))
        self.assertEqual(u'a smart group', group.name)
        self.assertEqual(u'msisdn:\+12*', group.query)
        self.assertEqual(
            self.contact_store.get_contacts_for_conversation(conversation),
            [])

    def test_smart_groups_with_matches_results(self):
        response = self.client.post(reverse('contacts:groups'), {
            'name': 'a smart group',
            'query': 'msisdn:\+12*',
            '_new_smart_group': '1',
        })

        contact = self.mkcontact()
        group = newest(self.contact_store.list_groups())
        conversation = self.user_helper.create_conversation(u'bulk_message')
        conversation.groups.add(group)
        conversation.save()

        self.assertRedirects(response, group_url(group.key))
        self.assertEqual(u'a smart group', group.name)
        self.assertEqual(u'msisdn:\+12*', group.query)
        self.assertEqual(
            list(self.contact_store.get_static_contact_keys_for_group(group)),
            [])
        self.assertEqual(
            list(self.contact_store.get_dynamic_contact_keys_for_group(group)),
            [contact.key])
        self.assertEqual(
            self.contact_store.get_contacts_for_conversation(conversation),
            [contact.key])

    def test_smart_groups_with_matches_AND_query_results(self):
        self.client.post(reverse('contacts:groups'), {
            'name': 'a smart group',
            'query': 'name:foo AND surname:bar',
            '_new_smart_group': '1',
        })

        self.mkcontact(surname='bar'),
        self.mkcontact(name='foo'),
        match = self.mkcontact(name='foo', surname='bar')

        group = newest(self.contact_store.list_groups())
        conversation = self.user_helper.create_conversation(u'bulk_message')
        conversation.groups.add(group)
        conversation.save()

        contacts = self.contact_store.get_contacts_for_conversation(
            conversation)
        self.assertEqual(contacts, [match.key])

    def test_smart_groups_with_matches_OR_query_results(self):
        self.client.post(reverse('contacts:groups'), {
            'name': 'a smart group',
            'query': 'name:foo OR surname:bar',
            '_new_smart_group': '1',
        })

        contact1 = self.mkcontact(surname='bar')
        contact2 = self.mkcontact(name='foo')
        contact3 = self.mkcontact(name='foo', surname='bar')

        group = newest(self.contact_store.list_groups())
        conv = self.user_helper.create_conversation(u'bulk_message')
        conv.groups.add(group)
        conv.save()

        self.assertEqual(
            set(self.contact_store.get_contacts_for_conversation(conv)),
            set([contact1.key, contact2.key, contact3.key]))

    def test_smart_group_limit(self):
        for i in range(10):
            self.mkcontact(name=u'Ben')

        self.client.post(reverse('contacts:groups'), {
            'name': 'a smart group',
            'query': 'name:Ben',
            '_new_smart_group': '1',
        })
        group = newest(self.contact_store.list_groups())

        default_limit = self.client.get('%s?query=Ben' % (
            reverse('contacts:group', kwargs={
                'group_key': group.key,
            }),))
        self.assertContains(
            default_limit,
            escape("Showing 10 of the group's 10 contact(s)"))

        custom_limit = self.client.get('%s?query=Ben&limit=5' % (
            reverse('contacts:group', kwargs={
                'group_key': group.key,
            }),))
        self.assertContains(
            custom_limit,
            escape("Showing 5 of the group's 10 contact(s)"))

    def test_smartgroup_contact_export(self):
        self.client.post(reverse('contacts:groups'), {
            'name': 'a smart group',
            'query': 'name:foo OR surname:bar',
            '_new_smart_group': '1',
        })

        self.mkcontact(surname='bar')
        self.mkcontact(name='foo')
        self.mkcontact(name='foo', surname='bar')

        group = newest(self.contact_store.list_groups())
        group_url = reverse('contacts:group', kwargs={
            'group_key': group.key,
        })
        self.assertEqual(group.name, 'a smart group')
        response = self.client.post(group_url, {'_export': True})

        contacts = get_all_contact_keys_for_group(self.contact_store, group)
        self.assertEqual(len(contacts), 3)

        self.assertRedirects(response, group_url)
        self.assertEqual(len(mail.outbox), 1)
        [email] = mail.outbox
        [(file_name, contents, mime_type)] = email.attachments

        self.assertEqual(file_name, 'contacts-export.zip')

        zipfile = ZipFile(StringIO(contents), 'r')
        csv_contents = zipfile.open('contacts-export.csv', 'r').read()

        self.assertEqual(email.recipients(), [self.user_email])
        self.assertTrue(
            '%s contacts export' % (group.name,) in email.subject)
        self.assertTrue(
            '%s contact(s) from group "%s" attached' % (
                len(contacts), group.name) in email.body)
        self.assertEqual(file_name, 'contacts-export.zip')
        self.assertTrue(csv_contents)
        self.assertEqual(mime_type, 'application/zip')

class TestFieldNormalizer(GoDjangoTestCase):

    def setUp(self):
        self.fn = FieldNormalizer()

    def assertNormalizedMsisdn(self, country_code, value, expected,
                               instance_type=unicode):
        normalized = self.fn.do_msisdn(value, country_code)
        self.assertEqual(normalized, expected)
        self.assertTrue(isinstance(normalized, instance_type))

    def assertNormalized(self, name, value, expected, instance_type=None):
        normalized = self.fn.normalize(name, value)
        self.assertEqual(normalized, expected)
        if instance_type:
            self.assertTrue(isinstance(normalized, instance_type))

    def test_msisdns(self):
        self.assertNormalizedMsisdn('27', '761234567', '+27761234567')
        self.assertNormalizedMsisdn('27', '761234567.0', '+27761234567')
        self.assertNormalizedMsisdn('27', '0761234567', '+27761234567')
        self.assertNormalizedMsisdn('27', '27761234567', '+27761234567')
        self.assertNormalizedMsisdn('27', '0027761234567', '+27761234567')
        self.assertNormalizedMsisdn('27', '+27761234567', '+27761234567')
        self.assertNormalizedMsisdn('27', 761234567, '+27761234567')
        self.assertNormalizedMsisdn('27', 761234567.0, '+27761234567')
        self.assertNormalizedMsisdn('27', 27761234567, '+27761234567')
        self.assertNormalizedMsisdn('27', 2.74727E+10, '+27472700000')
        self.assertNormalizedMsisdn('27', '2.74727E+10', '+27472700000')

    def test_internationalized_msisdn(self):
        self.assertNormalized('msisdn_int', '0027761234567', '+27761234567',
                              unicode)
        self.assertNormalized('msisdn_int', '27761234567', '+27761234567',
                              unicode)
        self.assertNormalized('msisdn_int', 27761234567, '+27761234567',
                              unicode)
        self.assertNormalized('msisdn_int', '+27761234567', '+27761234567',
                              unicode)
        self.assertNormalized('msisdn_int', '2.74727E+10', '+27472700000')
        self.assertNormalized('msisdn_int', 2.74727E+10, '+27472700000')

    def test_integer(self):
        self.assertNormalized('integer', '0.1', 0, int)
        self.assertNormalized('integer', '1.1', 1, int)
        self.assertNormalized('integer', 2.1, 2, int)
        self.assertNormalized('integer', '', '', str)
        self.assertNormalized('integer', 'None', 'None', str)
        self.assertNormalized('integer', None, None)

    def test_float(self):
        self.assertNormalized('float', '0.1', 0.1, float)
        self.assertNormalized('float', '1.1', 1.1, float)
        self.assertNormalized('float', 2.1, 2.1, float)
        self.assertNormalized('float', '', '', str)
        self.assertNormalized('float', 'None', 'None', str)
        self.assertNormalized('float', None, None)

    def test_string(self):
        self.assertNormalized('string', 761234567.0, '761234567.0', unicode)
        self.assertNormalized('string', '1.1', '1.1', unicode)
        self.assertNormalized('string', '', '', unicode)
        self.assertNormalized('string', 'None', 'None', unicode)
        self.assertNormalized('string', None, None)

    def test_unknown(self):
        self.assertNormalized('foo', 761234567.0, 761234567.0, float)
        self.assertNormalized('bar', '1.1', '1.1', str)
        self.assertNormalized('bas', u'1.1', u'1.1', unicode)
        self.assertNormalized('baz', '', '', str)
        self.assertNormalized('fubar', 'None', 'None', str)
        self.assertNormalized('zab', None, None)


text:
package coverage

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"strings"
	"time"

	"github.com/google/go-github/v26/github"
	"golang.org/x/tools/cover"

	"istio.io/bots/policybot/pkg/blobstorage"
	"istio.io/bots/policybot/pkg/gh"
	"istio.io/bots/policybot/pkg/storage"
	"istio.io/pkg/log"
)

var scope = log.RegisterScope("coverage", "Coverage client", 0)

var (
	Pending    = "pending"
	Success    = "success"
	Error      = "error"
	Failure    = "failure"
	statusName = "istio-testing/coverage"
)

type profiles map[string]*cover.Profile

const (
	e2eType   = "e2e"
	integType = "integ"
	unitType  = "unit"
	covPassed = "All coverage checks passed."

	signature = "\n\nCoverage checks brought to you by your local testing bot."
)

// Client handles all aspects of gathering and reporting code coverage.
type Client struct {
	OrgLogin, Repo string
	Bucket         string
	BlobClient     blobstorage.Store
	StorageClient  storage.Store
	GithubClient   *gh.ThrottledClient
}

func logTime(sha string, start time.Time) {
	scope.Infof("coverage check for %s completed in %s", sha, time.Since(start))
}

// CheckCoverage checks to see if all pending tests have completed and compiles
// coverage information if so.
func (c *Client) CheckCoverage(ctx context.Context, pr *github.PullRequest, sha string) error {
	defer logTime(sha, time.Now())
	resp, _, err := c.GithubClient.ThrottledCall(
		func(client *github.Client) (interface{}, *github.Response, error) {
			return client.Repositories.GetCombinedStatus(ctx, c.OrgLogin, c.Repo, sha, nil)
		})
	if err != nil {
		return fmt.Errorf("coverage: error looking up statuses for %s/%s commit %s: %v",
			c.OrgLogin, c.Repo, sha, err)
	}
	statuses := resp.(*github.CombinedStatus)
	hasCoverageStatus := false
	for _, status := range statuses.Statuses {
		if status.GetContext() == statusName {
			hasCoverageStatus = true
		} else if status.GetCreator().GetLogin() == "istio-testing" &&
			status.GetContext() != "tide" && status.GetState() == "pending" {
			scope.Infof("skipping coverage check for %s/%s commit %s, which has pending statuses",
				c.OrgLogin, c.Repo, sha)
			return nil
		}
	}
	if !hasCoverageStatus {
		scope.Infof("skipping coverage check for %s/%s commit %s, which has no coverage status",
			c.OrgLogin, c.Repo, sha)
		c.SetCoverageStatus(ctx, sha, Success, covPassed)
		return nil
	}

	// Get all test results for the commit from the database.
	var coverageResults []*storage.TestResult
	err = c.StorageClient.QueryTestResultsBySHA(
		ctx, c.OrgLogin, c.Repo, sha,
		func(result *storage.TestResult) error {
			if hasCoverage(result) {
				coverageResults = append(coverageResults, result)
			}
			return nil
		})
	if err != nil {
		c.SetCoverageStatus(ctx, sha, Error, "Could not fetch test results.")
		return fmt.Errorf("coverage: error fetching test results for %s/%s commit %s: %v",
			c.OrgLogin, c.Repo, sha, err)
	}

	// Download, parse, and synthesize all coverage files. Unfortunately, we
	// have to download to temp files to process them with Go's coverage
	// package.
	b := c.BlobClient.Bucket(c.Bucket)
	tmpDir, err := os.MkdirTemp("", "coverage")
	if err != nil {
		c.SetCoverageStatus(ctx, sha, Error, "Could not create temp directory for coverage files.")
		return fmt.Errorf("coverage: error creating temp dir for coverage files")
	}
	defer os.RemoveAll(tmpDir)
	covMap := make(map[*storage.TestResult]profiles)
	for _, result := range coverageResults {
		covMap[result], err = c.getProfilesForTestResult(ctx, result, b, tmpDir)
		if err != nil {
			return err
		}
	}

	covData, err := c.getCoverageDataFromProfiles(sha, covMap, pr)
	if err != nil {
		c.SetCoverageStatus(ctx, sha, Error, "Could not generate coverage data from profiles.")
		return fmt.Errorf("coverage: error generating coverage data from profiles: %v", err)
	}
	err = c.StorageClient.WriteCoverageData(ctx, covData)
	if err != nil {
		c.SetCoverageStatus(ctx, sha, Error, "Could not write coverage data to storage.")
		return fmt.Errorf("coverage: error writing coverage data to storage: %v", err)
	}
	res := c.checkCoverageDiff(ctx, pr, sha)
	c.SetCoverageStatus(ctx, sha, res.GetGithubStatus(), res.GetDescription())
	comment := res.GetComment()
	if comment != "" {
		if err := c.GithubClient.AddOrReplaceBotComment(ctx, c.OrgLogin, c.Repo, pr.GetNumber(), pr.User.GetLogin(), comment, signature); err != nil {
			return fmt.Errorf("coverage: error writing Github comment: %v", err)
		}
	}
	return nil
}

func (c *Client) getProfilesForTestResult(
	ctx context.Context,
	r *storage.TestResult,
	b blobstorage.Bucket,
	tmpDir string,
) (profiles, error) {
	merged := make(profiles)
	for _, artifact := range r.Artifacts {
		if isCoverageArtifact(artifact) {
			f, err := os.CreateTemp(tmpDir, "coverage-*.cov")
			if err != nil {
				return nil, fmt.Errorf("coverage: error creating coverage temp file: %v", err)
			}
			r, err := b.Reader(ctx, artifact)
			if err != nil {
				return nil, fmt.Errorf("coverage: error reading coverage file %s/%s from blob storage: %v",
					c.Bucket, artifact, err)
			}
			_, err = io.Copy(f, r)
			if err != nil {
				return nil, fmt.Errorf("coverage: error writing coverage file %s/%s: %v",
					c.Bucket, artifact, err)
			}
			profs, err := cover.ParseProfiles(f.Name())
			if err != nil {
				return nil, fmt.Errorf("coverage: error parsing coverage file for %s/%s: %v",
					c.Bucket, artifact, err)
			}
			err = mergeProfiles(merged, indexProfiles(profs))
			if err != nil {
				return nil, fmt.Errorf("coverage: error merging coverage profiles for %s/%s: %v",
					c.Bucket, artifact, err)
			}
		}
	}
	return merged, nil
}

type aggregate struct {
	p profiles
	c time.Time
}

func (c *Client) getCoverageDataFromProfiles(
	sha string,
	covMap map[*storage.TestResult]profiles,
	pr *github.PullRequest,
) ([]*storage.CoverageData, error) {
	var covs []*storage.CoverageData
	testTypeCov := map[string]*aggregate{
		e2eType:   {make(profiles), time.Time{}},
		integType: {make(profiles), time.Time{}},
		unitType:  {make(profiles), time.Time{}},
	}
	// Ignore errors, we just won't generate any extra aggregate profiles.
	cfg, _ := GetConfig(c.OrgLogin, c.Repo)
	addlLabels := getCustomLabels(cfg)
	for _, label := range addlLabels {
		testTypeCov[label] = &aggregate{make(profiles), time.Time{}}
	}

	// In theory, we can probably just use GetRef instead of operating over GetLabel, but
	// there really is not documentation explaining the difference between the second part
	// of the label and the ref. The best I can tell is that perhaps the PR can be based
	// on an older ref in the branch, which would cause the ref to be a SHA instead of
	// a branch name.
	label := pr.GetBase().GetLabel()
	branch := label[strings.Index(label, ":")+1:]
	// Create a merged profile for each test type and generate CoverageData entries for each
	// individual test.
	for r, profs := range covMap {
		testType := getTestType(r.TestName)
		for label, info := range testTypeCov {
			if strings.Contains(label, testType) {
				if err := mergeProfiles(info.p, profs); err != nil {
					return nil, err
				}
				// Use the last finish time for a given test type as the aggregate finish time.
				if r.FinishTime.After(info.c) {
					info.c = r.FinishTime
				}
			}
		}
		covs = append(covs,
			getCoverageDataFromProfiles(c.OrgLogin, c.Repo, branch, sha, r.TestName, r.FinishTime, profs)...)
	}
	// Now create CoverageData entries representing each test type aggregation.
	for testType, info := range testTypeCov {
		covs = append(covs,
			getCoverageDataFromProfiles(c.OrgLogin, c.Repo, branch, sha, testType, info.c, info.p)...)
	}
	return covs, nil
}

func getCoverageDataFromProfiles(
	org, repo, branch, sha, test string,
	completedAt time.Time,
	profs profiles,
) []*storage.CoverageData {
	packageCov := make(map[string]*storage.CoverageData)
	for name, p := range profs {
		pkg := name[0:strings.LastIndex(name, "/")]
		cov, ok := packageCov[pkg]
		if !ok {
			cov = &storage.CoverageData{
				OrgLogin:    org,
				RepoName:    repo,
				BranchName:  branch,
				PackageName: pkg,
				Sha:         sha,
				TestName:    test,
				Type:        getTestType(test),
				CompletedAt: completedAt,
			}
			packageCov[pkg] = cov
		}
		for _, block := range p.Blocks {
			cov.StmtsTotal += int64(block.NumStmt)
			if block.Count > 0 {
				cov.StmtsCovered += int64(block.NumStmt)
			}
		}
	}
	covs := make([]*storage.CoverageData, 0, len(packageCov))
	for _, cov := range packageCov {
		covs = append(covs, cov)
	}
	return covs
}

// SetCoverageStatus sets a pending coverage status for a given commit.
func (c *Client) SetCoverageStatus(ctx context.Context, sha, state, details string) {
	_, _, err := c.GithubClient.ThrottledCall(
		func(client *github.Client) (interface{}, *github.Response, error) {
			return client.Repositories.CreateStatus(ctx, c.OrgLogin, c.Repo, sha, &github.RepoStatus{
				State:       &state,
				Context:     &statusName,
				Description: &details,
			})
		})
	if err != nil {
		scope.Errorf("Failed to set pending coverage status on %s/%s for commit %s: %v",
			c.OrgLogin, c.Repo, sha, err)
	}
}

func hasCoverage(r *storage.TestResult) bool {
	for _, artifact := range r.Artifacts {
		if isCoverageArtifact(artifact) {
			return true
		}
	}
	return false
}

func isCoverageArtifact(artifact string) bool {
	return strings.HasSuffix(artifact, ".cov")
}

func getTestType(name string) string {
	if strings.HasPrefix(name, "e2e") {
		return e2eType
	} else if strings.HasPrefix(name, "integ") {
		return integType
	}
	return unitType
}

func indexProfiles(profs []*cover.Profile) profiles {
	m := make(profiles)
	for _, p := range profs {
		m[p.FileName] = p
	}
	return m
}

func mergeProfiles(a, b profiles) error {
	if len(b) == 0 {
		return nil
	}
	for _, p := range a {
		if q, ok := b[p.FileName]; ok {
			if err := mergeProfile(p, q); err != nil {
				return err
			}
		}
	}
	for _, p := range b {
		if _, ok := a[p.FileName]; !ok {
			a[p.FileName] = p
		}
	}
	return nil
}

func mergeProfile(p, q *cover.Profile) error {
	if len(p.Blocks) != len(q.Blocks) {
		return errors.New("profiles have different block lengths")
	}
	for i := 0; i < len(p.Blocks); i++ {
		bp := &p.Blocks[i]
		bq := q.Blocks[i]
		if bp.StartLine != bq.StartLine ||
			bp.StartCol != bq.StartCol ||
			bp.EndLine != bq.EndLine ||
			bp.EndCol != bq.EndCol {
			return fmt.Errorf("blocks have different boundaries: %+v %+v", bp, bq)
		}
		bp.Count += bq.Count
	}
	return nil
}

redpajama_set_name: RedPajamaGithub


text:
package io.amient.affinity.kafka

import io.amient.affinity.avro.record.AvroRecord
import org.apache.kafka.common.serialization.{Deserializer, Serde, Serializer}

import scala.reflect.runtime.universe._

/**
  * This Serde can be used for internal topics of processors. These schemas won't be registered
  * in the schema registry.
  *
  * @tparam T
  */
class InternalKafkaAvroSerde[T: TypeTag] extends Serde[T] {
  val schema = AvroRecord.inferSchema[T]
    override def configure(configs: java.util.Map[String, _], isKey: Boolean) = ()
    override def close() = ()

    override def deserializer() = new Deserializer[T] {
      override def configure(configs: java.util.Map[String, _], isKey: Boolean) = ()
      override def close() = ()
      override def deserialize(topic: String, data: Array[Byte]) = AvroRecord.read(data, schema)
    }

    override def serializer() = new Serializer[T] {
      override def configure(configs: java.util.Map[String, _], isKey: Boolean) = ()
      override def close() = ()
      override def serialize(topic: String, data: T) = AvroRecord.write(data, schema)
    }

}


text:
package com.elastisys.scale.cloudpool.vsphere.util;

import com.vmware.vim25.*;
import com.vmware.vim25.mo.ResourcePool;
import com.vmware.vim25.mo.VirtualMachine;
import org.joda.time.DateTime;

import java.rmi.RemoteException;
import java.util.Locale;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

public class MockedVm {

    private VirtualMachine vm;
    private VirtualMachineRuntimeInfo runtimeInfo;
    private int key = 1;

    public MockedVm() {
        vm = mock(VirtualMachine.class);
        runtimeInfo = mock(VirtualMachineRuntimeInfo.class);
        when(vm.getRuntime()).thenReturn(runtimeInfo);
    }

    public MockedVm withName(String name) {
        when(vm.getName()).thenReturn(name);
        return this;
    }

    public MockedVm withLaunchTime(DateTime launchTime) {
        when(runtimeInfo.getBootTime()).thenReturn(launchTime.toCalendar(Locale.ENGLISH));
        return this;
    }

    public MockedVm withPowerState(VirtualMachinePowerState powerState) {
        when(runtimeInfo.getPowerState()).thenReturn(powerState);
        return this;
    }

    public MockedVm withResourcePool(String poolName) throws RemoteException {
        ResourcePool resourcePool = mock(ResourcePool.class);
        when(vm.getResourcePool()).thenReturn(resourcePool);
        when(resourcePool.getName()).thenReturn(poolName);
        return this;
    }

    public MockedVm withMachineSize(String machineSize) {
        VirtualMachineConfigInfo config = mock(VirtualMachineConfigInfo.class);
        VirtualHardware hardware = mock(VirtualHardware.class);
        when(vm.getConfig()).thenReturn(config);
        when(config.getHardware()).thenReturn(hardware);
        when(hardware.getNumCPU()).thenReturn(2);
        when(hardware.getMemoryMB()).thenReturn(1024);
        return this;
    }

    public MockedVm withPrivateIp(String privateIp) {
        GuestInfo guestInfo = mock(GuestInfo.class);
        when(vm.getGuest()).thenReturn(guestInfo);
        when(guestInfo.getIpAddress()).thenReturn(privateIp);
        return this;
    }

    public MockedVm withTag(com.elastisys.scale.cloudpool.vsphere.tag.Tag tag) throws RemoteException {
        CustomFieldDef cfd = new CustomFieldDef();
        cfd.setName(tag.getKey());
        cfd.setKey(key);
        CustomFieldDef[] cfdArr = { cfd };
        CustomFieldStringValue cfsv = new CustomFieldStringValue();
        cfsv.setKey(key);
        cfsv.setValue(tag.getValue());
        CustomFieldStringValue[] cfsvArr = { cfsv };
        when(vm.getAvailableField()).thenReturn(cfdArr);
        when(vm.getCustomValue()).thenReturn(cfsvArr);
        return this;
    }

    public VirtualMachine build() {
        return vm;
    }
}


text:
typedef void *WorldRef;
typedef void *BodyRef;

FOUNDATION_EXTERN WorldRef B2DCreateWorld();
FOUNDATION_EXTERN void B2DReleaseWorld(WorldRef world);

FOUNDATION_EXTERN void B2DStep(WorldRef world);

FOUNDATION_EXTERN BodyRef B2DCreateStaticBody(WorldRef world, B2DVector position, B2DVector size);
FOUNDATION_EXTERN BodyRef B2DCreateDynamicBody(WorldRef world, B2DVector position, B2DVector size);

FOUNDATION_EXTERN B2DVector B2DBodyPosition(BodyRef body);
FOUNDATION_EXTERN B2DVector B2DBodyVelocity(BodyRef body);
FOUNDATION_EXTERN float B2DBodyMass(BodyRef body);

FOUNDATION_EXTERN void B2DBodyApplyImpulse(BodyRef body, B2DVector impulse);


text:
import imaplib
import email
import re
from dateutil import parser
from Stephanie.Modules.base_module import BaseModule

class GmailModule(BaseModule):
    def __init__(self, *args):
        super(GmailModule, self).__init__(*args)
        self.gmail_address = self.get_configuration('gmail_address')
        self.password = self.get_configuration('gmail_password')
        self.conn = None
        if self.gmail_address and self.password:
            self.do_init()
        else:
            return False

    def do_init(self):
        try:
            self.conn = imaplib.IMAP4_SSL('imap.gmail.com')
            self.conn.debug = 0
            self.conn.login(self.gmail_address, self.password)
        except:
            response = ("Either your credentials are wrong mate, or there is some problem going on, do me a favor, I know "
                        "you won't but whatever, just inform me in the forums.")
            print(response)
            return response

    def fetch_unread_emails(self, since=None, markRead=False, limit=None):
        """
            Fetches a list of unread email objects from a user's Gmail inbox.
            Arguments:
            profile -- contains information related to the user (e.g., Gmail
                       address)
            since -- if provided, no emails before this date will be returned
            markRead -- if True, marks all returned emails as read in target inbox
            Returns:
            A list of unread email objects.
        """
        self.conn.select(readonly=(not markRead))

        msgs = []
        (retcode, messages) = self.conn.search(None, '(UNSEEN)')

        if retcode == 'OK' and messages != ['']:
            num_unread = len(messages[0].split())
            # if limit and num_unread > limit:
            #     return num_unread
            i = 0
            for num in messages[0].split():
                # parse email RFC822 format
                if i > 5:
                    break
                ret, data = self.conn.fetch(num, '(RFC822)')
                correct_format_message = data[0][1].decode("utf-8")
                msg = email.message_from_string(correct_format_message)

                if not since or self.get_date(msg) > since:
                    msgs.append(msg)
                i += 1
        self.conn.close()
        self.conn.logout()

        return num_unread, msgs

    @staticmethod
    def get_sender(email_found):
        """
            Returns the best-guess sender of an email.
            Arguments:
            email -- the email whose sender is desired
            Returns:
            Sender of the email.
        """
        sender = email_found['From']
        m = re.match(r'(.*)\s<.*>', sender)
        if m:
            return m.group(1)
        return sender

    @staticmethod
    def get_date(email_found):
        return parser.parse(email_found.get('date'))

    @staticmethod
    def get_most_recent_date(emails):
        """
            Returns the most recent date of any email in the list provided.
            Arguments:
            emails -- a list of emails to check
            Returns:
            Date of the most recent email.
        """
        dates = [GmailModule.get_date(e) for e in emails]
        dates.sort(reverse=True)
        if dates:
            return dates[0]
        return None

    def handle(self):
        """
            Responds to user-input, typically speech text, with a summary of
            the user's Gmail inbox, reporting on the number of unread emails
            in the inbox, as well as their senders.
            Arguments:
            text -- user-input, typically transcribed speech
            self.assistant -- used to interact with the user (for both input and output)
            profile -- contains information related to the user (e.g., Gmail
                       address)
        """
        try:
            num_unread, msgs = self.fetch_unread_emails(limit=5)

            if num_unread > 5:
                response = "You have %d unread emails, out of which 5 latest ones are as follows, please wait a second, as I process" % num_unread
                self.assistant.say(response)
            senders = []
            for e in msgs:
                senders.append(self.get_sender(e))
        except imaplib.IMAP4.error:
            return "I'm sorry. I'm not authenticated to work with your Gmail."

        if not senders:
            return "You have no unread emails."
        elif len(senders) == 1:
            return "You have one unread email from " + senders[0] + "."
        else:
            response = "You have %d unread emails" % len(
                senders)
            unique_senders = list(set(senders))
            if len(unique_senders) > 1:
                unique_senders[-1] = 'and ' + unique_senders[-1]
                response += ". Senders include: "
                response += '...'.join(senders)
            else:
                response += " from " + unique_senders[0]

            return response


text:
layout: post
title: Earth/Sound/Apps
date: '2014-04-22T15:42:00-04:00'
tags:
- Sound
- environment
- earth day
- teaching
redirect_from: /post/83541782161/help-scientists-record-one-day-of-sound-on-earth/
---

From [Help Scientists Record One Day of Sound on Earth](http://www.wired.com/2014/04/soundscape-recorder/):

> Bryan Pijanowski wants to capture the sounds of the world on a single  day, and he needs your help. Beginning on Earth Day of this year,  Pijanowski hopes to enlist thousands of people in recording a few  minutes of their everyday surroundings with his Soundscape Recorder  smartphone app.

Though I have yet to meet Bryan, I've been lucky enough to recently meet some of the people working directly with him (some at [the recent workshop that I co-organized](http://sites.psu.edu/soundscienceenvironment2014/)). I actually used the [Soundscape Recorder](https://itunes.apple.com/us/app/soundscape-recorder/id836741158?mt=8) app this morning with my students, and it was a really successful and interesting exercise. Luckily, most of my students have smart phones—though not all of them had the up-to-date operating systems required for the app—so we paired up and walked around Penn State's campus capturing some local sounds, from the Creamery to the Arboretum.

After the recording process, the app asks you a bunch of questions, ranging from the factual (what did you hear?) to the loaded (how did it make you feel?). This gave the class a chance to discuss both what we recorded and the assumptions inherent in the act of recording and built-in to the app itself.


text:
<!--
  ~ Licensed to the Apache Software Foundation (ASF) under one or more
  ~ contributor license agreements.  See the NOTICE file distributed with
  ~ this work for additional information regarding copyright ownership.
  ~ The ASF licenses this file to You under the Apache License, Version 2.0
  ~ (the "License"); you may not use this file except in compliance with
  ~ the License.  You may obtain a copy of the License at
  ~
  ~     http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  ~
  -->

<code_scheme name="sky-walking">
    <option name="USE_SAME_INDENTS" value="true"/>
    <option name="IGNORE_SAME_INDENTS_FOR_LANGUAGES" value="true"/>
    <option name="OTHER_INDENT_OPTIONS">
        <value>
            <option name="INDENT_SIZE" value="4"/>
            <option name="CONTINUATION_INDENT_SIZE" value="4"/>
            <option name="TAB_SIZE" value="4"/>
            <option name="USE_TAB_CHARACTER" value="false"/>
            <option name="SMART_TABS" value="false"/>
            <option name="LABEL_INDENT_SIZE" value="0"/>
            <option name="LABEL_INDENT_ABSOLUTE" value="false"/>
            <option name="USE_RELATIVE_INDENTS" value="false"/>
        </value>
    </option>
    <option name="PREFER_LONGER_NAMES" value="false"/>
    <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="1000"/>
    <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="1000"/>
    <option name="PACKAGES_TO_USE_IMPORT_ON_DEMAND">
        <value/>
    </option>
    <option name="IMPORT_LAYOUT_TABLE">
        <value>
            <package name="" withSubpackages="true" static="false"/>
            <emptyLine/>
            <package name="" withSubpackages="true" static="true"/>
        </value>
    </option>
    <option name="JD_ALIGN_PARAM_COMMENTS" value="false"/>
    <option name="JD_ALIGN_EXCEPTION_COMMENTS" value="false"/>
    <option name="JD_P_AT_EMPTY_LINES" value="false"/>
    <option name="JD_KEEP_INVALID_TAGS" value="false"/>
    <option name="JD_DO_NOT_WRAP_ONE_LINE_COMMENTS" value="true"/>
    <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
    <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
    <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
    <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
    <option name="WHILE_ON_NEW_LINE" value="true"/>
    <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
    <option name="ALIGN_MULTILINE_FOR" value="false"/>
    <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
    <option name="SPACE_BEFORE_ARRAY_INITIALIZER_LBRACE" value="true"/>
    <option name="METHOD_PARAMETERS_WRAP" value="1"/>
    <option name="ARRAY_INITIALIZER_LBRACE_ON_NEXT_LINE" value="true"/>
    <option name="LABELED_STATEMENT_WRAP" value="1"/>
    <option name="WRAP_COMMENTS" value="true"/>
    <option name="METHOD_ANNOTATION_WRAP" value="1"/>
    <option name="CLASS_ANNOTATION_WRAP" value="1"/>
    <option name="FIELD_ANNOTATION_WRAP" value="1"/>
    <JavaCodeStyleSettings>
        <option name="CLASS_NAMES_IN_JAVADOC" value="3"/>
    </JavaCodeStyleSettings>
    <XML>
        <option name="XML_LEGACY_SETTINGS_IMPORTED" value="true"/>
    </XML>
    <ADDITIONAL_INDENT_OPTIONS fileType="haml">
        <option name="INDENT_SIZE" value="2"/>
    </ADDITIONAL_INDENT_OPTIONS>
    <codeStyleSettings language="Groovy">
        <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
        <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
        <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
        <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
        <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
        <option name="ALIGN_MULTILINE_FOR" value="false"/>
        <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
        <option name="METHOD_PARAMETERS_WRAP" value="1"/>
        <option name="METHOD_ANNOTATION_WRAP" value="1"/>
        <option name="CLASS_ANNOTATION_WRAP" value="1"/>
        <option name="FIELD_ANNOTATION_WRAP" value="1"/>
        <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
        <indentOptions>
            <option name="CONTINUATION_INDENT_SIZE" value="4"/>
        </indentOptions>
    </codeStyleSettings>
    <codeStyleSettings language="HOCON">
        <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
        <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
    </codeStyleSettings>
    <codeStyleSettings language="JAVA">
        <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
        <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
        <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
        <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
        <option name="WHILE_ON_NEW_LINE" value="true"/>
        <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
        <option name="ALIGN_MULTILINE_FOR" value="false"/>
        <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
        <option name="SPACE_BEFORE_ARRAY_INITIALIZER_LBRACE" value="true"/>
        <option name="METHOD_PARAMETERS_WRAP" value="1"/>
        <option name="ARRAY_INITIALIZER_LBRACE_ON_NEXT_LINE" value="true"/>
        <option name="LABELED_STATEMENT_WRAP" value="1"/>
        <option name="METHOD_ANNOTATION_WRAP" value="1"/>
        <option name="CLASS_ANNOTATION_WRAP" value="1"/>
        <option name="FIELD_ANNOTATION_WRAP" value="1"/>
        <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
        <indentOptions>
            <option name="CONTINUATION_INDENT_SIZE" value="4"/>
        </indentOptions>
    </codeStyleSettings>
    <codeStyleSettings language="JSON">
        <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
        <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
    </codeStyleSettings>
    <codeStyleSettings language="Scala">
        <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
        <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
        <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
        <option name="WHILE_ON_NEW_LINE" value="true"/>
        <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
        <option name="ALIGN_MULTILINE_FOR" value="false"/>
        <option name="METHOD_PARAMETERS_WRAP" value="1"/>
        <option name="METHOD_ANNOTATION_WRAP" value="1"/>
        <option name="CLASS_ANNOTATION_WRAP" value="1"/>
        <option name="FIELD_ANNOTATION_WRAP" value="1"/>
        <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
        <indentOptions>
            <option name="INDENT_SIZE" value="4"/>
            <option name="CONTINUATION_INDENT_SIZE" value="4"/>
            <option name="TAB_SIZE" value="4"/>
        </indentOptions>
    </codeStyleSettings>
    <codeStyleSettings language="XML">
        <indentOptions>
            <option name="CONTINUATION_INDENT_SIZE" value="4"/>
        </indentOptions>
    </codeStyleSettings>
</code_scheme>


text:
int main()
{
  int size; scanf("%d", &size);
  Graph_t *graph = createGraph(size);

  int ini, fim;
  while (scanf("%d %d", &ini, &fim) != EOF)
    addEdge(graph, ini, fim);

  //printGraph(graph);
  //printf("\nDFS:\n"); dfs(graph, 1);

  printf("\nTopological Sort with DFS:\n"); topologicalSortDFS(graph);
  printf("\nTopological Sort with BFS:\n"); topologicalSortBFS(graph);

  destroyGraph(graph);
  return(0);
}


text:
import { Injectable } from '@angular/core';
import { of as observableOf,  Observable } from 'rxjs';
import { TemperatureHumidityData, Temperature } from '../data/temperature-humidity';

@Injectable()
export class TemperatureHumidityService extends TemperatureHumidityData {

  private temperatureDate: Temperature = {
    value: 24,
    min: 12,
    max: 30,
  };

  private humidityDate: Temperature = {
    value: 87,
    min: 0,
    max: 100,
  };

  getTemperatureData(): Observable<Temperature> {
    return observableOf(this.temperatureDate);
  }

  getHumidityData(): Observable<Temperature> {
    return observableOf(this.humidityDate);
  }
}


text:
Latex Templates — Geodäsie und Geoinformatik, Universität Stuttgart

##usage
See `example_document.pdf` for more details.


text:
<?php

namespace Byteout\QuizBundle\Controller;

use Sensio\Bundle\FrameworkExtraBundle\Configuration\Route;
use Symfony\Bundle\FrameworkBundle\Controller\Controller;

class DefaultController extends Controller
{
    /**
     * @Route("/")
     *
     * @param $name
     * @return \Symfony\Component\HttpFoundation\Response
     */
    public function indexAction()
    {
        return $this->render('ByteoutQuizBundle:Default:index.html.twig', array('name' => 'test'));
    }
}