# HG changeset patch -- Bitbucket.org # Project django-roa # URL http://bitbucket.org/david/django-roa/overview # User David Larlet # Date 1253004573 -7200 # Node ID b9938bcba5738c00e5338d0c8f091010df4f4562 # Parent 668d9e16c459dca82139b002c3be883d0519386a Missing files from piston's update --- /dev/null +++ b/piston/signals.py @@ -0,0 +1,14 @@ +# Django imports +import django.dispatch + +# Piston imports +from utils import send_consumer_mail + +def consumer_post_save(sender, instance, created, **kwargs): + send_consumer_mail(instance) + +def consumer_post_delete(sender, instance, **kwargs): + instance.status = 'canceled' + send_consumer_mail(instance) + + --- /dev/null +++ b/piston/fixtures/oauth.json @@ -0,0 +1,27 @@ +[ + { + "pk": 1, + "model": "piston.consumer", + "fields": { + "status": "accepted", + "name": "Piston Test Consumer", + "secret": "T5XkNMkcjffDpC9mNQJbyQnJXGsenYbz", + "user": 2, + "key": "8aZSFj3W54h8J8sCpx", + "description": "A test consumer record for Piston unit tests." + } + }, + { + "pk": 1, + "model": "piston.token", + "fields": { + "is_approved": true, + "timestamp": 1249347414, + "token_type": 2, + "secret": "qSWZq36t7yvkBquetYBkd8JxnuCu9jKk", + "user": 2, + "key": "Y7358vL5hDBbeP3HHL", + "consumer": 1 + } + } +] --- /dev/null +++ b/piston/tests.py @@ -0,0 +1,44 @@ +# Django imports +from django.core import mail +from django.contrib.auth.models import User +from django.conf import settings + +# Piston imports +from test import TestCase +from models import Consumer + +class ConsumerTest(TestCase): + fixtures = ['models.json'] + + def setUp(self): + self.consumer = Consumer() + self.consumer.name = "Piston Test Consumer" + self.consumer.description = "A test consumer for Piston." + self.consumer.user = User.objects.get(pk=3) + self.consumer.generate_random_codes() + + def test_create_pending(self): + """ Ensure creating a pending Consumer sends proper emails """ + # If it's pending we should have two messages in the outbox; one + # to the consumer and one to the site admins. + if len(settings.ADMINS): + self.assertEquals(len(mail.outbox), 2) + else: + self.assertEquals(len(mail.outbox), 1) + + expected = "Your API Consumer for example.com is awaiting approval." + self.assertEquals(mail.outbox[0].subject, expected) + + def test_delete_consumer(self): + """ Ensure deleting a Consumer sends a cancel email """ + + # Clear out the outbox before we test for the cancel email. + mail.outbox = [] + + # Delete the consumer, which should fire off the cancel email. + self.consumer.delete() + + self.assertEquals(len(mail.outbox), 1) + expected = "Your API Consumer for example.com has been canceled." + self.assertEquals(mail.outbox[0].subject, expected) + --- /dev/null +++ b/piston/fixtures/models.json @@ -0,0 +1,46 @@ +[ + { + "pk": 2, + "model": "auth.user", + "fields": { + "username": "pistontestuser", + "first_name": "Piston", + "last_name": "User", + "is_active": true, + "is_superuser": false, + "is_staff": false, + "last_login": "2009-08-03 13:11:53", + "groups": [], + "user_permissions": [], + "password": "sha1$b6c1f$83d5879f3854f6e9d27f393e3bcb4b8db05cf671", + "email": "pistontestuser@example.com", + "date_joined": "2009-08-03 13:11:53" + } + }, + { + "pk": 3, + "model": "auth.user", + "fields": { + "username": "pistontestconsumer", + "first_name": "Piston", + "last_name": "Consumer", + "is_active": true, + "is_superuser": false, + "is_staff": false, + "last_login": "2009-08-03 13:11:53", + "groups": [], + "user_permissions": [], + "password": "sha1$b6c1f$83d5879f3854f6e9d27f393e3bcb4b8db05cf671", + "email": "pistontestconsumer@example.com", + "date_joined": "2009-08-03 13:11:53" + } + }, + { + "pk": 1, + "model": "sites.site", + "fields": { + "domain": "example.com", + "name": "example.com" + } + } +] --- /dev/null +++ b/piston/test.py @@ -0,0 +1,62 @@ +# Django imports +import django.test.client as client +import django.test as test +from django.utils.http import urlencode + +# Piston imports +from piston import oauth +from piston.models import Consumer, Token + +# 3rd/Python party imports +import httplib2, urllib, cgi + +URLENCODED_FORM_CONTENT = 'application/x-www-form-urlencoded' + +class OAuthClient(client.Client): + def __init__(self, consumer, token): + self.token = oauth.OAuthToken(token.key, token.secret) + self.consumer = oauth.OAuthConsumer(consumer.key, consumer.secret) + self.signature = oauth.OAuthSignatureMethod_HMAC_SHA1() + + super(OAuthClient, self).__init__() + + def request(self, **request): + # Figure out parameters from request['QUERY_STRING'] and FakePayload + params = {} + if request['REQUEST_METHOD'] in ('POST', 'PUT'): + if request['CONTENT_TYPE'] == URLENCODED_FORM_CONTENT: + payload = request['wsgi.input'].read() + request['wsgi.input'] = client.FakePayload(payload) + params = cgi.parse_qs(payload) + + url = "http://testserver" + request['PATH_INFO'] + + req = oauth.OAuthRequest.from_consumer_and_token( + self.consumer, token=self.token, + http_method=request['REQUEST_METHOD'], http_url=url, + parameters=params + ) + + req.sign_request(self.signature, self.consumer, self.token) + headers = req.to_header() + request['HTTP_AUTHORIZATION'] = headers['Authorization'] + + return super(OAuthClient, self).request(**request) + + def post(self, path, data={}, content_type=None, follow=False, **extra): + if content_type is None: + content_type = URLENCODED_FORM_CONTENT + + if isinstance(data, dict): + data = urlencode(data) + + return super(OAuthClient, self).post(path, data, content_type, follow, **extra) + +class TestCase(test.TestCase): + pass + +class OAuthTestCase(TestCase): + @property + def oauth(self): + return OAuthClient(self.consumer, self.token) +