clean up bash scripts, add url type checking

Author: rnpridgeon

confluent_kafka/avro/__init__.py

@@ -30,7 +30,7 @@ def __init__(self, config, default_key_schema=None,
                  default_value_schema=None, schema_registry=None):
 
         sr_conf = {key.replace("schema.registry.", ""): value
-                   for key, value in config.items() if key.startswith("schema.registry.")}
+                   for key, value in config.items() if key.startswith("schema.registry")}
 
         if config.get("schema.registry.basic.auth.credentials.source") == 'SASL_INHERIT':
             sr_conf['sasl.mechanisms'] = config.get('sasl.mechanisms', '')
@@ -103,7 +103,7 @@ class AvroConsumer(Consumer):
     def __init__(self, config, schema_registry=None):
 
         sr_conf = {key.replace("schema.registry.", ""): value
-                   for key, value in config.items() if key.startswith("schema.registry.")}
+                   for key, value in config.items() if key.startswith("schema.registry")}
 
         if config.get("schema.registry.basic.auth.credentials.source") == 'SASL_INHERIT':
             sr_conf['sasl.mechanisms'] = config.get('sasl.mechanisms', '')

confluent_kafka/avro/cached_schema_registry_client.py

@@ -29,6 +29,16 @@
 from .error import ClientError
 from . import loads
 
+import sys
+
+PY3 = sys.version_info[0] == 3
+
+# Python 2 considers int an instance of str
+if PY3:
+    string_types = str,
+else:
+    string_types = basestring,
+
 VALID_LEVELS = ['NONE', 'FULL', 'FORWARD', 'BACKWARD']
 VALID_METHODS = ['GET', 'POST', 'PUT', 'DELETE']
 VALID_AUTH_PROVIDERS = ['URL', 'USER_INFO', 'SASL_INHERIT']
@@ -62,6 +72,7 @@ def __init__(self, url, max_schemas_per_subject=1000, ca_location=None, cert_loc
         # In order to maintain compatibility the url(conf in future versions) param has been preserved for now.
         conf = url
         if not isinstance(url, dict):
+
             conf = {
                 'url': url,
                 'ssl.ca.location': ca_location,
@@ -79,7 +90,8 @@ def __init__(self, url, max_schemas_per_subject=1000, ca_location=None, cert_loc
         """Construct a Schema Registry client"""
 
         # Ensure URL valid scheme is included; http[s]
-        if not conf.get('url', '').startswith('http'):
+        url = conf.get('url', '')
+        if not isinstance(url, string_types) or not url.startswith('http'):
             raise ValueError("Invalid URL provided for Schema Registry")
 
         # subj => { schema => id }
@@ -98,7 +110,7 @@ def __init__(self, url, max_schemas_per_subject=1000, ca_location=None, cert_loc
         self._session = s
 
         if len(conf) > 0:
-            raise ValueError("Unrecognized configuration key(s): {}".format(conf.keys()))
+            raise ValueError("Unrecognized configuration properties: {}".format(conf.keys()))
 
     def __del__(self):
         self.close()

docker/bin/certify.sh

@@ -1,20 +1,15 @@
-#!/usr/bin/env bash
-set -eu
+#!/usr/bin/env bash -eu
 
 DOCKER_BIN="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd )"
-export PASS="abcdefgh"
+PASS="abcdefgh"
+
 source ${DOCKER_BIN}/../.env
 
 if [[ -f ${TLS}/ca-cert ]]; then
     echo "${TLS}/ca-cert found; skipping certificate generation.."
     exit 0
 fi
 
-# Clean up old certs, warning this is
-#for file in $(ls ${TLS});do
-#        rm ${TLS}/${file}
-#done
-
 HOST=$(hostname -f)
 
 echo "Creating ca-cert..."

docker/bin/cluster_up.sh

@@ -11,7 +11,7 @@ await_http() {
     local exit_code
     local attempt=0
 
-    until [[ $(curl ${2}) ]] || [[ ${attempt} -gt 5 ]]; do
+    until [[ `curl ${2}` ]] || [[ ${attempt} -gt 5 ]]; do
         echo "awaiting $1..."
         let "attempt+=1"
         sleep 6
@@ -25,10 +25,10 @@ await_http() {
     exit 1
 }
 
-echo "Configure Environment..."
+echo "Configuring Environment..."
 source ${DOCKER_SOURCE}/.env
 
-echo "Generate SSL certs..."
+echo "Generating SSL certs..."
 ${DOCKER_BIN}/certify.sh
 
 echo "Deploying cluster..."

docker/bin/gen-ssl-certs.sh

@@ -1,5 +1,4 @@
-#!/usr/bin/env bash
-set -eu
+#!/usr/bin/env bash -eu
 #
 #
 # This scripts generates:
@@ -10,9 +9,6 @@ set -eu
 # https://cwiki.apache.org/confluence/display/KAFKA/Deploying+SSL+for+Kafka
 #
 
-DOCKER_BIN="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd )"
-
-
 if [[ "$1" == "-k" ]]; then
     USE_KEYTOOL=1
     shift

setup.py

@@ -9,7 +9,7 @@
 
 if sys.version_info[0] < 3:
     avro = 'avro'
-    INSTALL_REQUIRES.extend(['futures', 'enum34'])
+    INSTALL_REQUIRES.extend(['futures', 'enum34', 'requests'])
 else:
     avro = 'avro-python3'
 

tests/README.md

@@ -48,7 +48,7 @@ To run the entire test suite:
     - ```./tests/run.sh tox [options]```
 
 - Without tox (will run against current interpreter)
-  - ```./tests/run_all.sh```
+  - ```./tests/run.sh```
 
 To run a specific `mode` or set of `modes` use the following syntax
 

tests/avro/test_cached_client.py

@@ -163,6 +163,17 @@ def test_empty_url(self):
                 'url': ''
             })
 
+    def test_invalid_type_url(self):
+        with self.assertRaises(ValueError):
+            self.client = CachedSchemaRegistryClient(
+                url=1)
+
+    def test_invalid_type_url_dict(self):
+        with self.assertRaises(ValueError):
+            self.client = CachedSchemaRegistryClient({
+                "url": 1
+                })
+
     def test_invalid_url(self):
         with self.assertRaises(ValueError):
             self.client = CachedSchemaRegistryClient({

tests/integration/integration_test.py

@@ -157,8 +157,7 @@ def verify_producer():
     # Producer config
     conf = {'bootstrap.servers': bootstrap_servers,
             'error_cb': error_cb,
-            'api.version.request': api_version_request,
-            'default.topic.config': {'produce.offset.report': True}}
+            'api.version.request': api_version_request}
 
     # Create producer
     p = confluent_kafka.Producer(conf)
@@ -756,10 +755,7 @@ def verify_avro():
                  'api.version.request': api_version_request,
                  'api.version.fallback.ms': 0,
                  'broker.version.fallback': '0.11.0.0',
-                 'schema.registry.url': schema_registry_url,
-                 'default.topic.config': {
-                     'produce.offset.report': True
-                 }}
+                 'schema.registry.url': schema_registry_url}
 
     consumer_conf = dict(base_conf, **{
         'group.id': 'test.py',
@@ -777,20 +773,16 @@ def verify_avro_https(mode_conf):
     if mode_conf is None:
         abort_on_missing_configuration('avro-https')
 
-    base_conf = dict({'bootstrap.servers': bootstrap_servers,
-                      'error_cb': error_cb,
-                      'api.version.request': api_version_request},
-                     **mode_conf)
+    base_conf = dict(mode_conf, **{'bootstrap.servers': bootstrap_servers,
+                                   'error_cb': error_cb,
+                                   'api.version.request': api_version_request})
 
-    consumer_conf = dict({
-        'group.id': generate_group_id(),
-        'session.timeout.ms': 6000,
-        'enable.auto.commit': False,
-        'api.version.request': api_version_request,
-        'on_commit': print_commit_result,
-        'default.topic.config': {
-            'auto.offset.reset': 'earliest'
-        }}, **base_conf)
+    consumer_conf = dict(base_conf, **{'group.id': generate_group_id(),
+                                       'session.timeout.ms': 6000,
+                                       'enable.auto.commit': False,
+                                       'api.version.request': api_version_request,
+                                       'on_commit': print_commit_result,
+                                       'auto.offset.reset': 'earliest'})
 
     run_avro_loop(base_conf, consumer_conf)
 
@@ -877,7 +869,7 @@ def run_avro_loop(producer_conf, consumer_conf):
 
     msgcount = 0
     while msgcount < len(combinations):
-        msg = c.poll(0)
+        msg = c.poll(100)
 
         if msg is None or msg.error():
             continue
@@ -1238,8 +1230,8 @@ def print_usage(exitcode, reason=None):
         modes = test_modes
 
     if bootstrap_servers is None or topic is None:
-        print_usage(1, "Properties bootstrap.servers and topic must be set. "
-                       "Use {} as a template when creating a new conf file.".format(testconf_file))
+        print_usage(1, "Missing property bootstrap.servers."
+                       "Ensure {} includes a valid bootstrap.servers property.".format(testconf_file))
 
     print('Using confluent_kafka module version %s (0x%x)' % confluent_kafka.version())
     print('Using librdkafka version %s (0x%x)' % confluent_kafka.libversion())

tests/integration/run.sh

@@ -1,5 +1,4 @@
-#!/usr/bin/env bash
-set -eu
+#!/usr/bin/env bash -eu
 
 cleanup() {
         ${DOCKER_BIN}/cluster_down.sh
@@ -18,17 +17,14 @@ fi
 #start cluster
 ${DOCKER_BIN}/cluster_up.sh
 
-if [[ $? -ne 0 ]]; then
-    exit 1
-fi
-
 run_tox() {
     echo "Executing tox $@"
     cd ${TEST_SOURCE}/..
     tox -r "$@"
 }
 
 run_native() {
+    python setup.py install
     for mode in "$@"; do
         modes="${modes:-} --${mode}"
     done