From 8b24b0e91e7ad981a4d873ddb715e4a55098e69b Mon Sep 17 00:00:00 2001 From: rtlCoIl Date: Mon, 18 Jun 2018 16:28:56 +0300 Subject: [PATCH 1/3] Improve PEP8 compliance --- cloudinary/__init__.py | 41 +++-- cloudinary/api.py | 92 +++++++--- cloudinary/auth_token.py | 21 ++- cloudinary/compat.py | 3 +- cloudinary/forms.py | 32 ++-- cloudinary/models.py | 6 +- cloudinary/poster/__init__.py | 11 +- cloudinary/poster/encode.py | 65 ++++--- cloudinary/poster/streaminghttp.py | 36 ++-- cloudinary/search.py | 5 +- cloudinary/templatetags/cloudinary.py | 13 +- cloudinary/uploader.py | 55 +++--- cloudinary/utils.py | 219 ++++++++++++++---------- django_tests/admin.py | 3 - django_tests/migrations/0001_initial.py | 2 +- django_tests/models.py | 6 +- django_tests/test_cloudinaryField.py | 13 +- django_tests/test_cloudinaryResource.py | 30 ++-- django_tests/test_user_agent.py | 2 +- django_tests/tests.py | 3 - django_tests/urls.py | 1 + samples/basic/basic.py | 121 +++++++------ samples/basic_flask/app.py | 3 +- samples/gae/main.py | 8 +- setup.py | 2 +- tests/api_test.py | 41 ++--- tests/archive_test.py | 11 +- tests/auth_token_test.py | 54 +++--- tests/config_test.py | 1 + tests/search_test.py | 24 +-- tests/streaming_profiles_test.py | 32 ++-- tests/test_helper.py | 1 - tests/uploader_test.py | 56 +++--- tests/utils_test.py | 78 ++++++--- tests/video_test.py | 3 +- 35 files changed, 638 insertions(+), 456 deletions(-) diff --git a/cloudinary/__init__.py b/cloudinary/__init__.py index 78bf8f44..1f380ff3 100644 --- a/cloudinary/__init__.py +++ b/cloudinary/__init__.py @@ -1,13 +1,12 @@ from __future__ import absolute_import +import os +import re import logging import numbers from math import ceil -import os -import re - from six import python_2_unicode_compatible, string_types @@ -17,13 +16,12 @@ ch.setFormatter(formatter) logger.addHandler(ch) -from platform import python_version - - from cloudinary import utils from cloudinary.compat import urlparse, parse_qs from cloudinary.search import Search +from platform import python_version + CF_SHARED_CDN = "d3jpl91pxevbkh.cloudfront.net" OLD_AKAMAI_SHARED_CDN = "cloudinary-a.akamaihd.net" AKAMAI_SHARED_CDN = "res.cloudinary.com" @@ -136,7 +134,7 @@ def _is_nested_key(self, key): def _put_nested_key(self, key, value): chain = re.split(r'[\[\]]+', key) - chain = [key for key in chain if key] + chain = [k for k in chain if k] outer = self.__dict__ last_key = chain.pop() for inner_key in chain: @@ -196,9 +194,11 @@ def get_prep_value(self): return None prep = '' prep = prep + self.resource_type + '/' + self.type + '/' - if self.version: prep = prep + 'v' + str(self.version) + '/' + if self.version: + prep = prep + 'v' + str(self.version) + '/' prep = prep + self.public_id - if self.format: prep = prep + '.' + self.format + if self.format: + prep = prep + '.' + self.format return prep def get_presigned(self): @@ -410,8 +410,10 @@ def video_thumbnail(self, **options): # Creates an HTML video tag for the provided +source+ # # ==== Options - # * source_types - Specify which source type the tag should include. defaults to webm, mp4 and ogv. - # * source_transformation - specific transformations to use for a specific source type. + # * source_types - Specify which source type the tag should include. + # defaults to webm, mp4 and ogv. + # * source_transformation - specific transformations to use + # for a specific source type. # * poster - override default thumbnail: # * url: provide an ad hoc url # * options: with specific poster transformations and/or Cloudinary +:public_id+ @@ -423,14 +425,15 @@ def video_thumbnail(self, **options): # CloudinaryResource("mymovie.webm").video(source_types = ['webm', 'mp4'], poster = {'effect': 'sepia'}) def video(self, **options): public_id = options.get('public_id', self.public_id) - source = re.sub("\.({0})$".format("|".join(self.default_source_types())), '', public_id) + source = re.sub(r"\.({0})$".format("|".join(self.default_source_types())), '', public_id) source_types = options.pop('source_types', []) source_transformation = options.pop('source_transformation', {}) fallback = options.pop('fallback_content', '') options['resource_type'] = options.pop('resource_type', self.resource_type or 'video') - if not source_types: source_types = self.default_source_types() + if not source_types: + source_types = self.default_source_types() video_options = options.copy() if 'poster' in video_options: @@ -439,11 +442,13 @@ def video(self, **options): if 'public_id' in poster_options: video_options['poster'] = utils.cloudinary_url(poster_options['public_id'], **poster_options)[0] else: - video_options['poster'] = self.video_thumbnail(public_id=source, **poster_options) + video_options['poster'] = self.video_thumbnail( + public_id=source, **poster_options) else: video_options['poster'] = self.video_thumbnail(public_id=source, **options) - if not video_options['poster']: del video_options['poster'] + if not video_options['poster']: + del video_options['poster'] nested_source_types = isinstance(source_types, list) and len(source_types) > 1 if not nested_source_types: @@ -453,8 +458,10 @@ def video(self, **options): video_options = video_url[1] if not nested_source_types: video_options['src'] = video_url[0] - if 'html_width' in video_options: video_options['width'] = video_options.pop('html_width') - if 'html_height' in video_options: video_options['height'] = video_options.pop('html_height') + if 'html_width' in video_options: + video_options['width'] = video_options.pop('html_width') + if 'html_height' in video_options: + video_options['height'] = video_options.pop('html_height') sources = "" if nested_source_types: diff --git a/cloudinary/api.py b/cloudinary/api.py index ee92fa0b..3e4f4493 100644 --- a/cloudinary/api.py +++ b/cloudinary/api.py @@ -4,26 +4,48 @@ import json import socket -import cloudinary +import urllib3 from six import string_types +from urllib3.exceptions import HTTPError -import urllib3 import certifi - +import cloudinary from cloudinary import utils -from urllib3.exceptions import HTTPError logger = cloudinary.logger + # intentionally one-liners -class Error(Exception): pass -class NotFound(Error): pass -class NotAllowed(Error): pass -class AlreadyExists(Error): pass -class RateLimited(Error): pass -class BadRequest(Error): pass -class GeneralError(Error): pass -class AuthorizationRequired(Error): pass +class Error(Exception): + pass + + +class NotFound(Error): + pass + + +class NotAllowed(Error): + pass + + +class AlreadyExists(Error): + pass + + +class RateLimited(Error): + pass + + +class BadRequest(Error): + pass + + +class GeneralError(Error): + pass + + +class AuthorizationRequired(Error): + pass EXCEPTION_CODES = { @@ -45,6 +67,7 @@ def __init__(self, result, response, **kwargs): self.rate_limit_reset_at = email.utils.parsedate(response.headers["x-featureratelimit-reset"]) self.rate_limit_remaining = int(response.headers["x-featureratelimit-remaining"]) + _http = urllib3.PoolManager( cert_reqs='CERT_REQUIRED', ca_certs=certifi.where() @@ -67,23 +90,26 @@ def resources(**options): resource_type = options.pop("resource_type", "image") upload_type = options.pop("type", None) uri = ["resources", resource_type] - if upload_type: uri.append(upload_type) - params = only(options, - "next_cursor", "max_results", "prefix", "tags", "context", "moderations", "direction", "start_at") + if upload_type: + uri.append(upload_type) + params = only(options, "next_cursor", "max_results", "prefix", "tags", + "context", "moderations", "direction", "start_at") return call_api("get", uri, params, **options) def resources_by_tag(tag, **options): resource_type = options.pop("resource_type", "image") uri = ["resources", resource_type, "tags", tag] - params = only(options, "next_cursor", "max_results", "tags", "context", "moderations", "direction") + params = only(options, "next_cursor", "max_results", "tags", + "context", "moderations", "direction") return call_api("get", uri, params, **options) def resources_by_moderation(kind, status, **options): resource_type = options.pop("resource_type", "image") uri = ["resources", resource_type, "moderations", kind, status] - params = only(options, "next_cursor", "max_results", "tags", "context", "moderations", "direction") + params = only(options, "next_cursor", "max_results", "tags", + "context", "moderations", "direction") return call_api("get", uri, params, **options) @@ -99,7 +125,8 @@ def resource(public_id, **options): resource_type = options.pop("resource_type", "image") upload_type = options.pop("type", "upload") uri = ["resources", resource_type, upload_type, public_id] - params = only(options, "exif", "faces", "colors", "image_metadata", "pages", "phash", "coordinates", "max_results") + params = only(options, "exif", "faces", "colors", "image_metadata", + "pages", "phash", "coordinates", "max_results") return call_api("get", uri, params, **options) @@ -114,9 +141,11 @@ def update(public_id, **options): if "tags" in options: params["tags"] = ",".join(utils.build_array(options["tags"])) if "face_coordinates" in options: - params["face_coordinates"] = utils.encode_double_array(options.get("face_coordinates")) + params["face_coordinates"] = utils.encode_double_array( + options.get("face_coordinates")) if "custom_coordinates" in options: - params["custom_coordinates"] = utils.encode_double_array(options.get("custom_coordinates")) + params["custom_coordinates"] = utils.encode_double_array( + options.get("custom_coordinates")) if "context" in options: params["context"] = utils.encode_context(options.get("context")) if "auto_tagging" in options: @@ -215,14 +244,15 @@ def delete_transformation(transformation, **options): return call_api("delete", uri, {}, **options) -# updates - currently only supported update is the "allowed_for_strict" boolean flag and unsafe_update +# updates - currently only supported update is the "allowed_for_strict" +# boolean flag and unsafe_update def update_transformation(transformation, **options): uri = ["transformations", transformation_string(transformation)] updates = only(options, "allowed_for_strict") if "unsafe_update" in options: updates["unsafe_update"] = transformation_string(options.get("unsafe_update")) - if not updates: raise Exception("No updates given") - + if not updates: + raise Exception("No updates given") return call_api("put", uri, updates, **options) @@ -361,7 +391,8 @@ def update_streaming_profile(name, **options): def call_json_api(method, uri, jsonBody, **options): logger.debug(jsonBody) data = json.dumps(jsonBody).encode('utf-8') - return _call_api(method, uri, body=data, headers={'Content-Type': 'application/json'}, **options) + return _call_api(method, uri, body=data, + headers={'Content-Type': 'application/json'}, **options) def call_api(method, uri, params, **options): @@ -372,11 +403,14 @@ def _call_api(method, uri, params=None, body=None, headers=None, **options): prefix = options.pop("upload_prefix", cloudinary.config().upload_prefix) or "https://api.cloudinary.com" cloud_name = options.pop("cloud_name", cloudinary.config().cloud_name) - if not cloud_name: raise Exception("Must supply cloud_name") + if not cloud_name: + raise Exception("Must supply cloud_name") api_key = options.pop("api_key", cloudinary.config().api_key) - if not api_key: raise Exception("Must supply api_key") + if not api_key: + raise Exception("Must supply api_key") api_secret = options.pop("api_secret", cloudinary.config().api_secret) - if not cloud_name: raise Exception("Must supply api_secret") + if not cloud_name: + raise Exception("Must supply api_secret") api_url = "/".join([prefix, "v1_1", cloud_name] + uri) processed_params = None @@ -437,10 +471,12 @@ def transformation_string(transformation): def __prepare_streaming_profile_params(**options): params = only(options, "display_name") if "representations" in options: - representations = [{"transformation": transformation_string(trans)} for trans in options["representations"]] + representations = [{"transformation": transformation_string(trans)} + for trans in options["representations"]] params["representations"] = json.dumps(representations) return params + def __delete_resource_params(options, **params): p = dict(transformations=utils.build_eager(options.get('transformations')), **only(options, "keep_original", "next_cursor", "invalidate")) diff --git a/cloudinary/auth_token.py b/cloudinary/auth_token.py index 72fc341e..37022bc7 100644 --- a/cloudinary/auth_token.py +++ b/cloudinary/auth_token.py @@ -3,14 +3,14 @@ import re import time from binascii import a2b_hex + from cloudinary.compat import quote_plus AUTH_TOKEN_NAME = "__cld_token__" - -def generate(url=None, acl=None, start_time=None, duration=None, expiration=None, ip=None, key=None, - token_name=AUTH_TOKEN_NAME): +def generate(url=None, acl=None, start_time=None, duration=None, + expiration=None, ip=None, key=None, token_name=AUTH_TOKEN_NAME): if expiration is None: if duration is not None: @@ -20,10 +20,13 @@ def generate(url=None, acl=None, start_time=None, duration=None, expiration=None raise Exception("Must provide either expiration or duration") token_parts = [] - if ip is not None: token_parts.append("ip=" + ip) - if start_time is not None: token_parts.append("st=%d" % start_time) + if ip is not None: + token_parts.append("ip=" + ip) + if start_time is not None: + token_parts.append("st=%d" % start_time) token_parts.append("exp=%d" % expiration) - if acl is not None: token_parts.append("acl=%s" % _escape_to_lower(acl)) + if acl is not None: + token_parts.append("acl=%s" % _escape_to_lower(acl)) to_sign = list(token_parts) if url is not None: to_sign.append("url=%s" % _escape_to_lower(url)) @@ -39,9 +42,5 @@ def _digest(message, key): def _escape_to_lower(url): escaped_url = quote_plus(url) - - def toLowercase(match): - return match.group(0).lower() - - escaped_url = re.sub(r'%..', toLowercase, escaped_url) + escaped_url = re.sub(r'%..', lambda x: x.group(0).lower(), escaped_url) return escaped_url diff --git a/cloudinary/compat.py b/cloudinary/compat.py index 430134ea..5b2b0689 100644 --- a/cloudinary/compat.py +++ b/cloudinary/compat.py @@ -1,5 +1,7 @@ # Copyright Cloudinary import six.moves.urllib.parse +from six import PY3, string_types, StringIO, BytesIO + urlencode = six.moves.urllib.parse.urlencode unquote = six.moves.urllib.parse.unquote urlparse = six.moves.urllib.parse.urlparse @@ -7,7 +9,6 @@ parse_qsl = six.moves.urllib.parse.parse_qsl quote_plus = six.moves.urllib.parse.quote_plus httplib = six.moves.http_client -from six import PY3, string_types, StringIO, BytesIO urllib2 = six.moves.urllib.request NotConnected = six.moves.http_client.NotConnected diff --git a/cloudinary/forms.py b/cloudinary/forms.py index 3465889a..1fba77e0 100644 --- a/cloudinary/forms.py +++ b/cloudinary/forms.py @@ -1,9 +1,10 @@ -from django import forms -from cloudinary import CloudinaryResource +import json +import re + import cloudinary.uploader import cloudinary.utils -import re -import json +from cloudinary import CloudinaryResource +from django import forms from django.utils.translation import ugettext_lazy as _ @@ -27,14 +28,16 @@ def render(self, name, value, attrs=None): else: params = cloudinary.utils.sign_request(params, options) - if 'resource_type' not in options: options['resource_type'] = 'auto' + if 'resource_type' not in options: + options['resource_type'] = 'auto' cloudinary_upload_url = cloudinary.utils.cloudinary_api_url("upload", **options) attrs["data-url"] = cloudinary_upload_url attrs["data-form-data"] = json.dumps(params) attrs["data-cloudinary-field"] = name chunk_size = options.get("chunk_size", None) - if chunk_size: attrs["data-max-chunk-size"] = chunk_size + if chunk_size: + attrs["data-max-chunk-size"] = chunk_size attrs["class"] = " ".join(["cloudinary-fileupload", attrs.get("class", "")]) widget = super(CloudinaryInput, self).render("file", None, attrs=attrs) @@ -53,8 +56,10 @@ class CloudinaryJsFileField(forms.Field): } def __init__(self, attrs=None, options=None, autosave=True, *args, **kwargs): - if attrs is None: attrs = {} - if options is None: options = {} + if attrs is None: + attrs = {} + if options is None: + options = {} self.autosave = autosave attrs = attrs.copy() attrs["options"] = options.copy() @@ -70,7 +75,8 @@ def enable_callback(self, request): def to_python(self, value): """Convert to CloudinaryResource""" - if not value: return None + if not value: + return None m = re.search(r'^([^/]+)/([^/]+)/v(\d+)/([^#]+)#([^/]+)$', value) if not m: raise forms.ValidationError("Invalid format") @@ -95,7 +101,8 @@ def validate(self, value): """Validate the signature""" # Use the parent's handling of required fields, etc. super(CloudinaryJsFileField, self).validate(value) - if not value: return + if not value: + return if not value.validate(): raise forms.ValidationError("Signature mismatch") @@ -108,7 +115,8 @@ def __init__(self, upload_preset, attrs=None, options=None, autosave=True, *args options = {} options = options.copy() options.update({"unsigned": True, "upload_preset": upload_preset}) - super(CloudinaryUnsignedJsFileField, self).__init__(attrs, options, autosave, *args, **kwargs) + super(CloudinaryUnsignedJsFileField, self).__init__( + attrs, options, autosave, *args, **kwargs) class CloudinaryFileField(forms.FileField): @@ -117,7 +125,7 @@ class CloudinaryFileField(forms.FileField): } default_error_messages = forms.FileField.default_error_messages.copy() default_error_messages.update(my_default_error_messages) - + def __init__(self, options=None, autosave=True, *args, **kwargs): self.autosave = autosave self.options = options or {} diff --git a/cloudinary/models.py b/cloudinary/models.py index 46319bf9..1d15122d 100644 --- a/cloudinary/models.py +++ b/cloudinary/models.py @@ -1,8 +1,6 @@ import re - from cloudinary import CloudinaryResource, forms, uploader - from django.core.files.uploadedfile import UploadedFile from django.db import models @@ -13,7 +11,9 @@ except ImportError: pass -CLOUDINARY_FIELD_DB_RE = r'(?:(?Pimage|raw|video)/(?Pupload|private|authenticated)/)?(?:v(?P\d+)/)?(?P.*?)(\.(?P[^.]+))?$' +CLOUDINARY_FIELD_DB_RE = r'(?:(?Pimage|raw|video)/' \ + r'(?Pupload|private|authenticated)/)?' \ + r'(?:v(?P\d+)/)?(?P.*?)(\.(?P[^.]+))?$' # Taken from six - https://pythonhosted.org/six/ diff --git a/cloudinary/poster/__init__.py b/cloudinary/poster/__init__.py index 9110fa42..9359a53d 100644 --- a/cloudinary/poster/__init__.py +++ b/cloudinary/poster/__init__.py @@ -1,17 +1,17 @@ # MIT licensed code copied from https://bitbucket.org/chrisatlee/poster # # Copyright (c) 2011 Chris AtLee -# +# # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: -# +# # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. -# +# # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE @@ -28,7 +28,4 @@ than an older version of poster. New in version 0.6.""" -import cloudinary.poster.streaminghttp -import cloudinary.poster.encode - -version = (0, 8, 2) # Thanks JP! +version = (0, 8, 2) # Thanks JP! diff --git a/cloudinary/poster/encode.py b/cloudinary/poster/encode.py index 4900eee4..90ef3069 100644 --- a/cloudinary/poster/encode.py +++ b/cloudinary/poster/encode.py @@ -6,9 +6,17 @@ multipart/form-data is the standard way to upload files over HTTP""" -__all__ = ['gen_boundary', 'encode_and_quote', 'MultipartParam', - 'encode_string', 'encode_file_header', 'get_body_size', 'get_headers', - 'multipart_encode'] +import mimetypes +import os +import re +from email.header import Header + +from cloudinary.compat import (PY3, advance_iterator, quote_plus, to_bytearray, + to_bytes, to_string) + +__all__ = [ + 'gen_boundary', 'encode_and_quote', 'MultipartParam', 'encode_string', + 'encode_file_header', 'get_body_size', 'get_headers', 'multipart_encode'] try: from io import UnsupportedOperation @@ -17,25 +25,19 @@ try: import uuid + def gen_boundary(): """Returns a random string to use as the boundary for a message""" return uuid.uuid4().hex except ImportError: - import random, sha + import random + import sha + def gen_boundary(): """Returns a random string to use as the boundary for a message""" bits = random.getrandbits(160) return sha.new(str(bits)).hexdigest() -import re, os, mimetypes -from cloudinary.compat import (PY3, string_types, to_bytes, to_string, - to_bytearray, quote_plus, advance_iterator) -try: - from email.header import Header -except ImportError: - # Python 2.4 - from email.Header import Header - if PY3: def encode_and_quote(data): if data is None: @@ -47,7 +49,7 @@ def encode_and_quote(data): """If ``data`` is unicode, return quote_plus(data.encode("utf-8")) otherwise return quote_plus(data)""" if data is None: return None - + if isinstance(data, unicode): data = data.encode("utf-8") return quote_plus(data) @@ -65,13 +67,15 @@ def _strify(s): return to_bytes(str(s)) else: def _strify(s): - """If s is a unicode string, encode it to UTF-8 and return the results, otherwise return str(s), or None if s is None""" + """If s is a unicode string, encode it to UTF-8 and return the results, + otherwise return str(s), or None if s is None""" if s is None: return None if isinstance(s, unicode): return s.encode("utf-8") return str(s) + class MultipartParam(object): """Represents a single parameter in a multipart/form-data request @@ -105,7 +109,7 @@ class MultipartParam(object): transferred, and the total size. """ def __init__(self, name, value=None, filename=None, filetype=None, - filesize=None, fileobj=None, cb=None): + filesize=None, fileobj=None, cb=None): self.name = Header(name).encode() self.value = _strify(value) if filename is None: @@ -141,7 +145,7 @@ def __init__(self, name, value=None, filename=None, filetype=None, fileobj.seek(0, 2) self.filesize = fileobj.tell() fileobj.seek(0) - except: + except Exception: raise ValueError("Could not determine filesize") def __cmp__(self, other): @@ -169,9 +173,9 @@ def from_file(cls, paramname, filename): """ return cls(paramname, filename=os.path.basename(filename), - filetype=mimetypes.guess_type(filename)[0], - filesize=os.path.getsize(filename), - fileobj=open(filename, "rb")) + filetype=mimetypes.guess_type(filename)[0], + filesize=os.path.getsize(filename), + fileobj=open(filename, "rb")) @classmethod def from_params(cls, params): @@ -204,7 +208,7 @@ def from_params(cls, params): filetype = None retval.append(cls(name=name, filename=filename, - filetype=filetype, fileobj=value)) + filetype=filetype, fileobj=value)) else: retval.append(cls(name, value)) return retval @@ -216,8 +220,8 @@ def encode_hdr(self, boundary): headers = ["--%s" % boundary] if self.filename: - disposition = 'form-data; name="%s"; filename="%s"' % (self.name, - to_string(self.filename)) + disposition = 'form-data; name="%s"; filename="%s"' % ( + self.name, to_string(self.filename)) else: disposition = 'form-data; name="%s"' % self.name @@ -267,8 +271,8 @@ def iter_encode(self, boundary, blocksize=4096): self.cb(self, current, total) last_block = to_bytearray("") encoded_boundary = "--%s" % encode_and_quote(boundary) - boundary_exp = re.compile(to_bytes("^%s$" % re.escape(encoded_boundary)), - re.M) + boundary_exp = re.compile( + to_bytes("^%s$" % re.escape(encoded_boundary)), re.M) while True: block = self.fileobj.read(blocksize) if not block: @@ -296,6 +300,7 @@ def get_size(self, boundary): return len(self.encode_hdr(boundary)) + 2 + valuesize + def encode_string(boundary, name, value): """Returns ``name`` and ``value`` encoded as a multipart/form-data variable. ``boundary`` is the boundary string used throughout @@ -303,8 +308,8 @@ def encode_string(boundary, name, value): return MultipartParam(name, value).encode(boundary) -def encode_file_header(boundary, paramname, filesize, filename=None, - filetype=None): + +def encode_file_header(boundary, paramname, filesize, filename=None, filetype=None): """Returns the leading data for a multipart/form-data field that contains file data. @@ -324,7 +329,8 @@ def encode_file_header(boundary, paramname, filesize, filename=None, """ return MultipartParam(paramname, filesize=filesize, filename=filename, - filetype=filetype).encode_hdr(boundary) + filetype=filetype).encode_hdr(boundary) + def get_body_size(params, boundary): """Returns the number of bytes that the multipart/form-data encoding @@ -332,6 +338,7 @@ def get_body_size(params, boundary): size = sum(p.get_size(boundary) for p in MultipartParam.from_params(params)) return size + len(boundary) + 6 + def get_headers(params, boundary): """Returns a dictionary with Content-Type and Content-Length headers for the multipart/form-data encoding of ``params``.""" @@ -341,6 +348,7 @@ def get_headers(params, boundary): headers['Content-Length'] = str(get_body_size(params, boundary)) return headers + class multipart_yielder: def __init__(self, params, boundary, cb): self.params = params @@ -396,6 +404,7 @@ def reset(self): for param in self.params: param.reset() + def multipart_encode(params, boundary=None, cb=None): """Encode ``params`` as multipart/form-data. diff --git a/cloudinary/poster/streaminghttp.py b/cloudinary/poster/streaminghttp.py index d8af5212..f5713cc0 100644 --- a/cloudinary/poster/streaminghttp.py +++ b/cloudinary/poster/streaminghttp.py @@ -27,15 +27,18 @@ ... {'Content-Length': str(len(s))}) """ -import sys, socket -from cloudinary.compat import httplib, urllib2, NotConnected +import socket +import sys + +from cloudinary.compat import NotConnected, httplib, urllib2 __all__ = ['StreamingHTTPConnection', 'StreamingHTTPRedirectHandler', - 'StreamingHTTPHandler', 'register_openers'] + 'StreamingHTTPHandler', 'register_openers'] if hasattr(httplib, 'HTTPS'): __all__.extend(['StreamingHTTPSHandler', 'StreamingHTTPSConnection']) + class _StreamingHTTPMixin: """Mixin class for HTTP and HTTPS connections that implements a streaming send method.""" @@ -62,7 +65,7 @@ def send(self, value): print("send:", repr(value)) try: blocksize = 8192 - if hasattr(value, 'read') : + if hasattr(value, 'read'): if hasattr(value, 'seek'): value.seek(0) if self.debuglevel > 0: @@ -86,10 +89,12 @@ def send(self, value): self.close() raise + class StreamingHTTPConnection(_StreamingHTTPMixin, httplib.HTTPConnection): """Subclass of `httplib.HTTPConnection` that overrides the `send()` method to support iterable body objects""" + class StreamingHTTPRedirectHandler(urllib2.HTTPRedirectHandler): """Subclass of `urllib2.HTTPRedirectHandler` that overrides the `redirect_request` method to properly handle redirected POST requests @@ -114,7 +119,7 @@ def redirect_request(self, req, fp, code, msg, headers, newurl): """ m = req.get_method() if (code in (301, 302, 303, 307) and m in ("GET", "HEAD") - or code in (301, 302, 303) and m == "POST"): + or code in (301, 302, 303) and m == "POST"): # Strictly (according to RFC 2616), 301 or 302 in response # to a POST MUST NOT cause a redirection without confirmation # from the user (of urllib2, in this case). In practice, @@ -125,14 +130,16 @@ def redirect_request(self, req, fp, code, msg, headers, newurl): newheaders = dict((k, v) for k, v in req.headers.items() if k.lower() not in ( "content-length", "content-type") - ) - return urllib2.Request(newurl, - headers=newheaders, - origin_req_host=req.get_origin_req_host(), - unverifiable=True) + ) + return urllib2.Request( + newurl, + headers=newheaders, + origin_req_host=req.get_origin_req_host(), + unverifiable=True) else: raise urllib2.HTTPError(req.get_full_url(), code, msg, headers, fp) + class StreamingHTTPHandler(urllib2.HTTPHandler): """Subclass of `urllib2.HTTPHandler` that uses StreamingHTTPConnection as its http connection class.""" @@ -156,9 +163,9 @@ def http_request(self, req): "No Content-Length specified for iterable body") return urllib2.HTTPHandler.do_request_(self, req) + if hasattr(httplib, 'HTTPS'): - class StreamingHTTPSConnection(_StreamingHTTPMixin, - httplib.HTTPSConnection): + class StreamingHTTPSConnection(_StreamingHTTPMixin, httplib.HTTPSConnection): """Subclass of `httplib.HTTSConnection` that overrides the `send()` method to support iterable body objects""" @@ -179,7 +186,7 @@ def https_request(self, req): if hasattr(data, 'read') or hasattr(data, 'next'): if not req.has_header('Content-length'): raise ValueError( - "No Content-Length specified for iterable body") + "No Content-Length specified for iterable body") return urllib2.HTTPSHandler.do_request_(self, req) @@ -188,7 +195,8 @@ def get_handlers(): if hasattr(httplib, "HTTPS"): handlers.append(StreamingHTTPSHandler) return handlers - + + def register_openers(): """Register the streaming http handlers in the global urllib2 default opener object. diff --git a/cloudinary/search.py b/cloudinary/search.py index 2decef84..91b7e9a6 100644 --- a/cloudinary/search.py +++ b/cloudinary/search.py @@ -1,5 +1,6 @@ import json from copy import deepcopy + from . import api @@ -46,7 +47,7 @@ def to_json(self): def execute(self, **options): """Execute the search and return results.""" options["content_type"] = 'application/json' - uri = ['resources','search'] + uri = ['resources', 'search'] return api.call_json_api('post', uri, self.as_dict(), **options) def _add(self, name, value): @@ -56,4 +57,4 @@ def _add(self, name, value): return self def as_dict(self): - return deepcopy(self.query) \ No newline at end of file + return deepcopy(self.query) diff --git a/cloudinary/templatetags/cloudinary.py b/cloudinary/templatetags/cloudinary.py index febe1e2f..0a669053 100644 --- a/cloudinary/templatetags/cloudinary.py +++ b/cloudinary/templatetags/cloudinary.py @@ -2,15 +2,14 @@ import json +import cloudinary +from cloudinary import CloudinaryResource, utils +from cloudinary.compat import PY3 +from cloudinary.forms import CloudinaryJsFileField, cl_init_js_callbacks from django import template from django.forms import Form from django.utils.safestring import mark_safe -import cloudinary -from cloudinary import CloudinaryResource, utils, uploader -from cloudinary.forms import CloudinaryJsFileField, cl_init_js_callbacks -from cloudinary.compat import PY3 - register = template.Library() @@ -57,9 +56,9 @@ def cloudinary_direct_upload_field(field_name="image", request=None): return value -"""Deprecated - please use cloudinary_direct_upload_field, or a proper form""" @register.inclusion_tag('cloudinary_direct_upload.html') def cloudinary_direct_upload(callback_url, **options): + """Deprecated - please use cloudinary_direct_upload_field, or a proper form""" params = utils.build_upload_params(callback=callback_url, **options) params = utils.sign_request(params, options) @@ -75,6 +74,8 @@ def cloudinary_includes(processing=False): CLOUDINARY_JS_CONFIG_PARAMS = ("api_key", "cloud_name", "private_cdn", "secure_distribution", "cdn_subdomain") + + @register.inclusion_tag('cloudinary_js_config.html') def cloudinary_js_config(): config = cloudinary.config() diff --git a/cloudinary/uploader.py b/cloudinary/uploader.py index c09a1842..c9f12e19 100644 --- a/cloudinary/uploader.py +++ b/cloudinary/uploader.py @@ -4,14 +4,14 @@ import socket from os.path import getsize -import cloudinary -import urllib3 +from six import string_types +from urllib3 import PoolManager +from urllib3.exceptions import HTTPError + import certifi +import cloudinary from cloudinary import utils from cloudinary.api import Error -from cloudinary.compat import string_types -from urllib3.exceptions import HTTPError -from urllib3 import PoolManager try: from urllib3.contrib.appengine import AppEngineManager, is_appengine_sandbox @@ -55,7 +55,8 @@ def upload_resource(file, **options): result = upload(file, **options) return cloudinary.CloudinaryResource( result["public_id"], version=str(result["version"]), - format=result.get("format"), type=result["type"], resource_type=result["resource_type"], metadata=result) + format=result.get("format"), type=result["type"], + resource_type=result["resource_type"], metadata=result) def upload_large(file, **options): @@ -74,9 +75,11 @@ def upload_large(file, **options): range = "bytes {0}-{1}/{2}".format(current_loc, current_loc + len(chunk) - 1, file_size) current_loc += len(chunk) - results = upload_large_part((file, chunk), - http_headers={"Content-Range": range, "X-Unique-Upload-Id": upload_id}, - **options) + results = upload_large_part( + (file, chunk), + http_headers={"Content-Range": range, + "X-Unique-Upload-Id": upload_id}, + **options) options["public_id"] = results.get("public_id") chunk = file_io.read(chunk_size) return results @@ -85,7 +88,8 @@ def upload_large(file, **options): def upload_large_part(file, **options): """ Upload large files. """ params = utils.build_upload_params(**options) - if 'resource_type' not in options: options['resource_type'] = "raw" + if 'resource_type' not in options: + options['resource_type'] = "raw" return call_api("upload", params, file=file, **options) @@ -94,7 +98,7 @@ def destroy(public_id, **options): "timestamp": utils.now(), "type": options.get("type"), "invalidate": options.get("invalidate"), - "public_id": public_id + "public_id": public_id } return call_api("destroy", params, **options) @@ -134,7 +138,8 @@ def generate_sprite(tag, **options): "tag": tag, "async": options.get("async"), "notification_url": options.get("notification_url"), - "transformation": utils.generate_transformation_string(fetch_format=options.get("format"), **options)[0] + "transformation": utils.generate_transformation_string( + fetch_format=options.get("format"), **options)[0] } return call_api("sprite", params, **options) @@ -230,17 +235,18 @@ def call_context_api(context, command, public_ids=None, **options): return call_api("context", params, **options) -TEXT_PARAMS = ["public_id", - "font_family", - "font_size", - "font_color", - "text_align", - "font_weight", - "font_style", - "background", - "opacity", - "text_decoration" - ] +TEXT_PARAMS = [ + "public_id", + "font_family", + "font_size", + "font_color", + "text_align", + "font_weight", + "font_style", + "background", + "opacity", + "text_decoration" +] def text(text, **options): @@ -325,4 +331,5 @@ def call_api(action, params, http_headers=None, return_error=False, unsigned=Fal return result finally: - if file_io: file_io.close() + if file_io: + file_io.close() diff --git a/cloudinary/utils.py b/cloudinary/utils.py index 0a3f3da4..f2589630 100644 --- a/cloudinary/utils.py +++ b/cloudinary/utils.py @@ -166,6 +166,7 @@ def recurse(bs): return generate_transformation_string(**bs)[0] else: return generate_transformation_string(transformation=bs)[0] + base_transformations = list(map(recurse, base_transformations)) named_transformation = None else: @@ -218,17 +219,17 @@ def recurse(bs): "fl": flags, "h": normalize_expression(height), "l": overlay, - "o": normalize_expression(options.pop('opacity',None)), - "q": normalize_expression(options.pop('quality',None)), - "r": normalize_expression(options.pop('radius',None)), + "o": normalize_expression(options.pop('opacity', None)), + "q": normalize_expression(options.pop('quality', None)), + "r": normalize_expression(options.pop('radius', None)), "so": normalize_expression(start_offset), "t": named_transformation, "u": underlay, "w": normalize_expression(width), - "x": normalize_expression(options.pop('x',None)), - "y": normalize_expression(options.pop('y',None)), + "x": normalize_expression(options.pop('x', None)), + "y": normalize_expression(options.pop('y', None)), "vc": video_codec, - "z": normalize_expression(options.pop('zoom',None)) + "z": normalize_expression(options.pop('zoom', None)) } simple_params = { "ac": "audio_codec", @@ -250,9 +251,9 @@ def recurse(bs): for param, option in simple_params.items(): params[param] = options.pop(option, None) - variables = options.pop('variables',{}) + variables = options.pop('variables', {}) var_params = [] - for key,value in options.items(): + for key, value in options.items(): if re.match(r'^\$', key): var_params.append(u"{0}_{1}".format(key, normalize_expression(str(value)))) @@ -306,15 +307,17 @@ def split_range(range): def norm_range_value(value): - if value is None: return None + if value is None: + return None match = re.match(RANGE_VALUE_RE, str(value)) - if match is None: return None + if match is None: + return None modifier = '' if match.group('modifier') is not None: - modifier = 'p' + modifier = 'p' return match.group('value') + modifier @@ -341,9 +344,11 @@ def cleanup_params(params): def sign_request(params, options): api_key = options.get("api_key", cloudinary.config().api_key) - if not api_key: raise ValueError("Must supply api_key") + if not api_key: + raise ValueError("Must supply api_key") api_secret = options.get("api_secret", cloudinary.config().api_secret) - if not api_secret: raise ValueError("Must supply api_secret") + if not api_secret: + raise ValueError("Must supply api_secret") params = cleanup_params(params) params["signature"] = api_sign_request(params, api_secret) @@ -380,11 +385,13 @@ def finalize_source(source, format, url_suffix): source_to_sign = source else: source = unquote(source) - if not PY3: source = source.encode('utf8') + if not PY3: + source = source.encode('utf8') source = smart_escape(source) source_to_sign = source if url_suffix is not None: - if re.search(r'[\./]', url_suffix): raise ValueError("url_suffix should not include . or /") + if re.search(r'[\./]', url_suffix): + raise ValueError("url_suffix should not include . or /") source = source + "/" + url_suffix if format is not None: source = source + "." + format @@ -406,7 +413,8 @@ def finalize_resource_type(resource_type, type, url_suffix, use_root_path, short raise ValueError("URL Suffix only supported for image/upload and raw/upload") if use_root_path: - if (resource_type == "image" and upload_type == "upload") or (resource_type == "images" and upload_type is None): + if (resource_type == "image" and upload_type == "upload") or ( + resource_type == "images" and upload_type is None): resource_type = None upload_type = None else: @@ -419,28 +427,33 @@ def finalize_resource_type(resource_type, type, url_suffix, use_root_path, short return resource_type, upload_type -def unsigned_download_url_prefix(source, cloud_name, private_cdn, cdn_subdomain, secure_cdn_subdomain, cname, secure, - secure_distribution): +def unsigned_download_url_prefix(source, cloud_name, private_cdn, cdn_subdomain, + secure_cdn_subdomain, cname, secure, secure_distribution): """cdn_subdomain and secure_cdn_subdomain 1) Customers in shared distribution (e.g. res.cloudinary.com) - if cdn_domain is true uses res-[1-5].cloudinary.com for both http and https. Setting secure_cdn_subdomain to false disables this for https. + if cdn_domain is true uses res-[1-5].cloudinary.com for both http and https. + Setting secure_cdn_subdomain to false disables this for https. 2) Customers with private cdn if cdn_domain is true uses cloudname-res-[1-5].cloudinary.com for http - if secure_cdn_domain is true uses cloudname-res-[1-5].cloudinary.com for https (please contact support if you require this) + if secure_cdn_domain is true uses cloudname-res-[1-5].cloudinary.com for https + (please contact support if you require this) 3) Customers with cname - if cdn_domain is true uses a[1-5].cname for http. For https, uses the same naming scheme as 1 for shared distribution and as 2 for private distribution.""" + if cdn_domain is true uses a[1-5].cname for http. For https, uses the same naming scheme + as 1 for shared distribution and as 2 for private distribution.""" shared_domain = not private_cdn shard = __crc(source) if secure: if secure_distribution is None or secure_distribution == cloudinary.OLD_AKAMAI_SHARED_CDN: - secure_distribution = cloud_name + "-res.cloudinary.com" if private_cdn else cloudinary.SHARED_CDN + secure_distribution = cloud_name + "-res.cloudinary.com" \ + if private_cdn else cloudinary.SHARED_CDN shared_domain = shared_domain or secure_distribution == cloudinary.SHARED_CDN if secure_cdn_subdomain is None and shared_domain: secure_cdn_subdomain = cdn_subdomain if secure_cdn_subdomain: - secure_distribution = re.sub('res.cloudinary.com', "res-" + shard + ".cloudinary.com", secure_distribution) + secure_distribution = re.sub('res.cloudinary.com', "res-" + shard + ".cloudinary.com", + secure_distribution) prefix = "https://" + secure_distribution elif cname: @@ -448,10 +461,12 @@ def unsigned_download_url_prefix(source, cloud_name, private_cdn, cdn_subdomain, prefix = "http://" + subdomain + cname else: subdomain = cloud_name + "-res" if private_cdn else "res" - if cdn_subdomain: subdomain = subdomain + "-" + shard + if cdn_subdomain: + subdomain = subdomain + "-" + shard prefix = "http://" + subdomain + ".cloudinary.com" - if shared_domain: prefix += "/" + cloud_name + if shared_domain: + prefix += "/" + cloud_name return prefix @@ -479,7 +494,8 @@ def cloudinary_url(source, **options): version = options.pop("version", None) format = options.pop("format", None) cdn_subdomain = options.pop("cdn_subdomain", cloudinary.config().cdn_subdomain) - secure_cdn_subdomain = options.pop("secure_cdn_subdomain", cloudinary.config().secure_cdn_subdomain) + secure_cdn_subdomain = options.pop("secure_cdn_subdomain", + cloudinary.config().secure_cdn_subdomain) cname = options.pop("cname", cloudinary.config().cname) shorten = options.pop("shorten", cloudinary.config().shorten) @@ -488,7 +504,8 @@ def cloudinary_url(source, **options): raise ValueError("Must supply cloud_name in tag or in configuration") secure = options.pop("secure", cloudinary.config().secure) private_cdn = options.pop("private_cdn", cloudinary.config().private_cdn) - secure_distribution = options.pop("secure_distribution", cloudinary.config().secure_distribution) + secure_distribution = options.pop("secure_distribution", + cloudinary.config().secure_distribution) sign_url = options.pop("sign_url", cloudinary.config().sign_url) api_secret = options.pop("api_secret", cloudinary.config().api_secret) url_suffix = options.pop("url_suffix", None) @@ -500,7 +517,8 @@ def cloudinary_url(source, **options): if (not source) or type == "upload" and re.match(r'^https?:', source): return original_source, options - resource_type, type = finalize_resource_type(resource_type, type, url_suffix, use_root_path, shorten) + resource_type, type = finalize_resource_type( + resource_type, type, url_suffix, use_root_path, shorten) source, source_to_sign = finalize_source(source, format, url_suffix) if source_to_sign.find("/") >= 0 \ @@ -508,7 +526,8 @@ def cloudinary_url(source, **options): and not re.match(r'^v[0-9]+', source_to_sign) \ and not version: version = "1" - if version: version = "v" + str(version) + if version: + version = "v" + str(version) transformation = re.sub(r'([^:])/+', r'\1/', transformation) @@ -516,35 +535,44 @@ def cloudinary_url(source, **options): if sign_url and not auth_token: to_sign = "/".join(__compact([transformation, source_to_sign])) signature = "s--" + to_string( - base64.urlsafe_b64encode(hashlib.sha1(to_bytes(to_sign + api_secret)).digest())[0:8]) + "--" - - prefix = unsigned_download_url_prefix(source, cloud_name, private_cdn, cdn_subdomain, secure_cdn_subdomain, cname, - secure, secure_distribution) - source = "/".join(__compact([prefix, resource_type, type, signature, transformation, version, source])) + base64.urlsafe_b64encode( + hashlib.sha1(to_bytes(to_sign + api_secret)).digest())[0:8]) + "--" + + prefix = unsigned_download_url_prefix( + source, cloud_name, private_cdn, cdn_subdomain, secure_cdn_subdomain, + cname, secure, secure_distribution) + source = "/".join(__compact( + [prefix, resource_type, type, signature, transformation, version, source])) if sign_url and auth_token: path = urlparse(source).path - token = cloudinary.auth_token.generate( **merge(auth_token, {"url": path})) + token = cloudinary.auth_token.generate(**merge(auth_token, {"url": path})) source = "%s?%s" % (source, token) return source, options def cloudinary_api_url(action='upload', **options): - cloudinary_prefix = options.get("upload_prefix", cloudinary.config().upload_prefix) or "https://api.cloudinary.com" + cloudinary_prefix = options.get("upload_prefix", cloudinary.config().upload_prefix)\ + or "https://api.cloudinary.com" cloud_name = options.get("cloud_name", cloudinary.config().cloud_name) - if not cloud_name: raise ValueError("Must supply cloud_name") + if not cloud_name: + raise ValueError("Must supply cloud_name") resource_type = options.get("resource_type", "image") return "/".join([cloudinary_prefix, "v1_1", cloud_name, resource_type, action]) # Based on ruby's CGI::unescape. In addition does not escape / : -def smart_escape(source,unsafe = r"([^a-zA-Z0-9_.\-\/:]+)"): +def smart_escape(source, unsafe=r"([^a-zA-Z0-9_.\-\/:]+)"): def pack(m): - return to_bytes('%' + "%".join(["%02X" % x for x in struct.unpack('B' * len(m.group(1)), m.group(1))]).upper()) + return to_bytes('%' + "%".join( + ["%02X" % x for x in struct.unpack('B' * len(m.group(1)), m.group(1))] + ).upper()) + return to_string(re.sub(to_bytes(unsafe), pack, to_bytes(source))) def random_public_id(): - return ''.join(random.SystemRandom().choice(string.ascii_lowercase + string.digits) for _ in range(16)) + return ''.join(random.SystemRandom().choice(string.ascii_lowercase + string.digits) + for _ in range(16)) def signed_preloaded_image(result): @@ -594,7 +622,8 @@ def download_archive_url(**options): params = options.copy() params.update(mode="download") cloudinary_params = sign_request(archive_params(**params), options) - return cloudinary_api_url("generate_archive", **options) + "?" + urlencode(bracketize_seq(cloudinary_params), True) + return cloudinary_api_url("generate_archive", **options) + "?" + \ + urlencode(bracketize_seq(cloudinary_params), True) def download_zip_url(**options): @@ -665,49 +694,54 @@ def build_custom_headers(headers): def build_upload_params(**options): - params = {"timestamp": now(), - "transformation": generate_transformation_string(**options)[0], - "public_id": options.get("public_id"), - "callback": options.get("callback"), - "format": options.get("format"), - "type": options.get("type"), - "backup": options.get("backup"), - "faces": options.get("faces"), - "image_metadata": options.get("image_metadata"), - "exif": options.get("exif"), - "colors": options.get("colors"), - "headers": build_custom_headers(options.get("headers")), - "eager": build_eager(options.get("eager")), - "use_filename": options.get("use_filename"), - "unique_filename": options.get("unique_filename"), - "discard_original_filename": options.get("discard_original_filename"), - "invalidate": options.get("invalidate"), - "notification_url": options.get("notification_url"), - "eager_notification_url": options.get("eager_notification_url"), - "eager_async": options.get("eager_async"), - "proxy": options.get("proxy"), - "folder": options.get("folder"), - "overwrite": options.get("overwrite"), - "tags": options.get("tags") and ",".join(build_array(options["tags"])), - "allowed_formats": options.get("allowed_formats") and ",".join(build_array(options["allowed_formats"])), - "face_coordinates": encode_double_array(options.get("face_coordinates")), - "custom_coordinates": encode_double_array(options.get("custom_coordinates")), - "context": encode_context(options.get("context")), - "moderation": options.get("moderation"), - "raw_convert": options.get("raw_convert"), - "quality_override": options.get("quality_override"), - "ocr": options.get("ocr"), - "categorization": options.get("categorization"), - "detection": options.get("detection"), - "similarity_search": options.get("similarity_search"), - "background_removal": options.get("background_removal"), - "upload_preset": options.get("upload_preset"), - "phash": options.get("phash"), - "return_delete_token": options.get("return_delete_token"), - "auto_tagging": options.get("auto_tagging") and str(options.get("auto_tagging")), - "responsive_breakpoints": generate_responsive_breakpoints_string(options.get("responsive_breakpoints")), - "async": options.get("async"), - "access_control": options.get("access_control") and json_encode(build_list_of_dicts(options.get("access_control")))} + params = { + "timestamp": now(), + "transformation": generate_transformation_string(**options)[0], + "public_id": options.get("public_id"), + "callback": options.get("callback"), + "format": options.get("format"), + "type": options.get("type"), + "backup": options.get("backup"), + "faces": options.get("faces"), + "image_metadata": options.get("image_metadata"), + "exif": options.get("exif"), + "colors": options.get("colors"), + "headers": build_custom_headers(options.get("headers")), + "eager": build_eager(options.get("eager")), + "use_filename": options.get("use_filename"), + "unique_filename": options.get("unique_filename"), + "discard_original_filename": options.get("discard_original_filename"), + "invalidate": options.get("invalidate"), + "notification_url": options.get("notification_url"), + "eager_notification_url": options.get("eager_notification_url"), + "eager_async": options.get("eager_async"), + "proxy": options.get("proxy"), + "folder": options.get("folder"), + "overwrite": options.get("overwrite"), + "tags": options.get("tags") and ",".join(build_array(options["tags"])), + "allowed_formats": options.get("allowed_formats") and ",".join( + build_array(options["allowed_formats"])), + "face_coordinates": encode_double_array(options.get("face_coordinates")), + "custom_coordinates": encode_double_array(options.get("custom_coordinates")), + "context": encode_context(options.get("context")), + "moderation": options.get("moderation"), + "raw_convert": options.get("raw_convert"), + "quality_override": options.get("quality_override"), + "ocr": options.get("ocr"), + "categorization": options.get("categorization"), + "detection": options.get("detection"), + "similarity_search": options.get("similarity_search"), + "background_removal": options.get("background_removal"), + "upload_preset": options.get("upload_preset"), + "phash": options.get("phash"), + "return_delete_token": options.get("return_delete_token"), + "auto_tagging": options.get("auto_tagging") and str(options.get("auto_tagging")), + "responsive_breakpoints": generate_responsive_breakpoints_string( + options.get("responsive_breakpoints")), + "async": options.get("async"), + "access_control": options.get("access_control") and json_encode( + build_list_of_dicts(options.get("access_control"))) + } return params @@ -790,12 +824,12 @@ def process_layer(layer, layer_parameter): if text is not None: var_pattern = VAR_NAME_RE - match = re.findall(var_pattern,text) + match = re.findall(var_pattern, text) - parts= filter(lambda p: p is not None, re.split(var_pattern,text)) + parts = filter(lambda p: p is not None, re.split(var_pattern, text)) encoded_text = [] for part in parts: - if re.match(var_pattern,part): + if re.match(var_pattern, part): encoded_text.append(part) else: encoded_text.append(smart_escape(smart_escape(part, r"([,/])"))) @@ -813,6 +847,7 @@ def process_layer(layer, layer_parameter): return ':'.join(components) + IF_OPERATORS = { "=": 'eq', "!=": 'ne', @@ -843,14 +878,14 @@ def process_layer(layer, layer_parameter): "width": "w" } -replaceRE = "((\\|\\||>=|<=|&&|!=|>|=|<|/|-|\\+|\\*)(?=[ _])|" + '|'.join(PREDEFINED_VARS.keys())+ ")" +replaceRE = "((\\|\\||>=|<=|&&|!=|>|=|<|/|-|\\+|\\*)(?=[ _])|" + '|'.join(PREDEFINED_VARS.keys()) + ")" def translate_if(match): name = match.group(0) return IF_OPERATORS.get(name, PREDEFINED_VARS.get(name, - name)) + name)) def process_conditional(conditional): @@ -861,7 +896,7 @@ def process_conditional(conditional): def normalize_expression(expression): - if re.match(r'^!.+!$',str(expression)): # quoted string + if re.match(r'^!.+!$', str(expression)): # quoted string return expression elif expression: result = str(expression) @@ -913,7 +948,7 @@ def base64_encode_url(url): try: url = unquote(url) - except: + except Exception: pass url = smart_escape(url) b64 = base64.b64encode(url.encode('utf-8')) @@ -929,4 +964,4 @@ def __json_serializer(obj): def is_remote_url(file): """Basic URL scheme check to define if it's remote URL""" - return isinstance(file, string_types) and re.match(REMOTE_URL_RE, file) \ No newline at end of file + return isinstance(file, string_types) and re.match(REMOTE_URL_RE, file) diff --git a/django_tests/admin.py b/django_tests/admin.py index 8c38f3f3..e69de29b 100644 --- a/django_tests/admin.py +++ b/django_tests/admin.py @@ -1,3 +0,0 @@ -from django.contrib import admin - -# Register your models here. diff --git a/django_tests/migrations/0001_initial.py b/django_tests/migrations/0001_initial.py index 663f6b7c..4bea7c4d 100644 --- a/django_tests/migrations/0001_initial.py +++ b/django_tests/migrations/0001_initial.py @@ -3,8 +3,8 @@ from __future__ import unicode_literals import cloudinary.models -from django.db import migrations, models import django.db.models.deletion +from django.db import migrations, models class Migration(migrations.Migration): diff --git a/django_tests/models.py b/django_tests/models.py index 38226a33..4bb266a2 100644 --- a/django_tests/models.py +++ b/django_tests/models.py @@ -1,7 +1,8 @@ -from django.db import models -from cloudinary.models import CloudinaryField from six import python_2_unicode_compatible +from cloudinary.models import CloudinaryField +from django.db import models + @python_2_unicode_compatible class Poll(models.Model): @@ -22,4 +23,3 @@ class Choice(models.Model): def __str__(self): return self.choice.encode() - diff --git a/django_tests/test_cloudinaryField.py b/django_tests/test_cloudinaryField.py index bdebf71b..d3f4d23f 100644 --- a/django_tests/test_cloudinaryField.py +++ b/django_tests/test_cloudinaryField.py @@ -1,15 +1,16 @@ import os import unittest -from django.core.files.uploadedfile import SimpleUploadedFile -from django.test import TestCase from mock import mock from urllib3.util import parse_url import cloudinary -from cloudinary import CloudinaryResource, CloudinaryImage, uploader +from cloudinary import CloudinaryImage, CloudinaryResource, uploader from cloudinary.forms import CloudinaryFileField from cloudinary.models import CloudinaryField +from django.test import TestCase +from django.core.files.uploadedfile import SimpleUploadedFile + from .models import Poll from .test_helper import SUFFIX @@ -21,6 +22,7 @@ class TestCloudinaryField(TestCase): + @classmethod def setUpTestData(cls): Poll.objects.create(question="with image", image="image/upload/v1234/{}.jpg".format(API_TEST_ID)) @@ -98,5 +100,8 @@ def test_image_field(self): field = Poll.objects.get(question="with image") self.assertIsNotNone(field) self.assertEqual(field.image.public_id, API_TEST_ID) - self.assertEqual(parse_url(field.image.url).path, "/{cloud}/image/upload/v1234/{name}.jpg".format(cloud=cloudinary.config().cloud_name, name=API_TEST_ID)) + self.assertEqual( + parse_url(field.image.url).path, + "/{cloud}/image/upload/v1234/{name}.jpg".format(cloud=cloudinary.config().cloud_name, name=API_TEST_ID) + ) self.assertTrue(False or field.image) diff --git a/django_tests/test_cloudinaryResource.py b/django_tests/test_cloudinaryResource.py index d768eb2f..a84b75fb 100644 --- a/django_tests/test_cloudinaryResource.py +++ b/django_tests/test_cloudinaryResource.py @@ -1,12 +1,9 @@ -import random +from urllib3 import disable_warnings import cloudinary -from cloudinary import CloudinaryResource -from cloudinary import uploader, api +from cloudinary import CloudinaryResource, api, uploader from django.test import TestCase -from urllib3 import disable_warnings - from .test_helper import SUFFIX disable_warnings() @@ -42,17 +39,19 @@ def test_validate(self): def test_get_prep_value(self): res = CloudinaryResource(metadata=self.uploaded) - value = "image/upload/v{version}/{id}.{format}".format(version=self.uploaded["version"], - id=self.uploaded["public_id"], - format=self.uploaded["format"]) + value = "image/upload/v{version}/{id}.{format}".format( + version=self.uploaded["version"], + id=self.uploaded["public_id"], + format=self.uploaded["format"]) self.assertEqual(value, res.get_prep_value()) def test_get_presigned(self): res = CloudinaryResource(metadata=self.uploaded) - value = "image/upload/v{version}/{id}.{format}#{signature}".format(version=self.uploaded["version"], - id=self.uploaded["public_id"], - format=self.uploaded["format"], - signature=self.uploaded["signature"]) + value = "image/upload/v{version}/{id}.{format}#{signature}".format( + version=self.uploaded["version"], + id=self.uploaded["public_id"], + format=self.uploaded["format"], + signature=self.uploaded["signature"]) self.assertEqual(value, res.get_presigned()) def test_url(self): @@ -65,6 +64,7 @@ def test_image(self): self.assertRegexpMatches(image, '[^-]src="{url}'.format(url=res.url)) self.assertNotRegexpMatches(image, 'data-src="{url}'.format(url=res.url)) image = res.image(responsive=True, width="auto", crop="scale") - self.assertNotRegexpMatches(image, '[^-]src="{url}'.format(url=res.build_url(width="auto", crop="scale"))) - self.assertRegexpMatches(image, 'data-src="{url}'.format(url=res.build_url(width="auto", crop="scale"))) - + self.assertNotRegexpMatches(image, '[^-]src="{url}'.format( + url=res.build_url(width="auto", crop="scale"))) + self.assertRegexpMatches(image, 'data-src="{url}'.format( + url=res.build_url(width="auto", crop="scale"))) diff --git a/django_tests/test_user_agent.py b/django_tests/test_user_agent.py index bad252c0..ea912c19 100644 --- a/django_tests/test_user_agent.py +++ b/django_tests/test_user_agent.py @@ -8,4 +8,4 @@ class TestUserAgent(TestCase): def test_django_user_agent(self): agent = cloudinary.get_user_agent() - six.assertRegex(self, agent, '^Django\/\d\.\d+\.\d+ CloudinaryPython\/\d\.\d+\.\d+ \(Python \d\.\d+\.\d+\)$') + six.assertRegex(self, agent, r'^Django\/\d\.\d+\.\d+ CloudinaryPython\/\d\.\d+\.\d+ \(Python \d\.\d+\.\d+\)$') diff --git a/django_tests/tests.py b/django_tests/tests.py index 7ce503c2..e69de29b 100644 --- a/django_tests/tests.py +++ b/django_tests/tests.py @@ -1,3 +0,0 @@ -from django.test import TestCase - -# Create your tests here. diff --git a/django_tests/urls.py b/django_tests/urls.py index 37d91e43..6b45e4df 100644 --- a/django_tests/urls.py +++ b/django_tests/urls.py @@ -1,4 +1,5 @@ from django.conf.urls import url + from .views import index urlpatterns = [ diff --git a/samples/basic/basic.py b/samples/basic/basic.py index 2d958319..b26e2333 100644 --- a/samples/basic/basic.py +++ b/samples/basic/basic.py @@ -1,9 +1,10 @@ #!/usr/bin/env python -import os, sys +import os +import sys +from cloudinary.api import delete_resources_by_tag, resources_by_tag from cloudinary.uploader import upload from cloudinary.utils import cloudinary_url -from cloudinary.api import delete_resources_by_tag, resources_by_tag # config os.chdir(os.path.join(os.path.dirname(sys.argv[0]), '.')) @@ -12,95 +13,107 @@ DEFAULT_TAG = "python_sample_basic" + def dump_response(response): print("Upload response:") for key in sorted(response.keys()): print(" %s: %s" % (key, response[key])) + def upload_files(): print("--- Upload a local file") - response = upload("pizza.jpg", tags = DEFAULT_TAG) + response = upload("pizza.jpg", tags=DEFAULT_TAG) dump_response(response) - url, options = cloudinary_url(response['public_id'], - format = response['format'], - width = 200, - height = 150, - crop = "fill" + url, options = cloudinary_url( + response['public_id'], + format=response['format'], + width=200, + height=150, + crop="fill" ) print("Fill 200x150 url: " + url) print("") print("--- Upload a local file with custom public ID") - response = upload("pizza.jpg", - tags = DEFAULT_TAG, - public_id = "custom_name", + response = upload( + "pizza.jpg", + tags=DEFAULT_TAG, + public_id="custom_name", ) dump_response(response) - url, options = cloudinary_url(response['public_id'], - format = response['format'], - width = 200, - height = 150, - crop = "fit" + url, options = cloudinary_url( + response['public_id'], + format=response['format'], + width=200, + height=150, + crop="fit" ) print("Fit into 200x150 url: " + url) print("") print("--- Upload a local file with eager transformation of scaling to 200x150") - response = upload("lake.jpg", - tags = DEFAULT_TAG, - public_id = "eager_custom_name", - eager = dict( - width = 200, - height = 150, - crop = "scale" + response = upload( + "lake.jpg", + tags=DEFAULT_TAG, + public_id="eager_custom_name", + eager=dict( + width=200, + height=150, + crop="scale" ), ) dump_response(response) - url, options = cloudinary_url(response['public_id'], - format = response['format'], - width = 200, - height = 150, - crop = "scale", + url, options = cloudinary_url( + response['public_id'], + format=response['format'], + width=200, + height=150, + crop="scale", ) print("scaling to 200x150 url: " + url) print("") print("--- Upload by fetching a remote image") - response = upload("http://res.cloudinary.com/demo/image/upload/couple.jpg", - tags = DEFAULT_TAG, + response = upload( + "http://res.cloudinary.com/demo/image/upload/couple.jpg", + tags=DEFAULT_TAG ) dump_response(response) - url, options = cloudinary_url(response['public_id'], - format = response['format'], - width = 200, - height = 150, - crop = "thumb", - gravity = "faces", + url, options = cloudinary_url( + response['public_id'], + format=response['format'], + width=200, + height=150, + crop="thumb", + gravity="faces", ) print("Face detection based 200x150 thumbnail url: " + url) print("") print("--- Fetch an uploaded remote image, fitting it into 500x500 and reducing saturation") - response = upload("http://res.cloudinary.com/demo/image/upload/couple.jpg", - tags = DEFAULT_TAG, - width = 500, - height = 500, - crop = "fit", - effect = "saturation:-70", + response = upload( + "http://res.cloudinary.com/demo/image/upload/couple.jpg", + tags=DEFAULT_TAG, + width=500, + height=500, + crop="fit", + effect="saturation:-70", ) dump_response(response) - url, options = cloudinary_url(response['public_id'], - format = response['format'], - width = 200, - height = 150, - crop = "fill", - gravity = "faces", - radius = 10, - effect = "sepia", + url, options = cloudinary_url( + response['public_id'], + format=response['format'], + width=200, + height=150, + crop="fill", + gravity="faces", + radius=10, + effect="sepia", ) print("Fill 200x150, round corners, apply the sepia effect, url: " + url) print("") + def cleanup(): response = resources_by_tag(DEFAULT_TAG) resources = response.get('resources', []) @@ -110,11 +123,13 @@ def cleanup(): print("Deleting {0:d} images...".format(len(resources))) delete_resources_by_tag(DEFAULT_TAG) print("Done!") - pass + if len(sys.argv) > 1: - if sys.argv[1] == 'upload': upload_files() - if sys.argv[1] == 'cleanup': cleanup() + if sys.argv[1] == 'upload': + upload_files() + if sys.argv[1] == 'cleanup': + cleanup() else: print("--- Uploading files and then cleaning up") print(" you can only one instead by passing 'upload' or 'cleanup' as an argument") diff --git a/samples/basic_flask/app.py b/samples/basic_flask/app.py index b5140321..301e1769 100644 --- a/samples/basic_flask/app.py +++ b/samples/basic_flask/app.py @@ -1,7 +1,6 @@ -import os -from flask import Flask, request, render_template from cloudinary.uploader import upload from cloudinary.utils import cloudinary_url +from flask import Flask, render_template, request app = Flask(__name__) diff --git a/samples/gae/main.py b/samples/gae/main.py index 24a12eb0..aaea3c9e 100644 --- a/samples/gae/main.py +++ b/samples/gae/main.py @@ -14,12 +14,13 @@ # See the License for the specific language governing permissions and # limitations under the License. # -import webapp2 import os -from cloudinary.compat import StringIO -from google.appengine.ext.webapp import template + +import webapp2 +from cloudinary.compat import StringIO from cloudinary.uploader import upload from cloudinary.utils import cloudinary_url +from google.appengine.ext.webapp import template class MainHandler(webapp2.RequestHandler): @@ -54,6 +55,7 @@ def post(self): path = os.path.join(os.path.dirname(__file__), 'index.html') self.response.write(template.render(path, template_values)) + app = webapp2.WSGIApplication([ ('/', MainHandler) ], debug=True) diff --git a/setup.py b/setup.py index c521eba9..d684fbfb 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,4 @@ -from setuptools import setup, find_packages +from setuptools import find_packages, setup version = '1.12.0' diff --git a/tests/api_test.py b/tests/api_test.py index a17968a2..b578979e 100644 --- a/tests/api_test.py +++ b/tests/api_test.py @@ -2,20 +2,21 @@ import unittest from collections import OrderedDict +import six from mock import patch +from urllib3 import HTTPResponse, disable_warnings from urllib3._collections import HTTPHeaderDict import cloudinary -import six -from cloudinary import uploader, api, utils - -from urllib3 import disable_warnings, HTTPResponse +from cloudinary import api, uploader, utils +from tests.test_helper import * -from .test_helper import * - -MOCK_HEADERS = HTTPHeaderDict({"x-featureratelimit-limit": '0', "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT', - "x-featureratelimit-remaining": '0', }) +MOCK_HEADERS = HTTPHeaderDict({ + "x-featureratelimit-limit": '0', + "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT', + "x-featureratelimit-remaining": '0', +}) if six.PY2: MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}', headers=MOCK_HEADERS) @@ -43,6 +44,7 @@ MAPPING_TEST_ID = "api_test_upload_mapping_{}".format(SUFFIX) RESTORE_TEST_ID = "api_test_restore_{}".format(SUFFIX) + class ApiTest(unittest.TestCase): @classmethod def setUpClass(cls): @@ -68,7 +70,9 @@ def tearDownClass(cls): except Exception: pass presets_response = api.upload_presets(max_results=200) - preset_names = [ preset["name"] for preset in presets_response.get('presets',[]) if UNIQUE_TAG in preset.get('settings',{}).get('tags','')] + preset_names = [ + preset["name"] for preset in presets_response.get('presets', []) + if UNIQUE_TAG in preset.get('settings', {}).get('tags', '')] for name in preset_names: try: api.delete_upload_preset(name) @@ -97,7 +101,6 @@ def test02_resources(self, mocker): api.resources() args, kargs = mocker.call_args self.assertTrue(get_uri(args).endswith('/resources/image')) - @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test03_resources_cursor(self): @@ -212,19 +215,17 @@ def test08a_delete_derived_by_transformation(self, mocker): mocker.return_value = MOCK_RESPONSE api.delete_derived_by_transformation( - [public_resource_id, public_resource_id2], - [transformation, transformation2], resource_type='raw', type='fetch', invalidate=True, foo='bar') + [public_resource_id, public_resource_id2], [transformation, transformation2], + resource_type='raw', type='fetch', invalidate=True, foo='bar') method, url, params = mocker.call_args[0][0:3] self.assertEqual('DELETE', method) self.assertTrue(url.endswith('/resources/raw/fetch')) self.assertIn(public_resource_id, get_list_param(mocker, 'public_ids')) self.assertIn(public_resource_id2, get_list_param(mocker, 'public_ids')) - self.assertEqual(get_param(mocker, 'transformations'), - utils.build_eager([transformation, transformation2])) + self.assertEqual(get_param(mocker, 'transformations'), utils.build_eager([transformation, transformation2])) self.assertTrue(get_param(mocker, 'keep_original')) self.assertTrue(get_param(mocker, 'invalidate')) - @patch('urllib3.request.RequestMethods.request') @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test09_delete_resources(self, mocker): @@ -550,11 +551,11 @@ def test_folder_listing(self): def test_CloudinaryImage_len(self): """Tests the __len__ function on CloudinaryImage""" metadata = { - "public_id": "test_id", - "format": "tst", - "version": "1234", - "signature": "5678", - } + "public_id": "test_id", + "format": "tst", + "version": "1234", + "signature": "5678", + } my_cloudinary_image = cloudinary.CloudinaryImage(metadata=metadata) self.assertEqual(len(my_cloudinary_image), len(metadata["public_id"])) diff --git a/tests/archive_test.py b/tests/archive_test.py index ac6f4a9f..a566adfe 100644 --- a/tests/archive_test.py +++ b/tests/archive_test.py @@ -1,4 +1,3 @@ -import random import tempfile import time import unittest @@ -8,7 +7,7 @@ import cloudinary import cloudinary.poster.streaminghttp -from cloudinary import uploader, utils, api +from cloudinary import api, uploader, utils from mock import patch import six @@ -16,7 +15,7 @@ from urllib3 import disable_warnings, HTTPResponse from urllib3._collections import HTTPHeaderDict -from .test_helper import SUFFIX, TEST_TAG +from .test_helper import SUFFIX disable_warnings() @@ -51,7 +50,8 @@ def test_create_archive(self): """should successfully generate an archive""" result = uploader.create_archive(tags=[TEST_TAG]) self.assertEqual(2, result.get("file_count")) - result2 = uploader.create_zip(tags=[TEST_TAG], transformations=[{"width": 0.5}, {"width": 2.0}]) + result2 = uploader.create_zip( + tags=[TEST_TAG], transformations=[{"width": 0.5}, {"width": 2.0}]) self.assertEqual(4, result2.get("file_count")) @patch('urllib3.request.RequestMethods.request') @@ -90,7 +90,8 @@ def test_archive_url(self): @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test_download_zip_url_options(self): - result = utils.download_zip_url(tags=[TEST_TAG], transformations=[{"width": 0.5}, {"width": 2.0}], cloud_name="demo") + result = utils.download_zip_url(tags=[TEST_TAG], transformations=[{"width": 0.5}, {"width": 2.0}], + cloud_name="demo") upload_prefix = cloudinary.config().upload_prefix or "https://api.cloudinary.com" six.assertRegex(self, result, r'^{0}/v1_1/demo/.*$'.format(upload_prefix)) diff --git a/tests/auth_token_test.py b/tests/auth_token_test.py index fd595dcc..1831e6c7 100644 --- a/tests/auth_token_test.py +++ b/tests/auth_token_test.py @@ -12,7 +12,7 @@ def setUp(self): self.url_backup = os.environ.get("CLOUDINARY_URL") os.environ["CLOUDINARY_URL"] = "cloudinary://a:b@test123" cloudinary.reset_config() - cloudinary.config(auth_token={"key":KEY, "duration": 300, "start_time": 11111111}) + cloudinary.config(auth_token={"key": KEY, "duration": 300, "start_time": 11111111}) def tearDown(self): os.environ["CLOUDINARY_URL"] = self.url_backup @@ -21,56 +21,70 @@ def tearDown(self): def test_generate_with_start_time_and_duration(self): token = cloudinary.utils.generate_auth_token(acl='/image/*', start_time=1111111111, duration=300) self.assertEqual('__cld_token__=st=1111111111~exp=1111111411~acl=%2fimage%2f%2a~hmac' - '=0d5b0c9c1485ee162c459879fe62e06caa23bc26fec92d58bd100f2e1592eac6', token) + '=0d5b0c9c1485ee162c459879fe62e06caa23bc26fec92d58bd100f2e1592eac6', token) - def test_should_add_token_if_authToken_is_globally_set_and_signed_is_True(self): cloudinary.config(private_cdn=True) - url, _ = cloudinary.utils.cloudinary_url( "sample.jpg", sign_url=True, resource_type="image", type="authenticated", version="1486020273") + url, _ = cloudinary.utils.cloudinary_url("sample.jpg", sign_url=True, resource_type="image", + type="authenticated", version="1486020273") self.assertEqual(url, "http://test123-res.cloudinary.com/image/authenticated/v1486020273/sample.jpg" "?__cld_token__=st=11111111~exp=11111411~hmac" - "=8db0d753ee7bbb9e2eaf8698ca3797436ba4c20e31f44527e43b6a6e995cfdb3") + "=8db0d753ee7bbb9e2eaf8698ca3797436ba4c20e31f44527e43b6a6e995cfdb3") def test_should_add_token_for_public_resource(self): cloudinary.config(private_cdn=True) - url, _ = cloudinary.utils.cloudinary_url( "sample.jpg", sign_url=True, resource_type="image", type="public", version="1486020273") + url, _ = cloudinary.utils.cloudinary_url("sample.jpg", sign_url=True, resource_type="image", type="public", + version="1486020273") self.assertEqual(url, "http://test123-res.cloudinary.com/image/public/v1486020273/sample.jpg?__cld_token__=st" - "=11111111~exp=11111411~hmac=c2b77d9f81be6d89b5d0ebc67b671557e88a40bcf03dd4a6997ff4b994ceb80e") + "=11111111~exp=11111411~hmac=" + "c2b77d9f81be6d89b5d0ebc67b671557e88a40bcf03dd4a6997ff4b994ceb80e") def test_should_not_add_token_if_signed_is_false(self): cloudinary.config(private_cdn=True) - url, _ = cloudinary.utils.cloudinary_url( "sample.jpg", type="authenticated", version="1486020273") + url, _ = cloudinary.utils.cloudinary_url("sample.jpg", type="authenticated", version="1486020273") self.assertEqual(url, "http://test123-res.cloudinary.com/image/authenticated/v1486020273/sample.jpg") def test_null_token(self): cloudinary.config(private_cdn=True) - url, _ = cloudinary.utils.cloudinary_url( "sample.jpg", auth_token=False, sign_url=True, type="authenticated", version="1486020273") - self.assertEqual(url, "http://test123-res.cloudinary.com/image/authenticated/s--v2fTPYTu--/v1486020273/sample.jpg") + url, _ = cloudinary.utils.cloudinary_url("sample.jpg", auth_token=False, sign_url=True, type="authenticated", + version="1486020273") + self.assertEqual( + url, + "http://test123-res.cloudinary.com/image/authenticated/s--v2fTPYTu--/v1486020273/sample.jpg" + ) def test_explicit_authToken_should_override_global_setting(self): cloudinary.config(private_cdn=True) - url, _ = cloudinary.utils.cloudinary_url( "sample.jpg", sign_url=True, auth_token={ "key": ALT_KEY, "start_time": 222222222, "duration": 100 }, type="authenticated", transformation={ "crop": "scale", "width": 300 }) + url, _ = cloudinary.utils.cloudinary_url("sample.jpg", sign_url=True, + auth_token={"key": ALT_KEY, "start_time": 222222222, "duration": 100}, + type="authenticated", transformation={"crop": "scale", "width": 300}) self.assertEqual(url, "http://test123-res.cloudinary.com/image/authenticated/c_scale," "w_300/sample.jpg?__cld_token__=st=222222222~exp=222222322~hmac" - "=7d276841d70c4ecbd0708275cd6a82e1f08e47838fbb0bceb2538e06ddfa3029") + "=7d276841d70c4ecbd0708275cd6a82e1f08e47838fbb0bceb2538e06ddfa3029") def test_should_compute_expiration_as_start_time_plus_duration(self): cloudinary.config(private_cdn=True) - token = { "key": KEY, "start_time": 11111111, "duration": 300 } - url, _ = cloudinary.utils.cloudinary_url( "sample.jpg", sign_url=True, auth_token=token, resource_type="image", type="authenticated", version="1486020273") + token = {"key": KEY, "start_time": 11111111, "duration": 300} + url, _ = cloudinary.utils.cloudinary_url("sample.jpg", sign_url=True, auth_token=token, resource_type="image", + type="authenticated", version="1486020273") self.assertEqual(url, "http://test123-res.cloudinary.com/image/authenticated/v1486020273/sample.jpg" "?__cld_token__=st=11111111~exp=11111411~hmac" - "=8db0d753ee7bbb9e2eaf8698ca3797436ba4c20e31f44527e43b6a6e995cfdb3") + "=8db0d753ee7bbb9e2eaf8698ca3797436ba4c20e31f44527e43b6a6e995cfdb3") def test_generate_token_string(self): - user = "foobar" # we can't rely on the default "now" value in tests - tokenOptions = { "key": KEY, "duration": 300, "acl": "/*/t_%s" % user } - tokenOptions["start_time"] = 222222222 # we can't rely on the default "now" value in tests - cookieToken = cloudinary.utils.generate_auth_token( **tokenOptions) - self.assertEqual(cookieToken, "__cld_token__=st=222222222~exp=222222522~acl=%2f%2a%2ft_foobar~hmac=1284376353c1c43d6f6a98f2813c5596f4ff6f34d837cd853fd8c3c9e7f8428c") + user = "foobar" # we can't rely on the default "now" value in tests + token_options = {"key": KEY, "duration": 300, "acl": "/*/t_%s" % user} + token_options["start_time"] = 222222222 # we can't rely on the default "now" value in tests + cookie_token = cloudinary.utils.generate_auth_token(**token_options) + self.assertEqual( + cookie_token, + "__cld_token__=st=222222222~exp=222222522~acl=%2f%2a%2ft_foobar~hmac=" + "1284376353c1c43d6f6a98f2813c5596f4ff6f34d837cd853fd8c3c9e7f8428c" + ) def test_must_provide_expiration_or_duration(self): self.assertRaises(Exception, cloudinary.utils.generate_auth_token, acl="*", expiration=None, duration=None) + if __name__ == '__main__': unittest.main() diff --git a/tests/config_test.py b/tests/config_test.py index 75de2002..2415296d 100644 --- a/tests/config_test.py +++ b/tests/config_test.py @@ -1,4 +1,5 @@ from unittest import TestCase + import cloudinary diff --git a/tests/search_test.py b/tests/search_test.py index b068a77a..8aed6be7 100644 --- a/tests/search_test.py +++ b/tests/search_test.py @@ -6,14 +6,16 @@ from urllib3 import disable_warnings import cloudinary -from cloudinary import uploader, api, logger, Search -from tests.test_helper import TEST_IMAGE, TEST_TAG, UNIQUE_TAG, SUFFIX +from cloudinary import api, logger, uploader +from cloudinary.search import Search +from tests.test_helper import SUFFIX, TEST_IMAGE, TEST_TAG, UNIQUE_TAG disable_warnings() TEST_TAG = 'search_{}'.format(TEST_TAG) UNIQUE_TAG = 'search_{}'.format(UNIQUE_TAG) + TEST_IMAGES_COUNT = 3 MAX_INDEX_RETRIES = 10 @@ -26,7 +28,8 @@ class SearchTest(unittest.TestCase): def setUpClass(cls): cls.ready = False cloudinary.reset_config() - if not cloudinary.config().api_secret: return + if not cloudinary.config().api_secret: + return for public_id in public_ids: res = uploader.upload(TEST_IMAGE, public_id=public_id, @@ -34,7 +37,6 @@ def setUpClass(cls): context="stage=value", eager=[{"width": 100, "crop": "scale"}]) upload_results.append(res) - attempt = 0 while attempt < MAX_INDEX_RETRIES: time.sleep(1) @@ -135,11 +137,11 @@ def test_should_paginate_resources_limited_by_tag_and_ordered_by_ascending_publi .next_cursor(results['next_cursor']) \ .execute() self.assertEqual(len(results['resources']), 1) - self.assertEqual(results['resources'][0]['public_id'], - public_ids[i], - "{0} found public_id {1} instead of {2} ".format(i, - results['resources'][0]['public_id'], - public_ids[i])) + self.assertEqual( + results['resources'][0]['public_id'], + public_ids[i], + "{0} found public_id {1} instead of {2} ".format( + i, results['resources'][0]['public_id'], public_ids[i])) self.assertEqual(results['total_count'], TEST_IMAGES_COUNT) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -159,11 +161,11 @@ def test_should_include_context(self): "Use env variable RUN_SEARCH_TESTS=1 if you really want to test it.") def test_should_include_context_tags_and_image_metadata(self): - results = Search().expression("tags={0}".format(UNIQUE_TAG)).with_field('context').with_field('tags').\ + results = Search().expression("tags={0}".format(UNIQUE_TAG)).\ + with_field('context').with_field('tags').\ with_field('image_metadata').execute() self.assertEqual(len(results['resources']), TEST_IMAGES_COUNT) - for res in results['resources']: self.assertEqual([key for key in iterkeys(res['context'])], [u'stage']) self.assertTrue('image_metadata' in res) diff --git a/tests/streaming_profiles_test.py b/tests/streaming_profiles_test.py index 08d0832d..f1067d51 100644 --- a/tests/streaming_profiles_test.py +++ b/tests/streaming_profiles_test.py @@ -1,8 +1,6 @@ import unittest -import time import cloudinary -from cloudinary import uploader, api, utils -import six +from cloudinary import api from urllib3 import disable_warnings from .test_helper import SUFFIX @@ -14,10 +12,12 @@ class StreamingProfilesTest(unittest.TestCase): initialized = False def setUp(self): - if StreamingProfilesTest.initialized: return + if StreamingProfilesTest.initialized: + return StreamingProfilesTest.initialized = True cloudinary.reset_config() - if not cloudinary.config().api_secret: return + if not cloudinary.config().api_secret: + return StreamingProfilesTest.test_id = "api_test_{}".format(SUFFIX) __predefined_sp = ["4k", "full_hd", "hd", "sd", "full_hd_wifi", "full_hd_lean", "hd_lean"] @@ -28,12 +28,14 @@ def test_create_streaming_profile(self): name = StreamingProfilesTest.test_id + "_streaming_profile" result = api.create_streaming_profile( name, - representations=[ - {"transformation": {"bit_rate": "5m", "height": 1200, "width": 1200, "crop": "limit"}}]) + representations=[{"transformation": { + "bit_rate": "5m", "height": 1200, "width": 1200, "crop": "limit" + }}]) self.assertIn("representations", result["data"]) reps = result["data"]["representations"] self.assertIsInstance(reps, list) - """should return transformation as an array""" + + # should return transformation as an array self.assertIsInstance(reps[0]["transformation"], list) tr = reps[0]["transformation"][0] @@ -46,7 +48,7 @@ def test_list_streaming_profiles(self): result = api.list_streaming_profiles() names = [sp["name"] for sp in result["data"]] self.assertTrue(len(names) >= len(self.__predefined_sp)) - """streaming profiles should include the predefined profiles""" + # streaming profiles should include the predefined profiles for name in self.__predefined_sp: self.assertIn(name, names) @@ -69,16 +71,18 @@ def test_update_delete_streaming_profile(self): name = StreamingProfilesTest.test_id + "_streaming_profile_delete" api.create_streaming_profile( name, - representations=[ - {"transformation": {"bit_rate": "5m", "height": 1200, "width": 1200, "crop": "limit"}}]) + representations=[{"transformation": { + "bit_rate": "5m", "height": 1200, "width": 1200, "crop": "limit" + }}]) result = api.update_streaming_profile( name, - representations=[ - {"transformation": {"bit_rate": "5m", "height": 1000, "width": 1000, "crop": "scale"}}]) + representations=[{"transformation": { + "bit_rate": "5m", "height": 1000, "width": 1000, "crop": "scale" + }}]) self.assertIn("representations", result["data"]) reps = result["data"]["representations"] self.assertIsInstance(reps, list) - """transformation is returned as an array""" + # transformation is returned as an array self.assertIsInstance(reps[0]["transformation"], list) tr = reps[0]["transformation"][0] diff --git a/tests/test_helper.py b/tests/test_helper.py index 8e7d5a89..aa3afee0 100644 --- a/tests/test_helper.py +++ b/tests/test_helper.py @@ -12,7 +12,6 @@ ZERO = timedelta(0) -# A UTC class. class UTC(tzinfo): """UTC""" diff --git a/tests/uploader_test.py b/tests/uploader_test.py index fd0ef996..b40f17e6 100644 --- a/tests/uploader_test.py +++ b/tests/uploader_test.py @@ -6,11 +6,11 @@ import six from mock import patch +import cloudinary +from cloudinary import api, uploader, utils + from urllib3 import disable_warnings, HTTPResponse from urllib3.util import parse_url - -import cloudinary -from cloudinary import uploader, utils, api from tests.test_helper import * if six.PY2: @@ -46,8 +46,9 @@ def test_upload(self): result = uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG]) self.assertEqual(result["width"], TEST_IMAGE_WIDTH) self.assertEqual(result["height"], TEST_IMAGE_HEIGHT) - expected_signature = utils.api_sign_request(dict(public_id=result["public_id"], version=result["version"]), - cloudinary.config().api_secret) + expected_signature = utils.api_sign_request( + dict(public_id=result["public_id"], version=result["version"]), + cloudinary.config().api_secret) self.assertEqual(result["signature"], expected_signature) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -67,7 +68,7 @@ def test_upload_file_io_without_filename(self): def test_upload_async(self, mocker): """should pass async value """ mocker.return_value = MOCK_RESPONSE - result = uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG], async=True) + uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG], async=True) params = mocker.call_args[0][2] self.assertTrue(params['async']) @@ -76,7 +77,7 @@ def test_upload_async(self, mocker): def test_ocr(self, mocker): """should pass ocr value """ mocker.return_value = MOCK_RESPONSE - result = uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG], ocr='adv_ocr') + uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG], ocr='adv_ocr') args, kargs = mocker.call_args self.assertEqual(get_params(args)['ocr'], 'adv_ocr') @@ -101,8 +102,9 @@ def test_upload_url(self): result = uploader.upload("http://cloudinary.com/images/old_logo.png", tags=[UNIQUE_TAG]) self.assertEqual(result["width"], TEST_IMAGE_WIDTH) self.assertEqual(result["height"], TEST_IMAGE_HEIGHT) - expected_signature = utils.api_sign_request(dict(public_id=result["public_id"], version=result["version"]), - cloudinary.config().api_secret) + expected_signature = utils.api_sign_request( + dict(public_id=result["public_id"], version=result["version"]), + cloudinary.config().api_secret) self.assertEqual(result["signature"], expected_signature) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -111,8 +113,9 @@ def test_upload_unicode_url(self): result = uploader.upload(u"http://cloudinary.com/images/old_logo.png", tags=[UNIQUE_TAG]) self.assertEqual(result["width"], TEST_IMAGE_WIDTH) self.assertEqual(result["height"], TEST_IMAGE_HEIGHT) - expected_signature = utils.api_sign_request(dict(public_id=result["public_id"], version=result["version"]), - cloudinary.config().api_secret) + expected_signature = utils.api_sign_request( + dict(public_id=result["public_id"], version=result["version"]), + cloudinary.config().api_secret) self.assertEqual(result["signature"], expected_signature) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -128,8 +131,9 @@ def test_upload_data_uri(self): tags=[UNIQUE_TAG]) self.assertEqual(result["width"], 16) self.assertEqual(result["height"], 16) - expected_signature = utils.api_sign_request(dict(public_id=result["public_id"], version=result["version"]), - cloudinary.config().api_secret) + expected_signature = utils.api_sign_request( + dict(public_id=result["public_id"], version=result["version"]), + cloudinary.config().api_secret) self.assertEqual(result["signature"], expected_signature) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -139,7 +143,8 @@ def test_rename(self): uploader.rename(result["public_id"], result["public_id"]+"2") self.assertIsNotNone(api.resource(result["public_id"]+"2")) result2 = uploader.upload("tests/favicon.ico", tags=[UNIQUE_TAG]) - self.assertRaises(api.Error, uploader.rename, result2["public_id"], result["public_id"]+"2") + self.assertRaises(api.Error, uploader.rename, + result2["public_id"], result["public_id"]+"2") uploader.rename(result2["public_id"], result["public_id"]+"2", overwrite=True) self.assertEqual(api.resource(result["public_id"]+"2")["format"], "ico") @@ -148,7 +153,8 @@ def test_use_filename(self): """should successfully take use file name of uploaded file in public id if specified use_filename """ result = uploader.upload(TEST_IMAGE, use_filename=True, tags=[UNIQUE_TAG]) six.assertRegex(self, result["public_id"], 'logo_[a-z0-9]{6}') - result = uploader.upload(TEST_IMAGE, use_filename=True, unique_filename=False, tags=[UNIQUE_TAG]) + result = uploader.upload( + TEST_IMAGE, use_filename=True, unique_filename=False, tags=[UNIQUE_TAG]) self.assertEqual(result["public_id"], 'logo') @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -156,8 +162,8 @@ def test_explicit(self): """should support explicit """ result = uploader.explicit("cloudinary", type="twitter_name", eager=[dict(crop="scale", width="2.0", format="png")], tags=[UNIQUE_TAG]) - url = utils.cloudinary_url("cloudinary", type="twitter_name", crop="scale", width="2.0", format="png", - version=result["version"])[0] + url = utils.cloudinary_url("cloudinary", type="twitter_name", crop="scale", width="2.0", + format="png", version=result["version"])[0] actual = result["eager"][0]["url"] self.assertEqual(parse_url(actual).path, parse_url(url).path) @@ -229,7 +235,8 @@ def test_face_coordinates(self): """should allow sending face coordinates""" coordinates = [[120, 30, 109, 150], [121, 31, 110, 151]] result_coordinates = [[120, 30, 109, 51], [121, 31, 110, 51]] - result = uploader.upload(TEST_IMAGE, face_coordinates=coordinates, faces=True, tags=[UNIQUE_TAG]) + result = uploader.upload(TEST_IMAGE, face_coordinates=coordinates, + faces=True, tags=[UNIQUE_TAG]) self.assertEqual(result_coordinates, result["faces"]) different_coordinates = [[122, 32, 111, 152]] @@ -294,7 +301,7 @@ def test_categorization(self): @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test_detection(self): """ should support requesting detection """ - with six.assertRaisesRegex(self, api.Error, 'illegal is not a valid'): + with six.assertRaisesRegex(self, api.Error, 'illegal is not a valid'): uploader.upload(TEST_IMAGE, detection="illegal", tags=[UNIQUE_TAG]) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -318,7 +325,7 @@ def test_upload_large(self): self.assertEqual(resource["resource_type"], "raw") resource2 = uploader.upload_large(temp_file_name, chunk_size=5243000, tags=["upload_large_tag", UNIQUE_TAG], - resource_type="image") + resource_type="image") self.assertEqual(resource2["tags"], ["upload_large_tag", UNIQUE_TAG]) self.assertEqual(resource2["resource_type"], "image") self.assertEqual(resource2["width"], 1400) @@ -332,7 +339,8 @@ def test_upload_large(self): resource4 = uploader.upload_large(REMOTE_TEST_IMAGE, tags=[UNIQUE_TAG]) self.assertEqual(resource4["width"], TEST_IMAGE_WIDTH) self.assertEqual(resource4["height"], TEST_IMAGE_HEIGHT) - expected_signature = utils.api_sign_request(dict(public_id=resource4["public_id"], version=resource4["version"]), + expected_signature = utils.api_sign_request(dict(public_id=resource4["public_id"], + version=resource4["version"]), cloudinary.config().api_secret) self.assertEqual(resource4["signature"], expected_signature) @@ -343,7 +351,7 @@ def test_upload_preset(self): """ should support unsigned uploading using presets """ preset = api.create_upload_preset(folder="upload_folder", unsigned=True, tags=[UNIQUE_TAG]) result = uploader.unsigned_upload(TEST_IMAGE, preset["name"], tags=[UNIQUE_TAG]) - six.assertRegex(self, result["public_id"], '^upload_folder\/[a-z0-9]+$') + six.assertRegex(self, result["public_id"], r'^upload_folder\/[a-z0-9]+$') api.delete_upload_preset(preset["name"]) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -403,8 +411,8 @@ def test_access_control(self, request_mock): # Should accept a dictionary of datetime objects acl_2 = OrderedDict((("access_type", "anonymous"), - ("start", datetime.strptime("2019-02-22 16:20:57Z", "%Y-%m-%d %H:%M:%SZ")), - ("end", datetime(2019, 3, 22, 0, 0, tzinfo=UTC())))) + ("start", datetime.strptime("2019-02-22 16:20:57Z", "%Y-%m-%d %H:%M:%SZ")), + ("end", datetime(2019, 3, 22, 0, 0, tzinfo=UTC())))) exp_acl_2 = '[{"access_type":"anonymous","start":"2019-02-22T16:20:57","end":"2019-03-22T00:00:00+00:00"}]' diff --git a/tests/utils_test.py b/tests/utils_test.py index 55f1e230..7b330299 100644 --- a/tests/utils_test.py +++ b/tests/utils_test.py @@ -1,4 +1,5 @@ # -*- coding: utf-8 -*- + import re import unittest from collections import OrderedDict @@ -12,6 +13,7 @@ from cloudinary.utils import build_list_of_dicts, json_encode from tests.test_helper import TEST_IMAGE, REMOTE_TEST_IMAGE + DEFAULT_ROOT_PATH = 'http://res.cloudinary.com/test123/' DEFAULT_UPLOAD_PATH = 'http://res.cloudinary.com/test123/image/upload/' VIDEO_UPLOAD_PATH = 'http://res.cloudinary.com/test123/video/upload/' @@ -151,15 +153,15 @@ def test_should_support_auto_value(self): options={"x": 1, "y": 2, "radius": 3, "gravity": "center", "quality": "auto:good", "prefix": "a"}, expected_url=DEFAULT_UPLOAD_PATH + "g_center,p_a,q_auto:good,r_3,x_1,y_2/test") self.__test_cloudinary_url( - options={"width":100, "height":100, "crop":'crop', "gravity":"auto:ocr_text"}, + options={"width": 100, "height": 100, "crop": 'crop', "gravity": "auto:ocr_text"}, expected_url=DEFAULT_UPLOAD_PATH + "c_crop,g_auto:ocr_text,h_100,w_100/test", expected_options={"width": 100, "height": 100}) self.__test_cloudinary_url( - options={"width":100, "height":100, "crop":'crop', "gravity":"ocr_text"}, + options={"width": 100, "height": 100, "crop": 'crop', "gravity": "ocr_text"}, expected_url=DEFAULT_UPLOAD_PATH + "c_crop,g_ocr_text,h_100,w_100/test", expected_options={"width": 100, "height": 100}) self.__test_cloudinary_url( - options={"width":100, "height":100, "crop":'crop', "gravity":"ocr_text:adv_ocr"}, + options={"width": 100, "height": 100, "crop": 'crop', "gravity": "ocr_text:adv_ocr"}, expected_url=DEFAULT_UPLOAD_PATH + "c_crop,g_ocr_text:adv_ocr,h_100,w_100/test", expected_options={"width": 100, "height": 100}) @@ -283,7 +285,7 @@ def test_fetch_overlay(self): self.__test_cloudinary_url( options={"overlay": "fetch:http://cloudinary.com/images/old_logo.png"}, expected_url=( - DEFAULT_UPLOAD_PATH + DEFAULT_UPLOAD_PATH + "l_fetch:aHR0cDovL2Nsb3VkaW5hcnkuY29tL2ltYWdlcy9vbGRfbG9nby5wbmc=/" + "test")) @@ -309,10 +311,11 @@ def test_underlay(self): def test_fetch_format(self): """should support format for fetch urls""" - self.__test_cloudinary_url(public_id="http://cloudinary.com/images/logo.png", - options={"format": "jpg", "type": "fetch"}, - expected_url=DEFAULT_ROOT_PATH + - "image/fetch/f_jpg/http://cloudinary.com/images/logo.png") + self.__test_cloudinary_url( + public_id="http://cloudinary.com/images/logo.png", + options={"format": "jpg", "type": "fetch"}, + expected_url=DEFAULT_ROOT_PATH + "image/fetch/f_jpg/http://cloudinary.com/images/logo.png" + ) def test_effect(self): """should support effect""" @@ -579,7 +582,7 @@ def test_start_offset(self): expected_url=VIDEO_UPLOAD_PATH + "so_auto/video_id") def test_end_offset(self): - # should support decimal seconds + # should support decimal seconds self.__test_cloudinary_url(public_id="video_id", options={'resource_type': 'video', 'end_offset': 2.63}, expected_url=VIDEO_UPLOAD_PATH + "eo_2.63/video_id") self.__test_cloudinary_url(public_id="video_id", options={'resource_type': 'video', 'end_offset': '2.63'}, @@ -613,13 +616,15 @@ def test_offset(self): 'eo_70.5p,so_35.5p': ['35.5p', '70.5p'] } for transformation, offset in test_cases.items(): - self.__test_cloudinary_url(public_id="video_id", options={'resource_type': 'video', 'offset': offset}, - expected_url=VIDEO_UPLOAD_PATH + transformation + "/video_id") + self.__test_cloudinary_url( + public_id="video_id", + options={'resource_type': 'video', 'offset': offset}, + expected_url=VIDEO_UPLOAD_PATH + transformation + "/video_id") def test_user_agent(self): with patch('cloudinary.USER_PLATFORM', ''): agent = cloudinary.get_user_agent() - six.assertRegex(self, agent, '^CloudinaryPython\/\d\.\d+\.\d+ \(Python \d\.\d+\.\d+\)$') + six.assertRegex(self, agent, r'^CloudinaryPython\/\d\.\d+\.\d+ \(Python \d\.\d+\.\d+\)$') platform = 'MyPlatform/1.2.3 (Test code)' with patch('cloudinary.USER_PLATFORM', platform): @@ -652,10 +657,11 @@ def test_overlay_options(self): ({'text': "Hello World, Nice to meet you?", 'font_family': "Arial", 'font_size': "18", 'font_weight': "bold", 'font_style': "italic", 'letter_spacing': 4, 'line_spacing': 3}, - "text:Arial_18_bold_italic_letter_spacing_4_line_spacing_3:Hello%20World%252C%20Nice%20to%20meet%20you%3F"), + "text:Arial_18_bold_italic_letter_spacing_4_line_spacing_3:Hello%20World" + "%252C%20Nice%20to%20meet%20you%3F"), ({'resource_type': "subtitles", 'public_id': "sample_sub_en.srt"}, "subtitles:sample_sub_en.srt"), - ({'resource_type': "subtitles", 'public_id': "sample_sub_he.srt", + ({'resource_type': "subtitles", 'public_id': "sample_sub_he.srt", 'font_family': "Arial", 'font_size': 40}, "subtitles:Arial_40:sample_sub_he.srt"), ({'url': "https://upload.wikimedia.org/wikipedia/commons/2/2b/고창갯벌.jpg"}, @@ -670,7 +676,8 @@ def test_overlay_options(self): def test_overlay_error_1(self): """ Must supply font_family for text in overlay """ with self.assertRaises(ValueError): - cloudinary.utils.cloudinary_url("test", overlay=dict(text="text", font_style="italic")) + cloudinary.utils.cloudinary_url( + "test", overlay=dict(text="text", font_style="italic")) def test_overlay_error_2(self): """ Must supply public_id for for non-text underlay """ @@ -686,7 +693,8 @@ def test_translate_if(self): all_operators += "_fc_lte_0_and" all_operators += "_w_gte_0" all_operators += ",e_grayscale" - condition = "width = 0 && height != 0 || aspect_ratio < 0 && page_count > 0 and face_count <= 0 and width >= 0" + condition = "width = 0 && height != 0 || aspect_ratio < 0 && page_count > 0 " \ + "and face_count <= 0 and width >= 0" options = {"if": condition, "effect": "grayscale"} transformation, options = cloudinary.utils.generate_transformation_string(**options) self.assertEqual({}, options) @@ -695,43 +703,60 @@ def test_translate_if(self): def test_merge(self): a = {"foo": "foo", "bar": "foo"} b = {"foo": "bar"} - self.assertIsNone(cloudinary.utils.merge( None,None)) + self.assertIsNone(cloudinary.utils.merge(None, None)) self.assertDictEqual(a, cloudinary.utils.merge(a, None)) self.assertDictEqual(a, cloudinary.utils.merge(None, a)) self.assertDictEqual({"foo": "bar", "bar": "foo"}, cloudinary.utils.merge(a, b)) self.assertDictEqual(a, cloudinary.utils.merge(b, a)) def test_array_should_define_a_set_of_variables(self): - options = { "if": "face_count > 2", "variables" : [ ["$z", 5], ["$foo", "$z * 2"] ], "crop" : "scale", "width" : "$foo * 200" } + options = { + "if": "face_count > 2", + "variables": [["$z", 5], ["$foo", "$z * 2"]], + "crop": "scale", + "width": "$foo * 200" + } transformation, options = cloudinary.utils.generate_transformation_string(**options) self.assertEqual('if_fc_gt_2,$z_5,$foo_$z_mul_2,c_scale,w_$foo_mul_200', transformation) def test_dollar_key_should_define_a_variable(self): - options = { "transformation":[ {"$foo":10 }, {"if":"face_count > 2"}, {"crop":"scale", "width":"$foo * 200 / face_count"}, {"if":"end"} ] } + options = {"transformation": [{"$foo": 10}, {"if": "face_count > 2"}, + {"crop": "scale", "width": "$foo * 200 / face_count"}, {"if": "end"}]} transformation, options = cloudinary.utils.generate_transformation_string(**options) self.assertEqual('$foo_10/if_fc_gt_2/c_scale,w_$foo_mul_200_div_fc/if_end', transformation) def test_should_sort_defined_variable(self): - options = { "$second": 1, "$first": 2} + options = {"$second": 1, "$first": 2} transformation, options = cloudinary.utils.generate_transformation_string(**options) self.assertEqual('$first_2,$second_1', transformation) def test_should_place_defined_variables_before_ordered(self): - options = {"variables" : [ ["$z", 5], ["$foo", "$z * 2"] ], "$second": 1, "$first": 2} + options = {"variables": [["$z", 5], ["$foo", "$z * 2"]], "$second": 1, "$first": 2} transformation, options = cloudinary.utils.generate_transformation_string(**options) self.assertEqual('$first_2,$second_1,$z_5,$foo_$z_mul_2', transformation) def test_should_support_text_values(self): public_id = "sample" - options = {"effect":"$efname:100", "$efname":"!blur!"} + options = {"effect": "$efname:100", "$efname": "!blur!"} url, options = cloudinary.utils.cloudinary_url(public_id, **options) - self.assertEqual(DEFAULT_UPLOAD_PATH+"$efname_!blur!,e_$efname:100/sample",url) + self.assertEqual(DEFAULT_UPLOAD_PATH + "$efname_!blur!,e_$efname:100/sample", url) def test_should_support_string_interpolation(self): public_id = "sample" - options = { "crop":"scale", "overlay":{"text":"$(start)Hello $(name)$(ext), $(no ) $( no)$(end)", "font_family":"Arial", "font_size":"18"}} + options = { + "crop": "scale", + "overlay": { + "text": "$(start)Hello $(name)$(ext), $(no ) $( no)$(end)", + "font_family": "Arial", + "font_size": "18" + } + } url, options = cloudinary.utils.cloudinary_url(public_id, **options) - self.assertEqual(DEFAULT_UPLOAD_PATH+"c_scale,l_text:Arial_18:$(start)Hello%20$(name)$(ext)%252C%20%24%28no%20%29%20%24%28%20no%29$(end)/sample",url) + self.assertEqual( + DEFAULT_UPLOAD_PATH + "c_scale,l_text:Arial_18:$(start)" + "Hello%20$(name)$(ext)%252C%20%24%28no%20%29" + "%20%24%28%20no%29$(end)/sample", + url) def test_encode_context(self): self.assertEqual("", cloudinary.utils.encode_context({})) @@ -741,7 +766,7 @@ def test_encode_context(self): # test that special characters are unchanged self.assertEqual("a=!@#$%^&*()_+<>?,./", cloudinary.utils.encode_context({"a": "!@#$%^&*()_+<>?,./"})) # check value escaping - self.assertEqual("a=b\|\|\=|c=d\=a\=\|", cloudinary.utils.encode_context(OrderedDict((("a", "b||="), + self.assertEqual(r"a=b\|\|\=|c=d\=a\=\|", cloudinary.utils.encode_context(OrderedDict((("a", "b||="), ("c", "d=a=|"))))) # check fallback self.assertEqual("not a dict", cloudinary.utils.encode_context("not a dict")) @@ -795,5 +820,6 @@ def test_is_remote_url(self): self.assertFalse(cloudinary.utils.is_remote_url(TEST_IMAGE)) self.assertTrue(cloudinary.utils.is_remote_url(REMOTE_TEST_IMAGE)) + if __name__ == '__main__': unittest.main() diff --git a/tests/video_test.py b/tests/video_test.py index f2fdae63..b0897dde 100644 --- a/tests/video_test.py +++ b/tests/video_test.py @@ -1,6 +1,7 @@ +import unittest + import cloudinary from cloudinary import CloudinaryVideo -import unittest VIDEO_UPLOAD_PATH = 'http://res.cloudinary.com/test123/video/upload/' DEFAULT_UPLOAD_PATH = 'http://res.cloudinary.com/test123/image/upload/' From 3e6bfb53ac59bcc0b5ae90fa5154a51bb1cfd83b Mon Sep 17 00:00:00 2001 From: Constantine Nathanson Date: Mon, 13 Aug 2018 16:15:44 +0300 Subject: [PATCH 2/3] Fix unit tests file names and folder name --- django_tests/helper_test.py | 12 ++++ django_tests/test_cloudinaryField.py | 6 +- django_tests/test_cloudinaryResource.py | 3 +- django_tests/test_cloudinary_file_field.py | 7 +- django_tests/test_helper.py | 7 -- setup.py | 4 +- {tests => test}/__init__.py | 0 tests/test_helper.py => test/helper_test.py | 33 +++++++++- {tests => test/resources}/docx.docx | Bin {tests => test/resources}/favicon.ico | Bin {tests => test/resources}/logo.png | Bin tests/api_test.py => test/test_api.py | 60 ++++++++---------- tests/archive_test.py => test/test_archive.py | 24 ++----- .../test_auth_token.py | 0 tests/config_test.py => test/test_config.py | 0 tests/image_test.py => test/test_image.py | 0 tests/search_test.py => test/test_search.py | 6 +- .../test_streaming_profiles.py | 12 ++-- .../uploader_test.py => test/test_uploader.py | 20 +++--- tests/utils_test.py => test/test_utils.py | 2 +- tests/video_test.py => test/test_video.py | 0 21 files changed, 100 insertions(+), 96 deletions(-) create mode 100644 django_tests/helper_test.py delete mode 100644 django_tests/test_helper.py rename {tests => test}/__init__.py (100%) rename tests/test_helper.py => test/helper_test.py (60%) rename {tests => test/resources}/docx.docx (100%) rename {tests => test/resources}/favicon.ico (100%) rename {tests => test/resources}/logo.png (100%) rename tests/api_test.py => test/test_api.py (94%) rename tests/archive_test.py => test/test_archive.py (82%) rename tests/auth_token_test.py => test/test_auth_token.py (100%) rename tests/config_test.py => test/test_config.py (100%) rename tests/image_test.py => test/test_image.py (100%) rename tests/search_test.py => test/test_search.py (99%) rename tests/streaming_profiles_test.py => test/test_streaming_profiles.py (91%) rename tests/uploader_test.py => test/test_uploader.py (98%) rename tests/utils_test.py => test/test_utils.py (99%) rename tests/video_test.py => test/test_video.py (100%) diff --git a/django_tests/helper_test.py b/django_tests/helper_test.py new file mode 100644 index 00000000..bd60ccbe --- /dev/null +++ b/django_tests/helper_test.py @@ -0,0 +1,12 @@ +import os +import random + +SUFFIX = os.environ.get('TRAVIS_JOB_ID') or random.randint(10000, 99999) + +RESOURCES_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "test", "resources") +TEST_IMAGE = os.path.join(RESOURCES_PATH, "logo.png") +TEST_TAG = "pycloudinary_test" +UNIQUE_TAG = "{0}_{1}".format(TEST_TAG, SUFFIX) + +TEST_IMAGE_W = 241 +TEST_IMAGE_H = 51 diff --git a/django_tests/test_cloudinaryField.py b/django_tests/test_cloudinaryField.py index d3f4d23f..2f178e62 100644 --- a/django_tests/test_cloudinaryField.py +++ b/django_tests/test_cloudinaryField.py @@ -12,14 +12,10 @@ from django.core.files.uploadedfile import SimpleUploadedFile from .models import Poll -from .test_helper import SUFFIX +from django_tests.helper_test import SUFFIX, TEST_IMAGE, TEST_IMAGE_W, TEST_IMAGE_H API_TEST_ID = "dj_test_{}".format(SUFFIX) -TEST_IMAGE = "tests/logo.png" -TEST_IMAGE_W = 241 -TEST_IMAGE_H = 51 - class TestCloudinaryField(TestCase): diff --git a/django_tests/test_cloudinaryResource.py b/django_tests/test_cloudinaryResource.py index a84b75fb..868de489 100644 --- a/django_tests/test_cloudinaryResource.py +++ b/django_tests/test_cloudinaryResource.py @@ -4,11 +4,10 @@ from cloudinary import CloudinaryResource, api, uploader from django.test import TestCase -from .test_helper import SUFFIX +from django_tests.helper_test import SUFFIX, TEST_IMAGE disable_warnings() -TEST_IMAGE = "tests/logo.png" TEST_TAG = "dj_pycloudinary_test_{0}".format(SUFFIX) diff --git a/django_tests/test_cloudinary_file_field.py b/django_tests/test_cloudinary_file_field.py index cd181e89..420a005f 100644 --- a/django_tests/test_cloudinary_file_field.py +++ b/django_tests/test_cloudinary_file_field.py @@ -8,15 +8,10 @@ from cloudinary import api, CloudinaryResource from cloudinary.forms import CloudinaryFileField -from django_tests.test_helper import SUFFIX +from django_tests.helper_test import SUFFIX, TEST_IMAGE, TEST_IMAGE_W, TEST_IMAGE_H API_TEST_ID = "dj_test_{}".format(SUFFIX) -TEST_IMAGES_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "tests") -TEST_IMAGE = os.path.join(TEST_IMAGES_PATH, "logo.png") -TEST_IMAGE_W = 241 -TEST_IMAGE_H = 51 - class TestCloudinaryFileField(TestCase): def setUp(self): diff --git a/django_tests/test_helper.py b/django_tests/test_helper.py deleted file mode 100644 index 46577163..00000000 --- a/django_tests/test_helper.py +++ /dev/null @@ -1,7 +0,0 @@ -import os -import random - -SUFFIX = os.environ.get('TRAVIS_JOB_ID') or random.randint(10000, 99999) -TEST_IMAGE = "tests/logo.png" -TEST_TAG = "pycloudinary_test" -UNIQUE_TAG = "{0}_{1}".format(TEST_TAG, SUFFIX) diff --git a/setup.py b/setup.py index d684fbfb..dbd28fda 100644 --- a/setup.py +++ b/setup.py @@ -14,7 +14,7 @@ author_email='info@cloudinary.com', url='http://cloudinary.com', license='MIT', - packages=find_packages(exclude=['ez_setup', 'examples', 'tests', 'django_tests', 'django_tests.*']), + packages=find_packages(exclude=['ez_setup', 'examples', 'test', 'django_tests', 'django_tests.*']), classifiers=[ "Development Status :: 5 - Production/Stable", "Environment :: Web Environment", @@ -49,7 +49,7 @@ ], include_package_data=True, zip_safe=False, - test_suite="tests", + test_suite="test", install_requires=[ "six", "mock", diff --git a/tests/__init__.py b/test/__init__.py similarity index 100% rename from tests/__init__.py rename to test/__init__.py diff --git a/tests/test_helper.py b/test/helper_test.py similarity index 60% rename from tests/test_helper.py rename to test/helper_test.py index aa3afee0..19c70606 100644 --- a/tests/test_helper.py +++ b/test/helper_test.py @@ -4,9 +4,20 @@ import re from datetime import timedelta, tzinfo +import six +from urllib3 import HTTPResponse +from urllib3._collections import HTTPHeaderDict + SUFFIX = os.environ.get('TRAVIS_JOB_ID') or random.randint(10000, 99999) + REMOTE_TEST_IMAGE = "http://cloudinary.com/images/old_logo.png" -TEST_IMAGE = "tests/logo.png" + +RESOURCES_PATH = "test/resources/" + +TEST_IMAGE = RESOURCES_PATH + "logo.png" +TEST_DOC = RESOURCES_PATH + "docx.docx" +TEST_ICON = RESOURCES_PATH + "favicon.ico" + TEST_TAG = "pycloudinary_test" UNIQUE_TAG = "{0}_{1}".format(TEST_TAG, SUFFIX) @@ -61,3 +72,23 @@ def get_list_param(mocker, name): params = get_params(args) reg = re.compile(r'{}\[\d*\]'.format(name)) return [params[key] for key in params.keys() if reg.match(key)] + + +def http_response_mock(body="", headers=None, status=200): + if headers is None: + headers = {} + + if not six.PY2: + body = body.encode("UTF-8") + + return HTTPResponse(body, HTTPHeaderDict(headers), status=status) + + +def api_response_mock(): + return http_response_mock('{"foo":"bar"}', {"x-featureratelimit-limit": '0', + "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT', + "x-featureratelimit-remaining": '0'}) + + +def uploader_response_mock(): + return http_response_mock('{"foo":"bar"}') diff --git a/tests/docx.docx b/test/resources/docx.docx similarity index 100% rename from tests/docx.docx rename to test/resources/docx.docx diff --git a/tests/favicon.ico b/test/resources/favicon.ico similarity index 100% rename from tests/favicon.ico rename to test/resources/favicon.ico diff --git a/tests/logo.png b/test/resources/logo.png similarity index 100% rename from tests/logo.png rename to test/resources/logo.png diff --git a/tests/api_test.py b/test/test_api.py similarity index 94% rename from tests/api_test.py rename to test/test_api.py index b578979e..6d507fa1 100644 --- a/tests/api_test.py +++ b/test/test_api.py @@ -4,28 +4,16 @@ import six from mock import patch -from urllib3 import HTTPResponse, disable_warnings -from urllib3._collections import HTTPHeaderDict +from urllib3 import disable_warnings import cloudinary from cloudinary import api, uploader, utils -from tests.test_helper import * +from test.helper_test import SUFFIX, TEST_IMAGE, get_uri, get_params, get_list_param, get_param, TEST_DOC, get_method, \ + UNIQUE_TAG, api_response_mock +MOCK_RESPONSE = api_response_mock() -MOCK_HEADERS = HTTPHeaderDict({ - "x-featureratelimit-limit": '0', - "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT', - "x-featureratelimit-remaining": '0', -}) - -if six.PY2: - MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}', headers=MOCK_HEADERS) -else: - MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}'.encode("UTF-8"), headers=MOCK_HEADERS) - -disable_warnings() - -UNIQUE_TAG = 'api_{}'.format(UNIQUE_TAG) +UNIQUE_API_TAG = 'api_{}'.format(UNIQUE_TAG) API_TEST_TAG = "api_test_{}_tag".format(SUFFIX) API_TEST_PREFIX = "api_test_{}".format(SUFFIX) API_TEST_ID = "api_test_{}".format(SUFFIX) @@ -44,6 +32,8 @@ MAPPING_TEST_ID = "api_test_upload_mapping_{}".format(SUFFIX) RESTORE_TEST_ID = "api_test_restore_{}".format(SUFFIX) +disable_warnings() + class ApiTest(unittest.TestCase): @classmethod @@ -53,7 +43,7 @@ def setUpClass(cls): return print("Running tests for cloud: {}".format(cloudinary.config().cloud_name)) for id in [API_TEST_ID, API_TEST_ID2]: - uploader.upload("tests/logo.png", + uploader.upload(TEST_IMAGE, public_id=id, tags=[API_TEST_TAG, ], context="key=value", eager=[{"width": 100, "crop": "scale"}], overwrite=True) @@ -72,14 +62,14 @@ def tearDownClass(cls): presets_response = api.upload_presets(max_results=200) preset_names = [ preset["name"] for preset in presets_response.get('presets', []) - if UNIQUE_TAG in preset.get('settings', {}).get('tags', '')] + if UNIQUE_API_TAG in preset.get('settings', {}).get('tags', '')] for name in preset_names: try: api.delete_upload_preset(name) except Exception: pass - cloudinary.api.delete_resources_by_tag(UNIQUE_TAG) - cloudinary.api.delete_resources_by_tag(UNIQUE_TAG, resource_type='raw') + cloudinary.api.delete_resources_by_tag(UNIQUE_API_TAG) + cloudinary.api.delete_resources_by_tag(UNIQUE_API_TAG, resource_type='raw') try: api.delete_upload_mapping(MAPPING_TEST_ID) @@ -389,7 +379,7 @@ def test18_usage(self): @unittest.skip("Skip delete all derived resources by default") def test19_delete_derived(self): """ should allow deleting all resource """ - uploader.upload("tests/logo.png", public_id=API_TEST_ID5, eager=[{"width": 101, "crop": "scale"}]) + uploader.upload(TEST_IMAGE, public_id=API_TEST_ID5, eager=[{"width": 101, "crop": "scale"}]) resource = api.resource(API_TEST_ID5) self.assertNotEqual(resource, None) self.assertEqual(len(resource["derived"]), 1) @@ -401,7 +391,7 @@ def test19_delete_derived(self): @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test20_manual_moderation(self): """ should support setting manual moderation status """ - resource = uploader.upload("tests/logo.png", moderation="manual", tags=[UNIQUE_TAG]) + resource = uploader.upload(TEST_IMAGE, moderation="manual", tags=[UNIQUE_API_TAG]) self.assertEqual(resource["moderation"][0]["status"], "pending") self.assertEqual(resource["moderation"][0]["kind"], "manual") @@ -422,7 +412,7 @@ def test21_notification_url(self, mocker): @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test22_raw_conversion(self): """ should support requesting raw_convert """ - resource = uploader.upload("tests/docx.docx", resource_type="raw", tags=[UNIQUE_TAG]) + resource = uploader.upload(TEST_DOC, resource_type="raw", tags=[UNIQUE_API_TAG]) with six.assertRaisesRegex(self, api.BadRequest, 'Illegal value'): api.update(resource["public_id"], raw_convert="illegal", resource_type="raw") @@ -473,9 +463,9 @@ def test27_start_at(self, mocker): @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test28_create_list_upload_presets(self): """ should allow creating and listing upload_presets """ - api.create_upload_preset(name=API_TEST_PRESET, folder="folder", tags=[UNIQUE_TAG]) - api.create_upload_preset(name=API_TEST_PRESET2, folder="folder2", tags=[UNIQUE_TAG]) - api.create_upload_preset(name=API_TEST_PRESET3, folder="folder3", tags=[UNIQUE_TAG]) + api.create_upload_preset(name=API_TEST_PRESET, folder="folder", tags=[UNIQUE_API_TAG]) + api.create_upload_preset(name=API_TEST_PRESET2, folder="folder2", tags=[UNIQUE_API_TAG]) + api.create_upload_preset(name=API_TEST_PRESET3, folder="folder3", tags=[UNIQUE_API_TAG]) api_response = api.upload_presets() presets = api_response["presets"] @@ -489,7 +479,7 @@ def test28_create_list_upload_presets(self): def test29_get_upload_presets(self): """ should allow getting a single upload_preset """ result = api.create_upload_preset(unsigned=True, folder="folder", width=100, crop="scale", - tags=["a", "b", "c", UNIQUE_TAG], context={"a": "b", "c": "d"}) + tags=["a", "b", "c", UNIQUE_API_TAG], context={"a": "b", "c": "d"}) name = result["name"] preset = api.upload_preset(name) self.assertEqual(preset["name"], name) @@ -498,7 +488,7 @@ def test29_get_upload_presets(self): self.assertEqual(settings["folder"], "folder") self.assertEqual(settings["transformation"], [{"width": 100, "crop": "scale"}]) self.assertEqual(settings["context"], {"a": "b", "c": "d"}) - self.assertEqual(settings["tags"], ["a", "b", "c", UNIQUE_TAG]) + self.assertEqual(settings["tags"], ["a", "b", "c", UNIQUE_API_TAG]) @patch('urllib3.request.RequestMethods.request') @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") @@ -535,10 +525,10 @@ def test32_background_removal(self): "Comment out this line if you really want to test it.") def test_folder_listing(self): """ should support listing folders """ - uploader.upload("tests/logo.png", public_id="{}1/item".format(PREFIX), tags=[UNIQUE_TAG]) - uploader.upload("tests/logo.png", public_id="{}2/item".format(PREFIX), tags=[UNIQUE_TAG]) - uploader.upload("tests/logo.png", public_id="{}1/test_subfolder1/item".format(PREFIX), tags=[UNIQUE_TAG]) - uploader.upload("tests/logo.png", public_id="{}1/test_subfolder2/item".format(PREFIX), tags=[UNIQUE_TAG]) + uploader.upload(TEST_IMAGE, public_id="{}1/item".format(PREFIX), tags=[UNIQUE_API_TAG]) + uploader.upload(TEST_IMAGE, public_id="{}2/item".format(PREFIX), tags=[UNIQUE_API_TAG]) + uploader.upload(TEST_IMAGE, public_id="{}1/test_subfolder1/item".format(PREFIX), tags=[UNIQUE_API_TAG]) + uploader.upload(TEST_IMAGE, public_id="{}1/test_subfolder2/item".format(PREFIX), tags=[UNIQUE_API_TAG]) result = api.root_folders() self.assertEqual(result["folders"][0]["name"], "{}1".format(PREFIX)) self.assertEqual(result["folders"][1]["name"], "{}2".format(PREFIX)) @@ -562,7 +552,7 @@ def test_CloudinaryImage_len(self): @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test_restore(self): """ should support restoring resources """ - uploader.upload("tests/logo.png", public_id=RESTORE_TEST_ID, backup=True, tags=[UNIQUE_TAG]) + uploader.upload(TEST_IMAGE, public_id=RESTORE_TEST_ID, backup=True, tags=[UNIQUE_API_TAG]) resource = api.resource(RESTORE_TEST_ID) self.assertNotEqual(resource, None) self.assertEqual(resource["bytes"], 3381) @@ -582,7 +572,7 @@ def test_restore(self): @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test_upload_mapping(self): - api.create_upload_mapping(MAPPING_TEST_ID, template="http://cloudinary.com", tags=[UNIQUE_TAG]) + api.create_upload_mapping(MAPPING_TEST_ID, template="http://cloudinary.com", tags=[UNIQUE_API_TAG]) result = api.upload_mapping(MAPPING_TEST_ID) self.assertEqual(result["template"], "http://cloudinary.com") api.update_upload_mapping(MAPPING_TEST_ID, template="http://res.cloudinary.com") diff --git a/tests/archive_test.py b/test/test_archive.py similarity index 82% rename from tests/archive_test.py rename to test/test_archive.py index a566adfe..b76be8ed 100644 --- a/tests/archive_test.py +++ b/test/test_archive.py @@ -12,33 +12,23 @@ from mock import patch import six import urllib3 -from urllib3 import disable_warnings, HTTPResponse -from urllib3._collections import HTTPHeaderDict +from urllib3 import disable_warnings -from .test_helper import SUFFIX +from test.helper_test import SUFFIX, TEST_IMAGE, api_response_mock -disable_warnings() - -MOCK_HEADERS = HTTPHeaderDict({ - "x-featureratelimit-limit": '0', - "x-featureratelimit-reset": 'Sat, 01 Apr 2017 22:00:00 GMT', - "x-featureratelimit-remaining": '0', -}) - -if six.PY2: - MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}', headers=MOCK_HEADERS) -else: - MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}'.encode("UTF-8"), headers=MOCK_HEADERS) +MOCK_RESPONSE = api_response_mock() TEST_TAG = "arch_pycloudinary_test_{}".format(SUFFIX) +disable_warnings() + class ArchiveTest(unittest.TestCase): @classmethod def setUpClass(cls): cloudinary.reset_config() - uploader.upload("tests/logo.png", tags=[TEST_TAG]) - uploader.upload("tests/logo.png", tags=[TEST_TAG], transformation=dict(width=10)) + uploader.upload(TEST_IMAGE, tags=[TEST_TAG]) + uploader.upload(TEST_IMAGE, tags=[TEST_TAG], transformation=dict(width=10)) @classmethod def tearDownClass(cls): diff --git a/tests/auth_token_test.py b/test/test_auth_token.py similarity index 100% rename from tests/auth_token_test.py rename to test/test_auth_token.py diff --git a/tests/config_test.py b/test/test_config.py similarity index 100% rename from tests/config_test.py rename to test/test_config.py diff --git a/tests/image_test.py b/test/test_image.py similarity index 100% rename from tests/image_test.py rename to test/test_image.py diff --git a/tests/search_test.py b/test/test_search.py similarity index 99% rename from tests/search_test.py rename to test/test_search.py index 8aed6be7..5746270b 100644 --- a/tests/search_test.py +++ b/test/test_search.py @@ -8,9 +8,7 @@ import cloudinary from cloudinary import api, logger, uploader from cloudinary.search import Search -from tests.test_helper import SUFFIX, TEST_IMAGE, TEST_TAG, UNIQUE_TAG - -disable_warnings() +from test.helper_test import SUFFIX, TEST_IMAGE, TEST_TAG, UNIQUE_TAG TEST_TAG = 'search_{}'.format(TEST_TAG) UNIQUE_TAG = 'search_{}'.format(UNIQUE_TAG) @@ -22,6 +20,8 @@ public_ids = ["api_test{0}_{1}".format(i, SUFFIX) for i in range(0, TEST_IMAGES_COUNT)] upload_results = ["++"] +disable_warnings() + class SearchTest(unittest.TestCase): @classmethod diff --git a/tests/streaming_profiles_test.py b/test/test_streaming_profiles.py similarity index 91% rename from tests/streaming_profiles_test.py rename to test/test_streaming_profiles.py index f1067d51..787e5462 100644 --- a/tests/streaming_profiles_test.py +++ b/test/test_streaming_profiles.py @@ -3,29 +3,29 @@ from cloudinary import api from urllib3 import disable_warnings -from .test_helper import SUFFIX +from test.helper_test import SUFFIX disable_warnings() class StreamingProfilesTest(unittest.TestCase): initialized = False + test_id = "streaming_profiles_test_{}".format(SUFFIX) def setUp(self): - if StreamingProfilesTest.initialized: + if self.initialized: return - StreamingProfilesTest.initialized = True + self.initialized = True cloudinary.reset_config() if not cloudinary.config().api_secret: return - StreamingProfilesTest.test_id = "api_test_{}".format(SUFFIX) __predefined_sp = ["4k", "full_hd", "hd", "sd", "full_hd_wifi", "full_hd_lean", "hd_lean"] @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") def test_create_streaming_profile(self): """should create a streaming profile with representations""" - name = StreamingProfilesTest.test_id + "_streaming_profile" + name = self.test_id + "_streaming_profile" result = api.create_streaming_profile( name, representations=[{"transformation": { @@ -68,7 +68,7 @@ def test_get_streaming_profile(self): self.assertIn("crop", tr) def test_update_delete_streaming_profile(self): - name = StreamingProfilesTest.test_id + "_streaming_profile_delete" + name = self.test_id + "_streaming_profile_delete" api.create_streaming_profile( name, representations=[{"transformation": { diff --git a/tests/uploader_test.py b/test/test_uploader.py similarity index 98% rename from tests/uploader_test.py rename to test/test_uploader.py index b40f17e6..86ae178b 100644 --- a/tests/uploader_test.py +++ b/test/test_uploader.py @@ -1,4 +1,5 @@ import io +import os import tempfile import unittest from collections import OrderedDict @@ -9,25 +10,22 @@ import cloudinary from cloudinary import api, uploader, utils -from urllib3 import disable_warnings, HTTPResponse +from urllib3 import disable_warnings from urllib3.util import parse_url -from tests.test_helper import * +from test.helper_test import uploader_response_mock, SUFFIX, TEST_IMAGE, get_params, TEST_TAG, TEST_ICON, TEST_DOC, \ + REMOTE_TEST_IMAGE, UTC -if six.PY2: - MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}') -else: - MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}'.encode("UTF-8")) - -disable_warnings() +MOCK_RESPONSE = uploader_response_mock() TEST_IMAGE_HEIGHT = 51 TEST_IMAGE_WIDTH = 241 UNIQUE_TAG = "up_test_uploader_{}".format(SUFFIX) TEST_DOCX_ID = "test_docx_{}".format(SUFFIX) +disable_warnings() -class UploaderTest(unittest.TestCase): +class UploaderTest(unittest.TestCase): def setUp(self): cloudinary.reset_config() @@ -142,7 +140,7 @@ def test_rename(self): result = uploader.upload(TEST_IMAGE, tags=[UNIQUE_TAG]) uploader.rename(result["public_id"], result["public_id"]+"2") self.assertIsNotNone(api.resource(result["public_id"]+"2")) - result2 = uploader.upload("tests/favicon.ico", tags=[UNIQUE_TAG]) + result2 = uploader.upload(TEST_ICON, tags=[UNIQUE_TAG]) self.assertRaises(api.Error, uploader.rename, result2["public_id"], result["public_id"]+"2") uploader.rename(result2["public_id"], result["public_id"]+"2", overwrite=True) @@ -289,7 +287,7 @@ def test_manual_moderation(self): def test_raw_conversion(self): """ should support requesting raw_convert """ with six.assertRaisesRegex(self, api.Error, 'Raw convert is invalid'): - uploader.upload("tests/docx.docx", public_id=TEST_DOCX_ID, raw_convert="illegal", + uploader.upload(TEST_DOC, public_id=TEST_DOCX_ID, raw_convert="illegal", resource_type="raw", tags=[UNIQUE_TAG]) @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret") diff --git a/tests/utils_test.py b/test/test_utils.py similarity index 99% rename from tests/utils_test.py rename to test/test_utils.py index 7b330299..8d813c1b 100644 --- a/tests/utils_test.py +++ b/test/test_utils.py @@ -11,7 +11,7 @@ import cloudinary.utils from cloudinary.utils import build_list_of_dicts, json_encode -from tests.test_helper import TEST_IMAGE, REMOTE_TEST_IMAGE +from test.helper_test import TEST_IMAGE, REMOTE_TEST_IMAGE DEFAULT_ROOT_PATH = 'http://res.cloudinary.com/test123/' diff --git a/tests/video_test.py b/test/test_video.py similarity index 100% rename from tests/video_test.py rename to test/test_video.py From 96eb5dd716d42b9aac1291c87e79c30eaf25336f Mon Sep 17 00:00:00 2001 From: Constantine Nathanson Date: Mon, 13 Aug 2018 16:54:46 +0300 Subject: [PATCH 3/3] Fix docstrings --- cloudinary/__init__.py | 39 ++++++++++++++++++++++----------------- cloudinary/api.py | 1 - cloudinary/models.py | 32 ++++++++++++++++++++++---------- cloudinary/uploader.py | 6 ++++++ cloudinary/utils.py | 9 ++++++++- 5 files changed, 58 insertions(+), 29 deletions(-) diff --git a/cloudinary/__init__.py b/cloudinary/__init__.py index 1f380ff3..fd59c7e8 100644 --- a/cloudinary/__init__.py +++ b/cloudinary/__init__.py @@ -46,7 +46,8 @@ def get_user_agent(): - """Provides the `USER_AGENT` string that is passed to the Cloudinary servers. + """ + Provides the `USER_AGENT` string that is passed to the Cloudinary servers. Prepends `USER_PLATFORM` if it is defined. :returns: the user agent @@ -407,23 +408,27 @@ def video_thumbnail(self, **options): self.default_poster_options(options) return self.build_url(**options) - # Creates an HTML video tag for the provided +source+ - # - # ==== Options - # * source_types - Specify which source type the tag should include. - # defaults to webm, mp4 and ogv. - # * source_transformation - specific transformations to use - # for a specific source type. - # * poster - override default thumbnail: - # * url: provide an ad hoc url - # * options: with specific poster transformations and/or Cloudinary +:public_id+ - # - # ==== Examples - # CloudinaryResource("mymovie.mp4").video() - # CloudinaryResource("mymovie.mp4").video(source_types = 'webm') - # CloudinaryResource("mymovie.ogv").video(poster = "myspecialplaceholder.jpg") - # CloudinaryResource("mymovie.webm").video(source_types = ['webm', 'mp4'], poster = {'effect': 'sepia'}) def video(self, **options): + """ + Creates an HTML video tag for the provided +source+ + + Examples: + CloudinaryResource("mymovie.mp4").video() + CloudinaryResource("mymovie.mp4").video(source_types = 'webm') + CloudinaryResource("mymovie.ogv").video(poster = "myspecialplaceholder.jpg") + CloudinaryResource("mymovie.webm").video(source_types = ['webm', 'mp4'], poster = {'effect': 'sepia'}) + + :param options: + * source_types - Specify which source type the tag should include. + defaults to webm, mp4 and ogv. + * source_transformation - specific transformations to use + for a specific source type. + * poster - override default thumbnail: + * url: provide an ad hoc url + * options: with specific poster transformations and/or Cloudinary +:public_id+ + + :return: Video tag + """ public_id = options.get('public_id', self.public_id) source = re.sub(r"\.({0})$".format("|".join(self.default_source_types())), '', public_id) diff --git a/cloudinary/api.py b/cloudinary/api.py index 3e4f4493..1516256b 100644 --- a/cloudinary/api.py +++ b/cloudinary/api.py @@ -15,7 +15,6 @@ logger = cloudinary.logger -# intentionally one-liners class Error(Exception): pass diff --git a/cloudinary/models.py b/cloudinary/models.py index 1d15122d..8a9a63a5 100644 --- a/cloudinary/models.py +++ b/cloudinary/models.py @@ -13,15 +13,21 @@ CLOUDINARY_FIELD_DB_RE = r'(?:(?Pimage|raw|video)/' \ r'(?Pupload|private|authenticated)/)?' \ - r'(?:v(?P\d+)/)?(?P.*?)(\.(?P[^.]+))?$' + r'(?:v(?P\d+)/)?' \ + r'(?P.*?)' \ + r'(\.(?P[^.]+))?$' -# Taken from six - https://pythonhosted.org/six/ def with_metaclass(meta, *bases): - """Create a base class with a metaclass.""" - # This requires a bit of explanation: the basic idea is to make a dummy - # metaclass for one level of class instantiation that replaces itself with - # the actual metaclass. + """ + Create a base class with a metaclass. + + This requires a bit of explanation: the basic idea is to make a dummy + metaclass for one level of class instantiation that replaces itself with + the actual metaclass. + + Taken from six - https://pythonhosted.org/six/ + """ class metaclass(meta): def __new__(cls, name, this_bases, d): return meta(name, bases, d) @@ -45,10 +51,16 @@ def get_internal_type(self): return 'CharField' def value_to_string(self, obj): - # We need to support both legacy `_get_val_from_obj` and new `value_from_object` models.Field methods. - # It would be better to wrap it with try -> except AttributeError -> fallback to legacy. - # Unfortunately, we can catch AttributeError exception from `value_from_object` function itself. - # Parsing exception string is an overkill here, that's why we check for attribute existence + """ + We need to support both legacy `_get_val_from_obj` and new `value_from_object` models.Field methods. + It would be better to wrap it with try -> except AttributeError -> fallback to legacy. + Unfortunately, we can catch AttributeError exception from `value_from_object` function itself. + Parsing exception string is an overkill here, that's why we check for attribute existence + + :param obj: Value to serialize + + :return: Serialized value + """ if hasattr(self, 'value_from_object'): value = self.value_from_object(obj) diff --git a/cloudinary/uploader.py b/cloudinary/uploader.py index c9f12e19..0f588842 100644 --- a/cloudinary/uploader.py +++ b/cloudinary/uploader.py @@ -185,8 +185,10 @@ def replace_tag(tag, public_ids=None, **options): def remove_all_tags(public_ids, **options): """ Remove all tags from the specified public IDs. + :param public_ids: the public IDs of the resources to update :param options: additional options passed to the request + :return: dictionary with a list of public IDs that were updated """ return call_tags_api(None, "remove_all", public_ids, **options) @@ -195,9 +197,11 @@ def remove_all_tags(public_ids, **options): def add_context(context, public_ids, **options): """ Add a context keys and values. If a particular key already exists, the value associated with the key is updated. + :param context: dictionary of context :param public_ids: the public IDs of the resources to update :param options: additional options passed to the request + :return: dictionary with a list of public IDs that were updated """ return call_context_api(context, "add", public_ids, **options) @@ -206,8 +210,10 @@ def add_context(context, public_ids, **options): def remove_all_context(public_ids, **options): """ Remove all custom context from the specified public IDs. + :param public_ids: the public IDs of the resources to update :param options: additional options passed to the request + :return: dictionary with a list of public IDs that were updated """ return call_context_api(None, "remove_all", public_ids, **options) diff --git a/cloudinary/utils.py b/cloudinary/utils.py index f2589630..bf4cac46 100644 --- a/cloudinary/utils.py +++ b/cloudinary/utils.py @@ -560,8 +560,15 @@ def cloudinary_api_url(action='upload', **options): return "/".join([cloudinary_prefix, "v1_1", cloud_name, resource_type, action]) -# Based on ruby's CGI::unescape. In addition does not escape / : def smart_escape(source, unsafe=r"([^a-zA-Z0-9_.\-\/:]+)"): + """ + Based on ruby's CGI::unescape. In addition does not escape / : + + :param source: Source string to escape + :param unsafe: Unsafe characters + + :return: Escaped string + """ def pack(m): return to_bytes('%' + "%".join( ["%02X" % x for x in struct.unpack('B' * len(m.group(1)), m.group(1))]