From 743587277b55a40d525d2877d44f752a205ae672 Mon Sep 17 00:00:00 2001 From: Mike Fiedler Date: Thu, 23 Dec 2021 17:49:26 -0500 Subject: [PATCH] lint: apply black Apply opinionated style to codebase. Signed-off-by: Mike Fiedler --- .flake8 | 4 +- .pre-commit-config.yaml | 4 ++ README.md | 1 + pyproject.toml | 2 + pytest_socket.py | 81 +++++++++++---------- tests/conftest.py | 2 +- tests/test_async.py | 18 +++-- tests/test_combinations.py | 7 +- tests/test_restrict_hosts.py | 116 ++++++++++++++++++------------ tests/test_socket.py | 132 +++++++++++++++++++++-------------- 10 files changed, 224 insertions(+), 143 deletions(-) diff --git a/.flake8 b/.flake8 index 1fd4893..d5a110d 100644 --- a/.flake8 +++ b/.flake8 @@ -1,2 +1,4 @@ [flake8] -max-line-length = 160 +; https://black.readthedocs.io/en/stable/guides/using_black_with_other_tools.html#flake8 +max-line-length = 88 +extend-ignore = E203 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ca67007..cdc16f5 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -20,3 +20,7 @@ repos: hooks: - id: isort name: isort (python) +- repo: https://github.com/psf/black + rev: 21.12b0 + hooks: + - id: black diff --git a/README.md b/README.md index 1c28a9d..bd165ba 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,7 @@ [![pre-commit.ci status](https://results.pre-commit.ci/badge/github/miketheman/pytest-socket/main.svg)](https://results.pre-commit.ci/latest/github/miketheman/pytest-socket/main) [![Maintainability](https://api.codeclimate.com/v1/badges/1608a75b1c3a20211992/maintainability)](https://codeclimate.com/github/miketheman/pytest-socket/maintainability) [![FOSSA Status](https://app.fossa.io/api/projects/git%2Bgithub.com%2Fmiketheman%2Fpytest-socket.svg?type=shield)](https://app.fossa.io/projects/git%2Bgithub.com%2Fmiketheman%2Fpytest-socket?ref=badge_shield) +[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black) A plugin to use with Pytest to disable or restrict `socket` calls during tests to ensure network calls are prevented. diff --git a/pyproject.toml b/pyproject.toml index 205db9c..bbb7c6a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,8 @@ socket = 'pytest_socket' [tool.isort] known_first_party = ['pytest_socket', 'conftest', 'test_socket'] +# https://black.readthedocs.io/en/stable/guides/using_black_with_other_tools.html#profilegcm +profile = "black" [build-system] requires = ["poetry-core>=1.0.0"] diff --git a/pytest_socket.py b/pytest_socket.py index 2cd92bd..86b5fdb 100644 --- a/pytest_socket.py +++ b/pytest_socket.py @@ -15,44 +15,45 @@ def __init__(self, *args, **kwargs): class SocketConnectBlockedError(RuntimeError): def __init__(self, allowed, host, *args, **kwargs): if allowed: - allowed = ','.join(allowed) + allowed = ",".join(allowed) super(SocketConnectBlockedError, self).__init__( - f'A test tried to use socket.socket.connect() with host "{host}" (allowed: "{allowed}").' + "A test tried to use socket.socket.connect() " + f'with host "{host}" (allowed: "{allowed}").' ) def pytest_addoption(parser): - group = parser.getgroup('socket') + group = parser.getgroup("socket") group.addoption( - '--disable-socket', - action='store_true', - dest='disable_socket', - help='Disable socket.socket by default to block network calls.' + "--disable-socket", + action="store_true", + dest="disable_socket", + help="Disable socket.socket by default to block network calls.", ) group.addoption( - '--allow-hosts', - dest='allow_hosts', - metavar='ALLOWED_HOSTS_CSV', - help='Only allow specified hosts through socket.socket.connect((host, port)).' + "--allow-hosts", + dest="allow_hosts", + metavar="ALLOWED_HOSTS_CSV", + help="Only allow specified hosts through socket.socket.connect((host, port)).", ) group.addoption( - '--allow-unix-socket', - action='store_true', - dest='allow_unix_socket', - help='Allow calls if they are to Unix domain sockets' + "--allow-unix-socket", + action="store_true", + dest="allow_unix_socket", + help="Allow calls if they are to Unix domain sockets", ) @pytest.fixture def socket_disabled(pytestconfig): - """ disable socket.socket for duration of this test function """ + """disable socket.socket for duration of this test function""" disable_socket(allow_unix_socket=pytestconfig.__socket_allow_unix_socket) yield @pytest.fixture def socket_enabled(pytestconfig): - """ enable socket.socket for duration of this test function """ + """enable socket.socket for duration of this test function""" enable_socket() yield @@ -67,11 +68,11 @@ def _is_unix_socket(family) -> bool: def disable_socket(allow_unix_socket=False): - """ disable socket.socket to disable the Internet. useful in testing. - """ + """disable socket.socket to disable the Internet. useful in testing.""" class GuardedSocket(socket.socket): - """ socket guard to disable socket creation (from pytest-socket) """ + """socket guard to disable socket creation (from pytest-socket)""" + def __new__(cls, family=-1, type=-1, proto=-1, fileno=None): if _is_unix_socket(family) and allow_unix_socket: return super().__new__(cls, family, type, proto, fileno) @@ -82,20 +83,26 @@ def __new__(cls, family=-1, type=-1, proto=-1, fileno=None): def enable_socket(): - """ re-enable socket.socket to enable the Internet. useful in testing. - """ + """re-enable socket.socket to enable the Internet. useful in testing.""" socket.socket = _true_socket def pytest_configure(config): - config.addinivalue_line("markers", "disable_socket(): Disable socket connections for a specific test") - config.addinivalue_line("markers", "enable_socket(): Enable socket connections for a specific test") - config.addinivalue_line("markers", "allow_hosts([hosts]): Restrict socket connection to defined list of hosts") + config.addinivalue_line( + "markers", "disable_socket(): Disable socket connections for a specific test" + ) + config.addinivalue_line( + "markers", "enable_socket(): Enable socket connections for a specific test" + ) + config.addinivalue_line( + "markers", + "allow_hosts([hosts]): Restrict socket connection to defined list of hosts", + ) # Store the global configs in the `pytest.Config` object. - config.__socket_disabled = config.getoption('--disable-socket') - config.__socket_allow_unix_socket = config.getoption('--allow-unix-socket') - config.__socket_allow_hosts = config.getoption('--allow-hosts') + config.__socket_disabled = config.getoption("--disable-socket") + config.__socket_allow_unix_socket = config.getoption("--allow-unix-socket") + config.__socket_allow_hosts = config.getoption("--allow-hosts") def pytest_runtest_setup(item) -> None: @@ -108,12 +115,16 @@ def pytest_runtest_setup(item) -> None: """ # If test has the `enable_socket` marker, we accept this as most explicit. - if 'socket_enabled' in item.fixturenames or item.get_closest_marker('enable_socket'): + if "socket_enabled" in item.fixturenames or item.get_closest_marker( + "enable_socket" + ): enable_socket() return # If the test has the `disable_socket` marker, it's explicitly disabled. - if 'socket_disabled' in item.fixturenames or item.get_closest_marker('disable_socket'): + if "socket_disabled" in item.fixturenames or item.get_closest_marker( + "disable_socket" + ): disable_socket(item.config.__socket_allow_unix_socket) return @@ -127,7 +138,7 @@ def pytest_runtest_setup(item) -> None: def _resolve_allow_hosts(item): """Resolve `allow_hosts` behaviors.""" - mark_restrictions = item.get_closest_marker('allow_hosts') + mark_restrictions = item.get_closest_marker("allow_hosts") cli_restrictions = item.config.__socket_allow_hosts hosts = None if mark_restrictions: @@ -156,10 +167,9 @@ def host_from_connect_args(args): def socket_allow_hosts(allowed=None, allow_unix_socket=False): - """ disable socket.socket.connect() to disable the Internet. useful in testing. - """ + """disable socket.socket.connect() to disable the Internet. useful in testing.""" if isinstance(allowed, str): - allowed = allowed.split(',') + allowed = allowed.split(",") if not isinstance(allowed, list): return @@ -174,7 +184,6 @@ def guarded_connect(inst, *args): def _remove_restrictions(): - """ restore socket.socket.* to allow access to the Internet. useful in testing. - """ + """restore socket.socket.* to allow access to the Internet. useful in testing.""" socket.socket = _true_socket socket.socket.connect = _true_connect diff --git a/tests/conftest.py b/tests/conftest.py index 4a4a488..68a3c16 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,7 @@ import pytest -pytest_plugins = 'pytester' +pytest_plugins = "pytester" unix_sockets_only = pytest.mark.skipif( diff --git a/tests/test_async.py b/tests/test_async.py index 7398750..6f63634 100644 --- a/tests/test_async.py +++ b/tests/test_async.py @@ -4,7 +4,8 @@ @unix_sockets_only def test_asynctest(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import socket import asynctest @@ -18,14 +19,16 @@ async def test_that_a_coroutine_runs(self): async def test_inet_is_blocked(self): socket.socket(socket.AF_INET, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest("--disable-socket", "--allow-unix-socket") result.assert_outcomes(passed=1, skipped=0, failed=1) @unix_sockets_only def test_starlette(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ from pytest_socket import disable_socket from starlette.responses import HTMLResponse from starlette.testclient import TestClient @@ -45,14 +48,16 @@ def test_app(): client = TestClient(app) response = client.get('/') assert response.status_code == 200 - """) + """ + ) result = testdir.runpytest("--disable-socket", "--allow-unix-socket") result.assert_outcomes(passed=1, skipped=0, failed=0) @unix_sockets_only def test_httpx_fails(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest import httpx @@ -64,6 +69,7 @@ def anyio_backend(): async def test_httpx(): async with httpx.AsyncClient() as client: await client.get("http://www.example.com/") - """) + """ + ) result = testdir.runpytest("--disable-socket", "--allow-unix-socket") assert_socket_blocked(result) diff --git a/tests/test_combinations.py b/tests/test_combinations.py index 86462b7..6fa1d25 100644 --- a/tests/test_combinations.py +++ b/tests/test_combinations.py @@ -44,7 +44,8 @@ def test_remote_not_allowed_fails(): result = testdir.runpytest() result.assert_outcomes(passed=4, failed=1) result.stdout.fnmatch_lines( - "*SocketConnectBlockedError: A test tried to use socket.socket.connect() with host*" + "*SocketConnectBlockedError: " + "A test tried to use socket.socket.connect() with host*" ) @@ -72,5 +73,7 @@ def test_inet_connect(): sock.connect(('{httpbin.host}', {httpbin.port})) """ ) - result = testdir.runpytest("--disable-socket", "--allow-unix-socket", f"--allow-hosts={httpbin.host}") + result = testdir.runpytest( + "--disable-socket", "--allow-unix-socket", f"--allow-hosts={httpbin.host}" + ) result.assert_outcomes(passed=2) diff --git a/tests/test_restrict_hosts.py b/tests/test_restrict_hosts.py index fe02bcd..529375a 100644 --- a/tests/test_restrict_hosts.py +++ b/tests/test_restrict_hosts.py @@ -3,7 +3,7 @@ import pytest -localhost = '127.0.0.1' +localhost = "127.0.0.1" connect_code_template = """ import socket @@ -34,7 +34,9 @@ def {2}(): def assert_host_blocked(result, host): - result.stdout.fnmatch_lines(f'*A test tried to use socket.socket.connect() with host "{host}"*') + result.stdout.fnmatch_lines( + f'*A test tried to use socket.socket.connect() with host "{host}"*' + ) @pytest.fixture @@ -43,10 +45,10 @@ def assert_socket_connect(should_pass, **kwargs): # get the name of the calling function test_name = inspect.stack()[1][3] - mark = '' - cli_arg = kwargs.get('cli_arg', None) - code_template = kwargs.get('code_template', connect_code_template) - mark_arg = kwargs.get('mark_arg', None) + mark = "" + cli_arg = kwargs.get("cli_arg", None) + code_template = kwargs.get("code_template", connect_code_template) + mark_arg = kwargs.get("mark_arg", None) if mark_arg: if isinstance(mark_arg, str): @@ -58,7 +60,7 @@ def assert_socket_connect(should_pass, **kwargs): testdir.makepyfile(code) if cli_arg: - result = testdir.runpytest(f'--allow-hosts={cli_arg}') + result = testdir.runpytest(f"--allow-hosts={cli_arg}") else: result = testdir.runpytest() @@ -67,28 +69,34 @@ def assert_socket_connect(should_pass, **kwargs): else: result.assert_outcomes(0, 0, 1) assert_host_blocked(result, httpbin.host) + return assert_socket_connect def test_help_message(testdir): result = testdir.runpytest( - '--help', + "--help", + ) + result.stdout.fnmatch_lines( + [ + "socket:", + "*--allow-hosts=ALLOWED_HOSTS_CSV", + "*Only allow specified hosts through", + "*socket.socket.connect((host, port)).", + ] ) - result.stdout.fnmatch_lines([ - 'socket:', - '*--allow-hosts=ALLOWED_HOSTS_CSV', - '*Only allow specified hosts through', - '*socket.socket.connect((host, port)).' - ]) def test_marker_help_message(testdir): result = testdir.runpytest( - '--markers', + "--markers", + ) + result.stdout.fnmatch_lines( + [ + "@pytest.mark.allow_hosts([[]hosts[]]): " + "Restrict socket connection to defined list of hosts", + ] ) - result.stdout.fnmatch_lines([ - '@pytest.mark.allow_hosts([[]hosts[]]): Restrict socket connection to defined list of hosts', - ]) def test_default_connect_enabled(assert_connect): @@ -104,7 +112,7 @@ def test_single_cli_arg_connect_unicode_enabled(assert_connect): def test_multiple_cli_arg_connect_enabled(assert_connect): - assert_connect(True, cli_arg=localhost + ',1.2.3.4') + assert_connect(True, cli_arg=localhost + ",1.2.3.4") def test_single_mark_arg_connect_enabled(assert_connect): @@ -112,39 +120,41 @@ def test_single_mark_arg_connect_enabled(assert_connect): def test_multiple_mark_arg_csv_connect_enabled(assert_connect): - assert_connect(True, mark_arg=localhost + ',1.2.3.4') + assert_connect(True, mark_arg=localhost + ",1.2.3.4") def test_multiple_mark_arg_list_connect_enabled(assert_connect): - assert_connect(True, mark_arg=[localhost, '1.2.3.4']) + assert_connect(True, mark_arg=[localhost, "1.2.3.4"]) def test_mark_cli_conflict_mark_wins_connect_enabled(assert_connect): - assert_connect(True, mark_arg=[localhost], cli_arg='1.2.3.4') + assert_connect(True, mark_arg=[localhost], cli_arg="1.2.3.4") def test_single_cli_arg_connect_disabled(assert_connect): - assert_connect(False, cli_arg='1.2.3.4', code_template=connect_unicode_code_template) + assert_connect( + False, cli_arg="1.2.3.4", code_template=connect_unicode_code_template + ) def test_multiple_cli_arg_connect_disabled(assert_connect): - assert_connect(False, cli_arg='5.6.7.8,1.2.3.4') + assert_connect(False, cli_arg="5.6.7.8,1.2.3.4") def test_single_mark_arg_connect_disabled(assert_connect): - assert_connect(False, mark_arg='1.2.3.4') + assert_connect(False, mark_arg="1.2.3.4") def test_multiple_mark_arg_csv_connect_disabled(assert_connect): - assert_connect(False, mark_arg='5.6.7.8,1.2.3.4') + assert_connect(False, mark_arg="5.6.7.8,1.2.3.4") def test_multiple_mark_arg_list_connect_disabled(assert_connect): - assert_connect(False, mark_arg=['5.6.7.8', '1.2.3.4']) + assert_connect(False, mark_arg=["5.6.7.8", "1.2.3.4"]) def test_mark_cli_conflict_mark_wins_connect_disabled(assert_connect): - assert_connect(False, mark_arg=['1.2.3.4'], cli_arg=localhost) + assert_connect(False, mark_arg=["1.2.3.4"], cli_arg=localhost) def test_default_urlopen_succeeds_by_default(assert_connect): @@ -152,46 +162,59 @@ def test_default_urlopen_succeeds_by_default(assert_connect): def test_single_cli_arg_urlopen_enabled(assert_connect): - assert_connect(True, cli_arg=localhost + ',1.2.3.4', code_template=urlopen_code_template) + assert_connect( + True, cli_arg=localhost + ",1.2.3.4", code_template=urlopen_code_template + ) def test_single_mark_arg_urlopen_enabled(assert_connect): - assert_connect(True, mark_arg=[localhost, '1.2.3.4'], code_template=urlopen_code_template) + assert_connect( + True, mark_arg=[localhost, "1.2.3.4"], code_template=urlopen_code_template + ) def test_global_restrict_via_config_fail(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import socket def test_global_restrict_via_config_fail(): socket.socket().connect(('127.0.0.1', 80)) - """) - testdir.makeini(""" + """ + ) + testdir.makeini( + """ [pytest] addopts = --allow-hosts=2.2.2.2 - """) + """ + ) result = testdir.runpytest() result.assert_outcomes(0, 0, 1) - assert_host_blocked(result, '127.0.0.1') + assert_host_blocked(result, "127.0.0.1") def test_global_restrict_via_config_pass(testdir, httpbin): - testdir.makepyfile(f""" + testdir.makepyfile( + f""" import socket def test_global_restrict_via_config_pass(): socket.socket().connect(('{httpbin.host}', {httpbin.port})) - """) - testdir.makeini(f""" + """ + ) + testdir.makeini( + f""" [pytest] addopts = --allow-hosts={httpbin.host} - """) + """ + ) result = testdir.runpytest() result.assert_outcomes(1, 0, 0) def test_test_isolation(testdir, httpbin): - testdir.makepyfile(f""" + testdir.makepyfile( + f""" import pytest import socket @@ -205,14 +228,16 @@ def test_fail(): def test_pass_2(): socket.socket().connect(('{httpbin.host}', {httpbin.port})) - """) + """ + ) result = testdir.runpytest() result.assert_outcomes(2, 0, 1) assert_host_blocked(result, httpbin.host) def test_conflicting_cli_vs_marks(testdir, httpbin): - testdir.makepyfile(f""" + testdir.makepyfile( + f""" import pytest import socket @@ -226,8 +251,9 @@ def test_fail(): def test_fail_2(): socket.socket().connect(('2.2.2.2', {httpbin.port})) - """) - result = testdir.runpytest('--allow-hosts=1.2.3.4') + """ + ) + result = testdir.runpytest("--allow-hosts=1.2.3.4") result.assert_outcomes(1, 0, 2) - assert_host_blocked(result, '2.2.2.2') + assert_host_blocked(result, "2.2.2.2") assert_host_blocked(result, httpbin.host) diff --git a/tests/test_socket.py b/tests/test_socket.py index d46b369..20e3cb6 100644 --- a/tests/test_socket.py +++ b/tests/test_socket.py @@ -4,34 +4,34 @@ from conftest import unix_sockets_only PYFILE_SOCKET_USED_IN_TEST_ARGS = """ - import socket + import socket - def test_socket(): - socket.socket(socket.AF_INET, socket.SOCK_STREAM) - """ + def test_socket(): + socket.socket(socket.AF_INET, socket.SOCK_STREAM) +""" PYFILE_SOCKET_USED_IN_TEST_KWARGS = """ - import socket + import socket - def test_socket(): - socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM) - """ + def test_socket(): + socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM) +""" PYFILE_SOCKET_NO_ARGS = """ - import socket + import socket - def test_socket(): - socket.socket() - """ + def test_socket(): + socket.socket() +""" def assert_socket_blocked(result): result.assert_outcomes(passed=0, skipped=0, failed=1) - result.stdout.fnmatch_lines(""" - *SocketBlockedError: A test tried to use socket.socket.* - """) + result.stdout.fnmatch_lines( + "*SocketBlockedError: A test tried to use socket.socket.*" + ) @pytest.mark.parametrize( @@ -39,8 +39,8 @@ def assert_socket_blocked(result): [ PYFILE_SOCKET_USED_IN_TEST_ARGS, PYFILE_SOCKET_USED_IN_TEST_KWARGS, - PYFILE_SOCKET_NO_ARGS - ] + PYFILE_SOCKET_NO_ARGS, + ], ) def test_socket_enabled_by_default(testdir, pyfile): testdir.makepyfile(pyfile) @@ -51,7 +51,8 @@ def test_socket_enabled_by_default(testdir, pyfile): def test_global_disable_via_fixture(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest import pytest_socket import socket @@ -62,7 +63,8 @@ def disable_socket_for_all(): def test_socket(): socket.socket(socket.AF_INET, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest() assert_socket_blocked(result) @@ -72,8 +74,8 @@ def test_socket(): [ PYFILE_SOCKET_USED_IN_TEST_ARGS, PYFILE_SOCKET_USED_IN_TEST_KWARGS, - PYFILE_SOCKET_NO_ARGS - ] + PYFILE_SOCKET_NO_ARGS, + ], ) def test_global_disable_via_cli_flag(testdir, pyfile): testdir.makepyfile(pyfile) @@ -83,12 +85,14 @@ def test_global_disable_via_cli_flag(testdir, pyfile): def test_help_message(testdir): result = testdir.runpytest( - '--help', + "--help", + ) + result.stdout.fnmatch_lines( + [ + "socket:", + "*--disable-socket*Disable socket.socket by default to block network*", + ] ) - result.stdout.fnmatch_lines([ - 'socket:', - '*--disable-socket*Disable socket.socket by default to block network*' - ]) @pytest.mark.parametrize( @@ -96,41 +100,47 @@ def test_help_message(testdir): [ PYFILE_SOCKET_USED_IN_TEST_ARGS, PYFILE_SOCKET_USED_IN_TEST_KWARGS, - PYFILE_SOCKET_NO_ARGS - ] + PYFILE_SOCKET_NO_ARGS, + ], ) def test_global_disable_via_config(testdir, pyfile): testdir.makepyfile(pyfile) - testdir.makeini(""" + testdir.makeini( + """ [pytest] addopts = --disable-socket - """) + """ + ) result = testdir.runpytest() assert_socket_blocked(result) def test_disable_socket_marker(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest import socket @pytest.mark.disable_socket def test_socket(): socket.socket(socket.AF_INET, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest() assert_socket_blocked(result) def test_enable_socket_marker(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest import socket @pytest.mark.enable_socket def test_socket(): socket.socket(socket.AF_INET, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest("--disable-socket") result.assert_outcomes(1, 0, 0) with pytest.raises(BaseException): @@ -138,7 +148,8 @@ def test_socket(): def test_urllib_succeeds_by_default(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ try: from urllib.request import urlopen except ImportError: @@ -146,7 +157,8 @@ def test_urllib_succeeds_by_default(testdir): def test_disable_socket_urllib(): assert urlopen('http://httpbin.org/get').getcode() == 200 - """) + """ + ) result = testdir.runpytest() result.assert_outcomes(1, 0, 0) with pytest.raises(BaseException): @@ -154,7 +166,8 @@ def test_disable_socket_urllib(): def test_enabled_urllib_succeeds(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest import pytest_socket try: @@ -165,7 +178,8 @@ def test_enabled_urllib_succeeds(testdir): @pytest.mark.enable_socket def test_disable_socket_urllib(): assert urlopen('http://httpbin.org/get').getcode() == 200 - """) + """ + ) result = testdir.runpytest("--disable-socket") result.assert_outcomes(1, 0, 0) with pytest.raises(BaseException): @@ -173,7 +187,8 @@ def test_disable_socket_urllib(): def test_disabled_urllib_fails(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest try: from urllib.request import urlopen @@ -183,13 +198,15 @@ def test_disabled_urllib_fails(testdir): @pytest.mark.disable_socket def test_disable_socket_urllib(): assert urlopen('http://httpbin.org/get').getcode() == 200 - """) + """ + ) result = testdir.runpytest() assert_socket_blocked(result) def test_double_call_does_nothing(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest import pytest_socket import socket @@ -209,7 +226,8 @@ def test_disable_enable(): pytest_socket.disable_socket() pytest_socket.enable_socket() socket.socket(socket.AF_INET, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest() result.assert_outcomes(3, 0, 0) with pytest.raises(BaseException): @@ -217,11 +235,13 @@ def test_disable_enable(): def test_socket_enabled_fixture(testdir, socket_enabled): - testdir.makepyfile(""" + testdir.makepyfile( + """ import socket def test_socket_enabled(socket_enabled): socket.socket(socket.AF_INET, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest() result.assert_outcomes(1, 0, 0) with pytest.raises(BaseException): @@ -229,7 +249,8 @@ def test_socket_enabled(socket_enabled): def test_mix_and_match(testdir, socket_enabled): - testdir.makepyfile(""" + testdir.makepyfile( + """ import socket def test_socket1(): @@ -238,13 +259,15 @@ 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("--disable-socket") result.assert_outcomes(1, 0, 2) def test_socket_subclass_is_still_blocked(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import pytest import pytest_socket import socket @@ -256,32 +279,37 @@ class MySocket(socket.socket): pass MySocket(socket.AF_INET, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest() assert_socket_blocked(result) @unix_sockets_only def test_unix_domain_sockets_blocked_with_disable_socket(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import socket def test_unix_socket(): socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest("--disable-socket") assert_socket_blocked(result) @unix_sockets_only def test_enabling_unix_domain_sockets_with_disable_socket(testdir): - testdir.makepyfile(""" + testdir.makepyfile( + """ import socket def test_inet(): socket.socket(socket.AF_INET, socket.SOCK_STREAM) def test_unix_socket(): socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - """) + """ + ) result = testdir.runpytest("--disable-socket", "--allow-unix-socket") result.assert_outcomes(passed=1, skipped=0, failed=1)