mock response for testing

etcd3/client.py

@@ -75,6 +75,8 @@ def iter_response(resp):
             yield s
         elif bracket_flag < 0:
             raise Etcd3StreamError("Stream decode error", buf, resp)
+    if buf:
+        raise Etcd3StreamError("Stream decode error", buf, resp)
 
 
 class Etcd3APIClient(AuthAPI, ClusterAPI, KVAPI, LeaseAPI, MaintenanceAPI, WatchAPI):
@@ -109,7 +111,10 @@ def __init__(self, host='localhost', port=2379, protocol='http',
         self.token = token
 
     def __set_conn_pool(self, pool_size):
-        pass
+        from requests.adapters import HTTPAdapter
+        adapter = HTTPAdapter(pool_connections=pool_size, pool_maxsize=pool_size)
+        self.session.mount('http://', adapter)
+        self.session.mount('https://', adapter)
 
     def close(self):
         return self.session.close()

requirements_dev.txt

@@ -6,6 +6,7 @@ watchdog>=0.8.3
 flake8>=2.6.0
 tox>=2.3.1
 coverage>=4.1
+mock==2.0.0
 Sphinx>=1.4.8
 cryptography>=1.7
 PyYAML>=3.11

tests/etcd_go_cli.py

@@ -43,7 +43,7 @@ def find_executable(executable, path=None):
         return None
 
 
-etcdctl_path = find_executable('etcdctl')
+ETCDCTL_PATH = find_executable('etcdctl')
 
 
 def etcdctl(*args, **kwargs):
@@ -54,15 +54,23 @@ def etcdctl(*args, **kwargs):
     version = kwargs.get('version', 3)
 
     envs = {}
-    cmd = [etcdctl_path, '--endpoints', endpoint]
+    cmd = [ETCDCTL_PATH, '--endpoints', endpoint]
     if json:
         cmd.extend(['-w', 'json'])
     if version == 3:
         envs['ETCDCTL_API'] = '3'
     cmd.extend(args)
     p = Popen(cmd, stdout=PIPE, env=envs)
-    return p.communicate()[0]
+    out, err = p.communicate()
+    if not p.returncode == 0:
+        raise RuntimeError(err)
+    return out
 
+try:
+    if ETCDCTL_PATH and etcdctl('--dial-timeout=0.2s endpoint health'):
+        NO_ETCD_SERVICE = False
+except:
+    NO_ETCD_SERVICE = True
 
 if __name__ == '__main__':
     print(etcdctl('get foo'))

tests/mocks.py

@@ -0,0 +1,21 @@
+import json
+
+import six
+from mock import Mock
+
+
+def fake_post(status_code, content, resp_class):
+    mock_response = Mock(spec=resp_class)
+    mock_response.status_code = status_code
+    mock_response.content = content
+
+    def iter_content(chunk_size):
+        p = 0
+        while p < len(content):
+            yield content[p: p + chunk_size]
+            p += chunk_size
+
+    mock_response.iter_content = iter_content
+    mock_response.json = lambda: json.loads(six.text_type(content, encoding='utf-8'))
+
+    return Mock(return_value=mock_response)

tests/test_client.py

@@ -4,29 +4,31 @@
 from etcd3.client import Etcd3APIClient
 from etcd3.errors import Etcd3APIError
 from .envs import protocol, host, port
-from .etcd_go_cli import etcdctl
+from .etcd_go_cli import etcdctl, NO_ETCD_SERVICE
+from .mocks import fake_post
 
 
-@pytest.fixture()
+@pytest.fixture(scope='session')
 def client():
+    """
+    init Etcd3Client, close its connection-pool when teardown
+    """
     c = Etcd3APIClient(host, port, protocol)
     yield c
     c.close()
 
 
+@pytest.mark.skipif(NO_ETCD_SERVICE, reason="no etcd service available")
 def test_request_and_model(client):
     etcdctl('put test_key test_value')
     result = client.call_rpc('/v3alpha/kv/range', {'key': 'test_key'})
     # {"header":{"cluster_id":11588568905070377092,"member_id":128088275939295631,"revision":3,"raft_term":2},"kvs":[{"key":"dGVzdF9rZXk=","create_revision":3,"mod_revision":3,"version":1,"value":"dGVzdF92YWx1ZQ=="}],"count":1}'
-    if six.PY2:
-        assert result.kvs[0].key == 'test_key'
-        assert result.kvs[0].value == 'test_value'
-    else:
-        assert result.kvs[0].key == b'test_key'
-        assert result.kvs[0].value == b'test_value'
+    assert result.kvs[0].key == b'test_key'
+    assert result.kvs[0].value == b'test_value'
     etcdctl('del test_key')
 
 
+@pytest.mark.skipif(NO_ETCD_SERVICE, reason="no etcd service available")
 def test_stream(client):
     r = client.call_rpc('/v3alpha/watch', {'create_request': {'key': 'test_key'}}, stream=True)
     times = 3
@@ -36,19 +38,57 @@ def test_stream(client):
             break
         etcdctl('put test_key test_value')
         if hasattr(i, 'events'):
-            if six.PY2:
-                assert i.events[0].kv.key == 'test_key'
-                assert i.events[0].kv.value == 'test_value'
-            else:
-                assert i.events[0].kv.key == b'test_key'
-                assert i.events[0].kv.value == b'test_value'
+            assert i.events[0].kv.key == b'test_key'
+            assert i.events[0].kv.value == b'test_value'
             times -= 1
         else:
             header_times -= 1
             assert header_times >= 0
     r.close()
 
 
-def test_exception(client):
+@pytest.mark.skipif(NO_ETCD_SERVICE, reason="no etcd service available")
+def test_request_exception(client):
     with pytest.raises(Etcd3APIError, match=r".*'Not Found'.*"):
-        client.call_rpc('/v3alpha/kv/rang', {'key': 'foa'})
+        client.call_rpc('/v3alpha/kv/rag', {})  # non exist path
+
+
+def test_patched_stream(client, monkeypatch):
+    s = b'{"result": {"header": {"raft_term": 7, "member_id": 128088275939295631, "cluster_id": 11588568905070377092, "revision": 378}, "created": true}}' \
+        b'{"result": {"header": {"raft_term": 7, "member_id": 128088275939295631, "cluster_id": 11588568905070377092, "revision": 379}, "events": [{"kv": {"mod_revision": 379, "value": "dGVzdF92YWx1ZQ==", "create_revision": 379, "version": 1, "key": "dGVzdF9rZXk="}}]}}' \
+        b'{"result": {"header": {"raft_term": 7, "member_id": 128088275939295631, "cluster_id": 11588568905070377092, "revision": 380}, "events": [{"kv": {"mod_revision": 380, "value": "dGVzdF92YWx1ZQ==", "create_revision": 379, "version": 2, "key": "dGVzdF9rZXk="}}]}}' \
+        b'{"result": {"header": {"raft_term": 7, "member_id": 128088275939295631, "cluster_id": 11588568905070377092, "revision": 381}, "events": [{"kv": {"mod_revision": 381, "value": "dGVzdF92YWx1ZQ==", "create_revision": 379, "version": 3, "key": "dGVzdF9rZXk="}}]}}'
+    post = fake_post(200, s, client.response_class)
+    monkeypatch.setattr(client.session, 'post', post)
+    r = client.call_rpc('/v3alpha/watch', {'create_request': {'key': 'test_key'}}, stream=True)
+    times = 3
+    header_times = 1
+    for i in r:
+        if not times:
+            break
+        if hasattr(i, 'events'):
+            assert i.events[0].kv.key == b'test_key'
+            assert i.events[0].kv.value == b'test_value'
+            times -= 1
+        else:
+            header_times -= 1
+            assert header_times >= 0
+
+
+def test_patched_request_and_model(client, monkeypatch):
+    s = b'{"header":{"cluster_id":11588568905070377092,"member_id":128088275939295631,"revision":3,"raft_term":2},' \
+        b'"kvs":[{"key":"dGVzdF9rZXk=","create_revision":3,"mod_revision":3,"version":1,"value":"dGVzdF92YWx1ZQ=="}],' \
+        b'"count":1}'
+    post = fake_post(200, s, client.response_class)
+    monkeypatch.setattr(client.session, 'post', post)
+    result = client.call_rpc('/v3alpha/kv/range', {'key': 'test_key'})
+    assert post.call_args[1]['json']['key'] == 'dGVzdF9rZXk='
+    assert result.kvs[0].key == b'test_key'
+    assert result.kvs[0].value == b'test_value'
+
+
+def test_patched_request_exception(client, monkeypatch):
+    post = fake_post(404, 'Not Found', client.response_class)
+    monkeypatch.setattr(client.session, 'post', post)
+    with pytest.raises(Etcd3APIError, match=r".*'Not Found'.*"):
+        client.call_rpc('/v3alpha/kv/rag', {})  # non exist path

tox.ini

@@ -1,5 +1,6 @@
 [tox]
-envlist = py27, py34, py35, py36, flake8
+;envlist = py27, py34, py35, py36, flake8
+envlist = py27, py34, py35, py36
 
 [travis]
 python =
@@ -28,6 +29,7 @@ deps =
 commands =
     pip install -U pip
     pip install -r requirements.txt
+    pip install -r requirements_dev.txt
     py.test --basetemp={envtmpdir}