Add unit tests

Author: const-cloudinary

.gitignore

@@ -9,3 +9,4 @@ venv
 .vscode
 .idea
 
+.cld-sync

.travis.yml

@@ -0,0 +1,25 @@
+language: python
+dist: focal
+matrix:
+  include:
+  - python: 3.6
+    env: TOXENV=py36-core
+  - python: 3.7
+    env: TOXENV=py37-core
+  - python: 3.8
+    env: TOXENV=py38-core
+  - python: 3.9
+    env: TOXENV=py39-core
+install:
+- pip install tox
+
+before_script: >
+  export CLOUDINARY_URL=$(bash tools/get_test_cloud.sh);
+  echo cloud_name: "$(echo $CLOUDINARY_URL | cut -d'@' -f2)"
+script:
+- tox -e $TOXENV
+
+notifications:
+  email:
+    recipients:
+      - [email protected]

cloudinary_cli/core/config.py

@@ -21,7 +21,7 @@ def config(new, ls, show, rm, from_url):
         config_name, cloudinary_url = new or [None, from_url]
 
         if not verify_cloudinary_url(cloudinary_url):
-            return
+            return 1
 
         config_name = config_name or cloudinary.config().cloud_name
 
@@ -31,7 +31,7 @@ def config(new, ls, show, rm, from_url):
         logger.info("Example usage: cld -C {} <command>".format(config_name))
     elif rm:
         if remove_config_keys(rm):
-            logger.warn(f"Configuration '{rm}' not found.")
+            logger.warning(f"Configuration '{rm}' not found.")
         else:
             logger.info(f"Configuration '{rm}' deleted.")
     elif ls:

cloudinary_cli/samples/__init__.py

@@ -28,7 +28,7 @@ def dog(transformation, open_in_browser):
 
 def _handle_sample_command(source, transformation=None, open_in_browser=False, resource_type="image"):
     cloudinary.config(cloud_name="demo", secure_distribution=None, cname=None)
-    res = cloudinary_url(source, raw_transformation=transformation, resource_type=resource_type)[0]
+    res = cloudinary_url(source, raw_transformation=transformation, resource_type=resource_type)
     print(res)
     if open_in_browser:
         open_url(res)

setup.py

@@ -23,7 +23,7 @@
         'console_scripts': [
             'cld=cloudinary_cli.cli:main',
             'cloudinary=cloudinary_cli.cli:main',
-            ],
+        ],
     },
     long_description=long_description,
     long_description_content_type="text/markdown",
@@ -34,7 +34,7 @@
     license="MIT",
     python_requires='>=3.6.0',
     setup_requires=["pytest-runner"],
-    tests_require=["pytest"],
+    tests_require=["pytest", "mock", "urllib3"],
     install_requires=[
         "cloudinary>=1.25.0",
         "pygments",

test/helper_test.py

@@ -0,0 +1,76 @@
+import os
+import random
+import time
+from functools import wraps
+from pathlib import Path
+
+import cloudinary.api
+from cloudinary import logger
+from urllib3 import HTTPResponse, disable_warnings
+from urllib3._collections import HTTPHeaderDict
+
+SUFFIX = os.environ.get('TRAVIS_JOB_ID') or random.randint(10000, 99999)
+
+RESOURCES_DIR = Path.joinpath(Path(__file__).resolve().parent, "resources")
+TEST_FILES_DIR = str(Path.joinpath(RESOURCES_DIR, "test_sync"))
+
+disable_warnings()
+
+
+def unique_suffix(value):
+    return f"{value}_{SUFFIX}"
+
+
+def http_response_mock(body="", headers=None, status=200):
+    if headers is None:
+        headers = {}
+
+    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"}')
+
+
+def retry_assertion(num_tries=3, delay=3):
+    """
+    Helper for retrying inconsistent unit tests
+
+    :param num_tries: Number of tries to perform
+    :param delay: Delay in seconds between retries
+    """
+
+    def retry_decorator(func):
+        @wraps(func)
+        def retry_func(*args, **kwargs):
+            try_num = 1
+            while try_num < num_tries:
+                try:
+                    return func(*args, **kwargs)
+                except AssertionError:
+                    logger.warning("Assertion #{} out of {} failed, retrying in {} seconds".format(try_num, num_tries,
+                                                                                                   delay))
+                    time.sleep(delay)
+                    try_num += 1
+
+            return func(*args, **kwargs)
+
+        return retry_func
+
+    return retry_decorator
+
+
+def delete_cld_folder_if_exists(folder):
+    cloudinary.api.delete_resources_by_prefix(folder)
+    try:
+        cloudinary.api.delete_folder(folder)
+    except cloudinary.exceptions.NotFound:
+        pass