Revision be67d92e snf-astakos-app/astakos/im/tests/api.py
| b/snf-astakos-app/astakos/im/tests/api.py | ||
|---|---|---|
| 33 | 33 |
|
| 34 | 34 |
from astakos.im.tests.common import * |
| 35 | 35 |
from astakos.im.settings import astakos_services, BASE_HOST |
| 36 |
from astakos.oa2.backends import DjangoBackend |
|
| 37 |
|
|
| 36 | 38 |
from synnefo.lib.services import get_service_path |
| 37 | 39 |
from synnefo.lib import join_urls |
| 38 | 40 |
|
| ... | ... | |
| 43 | 45 |
#from xml.dom import minidom |
| 44 | 46 |
|
| 45 | 47 |
import json |
| 48 |
import time |
|
| 46 | 49 |
|
| 47 | 50 |
ROOT = "/%s/%s/%s/" % ( |
| 48 | 51 |
astakos_settings.BASE_PATH, astakos_settings.ACCOUNTS_PREFIX, 'v1.0') |
| ... | ... | |
| 446 | 449 |
e3.data.create(key='versionId', value='v2.0') |
| 447 | 450 |
e3.data.create(key='publicURL', value='http://localhost:8000/s3/v2.0') |
| 448 | 451 |
|
| 452 |
oa2_backend = DjangoBackend() |
|
| 453 |
self.token = oa2_backend.token_model.create( |
|
| 454 |
code='12345', |
|
| 455 |
expires_at=datetime.now() + timedelta(seconds=5), |
|
| 456 |
user=self.user1, |
|
| 457 |
client=oa2_backend.client_model.create(type='public'), |
|
| 458 |
redirect_uri='https://server.com/handle_code') |
|
| 459 |
|
|
| 449 | 460 |
def test_authenticate(self): |
| 450 | 461 |
client = Client() |
| 451 | 462 |
url = reverse('astakos.api.tokens.authenticate')
|
| ... | ... | |
| 570 | 581 |
r = client.post(url, post_data, content_type='application/json') |
| 571 | 582 |
self.assertEqual(r.status_code, 200) |
| 572 | 583 |
|
| 573 |
# Check successful json response |
|
| 584 |
# Check successful json response: user credential auth
|
|
| 574 | 585 |
post_data = """{"auth":{"passwordCredentials":{"username":"%s",
|
| 575 | 586 |
"password":"%s"}, |
| 576 | 587 |
"tenantName":"%s"}}""" % ( |
| ... | ... | |
| 594 | 605 |
self.assertEqual(user, self.user1.uuid) |
| 595 | 606 |
self.assertEqual(len(service_catalog), 3) |
| 596 | 607 |
|
| 608 |
# Check successful json response: token auth |
|
| 609 |
post_data = """{"auth":{"token":{"id":"%s"},
|
|
| 610 |
"tenantName":"%s"}}""" % ( |
|
| 611 |
self.user1.auth_token, self.user1.uuid) |
|
| 612 |
r = client.post(url, post_data, content_type='application/json') |
|
| 613 |
self.assertEqual(r.status_code, 200) |
|
| 614 |
self.assertTrue(r['Content-Type'].startswith('application/json'))
|
|
| 615 |
try: |
|
| 616 |
body = json.loads(r.content) |
|
| 617 |
except Exception, e: |
|
| 618 |
self.fail(e) |
|
| 619 |
|
|
| 620 |
try: |
|
| 621 |
token = body['access']['token']['id'] |
|
| 622 |
user = body['access']['user']['id'] |
|
| 623 |
service_catalog = body['access']['serviceCatalog'] |
|
| 624 |
except KeyError: |
|
| 625 |
self.fail('Invalid response')
|
|
| 626 |
|
|
| 627 |
self.assertEqual(token, self.user1.auth_token) |
|
| 628 |
self.assertEqual(user, self.user1.uuid) |
|
| 629 |
self.assertEqual(len(service_catalog), 3) |
|
| 630 |
|
|
| 597 | 631 |
# Check successful xml response |
| 598 | 632 |
headers = {'HTTP_ACCEPT': 'application/xml'}
|
| 599 | 633 |
post_data = """{"auth":{"passwordCredentials":{"username":"%s",
|
| ... | ... | |
| 609 | 643 |
# except Exception, e: |
| 610 | 644 |
# self.fail(e) |
| 611 | 645 |
|
| 646 |
# oath access token authorization |
|
| 647 |
post_data = """{"auth":{"token":{"id":"%s"},
|
|
| 648 |
"tenantName":"%s"}}""" % ( |
|
| 649 |
self.token.code, self.user1.uuid) |
|
| 650 |
r = client.post(url, post_data, content_type='application/json') |
|
| 651 |
self.assertEqual(r.status_code, 401) |
|
| 652 |
|
|
| 612 | 653 |
|
| 613 | 654 |
class UserCatalogsTest(TestCase): |
| 614 | 655 |
def test_get_uuid_displayname_catalogs(self): |
| ... | ... | |
| 707 | 748 |
json.loads(response.content) |
| 708 | 749 |
except ValueError: |
| 709 | 750 |
self.assertTrue(False) |
| 751 |
|
|
| 752 |
|
|
| 753 |
class ValidateAccessToken(TestCase): |
|
| 754 |
def setUp(self): |
|
| 755 |
self.oa2_backend = DjangoBackend() |
|
| 756 |
self.user = AstakosUser.objects.create(username="user@synnefo.org") |
|
| 757 |
self.token = self.oa2_backend.token_model.create( |
|
| 758 |
code='12345', |
|
| 759 |
expires_at=datetime.now() + timedelta(seconds=5), |
|
| 760 |
user=self.user, |
|
| 761 |
client=self.oa2_backend.client_model.create(type='public'), |
|
| 762 |
redirect_uri='https://server.com/handle_code') |
|
| 763 |
|
|
| 764 |
def test_validate_token(self): |
|
| 765 |
# invalid token |
|
| 766 |
url = reverse('astakos.api.tokens.validate_token',
|
|
| 767 |
kwargs={'token_id': 'invalid'})
|
|
| 768 |
r = self.client.get(url) |
|
| 769 |
self.assertEqual(r.status_code, 404) |
|
| 770 |
|
|
| 771 |
# valid token |
|
| 772 |
url = reverse('astakos.api.tokens.validate_token',
|
|
| 773 |
kwargs={'token_id': self.token.code})
|
|
| 774 |
|
|
| 775 |
r = self.client.head(url) |
|
| 776 |
self.assertEqual(r.status_code, 400) |
|
| 777 |
r = self.client.put(url) |
|
| 778 |
self.assertEqual(r.status_code, 400) |
|
| 779 |
r = self.client.post(url) |
|
| 780 |
self.assertEqual(r.status_code, 400) |
|
| 781 |
|
|
| 782 |
r = self.client.get(url) |
|
| 783 |
self.assertEqual(r.status_code, 200) |
|
| 784 |
self.assertTrue(r['Content-Type'].startswith('application/json'))
|
|
| 785 |
try: |
|
| 786 |
body = json.loads(r.content) |
|
| 787 |
user = body['access']['user']['id'] |
|
| 788 |
self.assertEqual(user, self.user.uuid) |
|
| 789 |
except Exception: |
|
| 790 |
self.fail('Unexpected response content')
|
|
| 791 |
|
|
| 792 |
# expired token |
|
| 793 |
sleep_time = (self.token.expires_at - datetime.now()).total_seconds() |
|
| 794 |
time.sleep(sleep_time) |
|
| 795 |
r = self.client.get(url) |
|
| 796 |
self.assertEqual(r.status_code, 404) |
|
| 797 |
# assert expired token has been deleted |
|
| 798 |
self.assertEqual(self.oa2_backend.token_model.count(), 0) |
|
| 799 |
|
|
| 800 |
# user authentication token |
|
| 801 |
url = reverse('astakos.api.tokens.validate_token',
|
|
| 802 |
kwargs={'token_id': self.user.auth_token})
|
|
| 803 |
self.assertEqual(r.status_code, 404) |
|
Also available in: Unified diff