diff --git a/cloudinary/__init__.py b/cloudinary/__init__.py
index 78bf8f44..fd59c7e8 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"
@@ -48,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
@@ -136,7 +135,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 +195,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):
@@ -407,30 +408,37 @@ 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("\.({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 +447,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 +463,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..1516256b 100644
--- a/cloudinary/api.py
+++ b/cloudinary/api.py
@@ -4,26 +4,47 @@
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 +66,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 +89,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 +124,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 +140,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 +243,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 +390,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 +402,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 +470,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..8a9a63a5 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,15 +11,23 @@
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+)/)?' \
+ 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/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..0f588842 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)
@@ -180,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)
@@ -190,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)
@@ -201,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)
@@ -230,17 +241,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 +337,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..bf4cac46 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,51 @@ 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_.\-\/:]+)"):
+ """
+ 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))]).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 +629,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 +701,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 +831,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 +854,7 @@ def process_layer(layer, layer_parameter):
return ':'.join(components)
+
IF_OPERATORS = {
"=": 'eq',
"!=": 'ne',
@@ -843,14 +885,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 +903,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 +955,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 +971,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/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/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..2f178e62 100644
--- a/django_tests/test_cloudinaryField.py
+++ b/django_tests/test_cloudinaryField.py
@@ -1,26 +1,24 @@
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
+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):
+
@classmethod
def setUpTestData(cls):
Poll.objects.create(question="with image", image="image/upload/v1234/{}.jpg".format(API_TEST_ID))
@@ -98,5 +96,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..868de489 100644
--- a/django_tests/test_cloudinaryResource.py
+++ b/django_tests/test_cloudinaryResource.py
@@ -1,17 +1,13 @@
-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
+from django_tests.helper_test import SUFFIX, TEST_IMAGE
disable_warnings()
-TEST_IMAGE = "tests/logo.png"
TEST_TAG = "dj_pycloudinary_test_{0}".format(SUFFIX)
@@ -42,17 +38,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 +63,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_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/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..dbd28fda 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'
@@ -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 8e7d5a89..19c70606 100644
--- a/tests/test_helper.py
+++ b/test/helper_test.py
@@ -4,15 +4,25 @@
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)
ZERO = timedelta(0)
-# A UTC class.
class UTC(tzinfo):
"""UTC"""
@@ -62,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 92%
rename from tests/api_test.py
rename to test/test_api.py
index a17968a2..6d507fa1 100644
--- a/tests/api_test.py
+++ b/test/test_api.py
@@ -2,29 +2,18 @@
import unittest
from collections import OrderedDict
+import six
from mock import patch
-from urllib3._collections import HTTPHeaderDict
+from urllib3 import disable_warnings
import cloudinary
-import six
-from cloudinary import uploader, api, utils
-
-from urllib3 import disable_warnings, HTTPResponse
-
-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', })
+from cloudinary import api, uploader, utils
+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
-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()
-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)
@@ -43,6 +32,9 @@
MAPPING_TEST_ID = "api_test_upload_mapping_{}".format(SUFFIX)
RESTORE_TEST_ID = "api_test_restore_{}".format(SUFFIX)
+disable_warnings()
+
+
class ApiTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
@@ -51,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)
@@ -68,14 +60,16 @@ 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_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)
@@ -97,7 +91,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 +205,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):
@@ -388,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)
@@ -400,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")
@@ -421,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")
@@ -472,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"]
@@ -488,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)
@@ -497,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")
@@ -534,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))
@@ -550,18 +541,18 @@ 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"]))
@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)
@@ -581,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 75%
rename from tests/archive_test.py
rename to test/test_archive.py
index ac6f4a9f..b76be8ed 100644
--- a/tests/archive_test.py
+++ b/test/test_archive.py
@@ -1,4 +1,3 @@
-import random
import tempfile
import time
import unittest
@@ -8,38 +7,28 @@
import cloudinary
import cloudinary.poster.streaminghttp
-from cloudinary import uploader, utils, api
+from cloudinary import api, uploader, utils
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, TEST_TAG
+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):
@@ -51,7 +40,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 +80,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/test/test_auth_token.py
similarity index 52%
rename from tests/auth_token_test.py
rename to test/test_auth_token.py
index fd595dcc..1831e6c7 100644
--- a/tests/auth_token_test.py
+++ b/test/test_auth_token.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/test/test_config.py
similarity index 99%
rename from tests/config_test.py
rename to test/test_config.py
index 75de2002..2415296d 100644
--- a/tests/config_test.py
+++ b/test/test_config.py
@@ -1,4 +1,5 @@
from unittest import TestCase
+
import cloudinary
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 92%
rename from tests/search_test.py
rename to test/test_search.py
index b068a77a..5746270b 100644
--- a/tests/search_test.py
+++ b/test/test_search.py
@@ -6,27 +6,30 @@
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
-
-disable_warnings()
+from cloudinary import api, logger, uploader
+from cloudinary.search import Search
+from test.helper_test import SUFFIX, TEST_IMAGE, TEST_TAG, UNIQUE_TAG
TEST_TAG = 'search_{}'.format(TEST_TAG)
UNIQUE_TAG = 'search_{}'.format(UNIQUE_TAG)
+
TEST_IMAGES_COUNT = 3
MAX_INDEX_RETRIES = 10
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
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/test/test_streaming_profiles.py
similarity index 72%
rename from tests/streaming_profiles_test.py
rename to test/test_streaming_profiles.py
index 08d0832d..787e5462 100644
--- a/tests/streaming_profiles_test.py
+++ b/test/test_streaming_profiles.py
@@ -1,39 +1,41 @@
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
+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: return
- StreamingProfilesTest.initialized = True
+ if self.initialized:
+ return
+ self.initialized = True
cloudinary.reset_config()
- if not cloudinary.config().api_secret: return
- StreamingProfilesTest.test_id = "api_test_{}".format(SUFFIX)
+ if not cloudinary.config().api_secret:
+ return
__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": {"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)
@@ -66,19 +68,21 @@ 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": {"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/uploader_test.py b/test/test_uploader.py
similarity index 90%
rename from tests/uploader_test.py
rename to test/test_uploader.py
index fd0ef996..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
@@ -6,28 +7,25 @@
import six
from mock import patch
-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 *
+from cloudinary import api, uploader, utils
-if six.PY2:
- MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}')
-else:
- MOCK_RESPONSE = HTTPResponse(body='{"foo":"bar"}'.encode("UTF-8"))
+from urllib3 import disable_warnings
+from urllib3.util import parse_url
+from test.helper_test import uploader_response_mock, SUFFIX, TEST_IMAGE, get_params, TEST_TAG, TEST_ICON, TEST_DOC, \
+ REMOTE_TEST_IMAGE, UTC
-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()
@@ -46,8 +44,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 +66,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 +75,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 +100,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 +111,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 +129,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")
@@ -138,8 +140,9 @@ 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])
- self.assertRaises(api.Error, uploader.rename, result2["public_id"], result["public_id"]+"2")
+ 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)
self.assertEqual(api.resource(result["public_id"]+"2")["format"], "ico")
@@ -148,7 +151,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 +160,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 +233,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]]
@@ -282,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")
@@ -294,7 +299,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 +323,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 +337,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 +349,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 +409,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/test/test_utils.py
similarity index 94%
rename from tests/utils_test.py
rename to test/test_utils.py
index 55f1e230..8d813c1b 100644
--- a/tests/utils_test.py
+++ b/test/test_utils.py
@@ -1,4 +1,5 @@
# -*- coding: utf-8 -*-
+
import re
import unittest
from collections import OrderedDict
@@ -10,7 +11,8 @@
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/'
DEFAULT_UPLOAD_PATH = 'http://res.cloudinary.com/test123/image/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/test/test_video.py
similarity index 99%
rename from tests/video_test.py
rename to test/test_video.py
index f2fdae63..b0897dde 100644
--- a/tests/video_test.py
+++ b/test/test_video.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/'