Improve unit tests.

.gitignore

@@ -29,3 +29,4 @@ __pycache__
 /winrm/tests/config.json
 .pytest_cache
 venv
+.tox

requirements-test.txt

@@ -4,3 +4,4 @@ pytest==4.4.2
 pytest-cov==2.7.1
 pytest-flake8==1.0.4
 mock==3.0.5
+requests_mock==1.5.2

requirements.txt

@@ -0,0 +1,4 @@
+xmltodict
+requests>=2.9.1
+requests_ntlm>=0.3.0
+six>=1.7.0

requirements/extras/requirements-credssp.txt

@@ -0,0 +1 @@
+requests-credssp>=1.0.0

requirements/extras/requirements-kerberos.txt

@@ -0,0 +1,2 @@
+winkerberos>=0.5.0 ; sys_platform=="win32"
+pykerberos>=1.2.1,<2.0.0 ; sys_platform!="win32"

requirements/requirements-testmin.txt

@@ -0,0 +1,13 @@
+xmltodict==0.3.0
+requests==2.9.1
+requests_ntlm==0.3.0
+six==1.7.0
+
+# credssp
+requests-credssp==1.0.0
+
+# kerberos
+winkerberos==0.5.0 ; sys_platform=="win32"
+pykerberos==1.2.1 ; sys_platform!="win32"
+
+-r ../requirements-test.txt

setup.py

@@ -1,3 +1,5 @@
+import os
+
 from setuptools import setup
 
 __version__ = '0.3.1.dev0'
@@ -11,6 +13,29 @@
 except ImportError:
     long_description = ''
 
+
+def install_deps():
+    default = open('requirements.txt', 'r').readlines()
+    pkg_list = []
+    for resource in default:
+        pkg_list.append(resource.strip())
+    return pkg_list
+
+
+def install_extras():
+    extras = dict()
+    for filename in os.listdir('requirements/extras'):
+        extra_key = filename.replace('requirements-', '').replace('.txt', '')
+        extras[extra_key] = list()
+        with open("requirements/extras/" + filename, 'r') as req_file:
+            for pkg_name in req_file.readlines():
+                extras[extra_key].append(pkg_name)
+    return extras
+
+
+req_deps_list = install_deps()
+extras_deps = install_extras()
+
 setup(
     name=project_name,
     version=__version__,
@@ -23,12 +48,8 @@
     license='MIT license',
     packages=('winrm', 'winrm.tests', 'winrm.vendor.requests_kerberos'),
     package_data={'winrm.tests': ['*.ps1']},
-    install_requires=['xmltodict', 'requests>=2.9.1', 'requests_ntlm>=0.3.0', 'six'],
-    extras_require={
-        'credssp': ['requests-credssp>=1.0.0'],
-        'kerberos:sys_platform=="win32"': ['winkerberos>=0.5.0'],
-        'kerberos:sys_platform!="win32"': ['pykerberos>=1.2.1,<2.0.0']
-    },
+    install_requires=req_deps_list,
+    extras_require=extras_deps,
     classifiers=[
         'Development Status :: 4 - Beta',
         'Environment :: Console',

tox.ini

@@ -0,0 +1,67 @@
+[tox]
+envlist = py27,py35,py36,py37,py38,py27min,py37min,py27base,py37base
+skip_missing_interpreters = True
+skipsdist = True
+sitepackages = False
+
+[testenv]
+ignore_errors = False
+deps =
+  -r{toxinidir}/requirements-test.txt
+  -r{toxinidir}/requirements/extras/requirements-credssp.txt
+  -r{toxinidir}/requirements/extras/requirements-kerberos.txt
+  -r{toxinidir}/requirements.txt
+setenv =
+  PYWINRM_TEST_CREDSSP=1
+  PYWINRM_TEST_KERBEROS=1
+commands =
+  pytest -v --flake8 --cov=winrm --cov-report=term-missing winrm/tests/
+whitelist_externals = bash
+
+# Test the base install, without extras.
+[testenv:py27base]
+ignore_errors = False
+deps =
+  -r{toxinidir}/requirements-test.txt
+  -r{toxinidir}/requirements.txt
+setenv =
+  PYWINRM_TEST_CREDSSP=0
+  PYWINRM_TEST_KERBEROS=0
+commands =
+  pytest -v --flake8 --cov=winrm --cov-report=term-missing winrm/tests/
+whitelist_externals = bash
+
+[testenv:py37base]
+ignore_errors = False
+deps =
+  -r{toxinidir}/requirements-test.txt
+  -r{toxinidir}/requirements.txt
+setenv =
+  PYWINRM_TEST_CREDSSP=0
+  PYWINRM_TEST_KERBEROS=0
+commands =
+  pytest -v --flake8 --cov=winrm --cov-report=term-missing winrm/tests/
+whitelist_externals = bash
+
+# Test the minimum requirements for the packages.
+[testenv:py27min]
+ignore_errors = False
+deps =
+  -r{toxinidir}/requirements/requirements-testmin.txt
+setenv =
+  PYWINRM_TEST_CREDSSP=1
+  PYWINRM_TEST_KERBEROS=1
+commands =
+  pytest -v --flake8 --cov=winrm --cov-report=term-missing winrm/tests/
+whitelist_externals = bash
+
+[testenv:py37min]
+ignore_errors = False
+deps =
+  -r{toxinidir}/requirements/requirements-testmin.txt
+setenv =
+  PYWINRM_TEST_CREDSSP=1
+  PYWINRM_TEST_KERBEROS=1
+commands =
+  pytest -v --flake8 --cov=winrm --cov-report=term-missing winrm/tests/
+whitelist_externals = bash

winrm/tests/base.py

@@ -0,0 +1,45 @@
+import os
+import unittest
+import requests_mock
+from winrm import transport
+
+if os.environ.get('PYWINRM_TEST_CREDSSP') == '1':
+    EXPECT_CREDSSP = True
+elif os.environ.get('PYWINRM_TEST_CREDSSP') == '0':
+    EXPECT_CREDSSP = False
+else:
+    EXPECT_CREDSSP = transport.HAVE_CREDSSP
+
+if os.environ.get('PYWINRM_TEST_KERBEROS') == '1':
+    EXPECT_KERBEROS = True
+elif os.environ.get('PYWINRM_TEST_KERBEROS') == '0':
+    EXPECT_KERBEROS = False
+else:
+    EXPECT_KERBEROS = transport.HAVE_KERBEROS
+
+
+class BaseTest(unittest.TestCase):
+    maxDiff = 2048
+    _old_env = None
+
+    def setUp(self):
+        super(BaseTest, self).setUp()
+        self.mocked_request = requests_mock.Mocker()
+        self.mocked_request.start()
+        self._old_env = {}
+        os.environ.pop('REQUESTS_CA_BUNDLE', None)
+        os.environ.pop('TRAVIS_APT_PROXY', None)
+        os.environ.pop('CURL_CA_BUNDLE', None)
+        os.environ.pop('HTTPS_PROXY', None)
+        os.environ.pop('HTTP_PROXY', None)
+        os.environ.pop('NO_PROXY', None)
+
+    def tearDown(self):
+        super(BaseTest, self).tearDown()
+        os.environ.pop('REQUESTS_CA_BUNDLE', None)
+        os.environ.pop('TRAVIS_APT_PROXY', None)
+        os.environ.pop('CURL_CA_BUNDLE', None)
+        os.environ.pop('HTTPS_PROXY', None)
+        os.environ.pop('HTTP_PROXY', None)
+        os.environ.pop('NO_PROXY', None)
+        self.mocked_request.stop()

winrm/tests/test_protocol.py

@@ -1,8 +1,22 @@
 import pytest
+import copy
+import xmltodict
 
+from winrm.tests import base as base_test
+from winrm.tests.winrm_responses import shells as shell_responses
 from winrm.protocol import Protocol
 
 
+def convert_to_dict(ordered_dict):
+    new_dict = dict()
+    for name, value in ordered_dict.items():
+        if isinstance(value, dict):
+            new_dict[name] = convert_to_dict(value)
+        else:
+            new_dict[name] = value
+    return new_dict
+
+
 def test_open_shell_and_close_shell(protocol_fake):
     shell_id = protocol_fake.open_shell()
     assert shell_id == '11111111-1111-1111-1111-111111111113'
@@ -71,3 +85,23 @@ def test_fail_set_operation_timeout_as_sec():
                  operation_timeout_sec='29a')
     assert str(exc.value) == "failed to parse operation_timeout_sec as int: " \
         "invalid literal for int() with base 10: '29a'"
+
+
+class TestTransportShells(base_test.BaseTest):
+
+    def test_open_shell(self):
+        self.mocked_request.post('https://example.com', text=shell_responses.OPEN_SHELL_RESPONSE)
+        server_conn = Protocol(endpoint="https://example.com",
+                               username='test',
+                               password='test',
+                               )
+        response = server_conn.open_shell()
+        self.assertEqual('5207F2DF-E6CA-4D10-8C7F-5380F01D6FDE', response)
+
+        # MessageID will be dynamic, no need to compare
+        expected_shell_request = xmltodict.parse(copy.deepcopy(shell_responses.OPEN_SHELL_REQUEST))
+        actual_shell_request = xmltodict.parse(copy.deepcopy(self.mocked_request.request_history[0].body))
+        del expected_shell_request['env:Envelope']['env:Header']['a:MessageID']
+        del actual_shell_request['env:Envelope']['env:Header']['a:MessageID']
+
+        self.assertEqual(convert_to_dict(expected_shell_request), convert_to_dict(actual_shell_request))

winrm/tests/test_transport.py

@@ -2,36 +2,23 @@
 import os
 import mock
 import unittest
+import requests
+from . import base as base_test
+from distutils.version import StrictVersion
 
 from winrm import transport
 from winrm.exceptions import WinRMError, InvalidCredentialsError
 
+REQUEST_VERSION = requests.__version__.split('.')
 
-class TestTransport(unittest.TestCase):
-    maxDiff = 2048
-    _old_env = None
+
+class TestTransport(base_test.BaseTest):
 
     def setUp(self):
         super(TestTransport, self).setUp()
-        self._old_env = {}
-        os.environ.pop('REQUESTS_CA_BUNDLE', None)
-        os.environ.pop('TRAVIS_APT_PROXY', None)
-        os.environ.pop('CURL_CA_BUNDLE', None)
-        os.environ.pop('HTTPS_PROXY', None)
-        os.environ.pop('HTTP_PROXY', None)
-        os.environ.pop('NO_PROXY', None)
         transport.DISPLAYED_PROXY_WARNING = False
         transport.DISPLAYED_CA_TRUST_WARNING = False
 
-    def tearDown(self):
-        super(TestTransport, self).tearDown()
-        os.environ.pop('REQUESTS_CA_BUNDLE', None)
-        os.environ.pop('TRAVIS_APT_PROXY', None)
-        os.environ.pop('CURL_CA_BUNDLE', None)
-        os.environ.pop('HTTPS_PROXY', None)
-        os.environ.pop('HTTP_PROXY', None)
-        os.environ.pop('NO_PROXY', None)
-
     def test_build_session_cert_validate_default(self):
         t_default = transport.Transport(endpoint="https://example.com",
                                         username='test',
@@ -144,6 +131,24 @@ def test_build_session_cert_ignore_2(self):
         t_default.build_session()
         self.assertIs(False, t_default.session.verify)
 
+    # TODO: I am not sure in which version changed specifically, but this can be updated if we need to find out.
+    @unittest.skipIf(StrictVersion(requests.__version__) > StrictVersion('2.9.1'), reason="Skipping for versions 2.9.1 or older")
+    def test_build_session_proxy_none_old_request(self):
+        os.environ['HTTP_PROXY'] = 'random_proxy'
+        os.environ['HTTPS_PROXY'] = 'random_proxy_2'
+
+        t_default = transport.Transport(endpoint="https://example.com",
+                                        server_cert_validation='validate',
+                                        username='test',
+                                        password='test',
+                                        auth_method='basic',
+                                        proxy=None
+                                        )
+
+        t_default.build_session()
+        self.assertEqual({'no_proxy': '*', 'http': 'random_proxy', 'https': 'random_proxy_2'}, t_default.session.proxies)
+
+    @unittest.skipIf(StrictVersion(requests.__version__) <= StrictVersion('2.9.1'), reason="Skipping for versions newer than 2.9.1")
     def test_build_session_proxy_none(self):
         os.environ['HTTP_PROXY'] = 'random_proxy'
         os.environ['HTTPS_PROXY'] = 'random_proxy_2'
@@ -308,3 +313,49 @@ def test_close_session_not_built(self, mock_session):
         t_default.close_session()
         self.assertFalse(mock_session.return_value.close.called)
         self.assertIsNone(t_default.session)
+
+
+class TestTransportCredSSP(base_test.BaseTest):
+
+    @unittest.skipIf(base_test.EXPECT_CREDSSP is False, reason="Only testing when CredSSP is available")
+    def test_with_credssp(self):
+        t_default = transport.Transport(endpoint="https://example.com",
+                                        username='test',
+                                        password='test',
+                                        auth_method='credssp',
+                                        )
+        t_default.build_session()
+
+    @unittest.skipIf(base_test.EXPECT_CREDSSP is True, reason="Only testing when CredSSP is unavailable")
+    def test_without_credssp(self):
+        t_default = transport.Transport(endpoint="https://example.com",
+                                        username='test',
+                                        password='test',
+                                        auth_method='credssp',
+                                        )
+        with self.assertRaises(WinRMError) as exc:
+            t_default.build_session()
+        self.assertEqual(str(exc.exception), 'requests auth method is credssp, but requests-credssp is not installed')
+
+
+class TestTransportKerberos(base_test.BaseTest):
+
+    @unittest.skipIf(base_test.EXPECT_KERBEROS is False, reason="Only testing when kerberos is available")
+    def test_with_kerberos(self):
+        t_default = transport.Transport(endpoint="https://example.com",
+                                        username='test',
+                                        password='test',
+                                        auth_method='kerberos',
+                                        )
+        t_default.build_session()
+
+    @unittest.skipIf(base_test.EXPECT_KERBEROS is True, reason="Only testing when kerberos is unavailable")
+    def test_without_kerberos(self):
+        t_default = transport.Transport(endpoint="https://example.com",
+                                        username='test',
+                                        password='test',
+                                        auth_method='kerberos',
+                                        )
+        with self.assertRaises(WinRMError) as exc:
+            t_default.build_session()
+        self.assertEqual(str(exc.exception), 'requested auth method is kerberos, but requests_kerberos is not installed')