Updated test suite

Author: sserrata

tests/test_DirectorySyncService.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""Tests for DirectorySyncService."""
+
+import os
+import sys
+
+import pytest
+
+curpath = os.path.dirname(os.path.abspath(__file__))
+sys.path[:0] = [os.path.join(curpath, os.pardir)]
+
+from pancloud.directorysync import DirectorySyncService
+from pancloud.httpclient import HTTPClient
+from pancloud.exceptions import RequiredKwargsError, \
+    UnexpectedKwargsError
+
+
+TARPIT = os.environ.get('TARPIT', 'http://10.255.255.1')
+
+
+class TestLoggingService:
+
+    def test_entry_points(self):
+
+        DirectorySyncService(url=TARPIT).session
+        DirectorySyncService(url=TARPIT).kwargs
+        DirectorySyncService(url=TARPIT).url
+        DirectorySyncService(url=TARPIT).attributes
+        DirectorySyncService(url=TARPIT).query
+        DirectorySyncService(url=TARPIT).domains
+        DirectorySyncService(url=TARPIT).count
+
+    def test_required_kwargs(self):
+        with pytest.raises(RequiredKwargsError):
+            DirectorySyncService()
+
+        with pytest.raises(RequiredKwargsError):
+            DirectorySyncService(session=None)
+
+    def test_unexpected_kwargs(self):
+        with pytest.raises(UnexpectedKwargsError):
+            DirectorySyncService(url=TARPIT, foo='foo')
+
+    def test_session(self):
+        session = HTTPClient(url=TARPIT)
+        DirectorySyncService(session=session)

tests/test_EventService.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""Tests for EventService."""
+
+import os
+import sys
+
+import pytest
+
+curpath = os.path.dirname(os.path.abspath(__file__))
+sys.path[:0] = [os.path.join(curpath, os.pardir)]
+
+from pancloud.event import EventService
+from pancloud.httpclient import HTTPClient
+from pancloud.exceptions import RequiredKwargsError, \
+    UnexpectedKwargsError
+
+
+TARPIT = os.environ.get('TARPIT', 'http://10.255.255.1')
+
+
+class TestLoggingService:
+
+    def test_entry_points(self):
+
+        EventService(url=TARPIT).session
+        EventService(url=TARPIT).kwargs
+        EventService(url=TARPIT).url
+        EventService(url=TARPIT).get_filters
+        EventService(url=TARPIT).set_filters
+        EventService(url=TARPIT).poll
+        EventService(url=TARPIT).ack
+        EventService(url=TARPIT).nack
+
+    def test_required_kwargs(self):
+        with pytest.raises(RequiredKwargsError):
+            EventService()
+
+        with pytest.raises(RequiredKwargsError):
+            EventService(session=None)
+
+    def test_unexpected_kwargs(self):
+        with pytest.raises(UnexpectedKwargsError):
+            EventService(url=TARPIT, foo='foo')
+
+    def test_session(self):
+        session = HTTPClient(url=TARPIT)
+        EventService(session=session)

tests/test_HTTPClient.py

@@ -3,50 +3,76 @@
 
 """Tests for Requests HTTPClient wrapper."""
 
-from __future__ import division
-import json
 import os
-import pickle
-import collections
-import contextlib
-import warnings
+import sys
 
-import io
 import pytest
-import os
-import sys
 
 curpath = os.path.dirname(os.path.abspath(__file__))
 sys.path[:0] = [os.path.join(curpath, os.pardir)]
 
-from pancloud import httpclient
-
-# from .compat import StringIO, u
-# from .utils import override_environ
-from urllib3.util import Timeout as Urllib3Timeout
-import socket
+from pancloud.httpclient import HTTPClient
+from pancloud.exceptions import HTTPError, RequiredKwargsError, \
+    UnexpectedKwargsError
 
-# Requests to this URL should always fail with a connection timeout (nothing
-# listening on that port)
-TARPIT = 'http://10.255.255.1'
 
-try:
-    from ssl import SSLContext
-    del SSLContext
-    HAS_MODERN_SSL = True
-except ImportError:
-    HAS_MODERN_SSL = False
-
-try:
-    httpclient.pyopenssl
-    HAS_PYOPENSSL = True
-except AttributeError:
-    HAS_PYOPENSSL = False
+HTTPBIN = os.environ.get('HTTPBIN_URL', 'http://httpbin.org')
+TARPIT = os.environ.get('TARPIT', 'http://10.255.255.1')
 
 
 class TestHTTPClient:
 
     def test_entry_points(self):
-        pass
+
+        HTTPClient(url=TARPIT).request
+        HTTPClient(url=TARPIT).pyopenssl
+
+    def test_invalid_url(self):
+        with pytest.raises(HTTPError):
+            HTTPClient(url='asdaksjhdakjsdh').request(method='GET')
+        with pytest.raises(HTTPError):
+            HTTPClient(url='http://').request(method='GET')
+
+    def test_required_kwargs(self):
+        with pytest.raises(RequiredKwargsError):
+            HTTPClient().request()
+
+    def test_connection_timeout(self):
+        with pytest.raises(HTTPError):
+            HTTPClient(url=TARPIT).request(
+                method='GET', timeout=(.1, None)
+            )
+
+    def test_read_timeout(self):
+        with pytest.raises(HTTPError):
+            HTTPClient(url=HTTPBIN, port=80).request(
+                method='GET', timeout=(None, .0001), path='/'
+            )
+
+    def test_httpclient_unexpected_kwargs(self):
+        with pytest.raises(UnexpectedKwargsError):
+            HTTPClient(url=TARPIT, foo='foo').request(method='GET')
+
+    def test_request_unexpected_kwargs(self):
+        with pytest.raises(UnexpectedKwargsError):
+            HTTPClient(url=TARPIT).request(method='GET', foo='foo')
+
+    def test_enforce_json(self):
+        with pytest.raises(HTTPError):
+            HTTPClient(
+                url=HTTPBIN,
+                port=80,
+                enforce_json=True,
+                headers={'Accept': 'application/json'}
+            ).request(method='GET', path='/')
+
+    def test_raise_for_status(self):
+        with pytest.raises(HTTPError):
+            HTTPClient(
+                url=HTTPBIN,
+                port=80,
+                raise_for_status=True
+            ).request(method='GET', path='/status/400')
+
 
 

tests/test_LoggingService.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""Tests for LoggingService."""
+
+import os
+import sys
+
+import pytest
+
+curpath = os.path.dirname(os.path.abspath(__file__))
+sys.path[:0] = [os.path.join(curpath, os.pardir)]
+
+from pancloud.logging import LoggingService
+from pancloud.httpclient import HTTPClient
+from pancloud.exceptions import RequiredKwargsError, \
+    UnexpectedKwargsError
+
+
+TARPIT = os.environ.get('TARPIT', 'http://10.255.255.1')
+
+
+class TestLoggingService:
+
+    def test_entry_points(self):
+
+        LoggingService(url=TARPIT).session
+        LoggingService(url=TARPIT).kwargs
+        LoggingService(url=TARPIT).url
+        LoggingService(url=TARPIT).poll
+        LoggingService(url=TARPIT).query
+        LoggingService(url=TARPIT).delete
+        LoggingService(url=TARPIT).iter_poll
+        LoggingService(url=TARPIT).poll_all
+        LoggingService(url=TARPIT).xpoll
+
+    def test_required_kwargs(self):
+        with pytest.raises(RequiredKwargsError):
+            LoggingService()
+
+        with pytest.raises(RequiredKwargsError):
+            LoggingService(session=None)
+
+    def test_unexpected_kwargs(self):
+        with pytest.raises(UnexpectedKwargsError):
+            LoggingService(url=TARPIT, foo='foo')
+
+    def test_session(self):
+        session = HTTPClient(url=TARPIT)
+        LoggingService(session=session)