Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_version_invalid_version_split():
version_file = create_file("version.py", "__version__")
with mock.patch("os.path.join", return_value=version_file), pytest.raises(
AssertionError
) as err:
get_version()
assert str(err.value) == "Cannot extract version from __version__"
def test_get_version():
version_file = create_file("version.py", '__version__ = "9.9.9"')
with mock.patch("os.path.join", return_value=version_file):
assert get_version() == "9.9.9"
def test_get_version_version_not_found():
version_file = create_file("version.py", 'version = "abc"')
with mock.patch("os.path.join", return_value=version_file), pytest.raises(
AssertionError
) as err:
get_version()
assert str(err.value) == "No __version__ assignment found"
def test_get_version_invalid_version():
version_file_a = create_file("versiona.py", "__version__ = a.1")
with mock.patch(
"os.path.join", return_value=version_file_a
), pytest.raises(AssertionError) as err:
get_version()
assert str(err.value) == "a.1 is not a valid version number"
version_file_b = create_file("versionb.py", "__version__ = a.1.1")
with mock.patch(
"os.path.join", return_value=version_file_b
), pytest.raises(AssertionError) as err:
get_version()
assert str(err.value) == "a.1.1 is not a valid version number"
version_file_c = create_file("versionc.py", "__version__ = 1.a.1")
with mock.patch(
"os.path.join", return_value=version_file_c
), pytest.raises(AssertionError) as err:
get_version()
assert str(err.value) == "1.a.1 is not a valid version number"
version_file_d = create_file("versiond.py", "__version__ = 1.1.a")
with mock.patch(
"os.path.join", return_value=version_file_d
def test_get_version_no_access():
with mock.patch("os.access", return_value=False), pytest.raises(
OSError
) as err:
get_version()
assert str(err.value) == "Cannot open __init__.py file for reading"
return collections.OrderedDict(headers), text
@property
def close(self):
if 'connection' in self.headers.keys():
conn = self.headers.get('connection')
if conn == 'close':
return True
if conn == 'keep-alive':
return False
return True
class Response:
banner = 'Blackhole HTTP/{}'.format(get_version())
request = None
headers = {}
_text = ''
content_type = 'text/html'
encoding = 'utf-8'
ok = True
code = 200
reason = 'OK'
def __init__(self, request):
self.request = request
self.headers = self.setup_headers()
def setup_headers(self):
headers = [('server', self.banner),
('date', self.date),
import collections
import datetime
import time
from wsgiref.handlers import format_date_time
from .status import error
from ..utils import get_version
__all__ = ('Response', 'BadRequest', 'NotFound', 'RequestTimeout',
'HttpVersionNotSupported')
class Response:
banner = 'HTTP/{}'.format(get_version())
request = None
headers = {}
_text = ''
content_type = 'text/html'
encoding = 'utf-8'
ok = True
code = 200
reason = 'OK'
def __init__(self, request):
self.request = request
self.headers = self.setup_headers()
def setup_headers(self):
headers = [('server', self.banner),
('date', self.date),
"(figuratively) pipes all mail to /dev/null. Blackhole is "
"built on top of asyncio and utilises async def and await "
"statements available in Python 3.5 and above."
)
epilog = (
"An explanation of all command line and all configuration "
"options is provided here -- "
"https://kura.github.io/blackhole/configuration.html"
)
parser = argparse.ArgumentParser(
"blackhole", description=description, epilog=epilog
)
parser.add_argument(
"-v", "--version", action="version", version=get_version()
)
parser.add_argument(
"-c",
"--conf",
type=str,
dest="config_file",
metavar="FILE",
help="override the default configuration options",
)
group = parser.add_mutually_exclusive_group()
group.add_argument(
"-t",
"--test",
dest="test",
action="store_true",
help="perform a configuration test",