Reorganize the API (#1)

* Test all Python versions

* Extensive changes to make the the plugin easier to use

* Do not disable the socket automatically.
* Add @pytest.mark.disable_socket and @pytest.mark.enable_socket to conveniently mark tests without importing the package
* Add socket_disabled and socket_enabled fixtures.
* Simplify socket blocking code.
* Remove printed messages.

* Relax py.test version requirement

* Add a command-line option, create a custom exception

* Fix Python 2 incompatibility

* Fix tests unintentionally modifying the socket status for other tests

* Formatting

Make flake8 happy

* Fix CI issues

* Fix pypy's HTTPS error

.travis.yml

@@ -2,21 +2,19 @@
 
 sudo: false
 language: python
-
-matrix:
- include:
- - python: 2.7
- env: TOX_ENV=py27
- - python: pypy
- env: TOX_ENV=pypy
- - python: 3.6
- env: TOX_ENV=flake8
+python:
+ - "2.7"
+ - "3.3"
+ - "3.4"
+ - "3.5"
+ - "3.6"
+ - "pypy"
 
 install:
- - pip install tox
+ - pip install tox-travis
 
 script:
- - tox -e $TOX_ENV
+ - tox -v
 
 before_cache:
  - rm -rf $HOME/.cache/pip/log

appveyor.yml

@@ -6,6 +6,18 @@ environment:
  - PYTHON: "C:\\Python27"
  TOX_ENV: "py27"
 
+ - PYTHON: "C:\\Python33"
+ TOX_ENV: "py33"
+
+ - PYTHON: "C:\\Python34"
+ TOX_ENV: "py34"
+
+ - PYTHON: "C:\\Python35"
+ TOX_ENV: "py35"
+
+ - PYTHON: "C:\\Python36"
+ TOX_ENV: "py36"
+
 init:
  - "%PYTHON%/python -V"
  - "%PYTHON%/python -c \"import struct;print( 8 * struct.calcsize(\'P\'))\""
@@ -18,7 +30,7 @@ install:
 build: false # Not a C# project, build stuff at the test step instead.
 
 test_script:
- - "%PYTHON%/Scripts/tox -e %TOX_ENV%"
+ - "%PYTHON%/Scripts/tox -v -e %TOX_ENV%"
 
 after_test:
  - "%PYTHON%/python setup.py bdist_wheel"

pytest_socket.py

@@ -1,45 +1,64 @@
 # -*- coding: utf-8 -*-
 import socket
-import sys
 
 import pytest
 
-_module = sys.modules[__name__]
+_true_socket = socket.socket
 
 
-def pytest_runtest_setup():
- """ disable the internet. test-cases can explicitly re-enable """
+class SocketBlockedError(RuntimeError):
+ def __init__(self, *args, **kwargs):
+ super(SocketBlockedError, self).__init__("A test tried to use socket.socket.")
+
+
+def pytest_addoption(parser):
+ group = parser.getgroup('socket')
+ group.addoption(
+ '--disable-socket',
+ action='store_true',
+ dest='disable_socket',
+ help='Disable socket.socket by default to block network calls.'
+ )
+
+
+@pytest.fixture(autouse=True)
+def _socket_marker(request):
+ if request.node.get_marker('disable_socket'):
+ request.getfixturevalue('socket_disabled')
+ if request.node.get_marker('enable_socket'):
+ request.getfixturevalue('socket_enabled')
+
+ if request.config.getoption('--disable-socket'):
+ request.getfixturevalue('socket_disabled')
+
+
+@pytest.fixture
+def socket_disabled():
+ """ disable socket.socket for duration of this test function """
  disable_socket()
+ yield
+ enable_socket()
 
 
-@pytest.fixture(scope='function')
-def socket_enabled(request):
- """ re-enable socket.socket for duration of this test function """
+@pytest.fixture
+def socket_enabled():
+ """ enable socket.socket for duration of this test function """
  enable_socket()
- request.addfinalizer(disable_socket)
+ yield
+ disable_socket()
 
 
 def disable_socket():
  """ disable socket.socket to disable the Internet. useful in testing.
  """
- setattr(_module, u'_socket_disabled', True)
 
  def guarded(*args, **kwargs):
- if getattr(_module, u'_socket_disabled', False):
- raise RuntimeError(
- u"A test tried to use socket.socket without explicitly un-blocking it.")
- else:
- # SocketType is a valid, public alias of socket.socket,
- # we use it here to avoid namespace collisions
- return socket.SocketType(*args, **kwargs)
+ raise SocketBlockedError()
 
  socket.socket = guarded
 
- print(u'[!] socket.socket is now blocked. The network should be inaccessible.')
-
 
 def enable_socket():
  """ re-enable socket.socket to enable the Internet. useful in testing.
  """
- setattr(_module, u'_socket_disabled', False)
- print(u'[!] socket.socket is UN-blocked, and the network can be accessed.')
+ socket.socket = _true_socket

setup.py

@@ -23,7 +23,7 @@ def read(fname):
  description='Pytest Plugin to disable socket calls during tests',
  long_description=read('README.rst'),
  py_modules=['pytest_socket'],
- install_requires=['pytest>=3.1.1'],
+ install_requires=['pytest>=3.0.0'],
  classifiers=[
  'Development Status :: 4 - Beta',
  'Framework :: Pytest',

tests/test_socket.py

@@ -1,41 +1,185 @@
 # -*- coding: utf-8 -*-
+import pytest
 
+from pytest_socket import enable_socket
 
-def test_disabled_socket_fails(testdir):
+
+@pytest.fixture(autouse=True)
+def reenable_socket():
+ # The tests can leave the socket disabled in the global scope.
+ # Fix that by automatically re-enabling it after each test
+ yield
+ enable_socket()
+
+
+def assert_socket_blocked(result):
+ result.stdout.fnmatch_lines("""
+ *SocketBlockedError: A test tried to use socket.socket.*
+ """)
+
+
+def test_socket_enabled_by_default(testdir):
  testdir.makepyfile("""
  import socket
+
+ def test_socket():
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ """)
+ result = testdir.runpytest("--verbose")
+ result.assert_outcomes(1, 0, 0)
+ with pytest.raises(Exception):
+ assert_socket_blocked(result)
+
+
+def test_global_disable_via_fixture(testdir):
+ testdir.makepyfile("""
+ import pytest
+ import pytest_socket
+ import socket
+
+ @pytest.fixture(autouse=True)
+ def disable_socket_for_all():
+ pytest_socket.disable_socket()
+
  def test_socket():
  socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  """)
  result = testdir.runpytest("--verbose")
  result.assert_outcomes(0, 0, 1)
- result.stdout.fnmatch_lines("""
- *RuntimeError: A test tried to use socket.socket without explicitly un-blocking it.*
+ assert_socket_blocked(result)
+
+
+def test_global_disable_via_cli_flag(testdir):
+ testdir.makepyfile("""
+ import socket
+
+ def test_socket():
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  """)
+ result = testdir.runpytest("--verbose", "--disable-socket")
+ result.assert_outcomes(0, 0, 1)
+ assert_socket_blocked(result)
 
 
-def test_disabled_httplib_fails(testdir):
+def test_global_disable_via_config(testdir):
  testdir.makepyfile("""
- import httplib
- def test_httplib():
- httplib.HTTPConnection("scanme.nmap.org:80").request("GET", "/")
+ import socket
+
+ def test_socket():
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ """)
+ testdir.makeini("""
+ [pytest]
+ addopts = --disable-socket
  """)
  result = testdir.runpytest("--verbose")
  result.assert_outcomes(0, 0, 1)
- result.stdout.fnmatch_lines("""
- *RuntimeError: A test tried to use socket.socket without explicitly un-blocking it.*
+ assert_socket_blocked(result)
+
+
+def test_disable_socket_marker(testdir):
+ testdir.makepyfile("""
+ import pytest
+ import socket
+
+ @pytest.mark.disable_socket
+ def test_socket():
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  """)
+ result = testdir.runpytest("--verbose")
+ result.assert_outcomes(0, 0, 1)
+ assert_socket_blocked(result)
+
+
+def test_enable_socket_marker(testdir):
+ testdir.makepyfile("""
+ import pytest
+ import socket
+
+ @pytest.mark.enable_socket
+ def test_socket():
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ """)
+ result = testdir.runpytest("--verbose", "--disable-socket")
+ result.assert_outcomes(0, 0, 1)
+ assert_socket_blocked(result)
+
+
+def test_urllib_succeeds_by_default(testdir):
+ testdir.makepyfile("""
+ try:
+ from urllib.request import urlopen
+ except ImportError:
+ from urllib2 import urlopen
+
+ def test_disable_socket_urllib():
+ assert urlopen('http:https://httpbin.org/get').getcode() == 200
+ """)
+ result = testdir.runpytest("--verbose")
+ result.assert_outcomes(1, 0, 0)
+
+
+def test_enabled_urllib_succeeds(testdir):
+ testdir.makepyfile("""
+ import pytest
+ import pytest_socket
+ try:
+ from urllib.request import urlopen
+ except ImportError:
+ from urllib2 import urlopen
+
+ @pytest.mark.enable_socket
+ def test_disable_socket_urllib():
+ assert urlopen('http:https://httpbin.org/get').getcode() == 200
+ """)
+ result = testdir.runpytest("--verbose", "--disable-socket")
+ result.assert_outcomes(0, 0, 1)
+ assert_socket_blocked(result)
+
+
+def test_disabled_urllib_fails(testdir):
+ testdir.makepyfile("""
+ import pytest
+ try:
+ from urllib.request import urlopen
+ except ImportError:
+ from urllib2 import urlopen
+
+ @pytest.mark.disable_socket
+ def test_disable_socket_urllib():
+ assert urlopen('http:https://httpbin.org/get').getcode() == 200
+ """)
+ result = testdir.runpytest("--verbose")
+ result.assert_outcomes(0, 0, 1)
+ assert_socket_blocked(result)
 
 
-def test_enabling_while_enabled_does_nothing(testdir):
+def test_double_call_does_nothing(testdir):
  testdir.makepyfile("""
+ import pytest
  import pytest_socket
- def test_enabled():
+ import socket
+
+ def test_double_enabled():
+ pytest_socket.enable_socket()
  pytest_socket.enable_socket()
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+ def test_double_disabled():
+ pytest_socket.disable_socket()
+ pytest_socket.disable_socket()
+ with pytest.raises(pytest_socket.SocketBlockedError):
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+ def test_disable_enable():
+ pytest_socket.disable_socket()
  pytest_socket.enable_socket()
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  """)
  result = testdir.runpytest("--verbose")
- result.assert_outcomes(1, 0, 0)
+ result.assert_outcomes(3, 0, 0)
+ with pytest.raises(Exception):
+ assert_socket_blocked(result)
 
 
 def test_socket_enabled_fixture(testdir, socket_enabled):
@@ -46,17 +190,20 @@ def test_socket_enabled(socket_enabled):
  """)
  result = testdir.runpytest("--verbose")
  result.assert_outcomes(1, 0, 0)
+ with pytest.raises(Exception):
+ assert_socket_blocked(result)
 
 
 def test_mix_and_match(testdir, socket_enabled):
  testdir.makepyfile("""
  import socket
+
  def test_socket1():
  socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  def test_socket_enabled(socket_enabled):
  socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  def test_socket2():
  socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  """)
- result = testdir.runpytest("--verbose")
+ result = testdir.runpytest("--verbose", "--disable-socket")
  result.assert_outcomes(1, 0, 2)

tox.ini

@@ -1,6 +1,6 @@
 # For more information about tox, see https://tox.readthedocs.io/en/latest/
 [tox]
-envlist = py27,pypy,flake8
+envlist = py27,py33,py34,py35,py36,pypy,flake8
 
 [testenv]
 deps = pytest