Merge lp:~canonical-isd-hackers/canonical-identity-provider/improve-coverage into lp:canonical-identity-provider/release

[Łukasz Czyżykowski]

Improve test coverage to 98%

[Szilveszter Farkas (phanatic)]

identityprovider/tests/test_admin.py: the 'as_administrator' decorator is very elegant, but for the sake of having a consistent coding style, I'd prefer using setUp/tearDown for that (as it's used in other test modules).

There are also a few points that make pylint/pyflakes unhappy, but I don't have a full report available, so I don't want to block this based on that.

review: Needs Fixing

[Szilveszter Farkas (phanatic)]

Thanks for the update!

review: Approve

=== modified file 'identityprovider/admin.py'
--- identityprovider/admin.py	2010-04-29 10:17:02 +0000
+++ identityprovider/admin.py	2010-06-01 12:03:28 +0000
@@ -1,17 +1,19 @@
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+from django import forms
 from django.contrib import admin
-from django import forms
 from django.utils.safestring import mark_safe
 from django.utils.translation import ugettext_lazy as _
 
+from identityprovider.const import SREG_LABELS
+from identityprovider.fields import CommaSeparatedField
 from identityprovider.models import (Account, AccountPassword, EmailAddress,
                                      OpenIDRPConfig, APIUser)
+from identityprovider.models.const import AccountStatus, EmailStatus
 from identityprovider.utils import encrypt_launchpad_password
-from identityprovider.const import SREG_LABELS
-from identityprovider.models.const import AccountStatus, EmailStatus
-from identityprovider.fields import CommaSeparatedField
+from identityprovider.widgets import (StatusWidget, ReadOnlyDateTimeWidget,
+    LPUsernameWidget, account_to_lp_link)
 
 
 class OpenIDRPConfigForm(forms.ModelForm):
@@ -100,35 +102,6 @@
     verbose_name_plural = _("Email addresses")
 
 
-class StatusWidget(forms.Select):
-
-    date_status_set = None
-
-    def render(self, name, value, attrs=None, choices=()):
-        select = super(StatusWidget, self).render(name, value, attrs, choices)
-        if self.date_status_set:
-            status_date = _("Set on %s") % self.date_status_set
-        else:
-            status_date = ""
-        return mark_safe("%s %s" % (select, status_date))
-
-
-class ReadOnlyDateTime(forms.widgets.Widget):
-
-    value = None
-
-    def render(self, name, value, attrs=None):
-        return str(self.value)
-
-
-class LPUsernameWidget(forms.widgets.Widget):
-
-    account = None
-
-    def render(self, name, value, attrs=None):
-        return mark_safe(account_to_lp_link(self.account))
-
-
 class LPUsernameField(forms.Field):
     widget = LPUsernameWidget
 
@@ -151,7 +124,8 @@
             widget.account = instance
         super(AccountAdminForm, self).__init__(*args, **kwargs)
 
-    date_created = forms.DateTimeField(widget=ReadOnlyDateTime, required=False)
+    date_created = forms.DateTimeField(widget=ReadOnlyDateTimeWidget,
+                                       required=False)
     lp_username = LPUsernameField(label=_("LP username"), required=False)
     status = forms.TypedChoiceField(widget=StatusWidget,
                                     choices=AccountStatus._get_choices())
@@ -162,13 +136,6 @@
                                         widget=forms.TextInput)
 
 
-def account_to_lp_link(account):
-    if account and account.person:
-        return ('<a href="https://launchpad.net/~%s">%s</a>' %
-                    (account.person.name, account.person.name))
-        return ""
-
-
 class AccountAdmin(admin.ModelAdmin):
     inlines = [AccountPasswordInline, EmailAddressInline]
 

=== modified file 'identityprovider/backend/base.py'
--- identityprovider/backend/base.py	2010-04-21 15:29:24 +0000
+++ identityprovider/backend/base.py	2010-06-01 12:03:28 +0000
@@ -86,8 +86,8 @@
         """django_session is always updated in the same way, so we can map
         that in to a row in the database.
         """
-        pkey_cond = '"%s"."%s" = %%s ' % (self.table, self.primary_key)
-        assert pkey_cond == match.group('cond')
+        pkey_cond = '"%s"."%s" = %%s' % (self.table, self.primary_key)
+        assert pkey_cond == match.group('cond').strip()
         update_format = '"session_data" = %s, "expire_date" = %s'
         assert update_format == match.group('cols')
         key = self.cache_key(params[2])
@@ -170,10 +170,7 @@
             return self.cursor.fetchmany(chunk)
 
     def __getattr__(self, attr):
-        if attr in self.__dict__:
-            return self.__dict__[attr]
-        else:
-            return getattr(self.cursor, attr)
+        return getattr(self.cursor, attr)
 
 
 class DatabaseWrapper(PostgresDatabaseWrapper):

=== modified file 'identityprovider/models/account.py'
--- identityprovider/models/account.py	2010-05-13 10:45:42 +0000
+++ identityprovider/models/account.py	2010-06-01 12:03:28 +0000
@@ -29,7 +29,7 @@
     an Account, and the necessary AccountPassword, EmailAddress objects. """
 
     def create_account(self, displayname, username, password,
-                       creation_rationale=None):
+                       creation_rationale=None, salt=None):
         from identityprovider.models import EmailAddress
         if creation_rationale is None:
             creation_rationale = \
@@ -44,7 +44,7 @@
             status=EmailStatus.PREFERRED)
         password = AccountPassword.objects.create(
             account=account,
-            password=encrypt_launchpad_password(password))
+            password=encrypt_launchpad_password(password, salt=salt))
         return account
 
     def get_by_email(self, email):

=== modified file 'identityprovider/models/api.py'
--- identityprovider/models/api.py	2010-04-22 15:29:52 +0000
+++ identityprovider/models/api.py	2010-06-01 12:03:28 +0000
@@ -15,8 +15,8 @@
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 
-    def set_password(self, password):
-        self.password = encrypt_launchpad_password(password)
+    def set_password(self, password, salt=None):
+        self.password = encrypt_launchpad_password(password, salt=salt)
 
     def verify_password(self, password):
         return validate_launchpad_password(password, self.password)

=== modified file 'identityprovider/models/authtoken.py'
--- identityprovider/models/authtoken.py	2010-05-13 14:53:17 +0000
+++ identityprovider/models/authtoken.py	2010-06-01 12:03:28 +0000
@@ -173,9 +173,11 @@
     """
     token = create_token(token_length)
     try:
-        while obj.objects.get(column=token) is not None:
+        params = {column: token}
+        while obj.objects.get(**params) is not None:
             token = create_token(token_length)
-    except:
+            params = {column: token}
+    except Exception, e:
         pass
     return token
 

=== added file 'identityprovider/models/fixtures/admin.json'
--- identityprovider/models/fixtures/admin.json	1970-01-01 00:00:00 +0000
+++ identityprovider/models/fixtures/admin.json	2010-06-01 12:03:28 +0000
@@ -0,0 +1,20 @@
+[
+    {
+        "pk": 1,
+        "model": "auth.user",
+        "fields": {
+            "username": "admin",
+            "first_name": "",
+            "last_name": "",
+            "is_active": true,
+            "is_superuser": true,
+            "is_staff": true,
+            "last_login": "2010-05-27 12:45:14",
+            "groups": [],
+            "user_permissions": [],
+            "password": "sha1$1639f$b6ab1ef76b7cf760e8431be7ab61a25dc4a433ac",
+            "email": "[e-mail]",
+            "date_joined": "2010-05-27 12:45:14"
+        }
+    }
+]

=== modified file 'identityprovider/models/openidmodels.py'
--- identityprovider/models/openidmodels.py	2010-04-21 15:29:24 +0000
+++ identityprovider/models/openidmodels.py	2010-06-01 12:03:28 +0000
@@ -270,4 +270,4 @@
     def cleanupAssociations(self):
         OpenIDAssociation.objects.extra(
             where=[
-                'issued + lifetimeint < (%s)' % time.time()]).delete()
+                '(issued + lifetime) < (%s)' % time.time()]).delete()

=== modified file 'identityprovider/models/person.py'
--- identityprovider/models/person.py	2010-04-21 15:29:24 +0000
+++ identityprovider/models/person.py	2010-06-01 12:03:28 +0000
@@ -88,8 +88,6 @@
         except TeamParticipation.DoesNotExist:
             if self.id == team.teamowner.id:
                 return True
-            elif team.is_team() and not team.teamowner.in_team(team):
-                return self.in_team(team.teamowner)
         return False
 
     @property

=== modified file 'identityprovider/readonly.py'
--- identityprovider/readonly.py	2010-05-05 17:47:40 +0000
+++ identityprovider/readonly.py	2010-06-01 12:03:28 +0000
@@ -80,10 +80,6 @@
         filename = self.marker_file_pattern % dbname
         return os.path.exists(filename)
 
-    def is_current_failed(self):
-        """Returns true if the current DB connection is failed."""
-        return self.is_failed(self.current_dbid())
-
     def get_connections(self):
         return settings.DB_CONNECTIONS
     connections = property(get_connections)

=== modified file 'identityprovider/tests/mockdb.py'
--- identityprovider/tests/mockdb.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/mockdb.py	2010-06-01 12:03:28 +0000
@@ -5,6 +5,21 @@
 from psycopg2 import DatabaseError
 
 
+class MockCursor(object):
+    def __init__(self, data=None, connection=None):
+        self.data = data
+        self.connection = connection
+
+    def execute(self, query, params):
+        return True
+
+    def fetchmany(self, size=None):
+        if size is not None:
+            return self.data[:size]
+        else:
+            return self.data
+
+
 class MockConnection(object):
     def __init__(self, connection, psycopg):
         self.connection = connection

=== modified file 'identityprovider/tests/test_admin.py'
--- identityprovider/tests/test_admin.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_admin.py	2010-06-01 12:03:28 +0000
@@ -3,15 +3,28 @@
 
 from django.contrib import admin
 from django.contrib.admin.sites import AlreadyRegistered, NotRegistered
-from django.test import TestCase
 
 from identityprovider.admin import (AccountPasswordInline,
     EmailAddressInline)
 from identityprovider.models import (Account, AccountPassword, EmailAddress,
     OpenIDRPConfig, APIUser)
-
-
-class AdminTestCase(TestCase):
+from identityprovider.models.const import EmailStatus
+from identityprovider.utils import validate_launchpad_password
+
+from utils import BasicAccountTestCase
+
+
+class AdminTestCase(BasicAccountTestCase):
+
+    fixtures = ['admin.json', 'test.json']
+
+    def setUp(self):
+        self.disable_csrf()
+        self.client.login(username="admin", password="admin007")
+
+    def tearDown(self):
+        self.reset_csrf()
+
     def test_registered_models(self):
         for model in (Account, OpenIDRPConfig, APIUser):
             self.assertRaises(AlreadyRegistered, admin.site.register, model)
@@ -23,3 +36,79 @@
         expected_inlines = [AccountPasswordInline, EmailAddressInline]
         registered_inlines = admin.site._registry[Account].inlines
         self.assertEqual(registered_inlines, expected_inlines)
+
+    def test_account_overview(self):
+        r = self.client.get('/admin/identityprovider/account/')
+        self.assertContains(r, '[e-mail]')
+
+    def test_account_change(self):
+        account_id = 1
+        url = '/admin/identityprovider/account/%s/' % account_id
+        account = Account.objects.get(pk=account_id)
+        email = account.preferredemail
+        new_email = '[e-mail]'
+        new_password = 'blah'
+        r = self.client.post(url, {
+            'creation_rationale': str(account.creation_rationale),
+            'status': str(account.status),
+            'displayname': account.displayname,
+            'openid_identifier': account.openid_identifier,
+            'accountpassword-TOTAL_FORMS': '1',
+            'accountpassword-INITIAL_FORMS': '1',
+            'accountpassword-0-id': str(account_id),
+            'accountpassword-0-account': str(account_id),
+            'accountpassword-0-password': new_password,
+            'emailaddress_set-TOTAL_FORMS': '1',
+            'emailaddress_set-INITIAL_FORMS': '1',
+            'emailaddress_set-0-id': str(email.id),
+            'emailaddress_set-0-account': str(account.id),
+            'emailaddress_set-0-email': new_email,
+            'emailaddress_set-0-status': str(email.status)})
+        # Any non-error status code would be fine here, but right now
+        # it redirects with a 302.
+        self.assertEqual(r.status_code, 302)
+        account = Account.objects.get(pk=1)
+        self.assertEqual(account.preferredemail.email, new_email)
+        self.assertTrue(validate_launchpad_password(
+            new_password, account.accountpassword.password))
+
+    def test_multiple_preferred_emails(self):
+        account_id = 1
+        url = '/admin/identityprovider/account/%s/' % account_id
+        account = Account.objects.get(pk=account_id)
+        email = account.preferredemail
+        r = self.client.post(url, {
+            'creation_rationale': str(account.creation_rationale),
+            'status': str(account.status),
+            'displayname': account.displayname,
+            'openid_identifier': account.openid_identifier,
+            'accountpassword-TOTAL_FORMS': '1',
+            'accountpassword-INITIAL_FORMS': '1',
+            'accountpassword-0-id': str(account_id),
+            'accountpassword-0-account': str(account_id),
+            'emailaddress_set-TOTAL_FORMS': '2',
+            'emailaddress_set-INITIAL_FORMS': '1',
+            'emailaddress_set-0-id': str(email.id),
+            'emailaddress_set-0-account': str(account.id),
+            'emailaddress_set-0-email': email.email,
+            'emailaddress_set-0-status': str(email.status),
+            'emailaddress_set-1-account': str(account.id),
+            'emailaddress_set-1-email': '[e-mail]',
+            'emailaddress_set-1-status': str(email.status)})
+        account = Account.objects.get(pk=1)
+        self.assertContains(r, 'Only one email address can be preferred.')
+        self.assertEqual(len(account.emailaddress_set.filter(
+            status=EmailStatus.PREFERRED)), 1)
+        self.assertEqual(account.preferredemail, email)
+
+    def test_inline_forms(self):
+        r = self.client.get('/admin/identityprovider/account/1/')
+        self.assertEqual(r.status_code, 200)
+        self.assertContains(r, '<input type="password" '
+                            'name="accountpassword-0-password"')
+
+    def test_add_api_user(self):
+        r = self.client.post('/admin/identityprovider/apiuser/add/', {
+            'username': 'api_user',
+            'password': 'backend'})
+        self.assertEqual(r.status_code, 302)

=== modified file 'identityprovider/tests/test_auth.py'
--- identityprovider/tests/test_auth.py	2010-05-10 14:07:45 +0000
+++ identityprovider/tests/test_auth.py	2010-06-01 12:03:28 +0000
@@ -20,10 +20,10 @@
     def test_authenticate_with_email_status_not_in_expected_one(self):
         email_address = EmailAddress.objects.get(
             email__iexact="[e-mail]")
-        email_address.status = EmailStatus.NEW
+        email_address.status = EmailStatus.OLD
         email_address.save()
 
-        result = self.backend.authenticate('[e-mail]', '')
+        result = self.backend.authenticate('[e-mail]', 'test')
 
         self.assertTrue(result is None)
 
@@ -120,3 +120,19 @@
 
         if created:
             consumer.delete()
+
+    def test_oauth_authenticate_stolen_token(self):
+        victim_account = Account.objects.get_by_email('[e-mail]')
+        token = create_oauth_token_for_account(victim_account, 'new-token')
+        oauth_token = token.oauth_token()
+
+        malicious_account = Account.objects.get_by_email('[e-mail]')
+        consumer, created = Consumer.objects.get_or_create(account=malicious_account)
+
+        # make sure the accounts are active
+        self.assertTrue(victim_account.status, AccountStatus.ACTIVE)
+        self.assertTrue(malicious_account.status, AccountStatus.ACTIVE)
+
+        # make sure authentication succeeds
+        response = oauth_authenticate(consumer, oauth_token, None)
+        self.assertFalse(response)

=== added file 'identityprovider/tests/test_backend_base.py'
--- identityprovider/tests/test_backend_base.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_backend_base.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,212 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import re
+import random
+
+from django.conf import settings
+from django.core.cache import cache
+from django.db.backends.postgresql_psycopg2.base import (
+    DatabaseWrapper as PostgresDatabaseWrapper)
+from django.test import TestCase
+
+from identityprovider.backend.base import (AuthUserCache, CursorReadOnlyWrapper,
+    DatabaseError, DatabaseWrapper, BaseCache, DjangoSessionCache)
+from identityprovider.readonly import readonly_manager
+from identityprovider.tests.mockdb import MockCursor
+from identityprovider.tests.test_backend_old_base import (
+    DatabaseWrapperTestCase as OldDatabaseWrapperTestCase)
+
+
+class BaseCacheTestCase(TestCase):
+
+    def setUp(self):
+        # initialize random number generator
+        random.seed(42)
+        # start with empty cache
+        cache._cache.flush_all()
+
+        self.cursor = MockCursor("'value'")
+
+        self.base_cache = BaseCache()
+        self.base_cache.table = "t"
+        self.base_cache.primary_key = "w"
+        self.match = re.match(CursorReadOnlyWrapper.select_pattern,
+                              'SELECT "a", "b" FROM "t" WHERE "t"."w" = %s')
+
+    def test_select_when_no_cached_value(self):
+        value = self.base_cache.select(self.match, ["1"], self.cursor)
+
+        self.assertEquals(value, "'value'")
+
+    def test_select_when_there_is_cached_value(self):
+        cache._cache.add(self.base_cache.cache_key("1"), "'VALUE'")
+        value = self.base_cache.select(self.match, ["1"], self.cursor)
+
+        self.assertEquals(value, "VALUE")
+
+
+class DjangoSessionCacheTestCase(TestCase):
+
+   def test_update(self):
+       mock = MockCursor("value")
+       match = re.match(
+           CursorReadOnlyWrapper.update_pattern,
+           'UPDATE "t" SET "session_data" = %s, "expire_date" = %s '
+           'WHERE "django_session"."session_key" = %s'
+       )
+       django_session_cache = DjangoSessionCache()
+       django_session_cache.update(match, ["a", "b", "c"], mock)
+
+       value = cache._cache.get(django_session_cache.cache_key("c"))
+       self.assertEquals(value, "[['c', 'a', 'b']]")
+
+
+class CursorReadOnlyWrapperTestCase(TestCase):
+    def setUp(self):
+        # initialize random number generator
+        random.seed(42)
+        # start with empty cache
+        cache._cache.flush_all()
+        self.cursor = MockCursor()
+        self.wrapper = CursorReadOnlyWrapper(self.cursor)
+
+    def test_init(self):
+        self.assertEqual(self.wrapper.cursor, self.cursor)
+        self.assertEqual(self.wrapper.cache, None)
+
+    def test_execute_cached_bad_command(self):
+        sql = ''
+        params = []
+        cached = self.wrapper.execute_cached('', sql, params)
+        self.assertFalse(cached)
+
+    def test_execute_cached_bad_sql(self):
+        sql = ''
+        params = []
+        cached = self.wrapper.execute_cached('select', sql, params)
+        self.assertFalse(cached)
+
+    def test_execute_cached_non_cached_table(self):
+        sql = 'SELECT * FROM "account" WHERE account_pkey = %s'
+        params = ['1']
+        cached = self.wrapper.execute_cached('select', sql, params)
+        self.assertFalse(cached)
+
+    def test_execute_cached_select(self):
+        sql = 'SELECT * FROM "auth_user" WHERE auth_user_pkey = %s'
+        params = ['1']
+        self.cursor.data = expected_values = [('auth_user_pkey', '1'),
+                                              ('username', 'user')]
+        cached = self.wrapper.execute_cached('select', sql, params)
+        self.assertTrue(cached)
+        self.assertEqual(self.wrapper._values, expected_values)
+
+    def test_execute_cached_insert(self):
+        sql = 'INSERT INTO "auth_user" (auth_user_pkey, username) VALUES (%s, %s)'
+        params = ['1', 'user']
+        cached = self.wrapper.execute_cached('insert', sql, params)
+        self.assertTrue(cached)
+        key = self.wrapper.cache.cache_key('1')
+        self.assertEqual(cache.get(key), str([params]))
+
+    def test_execute_cached_delete(self):
+        sql = 'DELETE FROM "auth_user" WHERE "auth_user_pkey" IN (%s)'
+        params = ['1']
+        cached = self.wrapper.execute_cached('delete', sql, params)
+        self.assertTrue(cached)
+        key = self.wrapper.cache.cache_key('1')
+        self.assertEqual(cache.get(key), '[]')
+
+    def test_execute_cached_update(self):
+        sql = 'UPDATE "auth_user" SET username=%s WHERE auth_user_pkey=%s'
+        params = ['user', '1']
+        cached = self.wrapper.execute_cached('update', sql, params)
+        self.assertTrue(cached)
+        key = self.wrapper.cache.cache_key('1')
+        self.assertEqual(cache.get(key), None)
+
+    def test_execute_when_cached(self):
+        sql = 'SELECT * FROM "auth_user" WHERE auth_user_pkey=%s'
+        params = ['1']
+        self.cursor.data = [('auth_user_pkey', '1'),
+                                              ('username', 'user')]
+        result = self.wrapper.execute(sql, params)
+        # query was cached
+        self.assertEqual(result, None)
+
+    def test_execute_select(self):
+        sql = 'SELECT * FROM "account"'
+        self.cursor.data = [('account_pkey', '1'),
+                                              ('displayname', 'Sample Person')]
+        result = self.wrapper.execute(sql)
+        # query was not cached
+        self.assertTrue(result)
+
+    def test_execute_modify(self):
+        sql = 'INSERT INTO "account" (account_pkey, displayname) VALUES (%s, %s)'
+        params = ['1', 'Sample Person']
+        self.assertRaises(DatabaseError, self.wrapper.execute, sql, params)
+
+    def test_fetchmany_not_cached(self):
+        self.cursor.data = expected_values = [('auth_user_pkey', '1'),
+                                              ('username', 'user')]
+        result = self.wrapper.fetchmany(100)
+        self.assertEqual(self.wrapper.cache, None)
+        self.assertEqual(result, expected_values)
+
+    def test_fetchmany_cached(self):
+        self.cursor.data = expected_values = [('auth_user_pkey', '1'),
+                                              ('username', 'user')]
+        sql = 'SELECT * FROM "auth_user" WHERE auth_user_pkey=%s'
+        params = ['1']
+        self.wrapper.execute_cached('select', sql, params)
+        result = self.wrapper.fetchmany(100)
+        self.assertTrue(isinstance(self.wrapper.cache, AuthUserCache))
+        self.assertEqual(result, expected_values)
+
+    def test_fetchmany_cache_empty(self):
+        self.cursor.data = []
+        sql = 'SELECT * FROM "auth_user" WHERE auth_user_pkey=%s'
+        params = ['1']
+        self.wrapper.execute_cached('select', sql, params)
+        self.assertRaises(StopIteration, self.wrapper.fetchmany, 100)
+
+    def test_getattr(self):
+        # test wrapper attribute
+        self.assertEqual(self.wrapper.cache, None)
+        # test cursor attribute
+        self.assertEqual(self.wrapper.connection, None)
+
+
+class DatabaseWrapperTestCase(OldDatabaseWrapperTestCase):
+    def setUp(self):
+        super(DatabaseWrapperTestCase, self).setUp()
+        settings.DATABASE_ENGINE = 'identityprovider.backend'
+
+    def test__cursor_readonly(self):
+        readonly_manager.set_readonly()
+
+        expected_cursor = PostgresDatabaseWrapper()._cursor(settings)
+        wrapper = DatabaseWrapper()
+        cursor = wrapper._cursor(settings)
+        self.assertTrue(isinstance(cursor, CursorReadOnlyWrapper))
+        self.assertEqual(type(cursor.cursor), type(expected_cursor))
+        self.assertEqual(cursor.cache, None)
+
+        expected_cursor.connection.close()
+        cursor.connection.close()
+
+        readonly_manager.clear_readonly()
+
+    def test__cursor_not_readonly(self):
+        readonly_manager.clear_readonly()
+
+        expected_cursor = PostgresDatabaseWrapper()._cursor(settings)
+        wrapper = DatabaseWrapper()
+        cursor = wrapper._cursor(settings)
+        self.assertEqual(type(cursor), type(expected_cursor))
+
+        expected_cursor.connection.close()
+        cursor.connection.close()
+

=== added file 'identityprovider/tests/test_backend_old_base.py'
--- identityprovider/tests/test_backend_old_base.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_backend_old_base.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,53 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import psycopg2.extensions
+from django.conf import settings
+from django.db.backends.postgresql_psycopg2 import base as base_pg
+from django.db.backends.util import CursorDebugWrapper
+from django.test import TestCase
+
+from identityprovider.backend.old import base
+
+
+class DatabaseWrapperTestCase(TestCase):
+    def setUp(self):
+        self.old_DATABASE_ENGINE = settings.DATABASE_ENGINE
+        settings.DATABASE_ENGINE = 'identityprovider.backend.old'
+        self.old_DEBUG = settings.DEBUG
+
+    def tearDown(self):
+        settings.DATABASE_ENGINE = self.old_DATABASE_ENGINE
+        settings.DEBUG = self.old_DEBUG
+
+    def test_cursor_when_debug(self):
+        settings.DEBUG = True
+
+        expected_cursor = base_pg.DatabaseWrapper().cursor()
+        wrapper = base.DatabaseWrapper()
+        cursor = wrapper.cursor()
+        self.assertTrue(isinstance(cursor, CursorDebugWrapper))
+        self.assertTrue(isinstance(expected_cursor, CursorDebugWrapper))
+        self.assertEqual(cursor.cursor.connection.dsn,
+                         expected_cursor.connection.dsn)
+        self.assertEqual(cursor.db, wrapper)
+
+        expected_cursor.connection.close()
+        cursor.connection.close()
+
+    def test_cursor_when_not_debug(self):
+        settings.DEBUG = False
+
+        expected_cursor = base_pg.DatabaseWrapper().cursor()
+        wrapper = base.DatabaseWrapper()
+        cursor = wrapper.cursor()
+        self.assertTrue(isinstance(cursor, CursorDebugWrapper))
+        self.assertTrue(isinstance(expected_cursor,
+                         psycopg2.extensions.cursor))
+        self.assertEqual(cursor.cursor.connection.dsn,
+                         expected_cursor.connection.dsn)
+        self.assertEqual(cursor.db, wrapper)
+
+        expected_cursor.connection.close()
+        cursor.connection.close()
+

=== modified file 'identityprovider/tests/test_command_sqlcachedflush.py'
--- identityprovider/tests/test_command_sqlcachedflush.py	2010-05-11 16:59:11 +0000
+++ identityprovider/tests/test_command_sqlcachedflush.py	2010-06-01 12:03:28 +0000
@@ -1,6 +1,7 @@
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+import os
 import signal
 from subprocess import Popen, PIPE
 
@@ -32,7 +33,7 @@
         if settings.DATABASE_USER:
             args.extend(['-U', settings.DATABASE_USER])
         if settings.DATABASE_PASSWORD:
-            _PGPASS = os.environ['PGPASSWORD']
+            _PGPASS = os.environ.get('PGPASSWORD')
             os.environ['PGPASSWORD'] = settings.DATABASE_PASSWORD
         if settings.DATABASE_HOST:
             args.extend(['-h', settings.DATABASE_HOST])

=== added file 'identityprovider/tests/test_command_sqlcachedloaddata.py'
--- identityprovider/tests/test_command_sqlcachedloaddata.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_command_sqlcachedloaddata.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,132 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import os
+import signal
+import tempfile
+from subprocess import Popen, PIPE
+
+from django.conf import settings
+from django.core import serializers
+from django.core.management import call_command
+from django.test import TestCase
+from django.utils import simplejson
+
+from identityprovider.tests.utils import create_pgsql_functions
+
+
+class Alarm(Exception):
+    pass
+
+
+def alarm_handler(signum, frame):
+    raise Alarm
+
+
+class SQLCachedLoadDataCommandTestCase(TestCase):
+    pgsql_functions = ['generate_openid_identifier']
+
+    def _pre_setup(self):
+        super(SQLCachedLoadDataCommandTestCase, self)._pre_setup()
+        create_pgsql_functions(self.pgsql_functions)
+
+    def setUp(self):
+        # set alarm handler
+        self.old_signal = signal.signal(signal.SIGALRM, alarm_handler)
+
+    def tearDown(self):
+        # reset alarm handler
+        signal.signal(signal.SIGALRM, self.old_signal)
+
+    def compare_sqlcachedloaddata_command(self, fixtures):
+        args = ['pg_dump']
+        if settings.DATABASE_USER:
+            args.extend(['-U', settings.DATABASE_USER])
+        if settings.DATABASE_PASSWORD:
+            _PGPASS = os.environ.get('PGPASSWORD')
+            os.environ['PGPASSWORD'] = settings.DATABASE_PASSWORD
+        if settings.DATABASE_HOST:
+            args.extend(['-h', settings.DATABASE_HOST])
+        if settings.DATABASE_PORT:
+            args.extend(['-p', settings.DATABASE_PORT])
+        args.append(settings.DATABASE_NAME)
+
+        # make sure we start off a clean db
+        call_command('flush', interactive=False, verbosity=0)
+
+        options = {'interactive': False, 'verbosity': 0, 'commit': True}
+        call_command('loaddata', *fixtures, **options)
+        loaddata_sql = Popen(args, stdout=PIPE).communicate()[0]
+
+        # clear db so we can load the data without triggering integrity errors
+        call_command('flush', interactive=False, verbosity=0)
+
+        call_command('sqlcachedloaddata', *fixtures, **options)
+        # timeout after 5 seconds
+        signal.alarm(5)
+        try:
+            sqlcachedloaddata_sql = Popen(args, stdout=PIPE).communicate()[0]
+            # reset the alarm
+            signal.alarm(0)
+        except Alarm:
+            self.fail()
+
+        self.assertEqual(sqlcachedloaddata_sql, loaddata_sql)
+
+        # play nice and leave the db clean
+        call_command('flush', interactive=False, verbosity=0)
+
+    def test_sqlcachedloaddata_command_not_cached(self):
+        self.compare_sqlcachedloaddata_command(['test'])
+
+    def test_sqlcachedloaddata_command_with_format(self):
+        self.compare_sqlcachedloaddata_command(['test.json'])
+
+    def test_sqlcachedloaddata_command_with_invalid_format(self):
+        self.compare_sqlcachedloaddata_command(['test.txt'])
+
+    def test_sqlcachedloaddata_command_with_fixture_dir(self):
+        self.compare_sqlcachedloaddata_command(['/tmp/test.json'])
+
+    def test_sqlcachedloaddata_command_conflicting_fixtures(self):
+        # create two fixtures with the same name
+        prefix = tempfile.gettempprefix()
+        fixtures = ["/tmp/%s.json" % prefix, "/tmp/%s.xml" % prefix]
+        simplejson.dump([], open(fixtures[0], 'a'))
+        f = open(fixtures[1], 'a')
+        f.write("""<?xml version="1.0" encoding="utf-8"?>""")
+        f.close()
+
+        self.compare_sqlcachedloaddata_command(["/tmp/%s" % prefix])
+
+        # remove fixture files
+        for fixture in fixtures:
+            os.unlink(fixture)
+
+    def test_sqlcachedloaddata_command_interrupt(self):
+        def mock_deserialize(format, fixture):
+            self.exception_raised = True
+            raise KeyboardInterrupt()
+
+        old_deserialize = serializers.deserialize
+        serializers.deserialize = mock_deserialize
+
+        self.exception_raised = False
+        self.compare_sqlcachedloaddata_command(['test'])
+        self.assertTrue(self.exception_raised)
+
+        serializers.deserialize = old_deserialize
+
+    def test_sqlcachedloaddata_command_show_traceback(self):
+        def mock_deserialize(format, fixture):
+            self.exception_raised = True
+            raise Exception()
+
+        old_deserialize = serializers.deserialize
+        serializers.deserialize = mock_deserialize
+
+        self.exception_raised = False
+        self.compare_sqlcachedloaddata_command(['test'])
+        self.assertTrue(self.exception_raised)
+
+        serializers.deserialize = old_deserialize

=== modified file 'identityprovider/tests/test_middleware.py'
--- identityprovider/tests/test_middleware.py	2010-05-07 23:31:41 +0000
+++ identityprovider/tests/test_middleware.py	2010-06-01 12:03:28 +0000
@@ -1,7 +1,9 @@
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+import functools
 import logging
+import re
 import sys
 
 from django.conf import settings
@@ -136,3 +138,44 @@
         #  Make sure public data is unaffected
         self.assertEquals(req.POST['post_public'], self.POST_PUBLIC)
         self.assertEquals(req.GET['get_public'], self.GET_PUBLIC)
+
+
+# The CSRF middleware requires a session cookie in order to activate.
+# The tests perform a login in order to acquire this session cookie.
+class CSRFMiddlewareTestCase(BasicAccountTestCase):
+
+    def login(self):
+        r = self.client.post('/+login', {
+            'email': '[e-mail]',
+            'password': 'test'})
+        self.assertTrue('sessionid' in self.client.cookies)
+
+    def test_allow_authorized(self):
+        self.login()
+        r = self.client.get('/')
+        csrf_field = re.search('<input [^>]*name=[\'"]csrfmiddlewaretoken[\'"][^>]*/>', r.content)
+        self.assertTrue(csrf_field)
+        csrf_token = re.search(' value=[\'"]([^\'"]+)[\'"]', csrf_field.group()).group(1)
+        r = self.client.post('/+edit', {
+            'displayname': 'Mark Shuttleworthy',
+            'csrfmiddlewaretoken': csrf_token
+            })
+        self.assertNotEquals(r.status_code, 403)
+
+    def test_forbid_unauthorized(self):
+        self.login()
+        r = self.client.post('/+edit', {'displayname': 'Mark Shuttleworthy'})
+        self.assertEquals(r.status_code, 403)
+
+    def test_forbid_forged(self):
+        self.login()
+        r = self.client.post('/+edit', {
+            'displayname': 'Mark Shuttleworthy',
+            'csrfmiddlewaretoken': '0'})
+        self.assertEquals(r.status_code, 403)
+
+    def test_ajax(self):
+        self.login()
+        r = self.client.post('/+edit', {'displayname': 'Mark Shuttleworthy'},
+                             HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertNotEquals(r.status_code, 403)

=== modified file 'identityprovider/tests/test_models_account.py'
--- identityprovider/tests/test_models_account.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_models_account.py	2010-06-01 12:03:28 +0000
@@ -5,10 +5,13 @@
 from django.contrib.auth.models import User
 from identityprovider.tests.utils import BasicAccountTestCase
 from identityprovider.models.account import (Account, AccountPassword,
-    AccountStatus)
+    AccountStatus, LPAccount)
 from identityprovider.models import account as a
 from identityprovider.models.emailaddress import EmailAddress
 from identityprovider.models.const import AccountCreationRationale, EmailStatus
+from identityprovider.readonly import readonly_manager
+from identityprovider.utils import encrypt_launchpad_password, generate_salt
+from identityprovider.webservice.models import AccountSet
 
 
 class MockBackend(object):
@@ -50,6 +53,31 @@
 
         self.assertTrue(account.last_login is not None)
 
+    def test_set_last_login_when_readonly(self):
+        account = self.get_existing_account()
+        account.last_login = last_login = datetime(2010, 01, 01)
+        self.assertEqual(account.last_login, last_login)
+
+        self.assertFalse(readonly_manager.is_readonly())
+        readonly_manager.set_readonly()
+
+        try:
+            account.last_login = now = datetime.now()
+            self.assertEqual(account.last_login, last_login)
+            self.assertNotEqual(account.last_login, now)
+        finally:
+            readonly_manager.clear_readonly()
+            self.assertFalse(readonly_manager.is_readonly())
+
+    def test_set_last_login_when_no_preferredemail(self):
+        account = self.get_new_account()
+        for email in account.emailaddress_set.all():
+            email.status = EmailStatus.NEW
+            email.save()
+        account.last_login = datetime.now()
+        self.assertRaises(User.DoesNotExist, User.objects.get,
+            username=account.openid_identifier)
+
     def test_is_active(self):
         account = self.get_new_account()
         account.status = AccountStatus.ACTIVE
@@ -168,3 +196,102 @@
 
         account_password = AccountPassword.objects.get(account=account)
         self.assertEqual(account_password.password, 'invalid')
+
+    def test_save_when_readonly(self):
+        account = self.get_existing_account()
+        status = account.status
+        self.assertNotEqual(status, AccountStatus.SUSPENDED)
+
+        self.assertFalse(readonly_manager.is_readonly())
+        readonly_manager.set_readonly()
+
+        try:
+            account.status = AccountStatus.SUSPENDED
+            account.save()
+
+            # refresh account from db
+            account = self.get_existing_account()
+            self.assertEqual(account.status, status)
+        finally:
+            readonly_manager.clear_readonly()
+            self.assertFalse(readonly_manager.is_readonly())
+
+    def test_save_when_no_password(self):
+        account = self.get_existing_account()
+        # remove all passwords
+        AccountPassword.objects.filter(account=account).delete()
+
+        # trigger test
+        account.status = AccountStatus.SUSPENDED
+        account.save()
+
+        password_count = AccountPassword.objects.filter(account=account).count()
+        self.assertEqual(password_count, 0)
+
+    def test_parent(self):
+        account = self.get_existing_account()
+        parent = getattr(account, '__parent__')
+        self.assertTrue(isinstance(parent, AccountSet))
+
+    def test_url_path(self):
+        account = self.get_existing_account()
+        url_path = getattr(account, '__url_path__')
+        self.assertEqual(url_path, str(account.id))
+
+    def test_create_account_with_rationale(self):
+        salt = generate_salt()
+        account = Account.objects.create_account('displayname', 'username',
+            'password', AccountCreationRationale.USER_CREATED, salt=salt)
+
+        self.assertEqual(account.displayname, 'displayname')
+        self.assertEqual(account.creation_rationale,
+            AccountCreationRationale.USER_CREATED)
+        self.assertEqual(account.status, AccountStatus.ACTIVE)
+
+        emails = EmailAddress.objects.filter(account=account)
+        self.assertEqual(emails.count(), 1)
+        email = emails[0]
+        self.assertEqual(email.email, 'username')
+        self.assertEqual(email.status, EmailStatus.PREFERRED)
+
+        passwords = AccountPassword.objects.filter(account=account)
+        self.assertEqual(passwords.count(), 1)
+        password = passwords[0]
+        self.assertEqual(password.password,
+            encrypt_launchpad_password('password', salt=salt))
+
+    def test_create_account_with_no_rationale(self):
+        salt = generate_salt()
+        account = Account.objects.create_account('displayname', 'username',
+            'password', salt=salt)
+
+        self.assertEqual(account.displayname, 'displayname')
+        self.assertEqual(account.creation_rationale,
+            AccountCreationRationale.OWNER_CREATED_LAUNCHPAD)
+        self.assertEqual(account.status, AccountStatus.ACTIVE)
+
+        emails = EmailAddress.objects.filter(account=account)
+        self.assertEqual(emails.count(), 1)
+        email = emails[0]
+        self.assertEqual(email.email, 'username')
+        self.assertEqual(email.status, EmailStatus.PREFERRED)
+
+        passwords = AccountPassword.objects.filter(account=account)
+        self.assertEqual(passwords.count(), 1)
+        password = passwords[0]
+        self.assertEqual(password.password,
+            encrypt_launchpad_password('password', salt=salt))
+
+
+class AccountPasswordTestCase(BasicAccountTestCase):
+    def test_unicode(self):
+        account = Account(openid_identifier='oid', displayname='displayname')
+        password = AccountPassword(account=account, password='password')
+        self.assertEqual(unicode(password), u'Password for displayname')
+
+
+class LPAccountTestCase(BasicAccountTestCase):
+    def test_unicode(self):
+        account = LPAccount(openid_identifier='oid')
+        self.assertEqual(unicode(account), u'LP account for oid')
+

=== added file 'identityprovider/tests/test_models_api.py'
--- identityprovider/tests/test_models_api.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_models_api.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,41 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from identityprovider.models.api import APIUser
+from identityprovider.tests.utils import SQLCachedTestCase
+from identityprovider.utils import encrypt_launchpad_password, generate_salt
+
+
+class APIUserTestCase(SQLCachedTestCase):
+    def setUp(self):
+        super(APIUserTestCase, self).setUp()
+
+        self.salt = generate_salt()
+        self.user = APIUser(username='username')
+        self.user.set_password('password', salt=self.salt)
+        self.user.save()
+
+    def test_set_password(self):
+        self.user.set_password('otherpassword', salt=self.salt)
+        self.assertEqual(self.user.password,
+            encrypt_launchpad_password('otherpassword', salt=self.salt))
+
+    def test_verify_password(self):
+        self.user.set_password('password', salt=self.salt)
+        self.assertFalse(self.user.verify_password('otherpassword'))
+        self.assertTrue(self.user.verify_password('password'))
+
+    def test_authenticate_user_exists(self):
+        user = APIUser.authenticate('username', 'password')
+        self.assertEqual(user, self.user)
+
+    def test_authenticate_wrong_password(self):
+        user = APIUser.authenticate('username', 'otherpassword')
+        self.assertEqual(user, None)
+
+    def test_authenticate_user_does_not_exist(self):
+        user = APIUser.authenticate('bad_user', 'password')
+        self.assertEqual(user, None)
+
+    def test_unicode(self):
+        self.assertEqual(unicode(self.user), u'username')

=== modified file 'identityprovider/tests/test_models_authtoken.py'
--- identityprovider/tests/test_models_authtoken.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_models_authtoken.py	2010-06-01 12:03:28 +0000
@@ -1,38 +1,60 @@
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+import random
 from django.conf import settings
 
 from identityprovider.tests.utils import BasicAccountTestCase
 from identityprovider.models.account import Account
 from identityprovider.models.authtoken import (AuthToken, AuthTokenFactory,
-    send_validation_email_request)
+    send_validation_email_request, create_unique_token_for_table, valid_email)
+from identityprovider.models.const import EmailStatus, LoginTokenType
 
 
 class AuthTokenTestCase(BasicAccountTestCase):
-
-    def test_sendEmailValidationRequest(self):
+    def setUp(self):
         def mock_send_email(self, from_name, subject, message):
             self.message = message
 
+        super(AuthTokenTestCase, self).setUp()
+
         # override method to capture message
-        _send_email = AuthToken._send_email
+        self.old_send_email = AuthToken._send_email
         AuthToken._send_email = mock_send_email
 
-        # make sure we start from a clean environment
-        self.assertFalse(hasattr(self, 'message'))
-
+    def tearDown(self):
+        AuthToken._send_email = self.old_send_email
+
+        super(AuthTokenTestCase, self).tearDown()
+
+    def test_sendEmailValidationRequest(self):
         account = Account.objects.get_by_email('[e-mail]')
         token = AuthTokenFactory().new_api_email_validation_token(
             account, '[e-mail]')
+
+        self.assertFalse(hasattr(token, 'message'))
         token.sendEmailValidationRequest()
-
         self.assertTrue(hasattr(token, 'message'))
         self.assertTrue(settings.SUPPORT_FORM_URL in token.message)
 
-        # cleanup
-        del token.message
-        AuthToken._send_email = _send_email
+    def test_sendEmailValidationToken(self):
+        account = Account.objects.get_by_email('[e-mail]')
+        token = AuthTokenFactory().new_api_email_validation_token(
+            account, '[e-mail]')
+        self.assertFalse(hasattr(token, 'message'))
+        token.sendEmailValidationToken()
+        self.assertTrue(hasattr(token, 'message'))
+        self.assertTrue(token.token in token.message)
+
+    def test_sendNewUserEmail(self):
+        account = Account.objects.get_by_email('[e-mail]')
+        token = AuthTokenFactory().new_api_email_validation_token(
+            account, '[e-mail]')
+
+        self.assertFalse(hasattr(token, 'message'))
+        token.sendNewUserEmail()
+        self.assertTrue(hasattr(token, 'message'))
+        self.assertTrue(token.get_absolute_url() in token.message)
 
     def test_send_validation_email_request_no_preferredemail(self):
         def mock_sendEmailValidationRequest(self):
@@ -40,16 +62,79 @@
         _sendEmailValidationRequest = AuthToken.sendEmailValidationRequest
         AuthToken.sendEmailValidationRequest = mock_sendEmailValidationRequest
         AuthToken.sendEmailValidationRequest_called = False
-        # replace preferredemail property
-        _preferredemail = Account.preferredemail
-        Account.preferredemail = None
 
         account = Account.objects.get_by_email('[e-mail]')
+        # remove preferredemail address
+        for email in account.emailaddress_set.all():
+            email.status = EmailStatus.NEW
+            email.save()
+
         email = '[e-mail]'
         redirection_url = 'http://localhost/'
         send_validation_email_request(account, email, redirection_url)
         self.assertTrue(AuthToken.sendEmailValidationRequest_called)
 
-        Account.preferredemail = _preferredemail
         del AuthToken.sendEmailValidationRequest_called
         AuthToken.sendEmailValidationRequest = _sendEmailValidationRequest
+
+    def test_consume(self):
+        email = '[e-mail]'
+        token_type = LoginTokenType.NEWACCOUNT
+        # need more than one token to fully test the method
+        token = AuthToken(email=email, token_type=token_type, token='a')
+        token.save()
+        token2 = AuthToken(email=email, token_type=token_type, token='b')
+        token2.save()
+        self.assertEqual(token.date_consumed, None)
+        self.assertEqual(token2.date_consumed, None)
+
+        token.consume()
+
+        tokens = AuthToken.objects.filter(email=token.email,
+            token_type=token.token_type, requester=token.requester)
+        self.assertEqual(tokens.count(), 2)
+        self.assertNotEqual(tokens[0].date_consumed, None)
+        self.assertNotEqual(tokens[1].date_consumed, None)
+
+    def test_new_token_invalid_type(self):
+        factory = AuthTokenFactory()
+        good_types = [LoginTokenType.PASSWORDRECOVERY,
+                      LoginTokenType.NEWACCOUNT,
+                      LoginTokenType.VALIDATEEMAIL,
+                      LoginTokenType.NEWPERSONLESSACCOUNT]
+        bad_types = [t[0] for t in LoginTokenType._get_choices() \
+                     if t[0] not in good_types]
+        for token_type in bad_types:
+            self.assertRaises(ValueError, factory.new, None, '',
+                '[e-mail]', token_type, '')
+
+    def test_create_unique_token_for_table(self):
+        # initialize randomizer
+        random.seed(42)
+
+        # create a token
+        token1 = create_unique_token_for_table(1, AuthToken, 'token')
+        # use that token within an AuthToken
+        AuthToken(email='[e-mail]',
+            token_type=LoginTokenType.NEWACCOUNT, token=token1).save()
+
+        # reinitialize randomizer to force same token
+        random.seed(42)
+        token2 = create_unique_token_for_table(1, AuthToken, 'token')
+        self.assertNotEqual(token1, token2)
+
+    def test_valid_email(self):
+        emails = {'valid': ['[e-mail]', '[e-mail]',
+                            '[e-mail]', '[e-mail]',
+                            '[e-mail]', '[e-mail]',
+                            '[e-mail]', 'i@tv', '[e-mail]',
+                            '[e-mail]', 'bob=[e-mail]',
+                            '[e-mail]'],
+                  'invalid': ['user@z..de', 'user@.z.de',
+                              '[e-mail]',
+                              '[e-mail]']}
+        for email in emails['valid']:
+            self.assertTrue(valid_email(email))
+        for email in emails['invalid']:
+            self.assertFalse(valid_email(email))
+

=== added file 'identityprovider/tests/test_models_captcha.py'
--- identityprovider/tests/test_models_captcha.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_models_captcha.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,157 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import unittest
+import urllib2
+from cStringIO import StringIO
+from django.conf import settings
+
+from identityprovider.models.captcha import Captcha, CaptchaResponse
+
+
+class CaptchaResponseTestCase(unittest.TestCase):
+    def test_no_data(self):
+        r = CaptchaResponse(200, None)
+        self.assertEqual(r.data(), None)
+
+    def test_data_from_response(self):
+        resp = StringIO('this is the response')
+        r = CaptchaResponse(200, resp)
+        self.assertEqual(r._data, None)
+        self.assertEqual(r.data(), 'this is the response')
+
+    def test_data_exists(self):
+        r = CaptchaResponse(200, None)
+        r._data = 'this is the data'
+        self.assertEqual(r.data(), 'this is the data')
+
+
+class CaptchaTestCase(unittest.TestCase):
+    def test_new_captcha_oops(self):
+        @classmethod
+        def mock_open(cls, request, env):
+            r = CaptchaResponse(500, None, 'a traceback')
+            return r
+        old_open = Captcha._open
+        Captcha._open = mock_open
+
+        c = Captcha.new({})
+        self.assertTrue(c.env['oops-dump'])
+
+        Captcha._open = old_open
+
+    def test_new_captcha_no_challenge(self):
+        @classmethod
+        def mock_open(cls, request, env):
+            r = CaptchaResponse(200, StringIO(''))
+            return r
+        old_open = Captcha._open
+        Captcha._open = mock_open
+
+        c = Captcha.new({})
+        self.assertEqual(c.captcha_id, None)
+        self.assertEqual(c.image_url, None)
+
+        Captcha._open = old_open
+
+    def test_captcha_open(self):
+        class MockOpener(object):
+            def open(self, request):
+                raise urllib2.URLError((-1, 'error'))
+        old_opener = Captcha.opener
+        Captcha.opener = MockOpener()
+
+        self.assertRaises(urllib2.URLError, Captcha.new, {})
+
+        Captcha.opener = old_opener
+
+class CaptchaVerifyTestCase(unittest.TestCase):
+    def setUp(self):
+        self.old_DISABLE_CAPTCHA_VERIFICATION = settings.DISABLE_CAPTCHA_VERIFICATION
+        settings.DISABLE_CAPTCHA_VERIFICATION = False
+
+    def tearDown(self):
+        settings.DISABLE_CAPTCHA_VERIFICATION = self.old_DISABLE_CAPTCHA_VERIFICATION
+
+    def test_verify_already_verified(self):
+        c = Captcha.new({})
+        c._verified = True
+        r = c.verify(None)
+        self.assertTrue(r)
+
+    def test_verify_no_verification(self):
+        settings.DISABLE_CAPTCHA_VERIFICATION = True
+
+        c = Captcha.new({})
+        r = c.verify(None)
+        self.assertTrue(r)
+
+    def test_verify_response_ok(self):
+        @classmethod
+        def mock_open(cls, request, env):
+            r = CaptchaResponse(200, StringIO('true\nok'))
+            return r
+        old_open = Captcha._open
+        Captcha._open = mock_open
+
+        c = Captcha.new({'REMOTE_ADDR': 'localhost'})
+        r = c.verify(None)
+        self.assertTrue(r)
+        self.assertEqual(c.message, 'ok')
+
+        Captcha._open = old_open
+
+    def test_verify_no_captcha_id(self):
+        @classmethod
+        def mock_open(cls, request, env):
+            r = CaptchaResponse(200, StringIO(''), 'a traceback')
+            return r
+        old_open = Captcha._open
+        Captcha._open = mock_open
+
+        c = Captcha.new({'REMOTE_ADDR': 'localhost'})
+        r = c.verify(None)
+        self.assertFalse(r)
+        self.assertEqual(c.message, 'no-challenge')
+
+        Captcha._open = old_open
+
+    def test_verify_error(self):
+        @classmethod
+        def mock_open(cls, request, env):
+            r = CaptchaResponse(500, None, 'a traceback')
+            return r
+        old_open = Captcha._open
+        Captcha._open = mock_open
+
+        class MockEnv(dict):
+            def __getattribute__(self, attr):
+                if attr == '__setitem__':
+                    raise AttributeError()
+                else:
+                    return super(MockEnv, self).__getattribute__(attr)
+        env = MockEnv({'REMOTE_ADDR': 'localhost'})
+
+        c = Captcha(env, 'challenge-id')
+        r = c.verify(None)
+        self.assertFalse(r)
+        self.assertFalse('oops-dump' in c.env)
+
+        Captcha._open = old_open
+
+    def test_verify_error_oops(self):
+        @classmethod
+        def mock_open(cls, request, env):
+            r = CaptchaResponse(500, None, 'a traceback')
+            return r
+        old_open = Captcha._open
+        Captcha._open = mock_open
+
+        c = Captcha.new({'REMOTE_ADDR': 'localhost'})
+        c.captcha_id = 'challenge-id'
+        r = c.verify(None)
+        self.assertFalse(r)
+        self.assertTrue(c.env['oops-dump'])
+
+        Captcha._open = old_open
+

=== modified file 'identityprovider/tests/test_models_oauthtoken.py'
--- identityprovider/tests/test_models_oauthtoken.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_models_oauthtoken.py	2010-06-01 12:03:28 +0000
@@ -3,8 +3,8 @@
 
 from identityprovider.tests.utils import BasicAccountTestCase
 from identityprovider.models.account import Account
-from identityprovider.models.oauthtoken import (Consumer,
-                                                create_oauth_token_for_account)
+from identityprovider.models.oauthtoken import (Consumer, DataStore, Nonce,
+    Token, OAuthToken, create_oauth_token_for_account)
 
 
 class ConsumerTestCase(BasicAccountTestCase):
@@ -34,3 +34,76 @@
         token = create_oauth_token_for_account(account, 'new-token')
 
         self.assertEquals(token.consumer.account.id, account.id)
+
+
+class TokenTestCase(BasicAccountTestCase):
+    def setUp(self):
+        self.account = Account.objects.get_by_email('[e-mail]')
+        self.consumer, _ = Consumer.objects.get_or_create(account=self.account)
+        self.token = Token(consumer=self.consumer, token='token',
+                           token_secret='secret', name='name')
+    def test_unicode(self):
+        self.assertEqual(unicode(self.token), u'token')
+
+    def test_serialize(self):
+        expected = {'consumer_key': self.consumer.key,
+                    'consumer_secret': self.consumer.secret,
+                    'token': 'token',
+                    'token_secret': 'secret',
+                    'name': 'name'}
+        self.assertEqual(self.token.serialize(), expected)
+
+
+class ConsumerTestCase(BasicAccountTestCase):
+    def test_unicode(self):
+        account = Account.objects.get_by_email('[e-mail]')
+        consumer, _ = Consumer.objects.get_or_create(account=account)
+        self.assertEqual(unicode(consumer), unicode(account.openid_identifier))
+
+
+class DataStoreTestCase(BasicAccountTestCase):
+    def setUp(self):
+        self.ds = DataStore()
+        account = Account.objects.get_by_email('[e-mail]')
+        self.consumer, _ = Consumer.objects.get_or_create(account=account)
+        self.token, _ = Token.objects.get_or_create(consumer=self.consumer,
+            token='token', token_secret='secret', name='name')
+        self.nonce, _ = Nonce.objects.get_or_create(consumer=self.consumer,
+            token=self.token, nonce='nonce')
+
+    def test_lookup_token_wrong_type(self):
+        self.assertRaises(AssertionError, self.ds.lookup_token, 'value',
+                          'token')
+
+    def test_lookup_token_exists(self):
+        # create a token
+        token = self.ds.lookup_token('access', 'token')
+        self.assertEqual(token.key, 'token')
+        self.assertEqual(token.secret, 'secret')
+
+    def test_lookup_token_not_exists(self):
+        self.token.delete()
+        token = self.ds.lookup_token('access', 'token')
+        self.assertEqual(token, None)
+
+    def test_lookup_consumer_exists(self):
+        consumer_key = self.consumer.key
+        result = self.ds.lookup_consumer(consumer_key)
+        self.assertEqual(result.key, self.consumer.key)
+        self.assertEqual(result.secret, self.consumer.secret)
+
+    def test_lookup_consumer_not_exists(self):
+        consumer_key = 'foo'
+        consumer = self.ds.lookup_consumer(consumer_key)
+        self.assertEqual(consumer, None)
+
+    def test_lookup_nonce_exists(self):
+        otoken = OAuthToken(self.token.token, self.token.token_secret)
+        r = self.ds.lookup_nonce(self.consumer, otoken, 'nonce')
+        self.assertTrue(r)
+
+    def test_lookup_nonce_not_exists(self):
+        self.nonce.delete()
+        otoken = OAuthToken(self.token.token, self.token.token_secret)
+        r = self.ds.lookup_nonce(self.consumer, otoken, 'nonce')
+        self.assertFalse(r)

=== modified file 'identityprovider/tests/test_models_openidmodels.py'
--- identityprovider/tests/test_models_openidmodels.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_models_openidmodels.py	2010-06-01 12:03:28 +0000
@@ -1,46 +1,52 @@
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+import base64
+from datetime import datetime, timedelta
 from time import time
 from openid.association import Association
 from openid.store.nonce import SKEW
+
+from identityprovider.const import NEVER_EXPIRES
+from identityprovider.models.account import Account
+from identityprovider.models.openidmodels import (DjangoOpenIDStore,
+    OpenIDAssociation, OpenIDAuthorization, OpenIDNonce, OpenIDRPConfig,
+    OpenIDRPSummary)
+from identityprovider.readonly import readonly_manager
 from identityprovider.tests.utils import BasicAccountTestCase
-from identityprovider.models.openidmodels import (DjangoOpenIDStore,
-                                                  OpenIDAssociation,
-                                                  OpenIDRPConfig,
-                                                  OpenIDNonce)
 
 
 class DjangoOpenIDStoreTestCase(BasicAccountTestCase):
 
     def setUp(self):
+        self.server_url = 'http://server.example.com'
         self.store = DjangoOpenIDStore()
-        OpenIDAssociation.objects.get_or_create(
-            server_url="http://server.example.com",
+        self.assoc, _ = OpenIDAssociation.objects.get_or_create(
+            server_url=self.server_url,
             handle="handle", secret="secret".encode("base64"),
             issued=time(), lifetime=1000, assoc_type='HMAC-SHA1')
 
     def test_store_association_when_association_already_exists(self):
         association = Association("handle", "secret", 2, 2, 'HMAC-SHA1')
 
-        self.store.storeAssociation("http://server.example.com", association)
+        self.store.storeAssociation(self.server_url, association)
         openid_association = OpenIDAssociation.objects.get(
-            server_url="http://server.example.com", handle="handle")
+            server_url=self.server_url, handle="handle")
 
         self.assertEqual(openid_association.issued, 2)
 
     def test_get_association_with_handle_none(self):
-        association = self.store.getAssociation("http://server.example.com")
+        association = self.store.getAssociation(self.server_url)
 
         self.assertEqual(association.handle, "handle")
 
     def test_get_association_when_lifetime_is_zero(self):
         openid_association = OpenIDAssociation.objects.get(
-            server_url="http://server.example.com", handle="handle")
+            server_url=self.server_url, handle="handle")
         openid_association.lifetime = 0
         openid_association.save()
 
-        association = self.store.getAssociation("http://server.example.com")
+        association = self.store.getAssociation(self.server_url)
 
         self.assertTrue(association is None)
 
@@ -61,6 +67,55 @@
         OpenIDNonce.objects.get(
             server_url="http://example.com", salt="salt").delete()
 
+    def test_getAssociation_not_existing(self):
+        OpenIDAssociation.objects.filter(server_url=self.server_url).delete()
+        assoc = self.store.getAssociation(self.server_url)
+        self.assertEqual(assoc, None)
+
+    def test_getAssociation_existing_no_handle(self):
+        expected = Association(self.assoc.handle,
+            base64.b64decode(str(self.assoc.secret)), int(self.assoc.issued),
+            self.assoc.lifetime, self.assoc.assoc_type)
+        assoc = self.store.getAssociation(self.server_url)
+        self.assertEqual(assoc, expected)
+
+    def test_getAssociation_expired(self):
+        # make sure the association has expired
+        self.assoc.lifetime = 0
+        self.assoc.save()
+
+        assoc = self.store.getAssociation(self.server_url)
+        self.assertEqual(assoc, None)
+        assocs = OpenIDAssociation.objects.filter(server_url=self.server_url)
+        self.assertEqual(assocs.count(), 0)
+
+    def test_getAssociation_existing_same_handle(self):
+        expected = Association(self.assoc.handle,
+            base64.b64decode(str(self.assoc.secret)), int(self.assoc.issued),
+            self.assoc.lifetime, self.assoc.assoc_type)
+        assoc = self.store.getAssociation(self.server_url, 'handle')
+        self.assertEqual(assoc, expected)
+
+    def test_getAssociation_existing_different_handle(self):
+        assoc = self.store.getAssociation(self.server_url, 'otherhandle')
+        self.assertEqual(assoc, None)
+
+    def test_cleanupNonce(self):
+        OpenIDNonce.objects.create(server_url=self.server_url, timestamp=0,
+                                   salt='')
+
+        self.assertEqual(OpenIDNonce.objects.all().count(), 1)
+        self.store.cleanupNonce()
+        self.assertEqual(OpenIDNonce.objects.all().count(), 0)
+
+    def test_cleanupAssociations(self):
+        self.assoc.lifetime = 0
+        self.assoc.save()
+
+        self.assertEqual(OpenIDAssociation.objects.all().count(), 1)
+        self.store.cleanupAssociations()
+        self.assertEqual(OpenIDAssociation.objects.all().count(), 0)
+
 
 class OpenIDRPConfigTestCase(BasicAccountTestCase):
 
@@ -77,6 +132,11 @@
 
         self.assertTrue(self.rp_config.logo_url.endswith('test.png'))
 
+    def test_unicode(self):
+        self.rp_config.displayname = 'MyOpenIDRPConfig'
+        self.rp_config.save()
+        self.assertEqual(unicode(self.rp_config), u'MyOpenIDRPConfig')
+
 
 class OpenIDAssociationTestCase(BasicAccountTestCase):
     fixtures = ['multiple_associations']
@@ -89,3 +149,130 @@
             server_url="http://test.com",
             handle="handle", secret="secret".encode("base64"),
             issued=time(), lifetime=1000, assoc_type='HMAC-SHA1')
+
+
+class OpenIDAuthorizationTestCase(BasicAccountTestCase):
+    def setUp(self):
+        self.account = Account.objects.get_by_email('[e-mail]')
+        self.trust_root = 'http://openid.launchpad.dev'
+        self.expires = datetime.now() + timedelta(1)
+
+    def create_auth(self):
+        return OpenIDAuthorization.objects.create( account=self.account,
+            client_id=None, date_expires=self.expires,
+            trust_root=self.trust_root)
+
+    def test_authorize_when_readonly(self):
+        readonly_manager.set_readonly()
+
+        try:
+            expires = datetime.now()
+            OpenIDAuthorization.objects.authorize(self.account, self.trust_root,
+                                                  expires)
+            self.assertRaises(OpenIDAuthorization.DoesNotExist,
+                OpenIDAuthorization.objects.get, account=self.account,
+                client_id=None, trust_root=self.trust_root)
+        finally:
+            readonly_manager.clear_readonly()
+
+    def test_authorize_expires(self):
+        OpenIDAuthorization.objects.authorize(self.account, self.trust_root,
+                                              None)
+        auth = OpenIDAuthorization.objects.get(account=self.account,
+            client_id=None, trust_root=self.trust_root)
+        self.assertEqual(auth.date_expires,
+            datetime.fromordinal(NEVER_EXPIRES.toordinal()))
+
+    def test_authorize_existing(self):
+        auth1 = self.create_auth()
+        OpenIDAuthorization.objects.authorize(self.account, self.trust_root,
+                                              None)
+        auth2 = OpenIDAuthorization.objects.get(pk=auth1.id)
+        self.assertEqual(auth2.date_expires,
+            datetime.fromordinal(NEVER_EXPIRES.toordinal()))
+
+    def test_authorize_not_existing(self):
+        OpenIDAuthorization.objects.authorize(self.account, self.trust_root,
+                                              None)
+        auth = OpenIDAuthorization.objects.get(account=self.account,
+            client_id=None, trust_root=self.trust_root)
+        self.assertEqual(auth.date_expires,
+            datetime.fromordinal(NEVER_EXPIRES.toordinal()))
+
+    def test_is_authorized_generic(self):
+        self.create_auth()
+        self.assertTrue(OpenIDAuthorization.objects.is_authorized(self.account,
+            self.trust_root, 'client'))
+
+    def test_is_authorized_client_id(self):
+        self.assertFalse(OpenIDAuthorization.objects.is_authorized(self.account,
+            self.trust_root, 'client'))
+        OpenIDAuthorization.objects.authorize(self.account, self.trust_root,
+            None, client_id='client')
+        self.assertTrue(OpenIDAuthorization.objects.is_authorized(self.account,
+            self.trust_root, 'client'))
+
+    def test_is_not_authorized(self):
+        self.assertFalse(OpenIDAuthorization.objects.is_authorized(self.account,
+            self.trust_root, 'client'))
+
+
+class OpenIDRPSummaryTestCase(BasicAccountTestCase):
+    def setUp(self):
+        self.account = Account.objects.get_by_email('[e-mail]')
+        self.trust_root = 'http://openid.launchpad.dev'
+
+    def create_summary(self):
+        return OpenIDRPSummary.objects.create(account=self.account,
+            trust_root=self.trust_root, openid_identifier='oid')
+
+    def test_record_when_readonly(self):
+        readonly_manager.set_readonly()
+
+        try:
+            summary = OpenIDRPSummary.objects.record(self.account,
+                                                     self.trust_root)
+            self.assertEqual(summary, None)
+            self.assertRaises(OpenIDRPSummary.DoesNotExist,
+                OpenIDRPSummary.objects.get, account=self.account,
+                trust_root=self.trust_root, openid_identifier=None)
+        finally:
+            readonly_manager.clear_readonly()
+
+    def test_record_with_existing_and_no_openid_identifier(self):
+        summary1 = self.create_summary()
+        summary2 = OpenIDRPSummary.objects.record(self.account, self.trust_root)
+        self.assertEqual(summary1.total_logins, 1)
+        self.assertEqual(summary2.total_logins, 1)
+        self.assertNotEqual(summary1, summary2)
+
+    def test_record_existing_same_oid(self):
+        summary1 = self.create_summary()
+        summary2 = OpenIDRPSummary.objects.record(self.account,
+            self.trust_root, openid_identifier='oid')
+        self.assertEqual(summary2.total_logins, 2)
+        self.assertEqual(summary1, summary2)
+
+    def test_record_existing_different_oid(self):
+        summary1 = self.create_summary()
+        summary2 = OpenIDRPSummary.objects.record(self.account,
+            self.trust_root, openid_identifier='oid1')
+        self.assertEqual(summary1.total_logins, 1)
+        self.assertEqual(summary2.total_logins, 1)
+        self.assertNotEqual(summary1, summary2)
+
+    def test_record_not_existing(self):
+        summary1 = OpenIDRPSummary.objects.record(self.account, self.trust_root,
+            openid_identifier='oid')
+        self.assertEqual(summary1.total_logins, 1)
+
+        summary2 = OpenIDRPSummary.objects.get(account=self.account,
+            trust_root=self.trust_root, openid_identifier='oid')
+        self.assertEqual(summary1, summary2)
+
+    def test_increment(self):
+        summary = self.create_summary()
+        self.assertEqual(summary.total_logins, 1)
+
+        summary.increment()
+        self.assertEqual(summary.total_logins, 2)

=== added file 'identityprovider/tests/test_models_person.py'
--- identityprovider/tests/test_models_person.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_models_person.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,57 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+
+from identityprovider.models.account import Account, LPAccount
+from identityprovider.models.const import (AccountCreationRationale,
+    AccountStatus)
+from identityprovider.models.person import Person
+from identityprovider.models.team import TeamParticipation
+from identityprovider.tests.utils import SQLCachedTestCase
+
+
+class PersonTestCase(SQLCachedTestCase):
+    pgsql_functions = ['generate_openid_identifier']
+
+    def setUp(self):
+        lp_account = LPAccount.objects.create(openid_identifier='oid')
+        self.person1 = Person.objects.create(displayname='Person',
+            name='person', lp_account=lp_account)
+        self.person2 = Person.objects.create(displayname='Other', name='other')
+        self.team1 = Person.objects.create(name='team', teamowner=self.person2)
+        self.team2 = Person.objects.create(name='other-team',
+                                           teamowner=self.person1)
+
+    def test_unicode(self):
+        self.assertEqual(unicode(self.person1), u'Person')
+
+    def test_in_team_no_team(self):
+        self.assertFalse(self.person1.in_team('no-team'))
+
+    def test_in_team(self):
+        tp = TeamParticipation.objects.create(team=self.team1,
+                                              person=self.person1)
+        self.assertTrue(self.person1.in_team('team'))
+
+    def test_in_team_object(self):
+        tp = TeamParticipation.objects.create(team=self.team1,
+                                              person=self.person1)
+        self.assertTrue(self.person1.in_team(self.team1))
+
+    def test_not_in_team(self):
+        self.assertFalse(self.person1.in_team('team'))
+
+    def test_in_team_no_teamparticipation_same_owner(self):
+        team = Person.objects.create(name='otherteam', teamowner=self.person1)
+        self.assertTrue(self.person1.in_team('otherteam'))
+
+    def test_account_when_no_account(self):
+        self.assertEqual(self.person1.account, None)
+
+    def test_account_when_account(self):
+        account = Account.objects.create(
+            creation_rationale=AccountCreationRationale.USER_CREATED,
+            status=AccountStatus.ACTIVE, displayname='Person',
+            openid_identifier='oid')
+        self.assertEqual(self.person1.account, account)
+

=== added file 'identityprovider/tests/test_models_team.py'
--- identityprovider/tests/test_models_team.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_models_team.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,14 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from identityprovider.models.person import Person
+from identityprovider.models.team import TeamParticipation
+from identityprovider.tests.utils import SQLCachedTestCase
+
+
+class TeamParticipationTestCase(SQLCachedTestCase):
+    def test_unicode(self):
+        team = Person.objects.create(displayname='Team', name='team')
+        person = Person.objects.create(displayname='Person', name='person')
+        tp = TeamParticipation.objects.create(team=team, person=person)
+        self.assertEqual(unicode(tp), u'Person in Team')

=== modified file 'identityprovider/tests/test_readonly.py'
--- identityprovider/tests/test_readonly.py	2010-05-05 17:47:40 +0000
+++ identityprovider/tests/test_readonly.py	2010-06-01 12:03:28 +0000
@@ -7,29 +7,35 @@
 import tempfile
 import socket
 import urllib2
+from datetime import datetime, timedelta
 from StringIO import StringIO
 
 from django.conf import settings
+from django.contrib.auth.models import User
+from django.contrib.sessions.models import Session
 from django.utils import simplejson
 
 from identityprovider.readonly import readonly_manager
-from identityprovider.tests.utils import SQLCachedTestCase
+from identityprovider.tests.utils import SQLCachedTestCase, BasicAccountTestCase
 from identityprovider import models
 from identityprovider.backend.base import DatabaseError
-from identityprovider.views.readonly import _remote_req
-
-
-class ReadOnlyTest(SQLCachedTestCase):
+from identityprovider.views.readonly import (_remote_req, get_server_atts,
+    update_server)
+
+
+class InReadOnlyTestCase(SQLCachedTestCase):
+
+    def setUp(self):
+        readonly_manager.set_readonly()
+
+    def tearDown(self):
+        readonly_manager.clear_readonly()
+
+
+class ReadOnlyTest(InReadOnlyTestCase):
     fixtures = ["test"]
     pgsql_functions = ['generate_openid_identifier']
 
-    def setUp(self):
-        self.readonly = getattr(settings, 'READ_ONLY_MODE', False)
-        settings.READ_ONLY_MODE = True
-
-    def tearDown(self):
-        settings.READ_ONLY_MODE = self.readonly
-
     def test_invalid_insert(self):
         tests = [(models.OpenIDRPConfig, {'trust_root': 'foo',
                     'displayname': 'foo', 'description': 'foo'}),
@@ -47,6 +53,26 @@
         p.displayname = 'Something Different'
         self.assertRaises(DatabaseError, p.save)
 
+    def test_current_dbid_with_settings(self):
+        old_DATABASE_ID = settings.DATABASE_ID
+        settings.DATABASE_ID = 'mydb'
+
+        self.assertEqual(readonly_manager.current_dbid(), 'mydb')
+
+        settings.DATABASE_ID = old_DATABASE_ID
+
+    def test_current_dbid_without_settings(self):
+        old_DATABASE_ID = settings.DATABASE_ID
+        del settings._target.DATABASE_ID
+
+        self.assertTrue(len(readonly_manager.connections) > 0)
+
+        self.assertEqual(readonly_manager.current_dbid(),
+                         readonly_manager.connections[0]['DATABASE_ID'])
+
+        settings.DATABASE_ID = old_DATABASE_ID
+
+
 
 class RemoteRequestTest(SQLCachedTestCase):
     msg = 'hello'
@@ -96,8 +122,20 @@
         self.assertEquals(self.host, self.req.get_host())
         self.restore_orig_urlopen()
 
+    def test_remote_req_urllib2_error(self):
+        def mock_urlopen(req, data=None):
+            raise urllib2.URLError((-1, 'error'))
+
+        self.setup_mock_urlopen(mock_urlopen)
+        server = {'host': self.host}
+        self.assertEquals(None, _remote_req(**server))
+        self.restore_orig_urlopen()
+
+
 class ReadOnlyBackUp(SQLCachedTestCase):
     """A base TestCase for backing up and restoring our readonly settings"""
+    pgsql_functions = ['generate_openid_identifier']
+
     def setUp(self):
         self.orig_readonly_secret = settings.READONLY_SECRET
         settings.READONLY_SECRET = 'test secret'
@@ -123,6 +161,7 @@
         readonly_manager.clear_failed('master')
         readonly_manager.set_db(settings.DB_CONNECTIONS[0])
 
+
 class ReadOnlyFlagFilesTest(ReadOnlyBackUp):
     def test_flag_files_in_right_directory(self):
         readonly_manager.set_readonly()
@@ -142,7 +181,10 @@
         self.assertTrue(mode & stat.S_IRGRP)
         self.assertTrue(mode & stat.S_IWGRP)
 
+
 class ReadOnlyDataTest(ReadOnlyBackUp):
+    pgsql_functions = ['generate_openid_identifier']
+
     def test_readonly_returns_404_for_get(self):
         response = self.client.get('/readonlydata')
         self.assertEquals(404, response.status_code)
@@ -199,3 +241,233 @@
         data = simplejson.loads(response.content)
         self.assertFalse(data['connections'][0]['failed'])
         self.assertTrue(data['readonly'])
+
+
+class ReadOnlyViewsTestCase(BasicAccountTestCase):
+    pgsql_functions = ['generate_openid_identifier']
+
+    def login_with_staff(self):
+        user = User.objects.create(username='admin')
+        user.is_staff = True
+        user.set_password('password')
+        user.save()
+        r = self.client.login(username='admin', password='password')
+        self.assertTrue(r)
+
+    def test_readonly_admin(self):
+        self.login_with_staff()
+
+        old_APP_SERVERS = settings.APP_SERVERS
+        settings.APP_SERVERS = [{'SERVER_ID': 'localhost', 'SCHEME': 'http',
+                                 'HOST': 'localhost', 'VIRTUAL_HOST': '',
+                                 'PORT': '8000'}]
+        expected = [{'appservers': [{'name': 'localhost', 'reachable': False}],
+                     'admin_media_prefix': settings.ADMIN_MEDIA_PREFIX,
+                     'clear_all_readonly': False,
+                     'set_all_readonly': False}]
+        r = self.client.get('/readonly')
+        self.assertTemplateUsed(r, 'admin/readonly.html')
+        for item in r.context:
+            self.assertEqual(item.dicts, expected)
+
+        settings.APP_SERVERS = old_APP_SERVERS
+
+    def test_get_server_atts_server_unreachable(self):
+        servers = [{'SERVER_ID': 'localhost', 'SCHEME': 'http',
+                    'HOST': 'localhost', 'VIRTUAL_HOST': '', 'PORT': '8000'}]
+        expected = {'appservers': [{'name': 'localhost', 'reachable': False}],
+                    'admin_media_prefix': settings.ADMIN_MEDIA_PREFIX,
+                    'clear_all_readonly': False,
+                    'set_all_readonly': False}
+        atts = get_server_atts(servers)
+        self.assertEqual(atts, expected)
+
+    def test_get_server_atts_data_error(self):
+        def mock_loads(data):
+            raise ValueError()
+        old_loads = simplejson.loads
+        simplejson.loads = mock_loads
+        def mock_urlopen(req, data=None):
+            data = {}
+            return StringIO(simplejson.dumps(data))
+        old_urlopen = urllib2.urlopen
+        urllib2.urlopen = mock_urlopen
+
+        servers = [{'SERVER_ID': 'localhost', 'SCHEME': 'http',
+                    'HOST': 'localhost', 'VIRTUAL_HOST': '', 'PORT': '8000'}]
+        expected = {'appservers': [{'name': 'localhost', 'reachable': False}],
+                    'admin_media_prefix': settings.ADMIN_MEDIA_PREFIX,
+                    'clear_all_readonly': False,
+                    'set_all_readonly': True}
+        atts = get_server_atts(servers)
+        self.assertEqual(atts, expected)
+
+        simplejson.loads = old_loads
+        urllib2.urlopen = old_urlopen
+
+    def test_get_server_atts_readonly(self):
+        def mock_urlopen(req, data=None):
+            data = {'readonly': True}
+            return StringIO(simplejson.dumps(data))
+        old_urlopen = urllib2.urlopen
+        urllib2.urlopen = mock_urlopen
+
+        servers = [{'SERVER_ID': 'localhost', 'SCHEME': 'http',
+                    'HOST': 'localhost', 'VIRTUAL_HOST': '', 'PORT': '8000'}]
+        expected = {'appservers': [{'name': 'localhost', 'reachable': True,
+                                    'readonly': True}],
+                    'admin_media_prefix': settings.ADMIN_MEDIA_PREFIX,
+                    'clear_all_readonly': True,
+                    'set_all_readonly': False}
+        atts = get_server_atts(servers)
+        self.assertEqual(atts, expected)
+
+        urllib2.urlopen = old_urlopen
+
+    def test_readonly_confirm_get(self):
+        self.login_with_staff()
+
+        r = self.client.get('/readonly/localhost/set')
+        self.assertTemplateUsed(r, 'admin/readonly_confirm.html')
+        for item in r.context:
+            self.assertEqual(item.dicts,
+                [{'appserver': 'localhost', 'action': 'set', 'conn': None}])
+
+    def test_readonly_confirm_post(self):
+        def mock_urlopen(req, data=None):
+            self.reqs.append(req)
+            return StringIO(simplejson.dumps({}))
+        old_urlopen = urllib2.urlopen
+        urllib2.urlopen = mock_urlopen
+
+        self.disable_csrf()
+        self.login_with_staff()
+
+        self.reqs = []
+        r = self.client.post('/readonly/localhost/set')
+        data = map(lambda x: x.data, self.reqs)
+        self.assertEqual(data, [
+            "action=set&conn=None&secret=%s" % settings.READONLY_SECRET
+        ])
+
+        self.assertRedirects(r, '/readonly')
+
+        self.reset_csrf()
+
+        urllib2.urlopen = mock_urlopen
+
+    def test_update_server_all_appservers(self):
+        def mock_urlopen(req, data=None):
+            self.reqs.append(req)
+            return StringIO(simplejson.dumps({}))
+        old_urlopen = urllib2.urlopen
+        urllib2.urlopen = mock_urlopen
+
+        old_APP_SERVERS = settings.APP_SERVERS
+        settings.APP_SERVERS = servers = [
+            {'SERVER_ID': 'localhost', 'SCHEME': 'http',
+             'HOST': 'localhost', 'VIRTUAL_HOST': '', 'PORT': '8000'},
+            {'SERVER_ID': 'otherhost', 'SCHEME': 'http',
+             'HOST': 'otherhost', 'VIRTUAL_HOST': '', 'PORT': '8000'},
+        ]
+
+        self.reqs = []
+        update_server('set')
+        data = map(lambda x: x.data, self.reqs)
+        self.assertEqual(data, [
+            "action=set&conn=None&secret=%s" % settings.READONLY_SECRET,
+            "action=set&conn=None&secret=%s" % settings.READONLY_SECRET
+        ])
+
+        settings.APP_SERVERS = old_APP_SERVERS
+        urllib2.urlopen = mock_urlopen
+
+    def test_update_server_one_appserver(self):
+        def mock_urlopen(req, data=None):
+            self.reqs.append(req)
+            return StringIO(simplejson.dumps({}))
+        old_urlopen = urllib2.urlopen
+        urllib2.urlopen = mock_urlopen
+
+        old_APP_SERVERS = settings.APP_SERVERS
+        settings.APP_SERVERS = servers = [
+            {'SERVER_ID': 'localhost', 'SCHEME': 'http',
+             'HOST': 'localhost', 'VIRTUAL_HOST': '', 'PORT': '8000'},
+            {'SERVER_ID': 'otherhost', 'SCHEME': 'http',
+             'HOST': 'otherhost', 'VIRTUAL_HOST': '', 'PORT': '8000'},
+        ]
+
+        self.reqs = []
+        update_server('set', 'localhost')
+        data = map(lambda x: x.data, self.reqs)
+        self.assertEqual(data, [
+            "action=set&conn=None&secret=%s" % settings.READONLY_SECRET
+        ])
+
+        settings.APP_SERVERS = old_APP_SERVERS
+        urllib2.urlopen = mock_urlopen
+
+    def test_update_server_one_connection(self):
+        def mock_urlopen(req, data=None):
+            self.reqs.append(req)
+            return StringIO(simplejson.dumps({}))
+        old_urlopen = urllib2.urlopen
+        urllib2.urlopen = mock_urlopen
+
+        old_APP_SERVERS = settings.APP_SERVERS
+        settings.APP_SERVERS = servers = [
+            {'SERVER_ID': 'localhost', 'SCHEME': 'http',
+             'HOST': 'localhost', 'VIRTUAL_HOST': '', 'PORT': '8000'},
+        ]
+
+        self.reqs = []
+        update_server('set', 'localhost', 'master')
+        data = map(lambda x: x.data, self.reqs)
+        self.assertEqual(data, [
+            "action=set&conn=master&secret=%s" % settings.READONLY_SECRET,
+        ])
+
+        settings.APP_SERVERS = old_APP_SERVERS
+        urllib2.urlopen = mock_urlopen
+
+    def test_update_server_clear_all(self):
+        def mock_urlopen(req, data=None):
+            self.reqs.append(req)
+            return StringIO(simplejson.dumps({}))
+        old_urlopen = urllib2.urlopen
+        urllib2.urlopen = mock_urlopen
+
+        old_APP_SERVERS = settings.APP_SERVERS
+        settings.APP_SERVERS = servers = [
+            {'SERVER_ID': 'localhost', 'SCHEME': 'http',
+             'HOST': 'localhost', 'VIRTUAL_HOST': '', 'PORT': '8000'},
+        ]
+
+        # create a dummy session for testing purposes
+        Session.objects.create(
+            session_key='session-key', session_data='session-data',
+            expire_date=datetime.now() + timedelta(1))
+
+
+        self.reqs = []
+        update_server('clear')
+        data = map(lambda x: x.data, self.reqs)
+        self.assertEqual(data, [
+            "action=clear&conn=None&secret=%s" % settings.READONLY_SECRET,
+        ])
+
+        # assert sessions were cleared
+        self.assertEqual(Session.objects.all().count(), 0)
+
+        settings.APP_SERVERS = old_APP_SERVERS
+        urllib2.urlopen = mock_urlopen
+
+
+class ReadOnlyViews(InReadOnlyTestCase):
+
+    pgsql_functions = ['generate_openid_identifier']
+
+    def test_new_account_is_rendered_using_read_only_template(self):
+        settings.READ_ONLY_MODE = True
+        r = self.client.get('/+new_account')
+        self.assertTemplateUsed(r, 'readonly.html')

=== added file 'identityprovider/tests/test_teams.py'
--- identityprovider/tests/test_teams.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_teams.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,200 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from openid.consumer.consumer import SuccessResponse
+from openid.consumer.discover import OpenIDServiceEndpoint
+from openid.message import IDENTIFIER_SELECT
+
+from identityprovider.teams import (supportsTeams, getTeamsNS, ns_uri,
+    TeamsNamespaceError, TeamsRequest, TeamsResponse)
+from identityprovider.tests.utils import SQLCachedTestCase
+from identityprovider.views import server
+
+
+class TeamsTestCase(SQLCachedTestCase):
+    def test_supportsTeams(self):
+        endpoint = OpenIDServiceEndpoint()
+        endpoint.type_uris.append(ns_uri)
+        r = supportsTeams(endpoint)
+        self.assertTrue(r)
+
+    def test_not_supportsTeams(self):
+        endpoint = OpenIDServiceEndpoint()
+        r = supportsTeams(endpoint)
+        self.assertFalse(r)
+
+
+class GetTeamsNSTestCase(SQLCachedTestCase):
+    def setUp(self):
+        super(GetTeamsNSTestCase, self).setUp()
+
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT}
+        openid_server = server._get_openid_server()
+        self.orequest = openid_server.decodeRequest(request)
+
+    def test_getTeamsNS(self):
+        message = self.orequest.message
+        self.assertEqual(message.namespaces.getAlias(ns_uri), None)
+        uri = getTeamsNS(message)
+        self.assertEqual(uri, ns_uri)
+        self.assertEqual(message.namespaces.getAlias(ns_uri), 'lp')
+
+    def test_getTeamsNS_alias_already_exists(self):
+        message = self.orequest.message
+        message.namespaces.addAlias('http://localhost/', 'lp')
+        self.assertEqual(message.namespaces.getAlias(ns_uri), None)
+        self.assertRaises(TeamsNamespaceError, getTeamsNS, message)
+
+
+class TeamsRequestTestCase(SQLCachedTestCase):
+    def setUp(self):
+        super(TeamsRequestTestCase, self).setUp()
+
+        self.req = TeamsRequest(
+            query_membership=['canonical-identity-provider'])
+
+    def test_init(self):
+        req = TeamsRequest()
+        self.assertEqual(req.query_membership, [])
+        self.assertEqual(req.ns_uri, ns_uri)
+
+        self.assertEqual(self.req.query_membership, ['canonical-identity-provider'])
+
+        self.assertRaises(TypeError, TeamsRequest,
+            query_membership='canonical-identity-provider')
+
+    def test_fromOpenIDRequest(self):
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT}
+        openid_server = server._get_openid_server()
+        orequest = openid_server.decodeRequest(request)
+        req = TeamsRequest.fromOpenIDRequest(orequest)
+        self.assertEqual(req.query_membership, [])
+        self.assertEqual(req.ns_uri, ns_uri)
+
+    def test_fromOpenIDRequest_with_query_membership(self):
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT,
+                   'openid.lp.query_membership': 'canonical-identity-provider'}
+        openid_server = server._get_openid_server()
+        orequest = openid_server.decodeRequest(request)
+        req = TeamsRequest.fromOpenIDRequest(orequest)
+        self.assertEqual(req.query_membership, ['canonical-identity-provider'])
+        self.assertEqual(req.ns_uri, ns_uri)
+
+    def test_parseExtensionArgs_no_strict(self):
+        req = TeamsRequest()
+        req.parseExtensionArgs(
+            {'query_membership':
+                'canonical-identity-provider,canonical-identity-provider'},
+            strict=False)
+        self.assertEqual(req.query_membership, ['canonical-identity-provider'])
+
+    def test_parseExtensionArgs_strict(self):
+        req = TeamsRequest()
+        self.assertRaises(ValueError, req.parseExtensionArgs,
+            {'query_membership':
+                'canonical-identity-provider,canonical-identity-provider'},
+            strict=True)
+
+    def test_allRequestedTeams(self):
+        self.assertEqual(self.req.allRequestedTeams(),
+                         ['canonical-identity-provider'])
+
+    def test_not_wereTeamsRequested(self):
+        req = TeamsRequest()
+        self.assertFalse(req.wereTeamsRequested())
+
+    def test_wereTeamsRequested(self):
+        self.assertTrue(self.req.wereTeamsRequested())
+
+    def test_contains(self):
+        self.assertTrue('canonical-identity-provider' in self.req)
+        self.assertFalse('other-team' in self.req)
+
+    def test_getExtensionArgs(self):
+        expected = {'query_membership': 'canonical-identity-provider'}
+        self.assertEqual(self.req.getExtensionArgs(), expected)
+
+        teams = ['canonical-identity-provider', 'other-team']
+        expected = {'query_membership': ','.join(teams)}
+        req = TeamsRequest(query_membership=teams)
+        self.assertEqual(req.getExtensionArgs(), expected)
+
+
+class TeamsResponseTestCase(SQLCachedTestCase):
+    def test_init(self):
+        resp = TeamsResponse()
+        self.assertEqual(resp.is_member, [])
+        self.assertEqual(resp.ns_uri, ns_uri)
+
+        resp = TeamsResponse(is_member=['canonical-identity-provider'])
+        self.assertEqual(resp.is_member, ['canonical-identity-provider'])
+
+    def test_addTeam(self):
+        resp = TeamsResponse()
+        self.assertEqual(resp.is_member, [])
+        resp.addTeam('canonical-identity-provider')
+        self.assertEqual(resp.is_member, ['canonical-identity-provider'])
+
+    def test_extractResponse(self):
+        req = TeamsRequest()
+        is_member_str = 'canonical-identity-provider'
+        resp = TeamsResponse.extractResponse(req, is_member_str)
+        self.assertEqual(resp.ns_uri, req.ns_uri)
+        self.assertEqual(resp.is_member, ['canonical-identity-provider'])
+
+    def test_fromSuccessResponse_signed_present(self):
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT,
+                   'openid.lp.is_member': 'canonical-identity-provider'}
+        openid_server = server._get_openid_server()
+        orequest = openid_server.decodeRequest(request)
+        signed_fields = ['openid.lp.is_member']
+        success_resp = SuccessResponse(orequest, orequest.message,
+                                       signed_fields=signed_fields)
+        resp = TeamsResponse.fromSuccessResponse(success_resp)
+        self.assertEqual(resp.is_member, ['canonical-identity-provider'])
+
+    def test_fromSuccessResponse_no_signed(self):
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT}
+        openid_server = server._get_openid_server()
+        orequest = openid_server.decodeRequest(request)
+        success_resp = SuccessResponse(orequest, orequest.message)
+        resp = TeamsResponse.fromSuccessResponse(success_resp)
+        self.assertEqual(resp.is_member, [])
+
+    def test_fromSuccessResponse_all(self):
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT,
+                   'openid.lp.is_member': 'canonical-identity-provider'}
+        openid_server = server._get_openid_server()
+        orequest = openid_server.decodeRequest(request)
+        success_resp = SuccessResponse(orequest, orequest.message)
+        resp = TeamsResponse.fromSuccessResponse(success_resp, False)
+        self.assertEqual(resp.is_member, ['canonical-identity-provider'])
+
+    def test_getExtensionArgs(self):
+        expected = {'is_member': 'canonical-identity-provider'}
+        req = TeamsResponse(is_member=['canonical-identity-provider'])
+        self.assertEqual(req.getExtensionArgs(), expected)
+
+        teams = ['canonical-identity-provider', 'other-team']
+        expected = {'is_member': ','.join(teams)}
+        req = TeamsResponse(is_member=teams)
+        self.assertEqual(req.getExtensionArgs(), expected)
+

=== modified file 'identityprovider/tests/test_views_account.py'
--- identityprovider/tests/test_views_account.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_views_account.py	2010-06-01 12:03:28 +0000
@@ -126,6 +126,10 @@
         self.account = email.account
 
     def test_deactivate_account(self):
+        # make sure we have a preferredemail
+        email = self.account.emailaddress_set.all()[0]
+        self.account.preferredemail = email
+
         # deactivate account
         r = self.client.post('/+deactivate')
         self.assertRedirects(r, '/+deactivated')
@@ -138,3 +142,37 @@
         self.assertEqual(None, self.account.preferredemail)
         for email in self.account.emailaddress_set.all():
             self.assertEqual(EmailStatus.NEW, email.status)
+
+    def test_deactivate_account_no_preferredemail(self):
+        # unset preferredemail
+        for email in self.account.emailaddress_set.all():
+            email.status = EmailStatus.NEW
+            email.save()
+        self.assertEqual(self.account.preferredemail, None)
+
+        # deactivate account
+        r = self.client.post('/+deactivate')
+        self.assertRedirects(r, '/+deactivated')
+
+        # re-request account object from db
+        self._refresh_account()
+
+        # test preferredemail status
+        self.assertEqual(self.account.preferredemail, None)
+
+    def test_deactivate_with_token(self):
+        token = 'a' * 16
+        # This strange dance with session is necessary to overcome way in which
+        # test.Client returns session (recreating it on every Client.session
+        # access)
+        session = self.client.session
+        session[token] = 'raw_orequest content'
+        session.save()
+
+        # deactivate account
+        r = self.client.post('/%s/+deactivate' % token)
+        self.assertRedirects(r, '/+deactivated')
+
+        # test token is preserved
+        self.assertEquals(self.client.session.get(token),
+                          'raw_orequest content')

=== modified file 'identityprovider/tests/test_views_consumer.py'
--- identityprovider/tests/test_views_consumer.py	2010-05-10 18:26:52 +0000
+++ identityprovider/tests/test_views_consumer.py	2010-06-01 12:03:28 +0000
@@ -5,6 +5,12 @@
 import unittest
 
 from openid import fetchers
+from openid.consumer import consumer
+from openid.consumer.discover import OpenIDServiceEndpoint, OPENID_2_0_TYPE
+from openid.message import IDENTIFIER_SELECT
+from openid.store.filestore import FileOpenIDStore
+from openid.yadis.constants import YADIS_CONTENT_TYPE
+from urllib import quote_plus
 
 # need to override fetcher before consumer module gets imported
 class MockFetcher(fetchers.Urllib2Fetcher):
@@ -14,8 +20,12 @@
         return super(MockFetcher, self).fetch(url, body, headers)
 
 fetchers.Urllib2Fetcher = MockFetcher
-from identityprovider.views.consumer import (getBaseURL, startOpenID,
-    renderIndexPage)
+from identityprovider.tests.utils import (SQLCachedTestCase,
+    OpenIDProviderTestCase)
+from identityprovider.views import server
+from identityprovider.views.consumer import (getBaseURL, normalDict,
+    getOpenIDStore, getConsumer, getViewURL, renderIndexPage, startOpenID,
+    finishOpenID, rpXRDS)
 
 
 class DummyRequest(object):
@@ -26,7 +36,7 @@
 class DummyDjangoRequest(object):
     META = {
         'HTTP_HOST': "localhost",
-        'SCRIPT_NAME': "http://localhost/consumer/",
+        'SCRIPT_NAME': "http://localhost",
         'SERVER_PROTOCOL': "http",
     }
     POST = {
@@ -64,6 +74,14 @@
         url = getBaseURL(self.https_req)
         self.assertEqual(url, self.https_consumer_url)
 
+    def test_consumer_on_port(self):
+        req = DummyRequest({'HTTP_HOST': 'localhost',
+                            'SERVER_PORT': '81',
+                            'SERVER_PROTOCOL': 'HTTP/1.1'})
+        base_url = 'http://localhost:81/'
+        url = getBaseURL(req)
+        self.assertEqual(url, base_url)
+
 
 class RenderIndexPageTestCase(unittest.TestCase):
     def test_renderIndexPage_utf8(self):
@@ -78,3 +96,159 @@
         headers = dict(response.items())
         self.assertEqual(headers['Content-Type'], 'text/html; charset=utf-8')
         self.assertTrue(fullname in response.content)
+
+
+class ConsumerTestCase(OpenIDProviderTestCase):
+    def setUp(self):
+        super(ConsumerTestCase, self).setUp()
+
+        self.req = DummyDjangoRequest()
+        self.endpoint = OpenIDServiceEndpoint()
+        self.endpoint.claimed_id = 'oid'
+        self.endpoint.server_url = 'http://localhost/'
+
+        class MockConsumer(consumer.Consumer):
+            def begin(this, url):
+                auth_request = consumer.AuthRequest(self.endpoint, None)
+                return auth_request
+        self.old_consumer = consumer.Consumer
+        consumer.Consumer = MockConsumer
+
+    def tearDown(self):
+        consumer.Consumer = self.old_consumer
+
+        super(ConsumerTestCase, self).tearDown()
+
+    def test_normalDict(self):
+        req = DummyDjangoRequest()
+        self.assertEqual(normalDict(req.POST),
+            {'openid_identifier': "http://localhost/+id/abcd123"})
+
+    def test_startOpenID_get(self):
+        r = self.client.get('/consumer/', **self.req.META)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'consumer/index.html')
+
+    def test_startOpenID_discovery_error(self):
+        # we don't want the mock consumer for this test
+        consumer.Consumer = self.old_consumer
+
+        r = self.client.post('/consumer/', self.req.POST, **self.req.META)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'consumer/index.html')
+        self.assertTrue(r.context['error'].startswith(
+            'OpenID discovery error'))
+
+    def test_startOpenID_sreg(self):
+        self.req.POST.update({'sreg': 'yes', 'sreg_nickname': 'yes'})
+        r = self.client.post('/consumer/', self.req.POST, **self.req.META)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.sreg.required'],
+                         quote_plus('fullname,email'))
+        self.assertEqual(query['openid.sreg.optional'], 'nickname')
+
+    def test_startOpenID_teams(self):
+        self.req.POST.update({'teams': 'yes'})
+        r = self.client.post('/consumer/', self.req.POST, **self.req.META)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.lp.query_membership'],
+                         quote_plus('myteam,hwdb-team,otherteam'))
+
+    def test_startOpenID_should_redirect(self):
+        r = self.client.post('/consumer/', self.req.POST, **self.req.META)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.trust_root'],
+                         quote_plus('http://localhost/consumer/'))
+        self.assertEqual(query['openid.return_to'],
+                         quote_plus('http://localhost/consumer/finish/'))
+
+    def test_startOpenID_render_template(self):
+        # replace consumer with OpenID 2.0 compliant one
+        class MockConsumer(consumer.Consumer):
+            def begin(self, url):
+                endpoint = OpenIDServiceEndpoint()
+                endpoint.claimed_id = 'oid'
+                endpoint.server_url = 'http://localhost/'
+                endpoint.type_uris = [OPENID_2_0_TYPE]
+                auth_request = consumer.AuthRequest(endpoint, None)
+                return auth_request
+        consumer.Consumer = MockConsumer
+
+        r = self.client.post('/consumer/', self.req.POST, **self.req.META)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'consumer/request_form.html')
+        self.assertTrue('openid_message'in r.context['html'])
+
+    def test_finishOpenID_get(self):
+        r = self.client.get('/consumer/finish/', **self.req.META)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'consumer/index.html')
+
+    def test_finishOpenID_post(self):
+        r = self.client.post('/consumer/finish/', **self.req.META)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'consumer/index.html')
+
+    def test_finishOpenID_consumer_success(self):
+        def mock_complete(this, request_args, return_to):
+            request = {'openid.mode': 'checkid_setup',
+                       'openid.trust_root': 'http://localhost/',
+                       'openid.return_to': 'http://localhost/',
+                       'openid.identity': IDENTIFIER_SELECT,
+                       'openid.sreg.nickname': 'mynickname',
+                       'openid.lp.is_member': 'myteam'}
+            openid_server = server._get_openid_server()
+            orequest = openid_server.decodeRequest(request)
+            response = consumer.SuccessResponse(
+                self.endpoint, orequest.message,
+                signed_fields=['openid.sreg.nickname',
+                               'openid.lp.is_member'])
+            return response
+        old_complete = consumer.Consumer.complete
+        consumer.Consumer.complete = mock_complete
+
+        r = self.client.post('/consumer/finish/', self.req.POST,
+                             **self.req.META)
+        self.assertEqual(r.context['url'], 'oid')
+        self.assertEqual(r.context['sreg'], [('nickname', 'mynickname')])
+        self.assertEqual(r.context['teams'], ['myteam'])
+
+        consumer.Consumer.complete = old_complete
+
+    def test_finishOpenID_consumer_cancel(self):
+        def mock_complete(this, request_args, return_to):
+            response = consumer.CancelResponse(self.endpoint)
+            return response
+        old_complete = consumer.Consumer.complete
+        consumer.Consumer.complete = mock_complete
+
+        r = self.client.post('/consumer/finish/', self.req.POST,
+                             **self.req.META)
+        self.assertEqual(r.context['message'],
+                         'OpenID authentication cancelled.')
+
+        consumer.Consumer.complete = old_complete
+
+    def test_finishOpenID_consumer_failure(self):
+        def mock_complete(this, request_args, return_to):
+            response = consumer.FailureResponse(self.endpoint,
+                                                message='some error')
+            return response
+        old_complete = consumer.Consumer.complete
+        consumer.Consumer.complete = mock_complete
+
+        r = self.client.post('/consumer/finish/', self.req.POST,
+                             **self.req.META)
+        self.assertEqual(r.context['error'],
+                         'OpenID authentication failed.')
+        self.assertEqual(r.context['failure_reason'], 'some error')
+
+        consumer.Consumer.complete = old_complete
+
+    def test_rpXRDS(self):
+        r = self.client.get('/consumer/xrds/', **self.req.META)
+        self.assertTemplateUsed(r, 'openidapplication-xrds.xml')
+        self.assertEqual(r['Content-Type'], YADIS_CONTENT_TYPE)

=== modified file 'identityprovider/tests/test_views_i18n.py'
--- identityprovider/tests/test_views_i18n.py	2010-05-13 15:01:03 +0000
+++ identityprovider/tests/test_views_i18n.py	2010-06-01 12:03:28 +0000
@@ -1,4 +1,5 @@
 from identityprovider.tests.utils import BasicAccountTestCase
+from identityprovider.models.account import Account
 
 
 class SetLanguageTestCase(BasicAccountTestCase):
@@ -23,3 +24,24 @@
         r = self.client.post('/set_language',
                              {'language': 'es', 'next': '/+login'})
         self.assertRedirects(r, '/+login')
+
+    def test_not_known_http_method_cases_404(self):
+        # Faking client.put method
+        r = self.client.get('/set_language', REQUEST_METHOD='PUT')
+        self.assertEquals(r.status_code, 404)
+
+    def test_setting_unsupported_language_raises_404(self):
+        r = self.client.post('/set_language', {'language': 'xx'})
+        self.assertEquals(r.status_code, 404)
+
+    def test_getting_renders_choose_page(self):
+        r = self.client.get('/set_language')
+        self.assertTemplateUsed(r, 'select_language.html')
+
+    def test_setting_language_for_authenticated_users_updates_db(self):
+        self.client.login(username='[e-mail]', password='test')
+        self.client.post('/set_language', {'language': 'es', 'next': '/'})
+
+        account = Account.objects.get_by_email('[e-mail]')
+
+        self.assertEquals(account.preferredlanguage, 'es')

=== modified file 'identityprovider/tests/test_views_server.py'
--- identityprovider/tests/test_views_server.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_views_server.py	2010-06-01 12:03:28 +0000
@@ -1,13 +1,28 @@
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+import datetime
+import urlparse
+from copy import deepcopy
 from django.conf import settings
+from django.contrib.auth.models import AnonymousUser
+from openid.extensions import pape
 from openid.extensions.sreg import SRegRequest
+from openid.message import Message, OPENID2_NS, IDENTIFIER_SELECT
+from openid.server.server import Server, ProtocolError
+from openid.yadis.constants import YADIS_HEADER_NAME
+from urllib import quote, quote_plus
 
-from identityprovider.models import Account, OpenIDRPConfig, Person
+import identityprovider.signed as signed
+from identityprovider.models import (Account, LPAccount, OpenIDAuthorization,
+    OpenIDRPConfig, Person)
+from identityprovider.models.const import (AccountStatus, EmailStatus,
+    AccountCreationRationale)
+from identityprovider.models.person import PersonLocation
+from identityprovider.models.authtoken import create_token
 from identityprovider.views import server
 from identityprovider.tests.utils import (SQLCachedTestCase,
-                                          BasicAccountTestCase)
+    BasicAccountTestCase, AuthenticatedTestCase, OpenIDProviderTestCase)
 from identityprovider.const import PERSON_VISIBILITY_PRIVATE_MEMBERSHIP
 
 __all__ = ['UntrustedRPTest', ]
@@ -16,14 +31,649 @@
 class DummyORequest(object):
     mode = 'checkid_setup'
     trust_root = 'http://localhost/'
+    identity = 'http://localhost/+id/mark_oid'
+
+    def idSelect(self):
+        return False
 
 
 class DummyRequest(object):
-    def __init__(self):
-        self.session = {}
-
-
-class UntrustedRPTest(SQLCachedTestCase):
+    def __init__(self, META=None, REQUEST=None):
+        class MockSession(dict):
+            def __init__(self):
+                self.session_key = 'session-key'
+            def flush(self):
+                pass
+
+        self.META = META if META is not None else {}
+        self.REQUEST = REQUEST if REQUEST is not None else {}
+        self.session = MockSession()
+
+
+class HandleOpenIDErrorTestCase(OpenIDProviderTestCase):
+
+    # tests for the _handle_openid_error method
+
+    def test_handle_openid_error_with_encode_url(self):
+        params = {'openid.return_to': 'http://localhost/'}
+        r = self.client.get('/+openid', params)
+        query = self.get_query(r)
+        error_msg = 'No+mode+value+in+message+'
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'error')
+        self.assertTrue(query['openid.error'].startswith(error_msg))
+
+    def test_handle_openid_error_other(self):
+        params = {'openid.mode': 'checkid_setup'}
+        r = self.client.get('/+openid', params)
+        error_mode = "mode:error"
+        error_msg = "error:Missing required field 'return_to'"
+        self.assertEqual(r.status_code, 200)
+        self.assertTrue(error_mode in r.content)
+        self.assertTrue(error_msg in r.content)
+
+
+class ProcessOpenIDRequestTestCase(OpenIDProviderTestCase):
+
+    # tests for the _process_openid_request method
+
+    def test_process_openid_request_no_orequest(self):
+        r = self.client.get('/+openid')
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'server_info.html')
+
+
+class HandleUserResponseTestCase(OpenIDProviderTestCase):
+    fixtures = ['test']
+
+    def setUp(self):
+        super(HandleUserResponseTestCase, self).setUp()
+
+        # create a trusted rpconfig
+        self.rpconfig = OpenIDRPConfig(trust_root='http://localhost/')
+        self.rpconfig.save()
+
+        self.params = {'openid.trust_root': 'http://localhost/',
+                       'openid.return_to': 'http://localhost/',
+                       'openid.identity': IDENTIFIER_SELECT,
+                       'openid.claimed_id': 'http://localhost/~userid',
+                       'openid.ns': OPENID2_NS,
+                       'openid.mode': 'checkid_setup'}
+
+    # tests for the _handle_user_response method
+
+    def test_handle_user_response_checkid_immediate(self):
+        self.params['openid.mode'] = 'checkid_immediate'
+
+        r = self.client.get('/+openid', self.params)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'setup_needed')
+
+    def test_handle_user_response_no_valid_openid(self):
+        self.params.update({'openid.identity': 'bogus',
+                            'openid.claimed_id': 'bogus'})
+        r = self.client.get('/+openid', self.params)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'invalid_identifier.html')
+
+    def test_handle_user_response_openid_is_authorized_idselect(self):
+        # update rp to auto authorize
+        self.rpconfig.auto_authorize = True
+        self.rpconfig.save()
+
+        account = Account.objects.get_by_email('[e-mail]')
+        self.client.login(username='[e-mail]', password='test')
+        r = self.client.get('/+openid', self.params)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'id_res')
+        self.assertEqual(query['openid.identity'],
+                         quote_plus(account.openid_identity_url))
+
+    def test_handle_user_response_openid_is_authorized_other_id(self):
+        self.rpconfig.auto_authorize = True
+        self.rpconfig.save()
+
+        self.params['openid.identity'] = 'http://openid.launchpad.dev/+id/mark_oid'
+        self.client.login(username='[e-mail]', password='test')
+        r = self.client.get('/+openid', self.params)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'id_res')
+        self.assertEqual(query['openid.identity'],
+                         quote_plus(self.params['openid.identity']))
+
+    def test_handle_user_response_user_is_authenticated(self):
+        self.params['openid.identity'] = 'http://openid.launchpad.dev/+id/other_oid'
+        self.client.login(username='[e-mail]', password='test')
+        r = self.client.get('/+openid', self.params)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'cancel')
+
+    def test_handle_user_response_decide(self):
+        r = self.client.get('/+openid', self.params)
+        self.assertEqual(r.status_code, 302)
+        self.assertTrue(r['Location'].endswith('+decide'))
+
+    def test_handle_user_response_with_referer(self):
+        META = {'HTTP_REFERER': 'http://localhost/'}
+        r = self.client.get('/+openid', self.params, **META)
+        openid_referer = self.client.cookies.get('openid_referer')
+        self.assertEqual(r.status_code, 302)
+        self.assertTrue(r['Location'].endswith('+decide'))
+        self.assertEqual(openid_referer.value, META['HTTP_REFERER'])
+
+
+class ValidOpenIDTestCase(OpenIDProviderTestCase):
+
+    def test_is_valid_openid_idselect(self):
+        valid = server._is_valid_openid_for_this_site(IDENTIFIER_SELECT)
+        self.assertTrue(valid)
+
+    def test_is_valid_openid_different_scheme(self):
+        srvparts = urlparse.urlparse(settings.SSO_ROOT_URL)
+        if srvparts.scheme == 'http':
+            scheme = 'https'
+        else:
+            scheme = 'http'
+        identity = settings.SSO_ROOT_URL.replace(srvparts.scheme, scheme)
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertFalse(valid)
+
+    def test_is_valid_openid_different_port(self):
+        srvparts = urlparse.urlparse(settings.SSO_ROOT_URL)
+        if srvparts.port is not None:
+            port = int(srvparts.port) + 1
+        else:
+            port = 81
+        identity = "%s:%s//%s%s" % (srvparts.scheme, port,
+            srvparts.hostname, srvparts.path)
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertFalse(valid)
+
+    def test_is_valid_openid_different_hostname(self):
+        identity = 'http://testserver/'
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertFalse(valid)
+
+    def test_is_valid_openid_path_patterns(self):
+        identity = settings.SSO_ROOT_URL
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertTrue(valid)
+
+        identity += '+id/foo'
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertTrue(valid)
+
+        identity = settings.SSO_ROOT_URL + '~foo'
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertTrue(valid)
+
+        # try a non-normalized uri
+        identity = settings.SSO_ROOT_URL.strip('/')
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertTrue(valid)
+
+    def test_is_valid_openid_other(self):
+        identity = settings.SSO_ROOT_URL + '+foo'
+        valid = server._is_valid_openid_for_this_site(identity)
+        self.assertFalse(valid)
+
+    def test_is_valid_openid_error(self):
+        valid = server._is_valid_openid_for_this_site(None)
+        self.assertFalse(valid)
+
+
+class DecideTestCase(AuthenticatedTestCase, OpenIDProviderTestCase):
+    def setUp(self):
+        super(DecideTestCase, self).setUp(disableCSRF=True)
+
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT}
+        openid_server = server._get_openid_server()
+        self.orequest = openid_server.decodeRequest(request)
+        self.token = create_token(16)
+        session = self.client.session
+        session[self.token] = signed.dumps(self.orequest,
+                                      settings.SECRET_KEY)
+        session.save()
+
+    def test_decide_invalid(self):
+        token = 'a' * 16
+        r = self.client.get("/%s/+decide" % token)
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.content, 'Invalid OpenID transaction')
+
+    def test_decide_authenticated(self):
+        r = self.client.post("/%s/+decide" % self.token, {'yes': 'yes'})
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'id_res')
+
+    def test_decide_auto_authorize(self):
+        # make sure rpconfig is set to auto authorize
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost/',
+                                  auto_authorize=True)
+        rpconfig.save()
+
+        r = self.client.post("/%s/+decide" % self.token)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'id_res')
+
+    def test_decide_process(self):
+        r = self.client.post("/%s/+decide" % self.token)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'decide.html')
+
+    def test_decide_not_authenticated(self):
+        self.client.logout()
+
+        # create a trusted rpconfig
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost/')
+        rpconfig.save()
+
+        # start openid request
+        params = {'openid.trust_root': 'http://localhost/',
+                  'openid.return_to': 'http://localhost/',
+                  'openid.identity': IDENTIFIER_SELECT,
+                  'openid.claimed_id': 'http://localhost/~userid',
+                  'openid.ns': OPENID2_NS,
+                  'openid.mode': 'checkid_setup'}
+        r = self.client.get('/+openid', params)
+        # follow redirect
+        path = r['Location'].split('http://testserver')[1]
+        self.assertTrue(path.endswith('+decide'))
+        r = self.client.get(path)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'registration/login.html')
+
+
+class PreAuthorizeTestCase(AuthenticatedTestCase):
+    def setUp(self):
+        super(PreAuthorizeTestCase, self).setUp(disableCSRF=True)
+
+    def test_pre_authorize_get(self):
+        r = self.client.get('/+pre-authorize-rp')
+        self.assertEqual(r.status_code, 400)
+
+    def test_pre_authorize_unauthorized(self):
+        r = self.client.post('/+pre-authorize-rp',
+            {'trust_root': 'http://localhost/',
+             'callback': 'http://localhost/'})
+        self.assertEqual(r.status_code, 400)
+
+    def test_pre_authorize_no_pre_authorized(self):
+        old_OPENID_PREAUTHORIZATION_ACL = settings.OPENID_PREAUTHORIZATION_ACL
+        settings.OPENID_PREAUTHORIZATION_ACL = []
+
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost/')
+        rpconfig.save()
+
+        extra = {'HTTP_REFERER': 'http://localhost/'}
+        r = self.client.post('/+pre-authorize-rp',
+            {'trust_root': 'http://localhost/',
+             'callback': 'http://localhost/'},
+            **extra)
+        self.assertEqual(r.status_code, 400)
+
+        settings.OPENID_PREAUTHORIZATION_ACL = old_OPENID_PREAUTHORIZATION_ACL
+
+    def test_pre_authorize_authenticated(self):
+        old_OPENID_PREAUTHORIZATION_ACL = settings.OPENID_PREAUTHORIZATION_ACL
+        settings.OPENID_PREAUTHORIZATION_ACL = [
+            ('http://localhost/', 'http://localhost/')
+        ]
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost/')
+        rpconfig.save()
+
+        extra = {'HTTP_REFERER': 'http://localhost/'}
+        r = self.client.post('/+pre-authorize-rp',
+            {'trust_root': 'http://localhost/',
+             'callback': 'http://localhost/'},
+            **extra)
+        self.assertRedirects(r, 'http://localhost/')
+
+        settings.OPENID_PREAUTHORIZATION_ACL = old_OPENID_PREAUTHORIZATION_ACL
+
+    def test_pre_authorize_not_authenticated(self):
+        old_OPENID_PREAUTHORIZATION_ACL = settings.OPENID_PREAUTHORIZATION_ACL
+        settings.OPENID_PREAUTHORIZATION_ACL = [
+            ('http://localhost/', 'http://localhost/')
+        ]
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost/')
+        rpconfig.save()
+
+        self.client.logout()
+        extra = {'HTTP_REFERER': 'http://localhost/'}
+        r = self.client.post('/+pre-authorize-rp',
+            {'trust_root': 'http://localhost/',
+             'callback': 'http://localhost/'},
+            **extra)
+        next_url = '/+login?next=' + quote('/+pre-authorize-rp?')
+        self.assertRedirects(r, next_url)
+
+        settings.OPENID_PREAUTHORIZATION_ACL = old_OPENID_PREAUTHORIZATION_ACL
+
+    def test_pre_authorize_after_login(self):
+        old_OPENID_PREAUTHORIZATION_ACL = settings.OPENID_PREAUTHORIZATION_ACL
+        settings.OPENID_PREAUTHORIZATION_ACL = [
+            ('http://localhost/', 'http://localhost/')
+        ]
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost/')
+        rpconfig.save()
+
+        # make sure we are logged out
+        self.client.logout()
+
+        # attempt to pre-authorize
+        extra = {'HTTP_REFERER': 'http://localhost/'}
+        data = {'trust_root': 'http://localhost/',
+                'callback': 'http://localhost/'}
+        r = self.client.post('/+pre-authorize-rp', data, **extra)
+        # we get redirected to login
+        next_url = '/+login?next=' + quote('/+pre-authorize-rp?')
+        self.assertRedirects(r, next_url)
+        # and the referer info is stored in the session
+        self.assertEqual(self.client.session['pre_auth_referer'],
+                         'http://localhost/')
+        self.assertTrue(self.client.session['pre_auth_referer_for'],
+                        'http://localhost/')
+
+        # login and redirect to pre-authorize again
+        data.update({'email': '[e-mail]', 'password': 'test',
+                     'next': '/+pre-authorize-rp'})
+        r = self.client.post('/+login', data, **extra)
+        r = self.client.post('/+pre-authorize-rp', data, **extra)
+        # we get effectively pre-authorized
+        self.assertRedirects(r, 'http://localhost/')
+        # and the pre-auth session data gets removed
+        self.assertFalse('pre_auth_referer' in self.client.session)
+        self.assertFalse('pre_auth_referer_for' in self.client.session)
+
+        settings.OPENID_PREAUTHORIZATION_ACL = old_OPENID_PREAUTHORIZATION_ACL
+
+    def test_pre_authorize_sess_referer_not_trust_root(self):
+        old_OPENID_PREAUTHORIZATION_ACL = settings.OPENID_PREAUTHORIZATION_ACL
+        settings.OPENID_PREAUTHORIZATION_ACL = [
+            ('http://otherhost/', 'http://localhost/')
+        ]
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost/')
+        rpconfig.save()
+
+        # make sure we are logged out
+        self.client.logout()
+
+        # attempt to pre-authorize
+        extra = {'HTTP_REFERER': 'http://otherhost/'}
+        data = {'trust_root': 'http://localhost/',
+                'callback': 'http://localhost/'}
+        r = self.client.post('/+pre-authorize-rp', data, **extra)
+        # we get redirected to login
+        next_url = '/+login?next=' + quote('/+pre-authorize-rp?')
+        self.assertRedirects(r, next_url)
+        # and the referer info is stored in the session
+        self.assertEqual(self.client.session['pre_auth_referer'],
+                         'http://otherhost/')
+        self.assertTrue(self.client.session['pre_auth_referer_for'],
+                        'http://localhost/')
+
+        # attempt to pre-authorize for a different trust_root
+        data['trust_root'] = 'http://otherhost/'
+        r = self.client.post('/+pre-authorize-rp', data, **extra)
+        # pre-authorization is denied
+        self.assertEqual(r.status_code, 400)
+
+        settings.OPENID_PREAUTHORIZATION_ACL = old_OPENID_PREAUTHORIZATION_ACL
+
+
+class CancelTestCase(AuthenticatedTestCase, OpenIDProviderTestCase):
+    def setUp(self):
+        super(CancelTestCase, self).setUp(disableCSRF=True)
+
+        self.params = request = {'openid.mode': 'checkid_setup',
+                                 'openid.trust_root': 'http://localhost/',
+                                 'openid.return_to': 'http://localhost/',
+                                 'openid.identity': IDENTIFIER_SELECT}
+        openid_server = server._get_openid_server()
+        self.orequest = openid_server.decodeRequest(request)
+        self.token = create_token(16)
+        session = self.client.session
+        session[self.token] = signed.dumps(self.orequest,
+                                      settings.SECRET_KEY)
+        session.save()
+
+        rpconfig = OpenIDRPConfig(trust_root='http://localhost')
+        rpconfig.save()
+
+    def test_cancel_invalid_openid_transaction(self):
+        token = 'a' * 16
+        r = self.client.get("/%s/+cancel" % token)
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.content, 'Invalid OpenID transaction')
+
+    def test_cancel_user_is_authenticated(self):
+        # cancel request
+        r = self.client.get("/%s/+cancel" % self.token)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'cancel')
+
+    def test_cancel_user_is_not_authenticated(self):
+        # save session data
+        session_data = self.client.session[self.token]
+
+        # logout
+        self.client.logout()
+        # request something so we get a real session in the client
+        self.client.get('/+openid', self.params)
+
+        # manipulate session
+        session = self.client.session
+        session[self.token] = session_data
+        session.save()
+
+        # cancel request
+        r = self.client.get("/%s/+cancel" % self.token)
+        query = self.get_query(r)
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(query['openid.mode'], 'cancel')
+
+
+class XRDSTestCase(OpenIDProviderTestCase):
+    fixtures = ['test']
+
+    def test_xrds(self):
+        r = self.client.get('/+xrds')
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r['Content-Type'], 'application/xrds+xml')
+        self.assertTemplateUsed(r, 'openidapplication-xrds.xml')
+
+    def test_identity_page_nonexisting_account(self):
+        r = self.client.get('/+id/aaaaaaaaaaaaaaaa')
+        self.assertEqual(r.status_code, 404)
+
+    def test_identity_page_inactive_account(self):
+        account = Account.objects.get_by_email('[e-mail]')
+        account.status = AccountStatus.DEACTIVATED
+        account.save()
+        r = self.client.get("/+id/%s" % account.openid_identifier)
+        self.assertEqual(r.status_code, 404)
+
+    def test_identity_page_active_account(self):
+        account = Account.objects.get_by_email('[e-mail]')
+        r = self.client.get("/+id/%s" % account.openid_identifier)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'person.html')
+        self.assertEqual(r[YADIS_HEADER_NAME],
+                         "%s/+xrds" % account.openid_identity_url)
+
+    def test_xrds_identity_page_nonexisting_account(self):
+        r = self.client.get('/+id/aaaaaaaaaaaaaaaa')
+        self.assertEqual(r.status_code, 404)
+
+    def test_xrds_identity_page_inactive_account(self):
+        account = Account.objects.get_by_email('[e-mail]')
+        account.status = AccountStatus.DEACTIVATED
+        account.save()
+        r = self.client.get("/+id/%s/+xrds" % account.openid_identifier)
+        self.assertEqual(r.status_code, 404)
+
+    def test_xrds_identity_page_active_account(self):
+        account = Account.objects.get_by_email('[e-mail]')
+        r = self.client.get("/+id/%s/+xrds" % account.openid_identifier)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'person-xrds.xml')
+        self.assertEqual(r['Content-Type'], 'application/xrds+xml')
+
+
+class OpenIDAuthorizedTestCase(AuthenticatedTestCase):
+    fixtures = ['test']
+
+    def setUp(self):
+        super(OpenIDAuthorizedTestCase, self).setUp()
+
+        self.request = DummyRequest()
+        self.request.user = Account.objects.get_by_email('[e-mail]')
+        self.request.user.last_login = datetime.datetime.utcnow()
+        self.orequest = DummyORequest()
+        self.orequest.identity = 'http://openid.launchpad.dev/+id/mark_oid'
+
+        @classmethod
+        def mock_fromOpenIDRequest(cls, request):
+            return None
+        self.old_fromOpenIDRequest = pape.Request.fromOpenIDRequest
+        pape.Request.fromOpenIDRequest = mock_fromOpenIDRequest
+
+    def tearDown(self):
+        pape.Request.fromOpenIDRequest = self.old_fromOpenIDRequest
+
+        super(OpenIDAuthorizedTestCase, self).tearDown()
+
+    def test_openid_is_authorized_not_authenticated(self):
+        self.request.user = AnonymousUser()
+        r = server._openid_is_authorized(self.request, self.orequest)
+        self.assertFalse(r)
+
+    def test_openid_is_authorized_id_owner(self):
+        self.orequest.identity = 'http://localhost/+id/mark_oid'
+        r = server._openid_is_authorized(self.request, self.orequest)
+        self.assertFalse(r)
+
+    def test_openid_is_authorized_should_reauthenticate(self):
+        @classmethod
+        def mock_fromOpenIDRequest(cls, request):
+            class MockPapeRequest(object):
+                max_auth_age = '0'
+            return MockPapeRequest()
+        old_fromOpenIDRequest = pape.Request.fromOpenIDRequest
+        pape.Request.fromOpenIDRequest = mock_fromOpenIDRequest
+
+        r  = server._openid_is_authorized(self.request, self.orequest)
+        self.assertFalse(r)
+
+        pape.Request.fromOpenIDRequest = old_fromOpenIDRequest
+
+    def test_openid_is_authorized_rpconfig(self):
+        rpconfig = OpenIDRPConfig(trust_root=self.orequest.trust_root)
+        rpconfig.auto_authorize = True
+        rpconfig.save()
+
+        r = server._openid_is_authorized(self.request, self.orequest)
+        self.assertTrue(r)
+
+    def test_openid_is_authorized_other(self):
+        r = server._openid_is_authorized(self.request, self.orequest)
+        self.assertFalse(r)
+
+        expires = datetime.datetime.utcnow() + datetime.timedelta(1)
+        OpenIDAuthorization.objects.authorize(self.request.user,
+            self.orequest.trust_root, expires,
+            self.request.session.session_key)
+
+        r = server._openid_is_authorized(self.request, self.orequest)
+        self.assertTrue(r)
+
+
+class ShouldReauthenticateTestCase(SQLCachedTestCase):
+    fixtures = ['test']
+
+    def setUp(self):
+        self.user = Account.objects.get_by_email('[e-mail]')
+        self.orequest = DummyORequest()
+        self.old_fromOpenIDRequest = pape.Request.fromOpenIDRequest
+
+    def tearDown(self):
+        pape.Request.fromOpenIDRequest = self.old_fromOpenIDRequest
+
+    def test_should_reauthenticate_no_pape(self):
+        # pape_request is None
+        @classmethod
+        def mock_fromOpenIDRequest(cls, orequest):
+            return None
+        pape.Request.fromOpenIDRequest = mock_fromOpenIDRequest
+
+        r = server._should_reauthenticate(self.orequest, self.user)
+        self.assertFalse(r)
+
+    def test_should_reauthenticate_no_pape_max_auth_age(self):
+        # pape_request.max_auth_age is None
+        @classmethod
+        def mock_fromOpenIDRequest(cls, request):
+            class MockPapeRequest(object):
+                max_auth_age = None
+            return MockPapeRequest()
+        pape.Request.fromOpenIDRequest = mock_fromOpenIDRequest
+
+        r = server._should_reauthenticate(self.orequest, self.user)
+        self.assertFalse(r)
+
+    def test_should_reauthenticate_invalid_max_auth_age(self):
+        # max_auth_age raise ValueError
+        @classmethod
+        def mock_fromOpenIDRequest(cls, request):
+            class MockPapeRequest(object):
+                max_auth_age = 'bad-value'
+            return MockPapeRequest()
+        pape.Request.fromOpenIDRequest = mock_fromOpenIDRequest
+
+        r = server._should_reauthenticate(self.orequest, self.user)
+        self.assertFalse(r)
+
+    def test_should_reauthenticate_true(self):
+        # last_login <= cutoff
+        @classmethod
+        def mock_fromOpenIDRequest(cls, request):
+            class MockPapeRequest(object):
+                max_auth_age = '0'
+            return MockPapeRequest()
+        pape.Request.fromOpenIDRequest = mock_fromOpenIDRequest
+        self.user.last_login = datetime.datetime.utcnow()
+
+        r = server._should_reauthenticate(self.orequest, self.user)
+        self.assertTrue(r)
+
+    def test_should_reauthenticate_false(self):
+        # last_login > cutoff
+        @classmethod
+        def mock_fromOpenIDRequest(cls, request):
+            class MockPapeRequest(object):
+                max_auth_age = '100'
+            return MockPapeRequest()
+        pape.Request.fromOpenIDRequest = mock_fromOpenIDRequest
+        self.user.last_login = datetime.datetime.utcnow()
+
+        r = server._should_reauthenticate(self.orequest, self.user)
+        self.assertFalse(r)
+
+
+class UntrustedRPTest(OpenIDProviderTestCase):
+    fixtures = ['test']
+
     def setUp(self):
         # Ensure that we're restricting RPs for these tests
         self.old_restrict = getattr(settings, 'SSO_RESTRICT_RP', True)
@@ -39,20 +689,68 @@
         self.assertEquals(302, response.status_code)
         self.assert_(response['Location'].endswith('+untrusted'))
 
+    def test_untrusted(self):
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT}
+        openid_server = server._get_openid_server()
+        orequest = openid_server.decodeRequest(request)
+        token = create_token(16)
+
+        # call up a session-modifying view to get a real session object
+        r = self.client.login(username='[e-mail]', password='test')
+        self.assertTrue(r)
+
+        session = self.client.session
+        session[token] = signed.dumps(orequest,
+                                      settings.SECRET_KEY)
+        session.save()
+
+        r = self.client.get("/%s/+untrusted" % token)
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'untrusted.html')
+
 
 class TestSregFields(BasicAccountTestCase):
+    def setUp(self):
+        super(TestSregFields, self).setUp()
+
+        self.account = Account(
+            creation_rationale=AccountCreationRationale.USER_CREATED,
+            status=AccountStatus.ACTIVE,
+            displayname='User')
+        self.account.save()
+        lp_account = LPAccount(
+            openid_identifier=self.account.openid_identifier)
+        lp_account.save()
+        person = Person(lp_account=lp_account)
+        person.save()
+        now = datetime.datetime.now()
+        personlocation = PersonLocation(date_created=now,
+                                        person=person, time_zone='UTC',
+                                        last_modified_by=person,
+                                        date_last_modified=now)
+        personlocation.save()
+        self.sreg_request = SRegRequest()
+        self.rpconfig = OpenIDRPConfig()
+
     def test_sreg_fields_no_preferredemail(self):
-        _preferredemail = Account.preferredemail
-        Account.preferredemail = None
+        for email in self.account.emailaddress_set.all():
+            email.status = EmailStatus.NEW
+            email.save()
 
-        account = Account()
-        sreg_request = SRegRequest()
-        rpconfig = OpenIDRPConfig()
         expected = []
-        result = server._sreg_fields(account, sreg_request, rpconfig)
+        result = server._sreg_fields(self.account, self.sreg_request,
+                                     self.rpconfig)
         self.assertEqual(expected, result)
 
-        Account.preferredemail = _preferredemail
+    def test_sreg_fields_timezone(self):
+        self.sreg_request.optional = ['timezone']
+        expected = [('timezone', 'UTC')]
+        result = server._sreg_fields(self.account, self.sreg_request,
+                                     self.rpconfig)
+        self.assertEqual(result, expected)
 
 
 class TestGetTeamMemberships(BasicAccountTestCase):

=== added file 'identityprovider/tests/test_views_testing.py'
--- identityprovider/tests/test_views_testing.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_views_testing.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,19 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from django.conf import settings
+from django.test import TestCase
+
+import identityprovider.urls
+
+
+class TestingViewsTestCase(TestCase):
+    def test_error(self):
+        old_DEBUG = settings.DEBUG
+        settings.DEBUG = True
+        reload(identityprovider.urls)
+
+        self.assertRaises(FloatingPointError, self.client.get, '/error')
+
+        settings.DEBUG = old_DEBUG
+

=== modified file 'identityprovider/tests/test_views_ui.py'
--- identityprovider/tests/test_views_ui.py	2010-05-13 16:36:02 +0000
+++ identityprovider/tests/test_views_ui.py	2010-06-01 12:03:28 +0000
@@ -8,11 +8,17 @@
 
 from django.conf import settings
 from django.core import mail
+from django.core import urlresolvers
 from django.test.client import Client
+from openid.message import IDENTIFIER_SELECT
 
+from identityprovider import signed
 from identityprovider.models import authtoken as at
-from identityprovider.views import ui
+from identityprovider.models import EmailAddress, Person, OpenIDRPConfig
+from identityprovider.models.const import EmailStatus
+from identityprovider.views import ui, server
 from identityprovider.models import Account, AuthToken, EmailAddress
+from identityprovider.models.authtoken import create_token
 from identityprovider.models.captcha import Captcha
 from identityprovider.models.const import (AccountStatus, EmailStatus,
     LoginTokenType)
@@ -26,6 +32,19 @@
     def test_is_safe_redirect_url_return_false(self):
         self.assertFalse(ui._is_safe_redirect_url('non-existing'))
 
+    def test_is_safe_redirect_url_with_params(self):
+        self.assertFalse(ui._is_safe_redirect_url('non-existing?q=some_value'))
+
+    def test_is_safe_redirect_url_404(self):
+        def mock_resolve(urlconf):
+            raise urlresolvers.Resolver404()
+        old_resolve = urlresolvers.resolve
+        urlresolvers.resolve = mock_resolve
+
+        self.assertFalse(ui._is_safe_redirect_url('non-existing'))
+
+        urlresolvers.resolve = old_resolve
+
 
 class UIViewsTestCase(BasicAccountTestCase):
 
@@ -88,6 +107,18 @@
             self.assertFormError(r, 'form', None,
                                  'Your account has been deactivated')
 
+    def test_login_without_next_with_token(self):
+        token = 'a'*16
+        r = self.client.post("/%s/+login" % token,
+            {'email': '[e-mail]', 'password': 'test'})
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(r['Location'], "http://testserver/%s/" % token)
+
+    def test_login_without_next_nor_token(self):
+        r = self.client.post('/+login', {'email': '[e-mail]',
+                                         'password': 'test'})
+        self.assertRedirects(r, '/')
+
     def test_logout(self):
         self.authenticate()
         r = self.client.get('/+logout')
@@ -107,6 +138,10 @@
         self.assertEquals(self.client.session.get(token),
                           'raw_orequest content')
 
+    def test_logout_no_preferredlanguage_attribute(self):
+        r = self.client.get('/+logout')
+        self.assertRedirects(r, '/+login')
+
     def create_token(self, token_type, email=None, redirection_url=None):
         token = at.AuthToken.objects.create(
             token_type=token_type,
@@ -219,6 +254,80 @@
         r = self.client.get(location)
         self.assertRedirects(r,  '/+logout-to-confirm')
 
+    def test_confirm_account_redirect_to_decide_token_error(self):
+        # get a valid session
+        token1 = 'a'*16
+        r = self.client.post("/%s/+new_account" % token1,
+                             {'email': '[e-mail]'})
+        self.assertRedirects(r, '/+email-sent')
+
+        # claim token
+        token2 = self.client.session['token_newaccount']
+
+        r = self.client.post("/token/%s/+newaccount" % token2,
+            {'displayname': 'Person', 'password': 'P4ssw0rd',
+             'passwordconfirm': 'P4ssw0rd'})
+        self.assertRedirects(r, "/%s/+decide" % token1)
+
+    def test_confirm_account_redirect_to_decide_token(self):
+        # get a valid session
+        token1 = create_token(16)
+        r = self.client.post("/%s/+new_account" % token1,
+                             {'email': '[e-mail]'})
+        self.assertRedirects(r, '/+email-sent')
+
+        # claim token
+        token2 = self.client.session['token_newaccount']
+
+        r = self.client.post("/token/%s/+newaccount" % token2,
+            {'displayname': 'Person', 'password': 'P4ssw0rd',
+             'passwordconfirm': 'P4ssw0rd'})
+        self.assertRedirects(r, "/%s/+decide" % token1)
+
+    def test_confirm_account_redirect_to_decide_with_rpconfig(self):
+        token1 = create_token(16)
+        r = self.client.post("/%s/+new_account" % token1,
+                             {'email': '[e-mail]'})
+        self.assertRedirects(r, '/+email-sent')
+
+        request = {'openid.mode': 'checkid_setup',
+                   'openid.trust_root': 'http://localhost/',
+                   'openid.return_to': 'http://localhost/',
+                   'openid.identity': IDENTIFIER_SELECT}
+        openid_server = server._get_openid_server()
+        orequest = openid_server.decodeRequest(request)
+        session = self.client.session
+        session[token1] = signed.dumps(orequest, settings.SECRET_KEY)
+        session.save()
+
+        # create rpconfig
+        rpconfig = OpenIDRPConfig.objects.create(trust_root='http://localhost/')
+
+        # claim token
+        token2 = self.client.session['token_newaccount']
+
+        r = self.client.post("/token/%s/+newaccount" % token2,
+            {'displayname': 'Person', 'password': 'P4ssw0rd',
+             'passwordconfirm': 'P4ssw0rd'})
+        self.assertRedirects(r, "/%s/+decide" % token1)
+
+        # verify account created
+        account = Account.objects.get(pk=self.client.session['_auth_user_id'])
+        self.assertEqual(account.creation_rationale,
+                         rpconfig.creation_rationale)
+
+    def test_confirm_account_invalid_form(self):
+        # setup session
+        token1 = create_token(16)
+        r = self.client.post("/%s/+new_account" % token1,
+                             {'email': '[e-mail]'})
+
+        # test view
+        token2 = self.client.session['token_newaccount']
+        r = self.client.post("/token/%s/+newaccount" % token2)
+        self.assertTemplateUsed(r, 'registration/confirm_new_account.html')
+        self.assertFormError(r, 'form', 'displayname', 'Required field.')
+
     def test_forgot_password_when_captcha_verification_fails(self):
         r = self.request_when_captcha_fails('/+forgot_password',
                                             {'email': '[e-mail]'})
@@ -276,6 +385,21 @@
         self.assertEqual(len(mail.outbox), 0)
         mail.outbox = []
 
+    def test_forgot_password_invalid_form(self):
+        r = self.client.post('/+forgot_password')
+        self.assertEqual(r.status_code, 200)
+        self.assertTemplateUsed(r, 'registration/forgot_password.html')
+        self.assertFormError(r, 'form', 'email', 'Required field.')
+
+    def test_forgot_password_with_token(self):
+        token1 = create_token(16)
+        r = self.client.post("/%s/+forgot_password" % token1,
+                             {'email': '[e-mail]'})
+
+        token2 = self.client.session['token_forgotpassword']
+        auth_token = AuthToken.objects.get(token=token2)
+        self.assertEqual(auth_token.redirection_url, "/%s/+decide" % token1)
+
     def test_reset_password_when_account_active(self):
         r = self.client.post('/+forgot_password',
                              {'email': '[e-mail]'})
@@ -391,6 +515,31 @@
         r = self.client.post('/token/%s/+resetpassword' % token, data)
         self.assertRedirects(r, '/+bad-token')
 
+    def test_reset_password_invalid_form(self):
+        # get valid session
+        r = self.client.post('/+forgot_password',
+                             {'email': '[e-mail]'})
+        # claim token
+        token = self.client.session['token_forgotpassword']
+
+        # test view
+        r = self.client.post("/token/%s/+resetpassword" % token)
+        self.assertTemplateUsed(r, 'registration/reset_password.html')
+        self.assertFormError(r, 'form', 'password', 'Required field.')
+
+    def test_reset_password_get(self):
+        # get valid session
+        r = self.client.post('/+forgot_password',
+                             {'email': '[e-mail]'})
+        # claim token
+        token = self.client.session['token_forgotpassword']
+
+        # test view
+        r = self.client.get("/token/%s/+resetpassword" % token)
+        self.assertTemplateUsed(r, 'registration/reset_password.html')
+        for context in r.context:
+            self.assertEqual(context['form'].errors, {})
+
     def request_when_captcha_fails(self, url, data):
         class MockCaptcha(object):
             def __init__(self, *args):
@@ -440,6 +589,27 @@
             self.assertEqual(len(mail.outbox), mails_sent)
             mail.outbox = []
 
+    def test_new_account_when_person_exists_and_account_not(self):
+        def mock_debug(logger, msg):
+            self.msg = msg
+        old_debug = logging.Logger.debug
+        logging.Logger.debug = mock_debug
+
+        person = Person.objects.create(displayname='Person')
+        emailaddress = EmailAddress.objects.create(email='[e-mail]',
+            lp_person=person, status=EmailStatus.VALIDATED)
+
+        r = self.client.post('/+new_account', {'email': '[e-mail]'})
+        self.assertTrue('account for person' in self.msg)
+
+        logging.Logger.debug = old_debug
+
+    def test_new_account_when_account_not_exists_no_token(self):
+        r = self.client.post('/+new_account', {'email': '[e-mail]'})
+        token = self.client.session['token_newaccount']
+        token_obj = AuthToken.objects.get(token=token)
+        self.assertEqual(token_obj.redirection_url, '/')
+
     def test_edit_account_template(self):
         r = self.client.post('/+login', {'email': '[e-mail]',
                                          'password': 'test',
@@ -546,6 +716,10 @@
         r = self.client.get("/token/%s/+newemail" % token.token)
         return (r, token)
 
+    def test_suspended(self):
+        r = self.client.get('/+suspended')
+        self.assertTemplateUsed(r, 'account/suspended.html')
+
 
 class UIViewsLPModelTestCase(LPAccountTestCase):
     def test_new_account_when_person_does_not_exist(self):

=== modified file 'identityprovider/tests/test_widgets.py'
--- identityprovider/tests/test_widgets.py	2010-04-21 15:29:24 +0000
+++ identityprovider/tests/test_widgets.py	2010-06-01 12:03:28 +0000
@@ -1,8 +1,74 @@
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+from datetime import datetime
+from django.conf import settings
 from unittest import TestCase
-from identityprovider.widgets import CommaSeparatedWidget
+
+from identityprovider.models import Account, LPAccount
+from identityprovider.tests.utils import SQLCachedTestCase
+from identityprovider.widgets import (CommaSeparatedWidget, ROAwareSelect,
+    ROAwareTextInput, StatusWidget, ReadOnlyDateTimeWidget, LPUsernameWidget)
+
+
+class ROAwareTextInputTestCase(TestCase):
+    def setUp(self):
+        self.widget = ROAwareTextInput()
+        self.old_READ_ONLY_MODE = getattr(settings, 'READ_ONLY_MODE', False)
+
+    def tearDown(self):
+        settings.READ_ONLY_MODE = self.old_READ_ONLY_MODE
+
+    def test_render_when_readonly(self):
+        settings.READ_ONLY_MODE = True
+
+        r = self.widget.render('test', None)
+        self.assertEqual(r, '<span class="rofield"></span>')
+
+        r = self.widget.render('test', 'value')
+        self.assertEqual(r, '<span class="rofield">value</span>')
+
+    def test_render_when_not_readonly(self):
+        settings.READ_ONLY_MODE = False
+
+        r = self.widget.render('test', None)
+        self.assertEqual(r, '<input type="text" name="test" />')
+
+        r = self.widget.render('test', 'value')
+        self.assertEqual(r, '<input type="text" name="test" value="value" />')
+
+
+class ROAwareSelectTestCase(TestCase):
+    def setUp(self):
+        self.choices = (('1', 'One'), ('2', 'Two'))
+        self.widget = ROAwareSelect()
+        self.old_READ_ONLY_MODE = getattr(settings, 'READ_ONLY_MODE', False)
+
+    def tearDown(self):
+        settings.READ_ONLY_MODE = self.old_READ_ONLY_MODE
+
+    def test_render_when_readonly(self):
+        settings.READ_ONLY_MODE = True
+
+        r = self.widget.render('test', 'value', choices=self.choices)
+        self.assertEqual(r, '<span class="rofield"></span>')
+
+    def test_render_when_readonly_selected(self):
+        settings.READ_ONLY_MODE = True
+
+        choices = self.choices + (('value', 'The Value'),)
+        r = self.widget.render('test', 'value', choices=choices)
+        self.assertEqual(r, '<span class="rofield">The Value</span>')
+
+    def test_render_when_not_readonly(self):
+        settings.READ_ONLY_MODE = False
+
+        r = self.widget.render('test', 'value', choices=self.choices)
+        expected = """<select name="test">
+<option value="1">One</option>
+<option value="2">Two</option>
+</select>"""
+        self.assertEqual(r, expected)
 
 
 class CommaSeparatedWidgetTestCase(TestCase):
@@ -22,3 +88,77 @@
     def test_render_when_value_is_comma_separated_list(self):
         r = self.widget.render('test', "1,2", choices=self.choices)
         self.assertEquals(r.count('selected='), 2)
+
+
+class StatusWidgetTestCase(TestCase):
+    def setUp(self):
+        self.choices = (('1', 'One'), ('2', 'Two'))
+        self.widget = StatusWidget()
+
+    def test_render_not_date_status_set(self):
+        r = self.widget.render('test', 'value', choices=self.choices)
+        expected = """<select name="test">
+<option value="1">One</option>
+<option value="2">Two</option>
+</select> """
+        self.assertEqual(r, expected)
+
+    def test_render_date_status_set(self):
+        self.widget.date_status_set = datetime(2010, 01, 01)
+        r = self.widget.render('test', 'value', choices=self.choices)
+        expected = """<select name="test">
+<option value="1">One</option>
+<option value="2">Two</option>
+</select> Set on 2010-01-01 00:00:00"""
+        self.assertEqual(r, expected)
+
+
+class ReadOnlyDateTimeWidgetTestCase(TestCase):
+    def setUp(self):
+        self.widget = ReadOnlyDateTimeWidget()
+
+    def test_render_string(self):
+        self.widget.value = 'value'
+        r = self.widget.render('test', None)
+        self.assertEqual(r, 'value')
+
+    def test_render_int(self):
+        self.widget.value = 4
+        r = self.widget.render('test', None)
+        self.assertEqual(r, '4')
+
+    def test_render_bool(self):
+        self.widget.value = False
+        r = self.widget.render('test', None)
+        self.assertEqual(r, 'False')
+
+    def test_render_datetime(self):
+        self.widget.value = datetime(2010, 01, 01)
+        r = self.widget.render('test', None)
+        self.assertEqual(r, '2010-01-01 00:00:00')
+
+
+class LPUsernameWidgetTestCase(SQLCachedTestCase):
+    pgsql_functions = ['generate_openid_identifier']
+    fixtures = ['test']
+
+    def setUp(self):
+        self.widget = LPUsernameWidget()
+        self.widget.account = Account.objects.get_by_email('[e-mail]')
+
+    def test_render_account(self):
+        r = self.widget.render('test', None)
+        self.assertEqual(r, '<a href="https://launchpad.net/~mark">mark</a>')
+
+    def test_render_no_account(self):
+        # unlink account
+        self.widget.account = None
+        r = self.widget.render('test', None)
+        self.assertEqual(r, '')
+
+    def test_render_account_no_person(self):
+        # unlink person
+        LPAccount.objects.filter(
+            openid_identifier=self.widget.account.openid_identifier).delete()
+        r = self.widget.render('test', None)
+        self.assertEqual(r, '')

=== added file 'identityprovider/tests/test_wsgi.py'
--- identityprovider/tests/test_wsgi.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_wsgi.py	2010-06-01 12:03:28 +0000
@@ -0,0 +1,43 @@
+from unittest import TestCase
+from identityprovider.wsgi import WSGIDispatch, make_app
+
+
+class StubWSGIApp(object):
+
+    def __init__(self):
+        self.called = False
+
+    def __call__(self, environ, start_response):
+        self.called = True
+        return []
+
+
+class WSGIDispatchTestCase(TestCase):
+
+    def setUp(self):
+        self.app_1 = StubWSGIApp()
+        self.app_2 = StubWSGIApp()
+        self.default_app = StubWSGIApp()
+        self.mapping = (('/a1', self.app_1), ('/a2', self.app_2))
+        self.dispatch = WSGIDispatch(self.mapping, self.default_app)
+
+    def call_dispatch_with_path(self, path):
+        self.dispatch({'PATH_INFO': path}, lambda x: x)
+
+    def test_call_is_routed_to_the_right_application(self):
+        self.call_dispatch_with_path('/a1')
+
+        self.assertTrue(self.app_1.called)
+
+    def test_call_is_routed_to_default_if_path_is_not_matching(self):
+        self.call_dispatch_with_path('/other')
+
+        self.assertTrue(self.default_app.called)
+
+
+class MakeAppTestCase(TestCase):
+
+    def test_make_sure_that_callable_is_returned(self):
+        app = make_app()
+
+        self.assertTrue(callable(app))

=== modified file 'identityprovider/tests/utils.py'
--- identityprovider/tests/utils.py	2010-05-13 13:30:24 +0000
+++ identityprovider/tests/utils.py	2010-06-01 12:03:28 +0000
@@ -4,6 +4,7 @@
 import logging
 import urllib
 import urllib2
+import sys
 from cStringIO import StringIO
 from types import MethodType
 from unittest import TestSuite, TextTestRunner, TestLoader
@@ -107,6 +108,18 @@
         settings.MIDDLEWARE_CLASSES = self.old_middlewares
 
 
+class OpenIDProviderTestCase(SQLCachedTestCase):
+    pgsql_functions = ['generate_openid_identifier']
+
+    def get_query(self, response):
+        query_start = response['Location'].find('?')
+        query_str = response['Location'][query_start+1:]
+        query_items = map(tuple,
+            [item.split('=') for item in query_str.split('&')])
+        query = dict(query_items)
+        return query
+
+
 def form_error_tester(cls, url, test_data):
     for test in test_data:
         response = cls.client.post(url, test['query'])
@@ -192,9 +205,19 @@
         return TestLoader().loadTestsFromModule(test)
 
 
+class NullDevice(object):
+    "A class that simulates writing to /dev/null."
+    def write(self, s):
+        pass
+
+
 # This function was adapted from the django project.
 # Please see the license file in the thirdparty/django directory.
 def run_tests(test_labels, verbosity=1, interactive=True, extra_tests=[]):
+    # disable stderr while running the tests
+    old_stderr = sys.stderr
+    sys.stderr = NullDevice()
+
     setup_test_environment()
     settings.DEBUG = False
     suite = TestSuite()
@@ -237,6 +260,9 @@
     settings.DATABASE_NAME = 'test_launchpad_dev'
     connection.creation.destroy_test_db(old_name, verbosity)
     teardown_test_environment()
+
+    # re-enable stderr
+    sys.stderr = old_stderr
     return len(result.failures) + len(result.errors)
 
 

=== modified file 'identityprovider/utils.py'
--- identityprovider/utils.py	2010-04-21 15:29:24 +0000
+++ identityprovider/utils.py	2010-06-01 12:03:28 +0000
@@ -100,7 +100,7 @@
 
 
 def get_person_and_account_by_email(email):
-    from identityprovider.models import EmailAddress, Person
+    from identityprovider.models import EmailAddress
     try:
         email = EmailAddress.objects.get(email__iexact=email)
         if email.account is None and email.person is None:
@@ -111,8 +111,6 @@
                 "to a team.") % email)
         else:
             return email.person, email.account
-    except Person.DoesNotExist:
-        return None, email.account
     except EmailAddress.DoesNotExist:
         raise PersonAndAccountNotFoundException()
 

=== modified file 'identityprovider/views/account.py'
--- identityprovider/views/account.py	2010-05-07 20:29:41 +0000
+++ identityprovider/views/account.py	2010-06-01 12:03:28 +0000
@@ -67,9 +67,6 @@
         email.status = EmailStatus.NEW
         email.save()
     account.status = AccountStatus.DEACTIVATED
-    if account.preferredemail is not None:
-        account.preferredemail.status = EmailStatus.NEW
-        account.preferredemail.save()
     account.save()
 
     # We don't want to lose session[token] when we log the user out

=== modified file 'identityprovider/views/consumer.py'
--- identityprovider/views/consumer.py	2010-05-10 18:26:52 +0000
+++ identityprovider/views/consumer.py	2010-06-01 12:03:28 +0000
@@ -251,6 +251,6 @@
     args = {'type_uris': [RP_RETURN_TO_URL_TYPE],
             'endpoint_uris': [getViewURL(request, finishOpenID)],
            }
-    response = direct_to_template(request, 'xrds.xml', args)
+    response = direct_to_template(request, 'openidapplication-xrds.xml', args)
     response['Content-Type'] = YADIS_CONTENT_TYPE
     return response

=== modified file 'identityprovider/views/server.py'
--- identityprovider/views/server.py	2010-04-21 15:29:24 +0000
+++ identityprovider/views/server.py	2010-06-01 12:03:28 +0000
@@ -129,7 +129,7 @@
 
 def _is_valid_openid_for_this_site(identity):
     try:
-        urinorm(identity)
+        identity = urinorm(identity)
         idparts = urlparse.urlparse(identity)
         srvparts = urlparse.urlparse(settings.SSO_ROOT_URL)
         if identity == IDENTIFIER_SELECT:
@@ -141,7 +141,7 @@
                 srvparts.hostname.endswith(".%s" % idparts.hostname)):
             return False
         accept_path_patterns = [
-            '/',
+            '^/$',
             '^/\+id/[a-zA-Z0-9\-_\.]+$',
             '^/~[a-zA-Z0-9\-_\.]+$',
         ]
@@ -383,8 +383,8 @@
         max_auth_age = int(pape_request.max_auth_age)
     except ValueError:
         logger.debug("pape:max_auth_age parameter should be an integer: %s" %
-                     max_auth_age)
-        raise HttpResponseBadRequest()
+                     pape_request.max_auth_age)
+        return False
 
     cutoff = datetime.utcnow() - timedelta(seconds=max_auth_age)
     logger.debug("%s" % user.last_login)

=== modified file 'identityprovider/views/ui.py'
--- identityprovider/views/ui.py	2010-05-13 16:01:26 +0000
+++ identityprovider/views/ui.py	2010-06-01 12:03:28 +0000
@@ -9,6 +9,7 @@
 
 from django.conf import settings
 from django.contrib.auth.decorators import login_required
+from django.core import urlresolvers
 from django.core.mail import send_mail
 from django.core.urlresolvers import Resolver404, resolve, reverse
 from django.http import Http404, HttpResponseRedirect, HttpResponseNotFound
@@ -81,7 +82,7 @@
     try:
         if url.find('?') >= 0:
             url = url[:url.find('?')]
-        response = resolve(url)
+        response = urlresolvers.resolve(url)
     except Resolver404:
         return False
     if response is None:
@@ -242,7 +243,7 @@
                         rpconfig = get_rpconfig(openid_request.trust_root)
                         if rpconfig is not None:
                             creation_rationale = rpconfig.creation_rationale
-                    except:
+                    except Exception, e:
                         pass
             Account.objects.create_account(
                 displayname, username, password, creation_rationale)
@@ -404,13 +405,7 @@
         form = ResetPasswordForm(request.POST)
         if form.is_valid():
             password = form.cleaned_data['password']
-            if not account.can_reset_password:
-                # the only possible status that can get us here is SUSPENDED,
-                # because we don't use NOACCOUNT
-                request.session['message'] = _(
-                    'You cannot reset the password of a suspended account.')
-                return HttpResponseRedirect('/+suspended')
-            elif not account.is_active and account.can_reactivate:
+            if not account.is_active and account.can_reactivate:
                 # if account can be reactivated, set the preferred email
                 # address
                 email_obj, created = EmailAddress.objects.get_or_create(

=== modified file 'identityprovider/views/utils.py'
--- identityprovider/views/utils.py	2010-04-21 15:29:24 +0000
+++ identityprovider/views/utils.py	2010-06-01 12:03:28 +0000
@@ -19,8 +19,5 @@
     else:
         alternatives.append(trust_root + '/')
 
-    try:
-        rpconfig = OpenIDRPConfig.objects.filter(trust_root__in=alternatives)
-        return rpconfig and rpconfig[0] or None
-    except OpenIDRPConfig.DoesNotExist:
-        return None
+    rpconfig = OpenIDRPConfig.objects.filter(trust_root__in=alternatives)
+    return rpconfig and rpconfig[0] or None

=== modified file 'identityprovider/widgets.py'
--- identityprovider/widgets.py	2010-04-21 15:29:24 +0000
+++ identityprovider/widgets.py	2010-06-01 12:03:28 +0000
@@ -2,9 +2,10 @@
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 from itertools import chain
-from django.forms.widgets import TextInput, Select, SelectMultiple
+from django.forms.widgets import TextInput, Select, SelectMultiple, Widget
 from django.conf import settings
 from django.utils.safestring import mark_safe
+from django.utils.translation import ugettext_lazy as _
 
 
 def read_only_markup(value):
@@ -13,6 +14,12 @@
     markup = '<span class="rofield">%s</span>' % value
     return mark_safe(markup)
 
+def account_to_lp_link(account):
+    if account and account.person:
+        return ('<a href="https://launchpad.net/~%s">%s</a>' %
+                    (account.person.name, account.person.name))
+    return ''
+
 
 class ROAwareTextInput(TextInput):
     def render(self, name, value, attrs=None):
@@ -47,3 +54,31 @@
             vals = value.split(',')
         return super(CommaSeparatedWidget, self).render(
             name, vals, attrs, choices)
+
+
+class StatusWidget(Select):
+    date_status_set = None
+
+    def render(self, name, value, attrs=None, choices=()):
+        select = super(StatusWidget, self).render(name, value, attrs, choices)
+        if self.date_status_set:
+            status_date = _("Set on %s") % self.date_status_set
+        else:
+            status_date = ""
+        return mark_safe("%s %s" % (select, status_date))
+
+
+class ReadOnlyDateTimeWidget(Widget):
+    value = None
+
+    def render(self, name, value, attrs=None):
+        return str(self.value)
+
+
+class LPUsernameWidget(Widget):
+    account = None
+
+    def render(self, name, value, attrs=None):
+        return mark_safe(account_to_lp_link(self.account))
+
+

=== modified file 'scripts/test'
--- scripts/test	2010-05-10 15:46:09 +0000
+++ scripts/test	2010-06-01 12:03:28 +0000
@@ -1,4 +1,4 @@
-#!/bin/sh
+#!/bin/bash
 
 # Copyright 2010 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
@@ -21,9 +21,11 @@
     echo "Probably you haven't activated right virtual environment."
     exit 1
 fi
-
+# Cleanup coverage only when running whole test suite
+cleanup="false"
 if [ ${#@} -eq 0 ]
 then
+    cleanup="true"
     unittest="true"
     doctest="true"
 fi
@@ -66,7 +68,18 @@
 fi
 
 # get rid of old coverage files to get a clean report
-find . -name '.coverage*' -exec rm -f {} \;
+if [ "$cleanup" = true ]
+then
+    find . -name '.coverage*' -exec rm -f {} \;
+fi
+
+function is_up() {
+    wget -q -O /dev/null "http://launchpad.dev/"
+}
+
+function wait_until_up() {
+    until is_up; do sleep 0.2; done
+}
 
 if [ "$doctest" = true ]
 then
@@ -77,6 +90,8 @@
     sudo $ENV `which python` wsgi_test_server.py &
     sso_pid=$!
 
+    wait_until_up
+
     (cd doctests && python runner.py)
 
     sudo kill $sso_pid
@@ -94,7 +109,7 @@
 
 if [ "$unittest" = true ]
 then
-    coverage run -p manage.py test $test_specification
+    coverage run -p manage.py test --noinput $test_specification
 fi
 
 # Creating nice (HTML) coverage report