Merge lp:~canonical-isd-hackers/canonical-identity-provider/813619-fix-private-teams into lp:~canonical-isd-hackers/canonical-identity-provider/stable

[Anthony Lenton]

Two small api bugfixes

A branch with mostly tests, completing missing unit tests for our api, and two one-liner bug fixes.

[Ricardo Kirkner (ricardokirkner)]

lovely!

review: Approve

[ISD Branch Mangler (isd-branches-mangler)]

Attempt to merge into lp:~canonical-isd-hackers/canonical-identity-provider/2.x-stable failed due to conflicts:

text conflict in identityprovider/tests/test_handlers.py

=== modified file 'identityprovider/api10/handlers.py'
--- identityprovider/api10/handlers.py	2011-07-21 20:54:07 +0000
+++ identityprovider/api10/handlers.py	2011-07-22 16:46:33 +0000
@@ -371,7 +371,7 @@
             account = accounts[0]
             memberships = (
                 TeamMembership.get_team_memberships_for_user(
-                    data['team_names'], account, False))
+                    data['team_names'], account, include_private=True))
             return memberships
         else:
             return []
@@ -409,8 +409,8 @@
         team_names = request.data['team_names']
 
         memberships = (
-            TeamMembership.get_team_memberships(
-                team_names, request.user, True))
+            TeamMembership.get_team_memberships_for_user(
+                team_names, request.user, include_private=True))
         return memberships
 
     @named_operation

=== added file 'identityprovider/tests/factory.py'
--- identityprovider/tests/factory.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/factory.py	2011-07-22 16:46:33 +0000
@@ -0,0 +1,105 @@
+from itertools import count
+
+from identityprovider.models import (
+    Account,
+    APIUser,
+    EmailAddress,
+    LPOpenIdIdentifier,
+    Person,
+    TeamParticipation,
+    )
+from identityprovider.const import (
+    PERSON_VISIBILITY_PUBLIC,
+    PERSON_VISIBILITY_PRIVATE_MEMBERSHIP,
+)
+
+from identityprovider.models.const import EmailStatus
+from identityprovider.tests.utils import SQLCachedTestCase
+
+__all__ = ['TestCaseWithFactory']
+
+
+class SSOObjectFactory(object):
+    """A factory for creating model objects for tests."""
+
+    def __init__(self):
+        self.counter = count()
+
+    def get_unique_string(self, prefix=None):
+        if prefix is None:
+            prefix = "generic-string"
+        return prefix + str(self.get_unique_integer())
+
+    def get_unique_integer(self):
+        """Return a unique integer for the test run."""
+        return self.counter.next()
+
+    def make_apiuser(self, username=None, password='test'):
+        if username is None:
+            username = self.get_unique_string(prefix='apiuser-')
+        user, created = APIUser.objects.get_or_create(username=username)
+        user.set_password(password)
+        user.save()
+        return user
+
+    def make_account(self, displayname=None, username=None, email=None):
+        if displayname is None:
+            displayname = self.get_unique_string(prefix='Account ')
+        if username is None:
+            username = self.get_unique_string(prefix='account-')
+        account = Account.objects.create_account(displayname, username,
+            password='test')
+        self.make_email_for_account(account, email=email,
+            status=EmailStatus.PREFERRED)
+        return account
+
+    def make_person(self, name=None, displayname=None, account=None):
+        if name is None:
+            name = self.get_unique_string(prefix='person-')
+        if displayname is None:
+            displayname = self.get_unique_string(prefix='Person ')
+        lp_account = None
+        if account is not None:
+            lp_account = account.id
+
+        person = Person.objects.create(name=name, displayname=displayname,
+            lp_account=lp_account)
+        if lp_account is not None:
+            LPOpenIdIdentifier.objects.create(
+                identifier=account.openid_identifier,
+                lp_account=lp_account)
+        return person
+
+    def make_team(self, name=None, private=False, owner=None):
+        if private:
+            visibility = PERSON_VISIBILITY_PRIVATE_MEMBERSHIP
+        else:
+            visibility = PERSON_VISIBILITY_PUBLIC
+        if owner is None:
+            owner = self.make_person()
+        if name is None:
+            name = self.get_unique_string(prefix='team-')
+        displayname = self.get_unique_string(prefix='Team ')
+        team = self.make_person(name, displayname=displayname)
+        team.visibility = visibility
+        team.teamowner = owner
+        team.save()
+        return team
+
+    def add_account_to_team(self, account, team):
+        if account.person is None:
+            self.make_person(account=account)
+        elif account.person.in_team(team.name):
+            return
+        TeamParticipation.objects.create(person=account.person, team=team)
+
+    def make_email_for_account(self, account, email=None, status=None):
+        if email is None:
+            email = self.get_unique_string(prefix='email-') + '@example.com'
+        if status is None:
+            status = EmailStatus.VALIDATED
+        return EmailAddress.objects.create(email=email, account=account,
+            status=status)
+
+class TestCaseWithFactory(SQLCachedTestCase):
+    factory = SSOObjectFactory()

=== modified file 'identityprovider/tests/test_handlers.py'
--- identityprovider/tests/test_handlers.py	2011-07-21 20:54:07 +0000
+++ identityprovider/tests/test_handlers.py	2011-07-22 16:46:33 +0000
@@ -1,9 +1,19 @@
 import base64
-
+import json
 from django.utils import simplejson
 from mock import patch
+from oauth.oauth import (
+    OAuthRequest,
+    OAuthConsumer,
+    OAuthToken,
+    OAuthSignatureMethod_PLAINTEXT,
+    )
 
-from identityprovider.tests.utils import SQLCachedTestCase
+from identityprovider.api10.handlers import (
+    AuthenticationHandler,
+    CanNotResetPasswordError,
+    RegistrationHandler,
+)
 from identityprovider.models import (
     Account,
     APIUser,
@@ -16,11 +26,8 @@
     EmailStatus,
     LoginTokenType,
 )
-from identityprovider.api10.handlers import (
-    AuthenticationHandler,
-    CanNotResetPasswordError,
-    RegistrationHandler,
-)
+from identityprovider.tests.utils import SQLCachedTestCase
+from identityprovider.tests.factory import TestCaseWithFactory
 
 
 class MockRequest(object):
@@ -31,6 +38,7 @@
         self.data = data
         self.environ = {'REMOTE_ADDR': '127.0.0.1'}
 
+
 class RegistrationHandlerTestCase(SQLCachedTestCase):
 
     fixtures = ["test"]
@@ -241,19 +249,21 @@
             email.account, '[e-mail]', None)
 
 
-class AuthenticationTestCase(SQLCachedTestCase):
-
-    fixtures = ["test"]
+class AuthenticationTestCase(TestCaseWithFactory):
+    pgsql_functions = ["generate_openid_identifier"]
 
     def setUp(self):
         self.authentication = AuthenticationHandler()
 
     def test_account_by_openid_with_valid_openid(self):
-        request = MockRequest(user=APIUser(), data={'openid': "mark_oid"})
+        account = self.factory.make_account()
+        openid = account.openid_identifier
+
+        request = MockRequest(user=APIUser(), data={'openid': openid})
 
         account = self.authentication.account_by_openid(request)
 
-        self.assertEquals(account['openid_identifier'], "mark_oid")
+        self.assertEquals(account['openid_identifier'], openid)
 
     def test_account_by_openid_with_invalid_openid(self):
         request = MockRequest(user=APIUser(), data={'openid': "bad-openid"})
@@ -263,10 +273,12 @@
         self.assertTrue(account is None)
 
     def test_authenticate_unquoted_token(self):
+        account = self.factory.make_account(email='[e-mail]')
         expected_keys = ['consumer_key', 'consumer_secret', 'name', 'token',
             'token_secret']
 
-        extra = {'HTTP_AUTHORIZATION': 'basic ' + base64.b64encode('[e-mail]:test')}
+        extra = {'HTTP_AUTHORIZATION': 'basic ' +
+            base64.b64encode('[e-mail]:test')}
         response = self.client.get('/api/1.0/authentications', data={
             'ws.op': 'authenticate', 'token_name': 'some-token'}, **extra)
 
@@ -275,13 +287,159 @@
         self.assertEqual(content['name'], 'some-token')
 
     def test_authenticate_quoted_token(self):
+        account = self.factory.make_account(email='[e-mail]')
         expected_keys = ['consumer_key', 'consumer_secret', 'name', 'token',
             'token_secret']
 
-        extra = {'HTTP_AUTHORIZATION': 'basic ' + base64.b64encode('[e-mail]:test')}
+        extra = {'HTTP_AUTHORIZATION': 'basic ' +
+            base64.b64encode('[e-mail]:test')}
         response = self.client.get('/api/1.0/authentications', data={
             'ws.op': 'authenticate', 'token_name': '"some-token"'}, **extra)
 
+
         content = simplejson.loads(response.content)
         self.assertEqual(set(content.keys()), set(expected_keys))
         self.assertEqual(content['name'], 'some-token')
+
+    def test_team_memberships_public_teams(self):
+        team = self.factory.make_team()
+        otherteam = self.factory.make_team()
+        apiuser = self.factory.make_apiuser(username='foobar')
+        account = self.factory.make_account()
+        self.factory.add_account_to_team(account, team)
+
+        # Request membership via API
+        extra = {'HTTP_AUTHORIZATION': 'basic ' +
+            base64.b64encode('foobar:test')}
+        response = self.client.get('/api/1.0/authentications', data={
+            'ws.op': 'team_memberships',
+            'team_names': '["%s", "%s"]' % (team.name, otherteam.name),
+            'openid_identifier': account.openid_identifier,
+            }, **extra)
+        self.assertContains(response, team.name)
+        self.assertNotContains(response, otherteam.name)
+
+    def test_team_membership_private_teams(self):
+        team = self.factory.make_team(private=True)
+        apiuser = self.factory.make_apiuser(username='foobar')
+        account = self.factory.make_account()
+        self.factory.add_account_to_team(account, team)
+
+        # Request membership via API
+        extra = {'HTTP_AUTHORIZATION': 'basic ' +
+            base64.b64encode('foobar:test')}
+        response = self.client.get('/api/1.0/authentications', data={
+            'ws.op': 'team_memberships',
+            'team_names': '["%s"]' % team.name,
+            'openid_identifier': account.openid_identifier,
+            }, **extra)
+        self.assertContains(response, team.name)
+
+    def test_team_membership_basic_auth_protected(self):
+        response = self.client.get('/api/1.0/authentications', data={
+            'ws.op': 'team_memberships',
+            'team_names': '["something"]',
+            'openid_identifier': 'test',
+            })
+        self.assertEqual(401, response.status_code)
+
+    def test_team_membership_fails_for_regular_user(self):
+        account = self.factory.make_account(email='[e-mail]')
+        extra = {'HTTP_AUTHORIZATION': 'basic ' +
+            base64.b64encode('[e-mail]:test')}
+        response = self.client.get('/api/1.0/authentications', data={
+            'ws.op': 'team_memberships',
+            'team_names': '["something"]',
+            'openid_identifier': 'test',
+            }, **extra)
+        self.assertEqual(403, response.status_code)
+
+
+class AccountsTestCase(TestCaseWithFactory):
+    pgsql_functions = ["generate_openid_identifier"]
+
+    def auth_header(self, token, url, realm='OAuth'):
+        oaconsumer = OAuthConsumer(token.consumer.key, token.consumer.secret)
+        oatoken = OAuthToken(token.token, token.token_secret)
+        oarequest = OAuthRequest.from_consumer_and_token(
+            oaconsumer, oatoken, http_url=url)
+        oarequest.sign_request(OAuthSignatureMethod_PLAINTEXT(),
+            oaconsumer, oatoken)
+        header = oarequest.to_header(realm)
+        return {'HTTP_AUTHORIZATION': header['Authorization']}
+
+
+    def test_me_is_oauth_protected(self):
+        extra = {'HTTP_AUTHORIZATION': 'basic ' +
+            base64.b64encode('[e-mail]:test')}
+        response = self.client.get('/api/1.0/accounts',
+            data={'ws.op': 'me'}, **extra)
+        self.assertEqual(401, response.status_code)
+
+    def test_me_success(self):
+        account = self.factory.make_account()
+        token = account.create_oauth_token('token-name')
+        url = '/api/1.0/accounts'
+        
+        response = self.client.get(url,
+            data={'ws.op': 'me'}, **self.auth_header(token, url))
+
+        data = json.loads(response.content)
+        expected = set(['username', 'preferred_email', 'displayname',
+            'unverified_emails', 'verified_emails', 'openid_identifier'])
+        self.assertEqual(expected, set(data.keys()))
+
+    def test_team_membership_public_teams(self):
+        account = self.factory.make_account()
+        team = self.factory.make_team()
+        otherteam = self.factory.make_team()
+        self.factory.add_account_to_team(account, team)
+        token = account.create_oauth_token('token-name')
+        url = '/api/1.0/accounts'
+
+        response = self.client.get(url, data={'ws.op': 'team_memberships',
+            'team_names': '["%s", "%s"]' % (team.name, otherteam.name)},
+            **self.auth_header(token, url))
+
+        self.assertContains(response, team.name)
+        self.assertNotContains(response, otherteam.name)
+
+    def test_team_membership_private_teams(self):
+        account = self.factory.make_account()
+        team = self.factory.make_team(private=True)
+        self.factory.add_account_to_team(account, team)
+        token = account.create_oauth_token('token-name')
+        url = '/api/1.0/accounts'
+
+        response = self.client.get(url, data={'ws.op': 'team_memberships',
+            'team_names': '["%s"]' % team.name},
+            **self.auth_header(token, url))
+
+        self.assertContains(response, team.name)
+
+    def test_validate_email_bad_token(self):
+        account = self.factory.make_account()
+        token = account.create_oauth_token('token-name')
+        url = '/api/1.0/accounts'
+
+        response = self.client.get(url, data={'ws.op': 'validate_email',
+            'email_token': 'ABCD'}, **self.auth_header(token, url))
+
+        self.assertContains(response, 'Bad email token!')
+
+    def test_validate_email_success(self):
+        account = self.factory.make_account()
+        email = self.factory.make_email_for_account(account,
+            status=EmailStatus.NEW)
+        authtoken = AuthTokenFactory().new_api_email_validation_token(
+                account, email.email)
+
+        token = account.create_oauth_token('token-name')
+        url = '/api/1.0/accounts'
+
+        response = self.client.get(url, data={'ws.op': 'validate_email',
+            'email_token': authtoken.token}, **self.auth_header(token, url))
+
+        self.assertContains(response, email.email)
+        updatedemail = EmailAddress.objects.get(email=email.email)
+        self.assertEqual(EmailStatus.VALIDATED, updatedemail.status)

=== modified file 'identityprovider/tests/test_views_preflight.py'
--- identityprovider/tests/test_views_preflight.py	2011-03-04 17:09:45 +0000
+++ identityprovider/tests/test_views_preflight.py	2011-07-22 16:46:33 +0000
@@ -13,12 +13,13 @@
         r = self.client.get(self.URL)
         self.assertEquals(r.status_code, 404)
 
-    def test_account_withtout_person_receives_404(self):
+    def test_account_without_person_receives_404(self):
         account = Account.objects.create_account(
             "test", "[e-mail]", "test")
         self.client.login(username="[e-mail]", password="test")
         r = self.client.get(self.URL)
         self.assertEquals(r.status_code, 404)
+        account.emailaddress_set.all().delete()
         account.delete()
 
     def test_person_not_in_the_required_team_will_get_404(self):