diff options
Diffstat (limited to 'mediagoblin/tests/test_api.py')
| -rw-r--r-- | mediagoblin/tests/test_api.py | 307 |
1 files changed, 255 insertions, 52 deletions
diff --git a/mediagoblin/tests/test_api.py b/mediagoblin/tests/test_api.py index 4e0cbd8f..55228edc 100644 --- a/mediagoblin/tests/test_api.py +++ b/mediagoblin/tests/test_api.py @@ -13,81 +13,284 @@ # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. +import json +import datetime - -import logging -import base64 - +import mock +import pytz import pytest +from webtest import AppError +from werkzeug.datastructures import FileStorage + +from .resources import GOOD_JPG from mediagoblin import mg_globals -from mediagoblin.tools import template, pluginapi +from mediagoblin.media_types import sniff_media +from mediagoblin.db.models import User, MediaEntry +from mediagoblin.submit.lib import new_upload_entry from mediagoblin.tests.tools import fixture_add_user -from .resources import GOOD_JPG, GOOD_PNG, EVIL_FILE, EVIL_JPG, EVIL_PNG, \ - BIG_BLUE +from mediagoblin.federation.task import collect_garbage +from mediagoblin.moderation.tools import take_away_privileges +class TestAPI(object): + """ Test mediagoblin's pump.io complient APIs """ -_log = logging.getLogger(__name__) + @pytest.fixture(autouse=True) + def setup(self, test_app): + self.test_app = test_app + self.db = mg_globals.database + self.user = fixture_add_user(privileges=[u'active', u'uploader']) -class TestAPI(object): - def setup(self): - self.db = mg_globals.database + def _activity_to_feed(self, test_app, activity, headers=None): + """ Posts an activity to the user's feed """ + if headers: + headers.setdefault("Content-Type", "application/json") + else: + headers = {"Content-Type": "application/json"} + + with mock.patch("mediagoblin.decorators.oauth_required", + new_callable=self.mocked_oauth_required): + response = test_app.post( + "/api/user/{0}/feed".format(self.user.username), + json.dumps(activity), + headers=headers + ) + + return response, json.loads(response.body) + + def _upload_image(self, test_app, image): + """ Uploads and image to MediaGoblin via pump.io API """ + data = open(image, "rb").read() + headers = { + "Content-Type": "image/jpeg", + "Content-Length": str(len(data)) + } + + + with mock.patch("mediagoblin.decorators.oauth_required", + new_callable=self.mocked_oauth_required): + response = test_app.post( + "/api/user/{0}/uploads".format(self.user.username), + data, + headers=headers + ) + image = json.loads(response.body) + + return response, image + + def _post_image_to_feed(self, test_app, image): + """ Posts an already uploaded image to feed """ + activity = { + "verb": "post", + "object": image, + } + + return self._activity_to_feed(test_app, activity) + + + def mocked_oauth_required(self, *args, **kwargs): + """ Mocks mediagoblin.decorator.oauth_required to always validate """ + + def fake_controller(controller, request, *args, **kwargs): + request.user = User.query.filter_by(id=self.user.id).first() + return controller(request, *args, **kwargs) + + def oauth_required(c): + return lambda *args, **kwargs: fake_controller(c, *args, **kwargs) + + return oauth_required + + def test_can_post_image(self, test_app): + """ Tests that an image can be posted to the API """ + # First request we need to do is to upload the image + response, image = self._upload_image(test_app, GOOD_JPG) + + # I should have got certain things back + assert response.status_code == 200 + + assert "id" in image + assert "fullImage" in image + assert "url" in image["fullImage"] + assert "url" in image + assert "author" in image + assert "published" in image + assert "updated" in image + assert image["objectType"] == "image" + + # Check that we got the response we're expecting + response, _ = self._post_image_to_feed(test_app, image) + assert response.status_code == 200 + + def test_upload_image_with_filename(self, test_app): + """ Tests that you can upload an image with filename and description """ + response, data = self._upload_image(test_app, GOOD_JPG) + response, data = self._post_image_to_feed(test_app, data) + + image = data["object"] + + # Now we need to add a title and description + title = "My image ^_^" + description = "This is my super awesome image :D" + license = "CC-BY-SA" + + image["displayName"] = title + image["content"] = description + image["license"] = license + + activity = {"verb": "update", "object": image} + + with mock.patch("mediagoblin.decorators.oauth_required", + new_callable=self.mocked_oauth_required): + response = test_app.post( + "/api/user/{0}/feed".format(self.user.username), + json.dumps(activity), + headers={"Content-Type": "application/json"} + ) + + image = json.loads(response.body)["object"] + + # Check everything has been set on the media correctly + media = MediaEntry.query.filter_by(id=image["id"]).first() + assert media.title == title + assert media.description == description + assert media.license == license + + # Check we're being given back everything we should on an update + assert image["id"] == media.id + assert image["displayName"] == title + assert image["content"] == description + assert image["license"] == license + + + def test_only_uploaders_post_image(self, test_app): + """ Test that only uploaders can upload images """ + # Remove uploader permissions from user + take_away_privileges(self.user.username, u"uploader") + + # Now try and upload a image + data = open(GOOD_JPG, "rb").read() + headers = { + "Content-Type": "image/jpeg", + "Content-Length": str(len(data)), + } + + with mock.patch("mediagoblin.decorators.oauth_required", + new_callable=self.mocked_oauth_required): + with pytest.raises(AppError) as excinfo: + test_app.post( + "/api/user/{0}/uploads".format(self.user.username), + data, + headers=headers + ) + + # Assert that we've got a 403 + assert "403 FORBIDDEN" in excinfo.value.message + + def test_object_endpoint(self, test_app): + """ Tests that object can be looked up at endpoint """ + # Post an image + response, data = self._upload_image(test_app, GOOD_JPG) + response, data = self._post_image_to_feed(test_app, data) + + # Now lookup image to check that endpoint works. + image = data["object"] + + assert "links" in image + assert "self" in image["links"] + + # Get URI and strip testing host off + object_uri = image["links"]["self"]["href"] + object_uri = object_uri.replace("http://localhost:80", "") + + with mock.patch("mediagoblin.decorators.oauth_required", + new_callable=self.mocked_oauth_required): + request = test_app.get(object_uri) + + image = json.loads(request.body) + entry = MediaEntry.query.filter_by(id=image["id"]).first() + + assert request.status_code == 200 + assert entry.id == image["id"] - self.user_password = u'4cc355_70k3N' - self.user = fixture_add_user(u'joapi', self.user_password, - privileges=[u'active',u'uploader']) + assert "image" in image + assert "fullImage" in image + assert "pump_io" in image + assert "links" in image - def login(self, test_app): - test_app.post( - '/auth/login/', { - 'username': self.user.username, - 'password': self.user_password}) + def test_post_comment(self, test_app): + """ Tests that I can post an comment media """ + # Upload some media to comment on + response, data = self._upload_image(test_app, GOOD_JPG) + response, data = self._post_image_to_feed(test_app, data) - def get_context(self, template_name): - return template.TEMPLATE_TEST_CONTEXT[template_name] + content = "Hai this is a comment on this lovely picture ^_^" - def http_auth_headers(self): - return {'Authorization': 'Basic {0}'.format( - base64.b64encode(':'.join([ - self.user.username, - self.user_password])))} + activity = { + "verb": "post", + "object": { + "objectType": "comment", + "content": content, + "inReplyTo": data["object"], + } + } - def do_post(self, data, test_app, **kwargs): - url = kwargs.pop('url', '/api/submit') - do_follow = kwargs.pop('do_follow', False) + response, comment_data = self._activity_to_feed(test_app, activity) + assert response.status_code == 200 - if not 'headers' in kwargs.keys(): - kwargs['headers'] = self.http_auth_headers() + # Find the objects in the database + media = MediaEntry.query.filter_by(id=data["object"]["id"]).first() + comment = media.get_comments()[0] - response = test_app.post(url, data, **kwargs) + # Tests that it matches in the database + assert comment.author == self.user.id + assert comment.content == content - if do_follow: - response.follow() + # Test that the response is what we should be given + assert comment.id == comment_data["object"]["id"] + assert comment.content == comment_data["object"]["content"] - return response + def test_profile(self, test_app): + """ Tests profile endpoint """ + uri = "/api/user/{0}/profile".format(self.user.username) + with mock.patch("mediagoblin.decorators.oauth_required", + new_callable=self.mocked_oauth_required): + response = test_app.get(uri) + profile = json.loads(response.body) - def upload_data(self, filename): - return {'upload_files': [('file', filename)]} + assert response.status_code == 200 - def test_1_test_test_view(self, test_app): - self.login(test_app) + assert profile["preferredUsername"] == self.user.username + assert profile["objectType"] == "person" - response = test_app.get( - '/api/test', - headers=self.http_auth_headers()) + assert "links" in profile - assert response.body == \ - '{"username": "joapi", "email": "joapi@example.com"}' + def test_garbage_collection_task(self, test_app): + """ Test old media entry are removed by GC task """ + # Create a media entry that's unprocessed and over an hour old. + entry_id = 72 + now = datetime.datetime.now(pytz.UTC) + file_data = FileStorage( + stream=open(GOOD_JPG, "rb"), + filename="mah_test.jpg", + content_type="image/jpeg" + ) - def test_2_test_submission(self, test_app): - self.login(test_app) + # Find media manager + media_type, media_manager = sniff_media(file_data, "mah_test.jpg") + entry = new_upload_entry(self.user) + entry.id = entry_id + entry.title = "Mah Image" + entry.slug = "slugy-slug-slug" + entry.media_type = 'image' + entry.uploaded = now - datetime.timedelta(days=2) + entry.save() - response = self.do_post( - {'title': 'Great JPG!'}, - test_app, - **self.upload_data(GOOD_JPG)) + # Validate the model exists + assert MediaEntry.query.filter_by(id=entry_id).first() is not None - assert response.status_int == 200 + # Call the garbage collection task + collect_garbage() - assert self.db.MediaEntry.query.filter_by(title=u'Great JPG!').first() + # Now validate the image has been deleted + assert MediaEntry.query.filter_by(id=entry_id).first() is None |