Merge lp:~ricardokirkner/ols-store-tests/more-acceptance-tests-1 into lp:~ubuntuone-pqm-team/ols-store-tests/store-acceptance-tests

[Ricardo Kirkner]

added more smoke tests

added tests for:

- search
- snaps
- recommendations

[Matias Bordese (matiasb)]

LGTM, with a suggestion.

review: Approve

[lp:~ricardokirkner/ols-store-tests/more-acceptance-tests-1]

30. By Ricardo Kirkner on 2016-07-07

decouple click from snap search endpoints

=== modified file 'tests/api/cpi/helpers.py'
--- tests/api/cpi/helpers.py	2016-07-07 15:41:03 +0000
+++ tests/api/cpi/helpers.py	2016-07-07 19:50:23 +0000
@@ -1,7 +1,10 @@
 import getpass
+import json
 import os
 import unittest
 
+import requests
+import requests_oauthlib
 
 CPI_ROOT_URL = os.getenv(
     'CPI_ROOT_URL', 'https://search.apps.staging.ubuntu.com')
@@ -26,6 +29,39 @@
 PRIVATE_SNAP_PACKAGE_REVISION = 1
 
 
+def get_oauth_token_data(email, password):
+    url = '{}/api/v2/tokens/oauth'.format(SSO_ROOT_URL)
+    data = {
+        'email': email,
+        'password': password,
+        'token_name': 'store-acceptance-tests',
+    }
+    headers = {
+        'Content-Type': 'application/json',
+        'Accept': 'application/json',
+    }
+    response = requests.post(url, data=json.dumps(data), headers=headers)
+    assert response.ok, 'Failed to get oauth token: %r' % response.content
+
+    data = response.json()
+    return {
+        'client_key': data['consumer_key'],
+        'client_secret': data['consumer_secret'],
+        'resource_owner_key': data['token_key'],
+        'resource_owner_secret': data['token_secret'],
+    }
+
+
+def get_oauth_client():
+    credentials = get_oauth_token_data(TEST_USER_EMAIL, TEST_USER_PASSWORD)
+    client = requests_oauthlib.OAuth1Session(**credentials)
+    client.headers.update({
+        'Content-Type': 'application/json',
+        'Cache-Control': 'no-cache',
+    })
+    return client
+
+
 class APITestCase(unittest.TestCase):
 
     def assert_success(self, response, content_type='application/hal+json'):
@@ -50,3 +86,8 @@
             'result': 'error',
             'errors': [message],
         })
+
+    def assert_unauthorized(self, response):
+        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.headers['Content-Type'], 'text/plain')
+        self.assertIn('WWW-Authenticate', response.headers)

=== modified file 'tests/api/cpi/test_metadata.py'
--- tests/api/cpi/test_metadata.py	2016-07-07 18:53:00 +0000
+++ tests/api/cpi/test_metadata.py	2016-07-07 19:50:23 +0000
@@ -85,19 +85,20 @@
             response, message='No JSON object could be decoded')
 
 
-def get_metadata(snaps=None, fields=None):
-    url = '{}/api/v1/metadata'.format(CPI_ROOT_URL)
-    data = {}
-    if snaps is not None:
-        data['snaps'] = snaps
-    if fields is not None:
-        data['fields'] = fields
-    response = requests.post(url, data=json.dumps(data),
-                             headers={'Content-Type': 'application/json'})
-    return response
-
-
 class MetadataTestCase(APITestCase):
+    endpoint = '/api/v1/metadata'
+
+    def get_metadata(self, snaps=None, fields=None):
+        url = '{}{}'.format(CPI_ROOT_URL, self.endpoint)
+        data = {}
+        if snaps is not None:
+            data['snaps'] = snaps
+        if fields is not None:
+            data['fields'] = fields
+        response = requests.post(
+            url, data=json.dumps(data),
+            headers={'Content-Type': 'application/json'})
+        return response
 
     def assert_metadata(self, response, fields=None, snap_id=None):
         body = response.json()
@@ -144,25 +145,25 @@
         self._assert_package_data(body, fields=[])
 
     def test_get_metadata(self):
-        response = get_metadata(
+        response = self.get_metadata(
             snaps=[{'snap_id': SNAP_PACKAGE_SNAP_ID}])
         self.assert_success(response)
         self.assert_metadata(response, snap_id=SNAP_PACKAGE_SNAP_ID)
 
     def test_get_metadata_no_snaps(self):
-        response = get_metadata()
+        response = self.get_metadata()
         self.assert_bad_request(
             response,
             message='Packages must be specified in parameter "snaps".')
 
     def test_get_metadata_missing_snap_id(self):
-        response = get_metadata(snaps=[{}])
+        response = self.get_metadata(snaps=[{}])
         self.assert_bad_request(
             response, message='Missing snap_id for snap.')
 
     def test_get_metadata_filter_fields(self):
         fields = ['channel', 'revision', 'snap_id']
-        response = get_metadata(
+        response = self.get_metadata(
             snaps=[{'snap_id': SNAP_PACKAGE_SNAP_ID}],
             fields=fields)
         self.assert_success(response)
@@ -170,7 +171,7 @@
             response, fields=fields, snap_id=SNAP_PACKAGE_SNAP_ID)
 
     def test_get_metadata_invalid_package(self):
-        response = get_metadata(
+        response = self.get_metadata(
             snaps=[{'snap_id': 'invalid'}])
         self.assert_success(response)
         self.assert_no_results(response)

=== modified file 'tests/api/cpi/test_package.py'
--- tests/api/cpi/test_package.py	2016-07-07 15:41:03 +0000
+++ tests/api/cpi/test_package.py	2016-07-07 19:50:23 +0000
@@ -1,7 +1,4 @@
-import json
-
 import requests
-import requests_oauthlib
 
 from .helpers import (
     CPI_ROOT_URL,
@@ -16,6 +13,7 @@
     TEST_USER_EMAIL,
     TEST_USER_PASSWORD,
     APITestCase,
+    get_oauth_client,
 )
 
 
@@ -35,39 +33,6 @@
     return response
 
 
-def get_oauth_token_data(email, password):
-    url = '{}/api/v2/tokens/oauth'.format(SSO_ROOT_URL)
-    data = {
-        'email': email,
-        'password': password,
-        'token_name': 'store-acceptance-tests',
-    }
-    headers = {
-        'Content-Type': 'application/json',
-        'Accept': 'application/json',
-    }
-    response = requests.post(url, data=json.dumps(data), headers=headers)
-    assert response.ok, 'Failed to get oauth token: %r' % response.content
-
-    data = response.json()
-    return {
-        'client_key': data['consumer_key'],
-        'client_secret': data['consumer_secret'],
-        'resource_owner_key': data['token_key'],
-        'resource_owner_secret': data['token_secret'],
-    }
-
-
-def get_oauth_client():
-    credentials = get_oauth_token_data(TEST_USER_EMAIL, TEST_USER_PASSWORD)
-    client = requests_oauthlib.OAuth1Session(**credentials)
-    client.headers.update({
-        'Content-Type': 'application/json',
-        'Cache-Control': 'no-cache',
-    })
-    return client
-
-
 class PackageBaseTestCase(APITestCase):
 
     def assert_package(self, response, snap_id=None, name=None,

=== added file 'tests/api/cpi/test_recommendation.py'
--- tests/api/cpi/test_recommendation.py	1970-01-01 00:00:00 +0000
+++ tests/api/cpi/test_recommendation.py	2016-07-07 19:50:23 +0000
@@ -0,0 +1,35 @@
+import requests
+
+from .helpers import (
+    CPI_ROOT_URL,
+    APITestCase,
+)
+
+
+def get_recommendation(slug):
+    url = '{}/api/v1/recommendations/{}'.format(CPI_ROOT_URL, slug)
+    response = requests.get(url)
+    return response
+
+
+class RecommendationsTestCase(APITestCase):
+
+    def assert_recommendation(self, response, slug=None):
+        body = response.json()
+        # assert result data structure
+        self.assertIn('applications', body)
+        self.assertGreater(len(body['applications']), 0)
+        self.assertIn('hotwords', body)
+        self.assertIn('slug', body)
+        if slug is not None:
+            self.assertEqual(body['slug'], slug)
+
+    def test_get_recommendation(self):
+        response = get_recommendation('hello')
+        self.assert_success(response)
+        self.assert_recommendation(response, 'hello')
+
+    def test_get_recommendation_not_found(self):
+        response = get_recommendation('missing')
+        self.assert_not_found(
+            response, message="No such recommendation u'missing'")

=== modified file 'tests/api/cpi/test_search.py'
--- tests/api/cpi/test_search.py	2016-06-28 19:20:09 +0000
+++ tests/api/cpi/test_search.py	2016-07-07 19:50:23 +0000
@@ -1,24 +1,50 @@
+from urllib.parse import urlencode
+
 import requests
 
 from .helpers import (
     CPI_ROOT_URL,
+    PRIVATE_SNAP_PACKAGE_NAME,
     APITestCase,
+    get_oauth_client,
 )
 
 
-def search(name=''):
-    url = '{}/api/v1/search?q=name:{}'.format(CPI_ROOT_URL, name)
-    response = requests.get(url)
-    return response
-
-
 class SearchTestCase(APITestCase):
+    endpoint = '/api/v1/search'
 
     def setUp(self):
         super(SearchTestCase, self).setUp()
         self.name = 'hello'
 
-    def assert_body_success(self, response, name):
+    def search(self, name=None, fields=None, auth=False):
+        query_data = {}
+        if name is not None:
+            query_data['name'] = name
+
+        query = {}
+        if query_data:
+            query['q'] = '&'.join(
+                '{}:{}'.format(key, value)
+                for key, value in query_data.items())
+        if fields:
+            query['fields'] = fields
+
+        return self._search(query=query, auth=auth)
+
+    def _search(self, query=None, auth=False):
+        url = '{}{}'.format(CPI_ROOT_URL, self.endpoint)
+        if query:
+            qs = urlencode(query)
+            url += '?' + qs
+        if auth:
+            with get_oauth_client() as client:
+                response = client.get(url)
+        else:
+            response = requests.get(url)
+        return response
+
+    def assert_results(self, response, name=None):
         body = response.json()
         # assert generic embedded data structure
         self.assertIn('_embedded', body)
@@ -27,9 +53,40 @@
         self.assertGreater(len(body['_embedded']['clickindex:package']), 0)
         # assert result data structure
         item = body['_embedded']['clickindex:package'][0]
-        self.assertIn(name, item['name'])
+        if name is not None:
+            self.assertIn(name, item['name'])
+
+    def assert_no_results(self, response, name):
+        body = response.json()
+        embedded = body.get('_embedded', {})
+        results = embedded.get('clickindex:package', [])
+        for item in results:
+            self.assertNotIn(name, item['name'])
+
+    def assert_recommendations(self, response):
+        body = response.json()
+        self.assertIn('_embedded', body)
+        self.assertIn('clickindex:recommendation', body['_embedded'])
+        # assert there is at least one result
+        self.assertGreater(
+            len(body['_embedded']['clickindex:recommendation']), 0)
 
     def test_search_by_name(self):
-        response = search(name=self.name)
-        self.assert_success(response)
-        self.assert_body_success(response, self.name)
+        response = self.search(name=self.name)
+        self.assert_success(response)
+        self.assert_results(response, self.name)
+
+    def test_search_all(self):
+        response = self.search()
+        self.assert_success(response)
+        self.assert_results(response)
+
+    def test_search_private_not_authorized(self):
+        response = self.search(name=PRIVATE_SNAP_PACKAGE_NAME)
+        self.assert_success(response)
+        self.assert_no_results(response, name=PRIVATE_SNAP_PACKAGE_NAME)
+
+    def test_search_private(self):
+        response = self.search(name=PRIVATE_SNAP_PACKAGE_NAME, auth=True)
+        self.assert_success(response)
+        self.assert_results(response, name=PRIVATE_SNAP_PACKAGE_NAME)

=== added file 'tests/api/cpi/test_snaps.py'
--- tests/api/cpi/test_snaps.py	1970-01-01 00:00:00 +0000
+++ tests/api/cpi/test_snaps.py	2016-07-07 19:50:23 +0000
@@ -0,0 +1,141 @@
+import requests
+
+from .helpers import (
+    CPI_ROOT_URL,
+    SNAP_PACKAGE_NAME,
+    PRIVATE_SNAP_PACKAGE_NAME,
+    APITestCase,
+    get_oauth_client,
+)
+from .test_metadata import MetadataTestCase
+from .test_search import SearchTestCase
+
+
+def get_snap_detail(name, series='16', auth=False):
+    url = '{}/api/v1/snaps/details/{}'.format(CPI_ROOT_URL, name)
+    headers = {}
+    if series:
+        headers['X-Ubuntu-Series'] = series
+    if auth:
+        with get_oauth_client() as client:
+            response = client.get(url, headers=headers)
+    else:
+        response = requests.get(url, headers=headers)
+    return response
+
+
+class SnapDetailTestCase(APITestCase):
+
+    def assert_snap_detail(self, response, name):
+        body = response.json()
+        fields = [
+            'alias',
+            'allow_unauthenticated',
+            'anon_download_url',
+            'architecture',
+            'binary_filesize',
+            'blacklist_country_codes',
+            'changelog',
+            'channel',
+            'channels',
+            'click_framework',
+            'click_version',
+            'company_name',
+            'confinement',
+            'content',
+            'date_published',
+            'department',
+            'description',
+            'developer_id',
+            'developer_name',
+            'download_sha512',
+            'download_url',
+            'epoch',
+            'framework',
+            'icon_url',
+            'icon_urls',
+            'id',
+            'is_published',
+            'keywords',
+            'last_updated',
+            'license',
+            'name',
+            'origin',
+            'package_name',
+            'plugs',
+            'price',
+            'prices',
+            'promoted',
+            'publisher',
+            'ratings_average',
+            'release',
+            'revision',
+            'screenshot_url',
+            'screenshot_urls',
+            'slots',
+            'snap_id',
+            'status',
+            'summary',
+            'support_url',
+            'terms_of_service',
+            'title',
+            'version',
+            'video_urls',
+            'website',
+            'whitelist_country_codes',
+        ]
+        for field in fields:
+            self.assertIn(field, body)
+        self.assertEqual(body['package_name'], name)
+
+    def test_get_snap_detail(self):
+        response = get_snap_detail(SNAP_PACKAGE_NAME)
+        self.assert_success(response)
+        self.assert_snap_detail(response, SNAP_PACKAGE_NAME)
+
+    def test_get_snap_detail_for_private_package(self):
+        response = get_snap_detail(PRIVATE_SNAP_PACKAGE_NAME, auth=True)
+        self.assert_success(response)
+        self.assert_snap_detail(response, PRIVATE_SNAP_PACKAGE_NAME)
+
+    def test_get_snap_detail_without_series(self):
+        response = get_snap_detail(SNAP_PACKAGE_NAME, series=None)
+        self.assert_bad_request(
+            response, message='X-Ubuntu-Series header is required.')
+
+    def test_get_snap_detail_unauthorized(self):
+        response = get_snap_detail(PRIVATE_SNAP_PACKAGE_NAME, auth=False)
+        self.assert_not_found(response, message='No such package')
+
+    def test_get_snap_detail_not_found(self):
+        response = get_snap_detail('missing')
+        self.assert_not_found(response, message='No such package')
+
+
+class SnapMetadataTestCase(MetadataTestCase):
+    endpoint = '/api/v1/snaps/metadata'
+
+
+class SnapSearchTestCase(SearchTestCase):
+    endpoint = '/api/v1/snaps/search'
+
+    def search(self, name=None, fields=None, private=False, auth=False):
+        query = {}
+        if name is not None:
+            query['name'] = name
+        if private:
+            query['private'] = 'true'
+        if fields is not None:
+            query['fields'] = fields
+        return super(SnapSearchTestCase, self)._search(
+            query=query, auth=auth)
+
+    def test_search_private_not_authorized(self):
+        response = self.search(name=PRIVATE_SNAP_PACKAGE_NAME, private=True)
+        self.assert_unauthorized(response)
+
+    def test_search_private(self):
+        response = self.search(name=PRIVATE_SNAP_PACKAGE_NAME, private=True,
+                               auth=True)
+        self.assert_success(response)
+        self.assert_results(response, name=PRIVATE_SNAP_PACKAGE_NAME)