Merge branch 'redis_fail_tests' into add_circleci_config

This commit is contained in:
2016-01-08 09:00:28 +05:30
10 changed files with 109 additions and 101 deletions
+2 -2
View File
@@ -1,10 +1,10 @@
machine:
python:
version: 2.7.10
version: 3.5.0
dependencies:
pre:
- sudo apt-get update
# - sudo apt-get update
- sudo apt-get install redis-server
override:
- pip install --upgrade setuptools
+3 -2
View File
@@ -1,11 +1,12 @@
import htmlPy
from .utils import render_template
from .utils_gui import render_template
import os
import redis
import dateutil.parser
from datetime import datetime
import humanfriendly
import os.path
from ..utils import s
env_redis_port = os.environ.get("PYTEST_STATUS_PORT")
@@ -49,7 +50,7 @@ class Controller(htmlPy.Object):
try:
# make sure that this is correct redis db
assert redis_db.get("PYTEST_STATUS_DB") == "1"
assert s(redis_db.get("PYTEST_STATUS_DB")) == "1"
except AssertionError:
self.app_gui.stop()
raise redis.exceptions.ConnectionError("Redis is running on this port, but it is not related to pytest status\n"
+1 -28
View File
@@ -6,6 +6,7 @@ import redis
import os
import psutil
import datetime
from ..utils import s
env_redis_port = os.environ.get("PYTEST_STATUS_PORT")
@@ -39,34 +40,6 @@ command_status_gui_gen = "pytest_gui_status \"{norm_dir_name}\""
# make bulk changes with pipeline
def s(input_):
''' Convert str or uncode or bytes to str.
If list, do it for all of them.
If others, return as is. '''
import sys
PY3 = sys.version_info > (3,)
if isinstance(input_, list):
return [s(ele) for ele in input_]
try:
if PY3:
assert(type(input_) in [str, bytes])
else:
assert(type(input_) in [str, unicode, bytes])
except AssertionError:
return input_
if PY3:
if type(input_) == bytes:
str_ = bytes.decode(input_)
return str_
# either str or unicode
str_ = str(input_)
return str_
class Helpers(object):
+27
View File
@@ -0,0 +1,27 @@
def s(input_):
''' Convert str or uncode or bytes to str.
If list, do it for all of them.
If others, return as is. '''
import sys
PY3 = sys.version_info > (3,)
if isinstance(input_, list):
return [s(ele) for ele in input_]
try:
if PY3:
assert(type(input_) in [str, bytes])
else:
assert(type(input_) in [str, unicode, bytes])
except AssertionError:
return input_
if PY3:
if type(input_) == bytes:
str_ = bytes.decode(input_)
return str_
# either str or unicode
str_ = str(input_)
return str_
View File
View File
+14 -12
View File
@@ -4,6 +4,8 @@ from mock import patch, MagicMock
import redis
import pytest
from ..utils_test import reload_2_3
REDIS_TEST_PORT = status_plugin.REDIS_PORT + 1
mock_htmlPy_module = MagicMock()
@@ -41,8 +43,8 @@ def redis_master(request, auto_shutdown=True):
def test_redis_fail_1(tmpdir):
# load new redis port
import pytest_gui_status.status_gui.gui_backend as gui_backend
reload(gui_backend)
reload(status_plugin)
reload_2_3(gui_backend)
reload_2_3(status_plugin)
# dont start redis. Also if redis running, stop it
redis_db = redis.StrictRedis(host='localhost', port=REDIS_TEST_PORT, db=0)
@@ -59,7 +61,7 @@ def test_redis_fail_1(tmpdir):
with pytest.raises(redis.exceptions.ConnectionError) as error_info:
gui_backend.Controller(fake_app_gui).redraw()
assert error_info.value.message == "Cant connect to this socket. Is redis running?\n" \
assert str(error_info.value) == "Cant connect to this socket. Is redis running?\n" \
+ "Stopping pytest_status_gui"
@@ -69,8 +71,8 @@ def test_redis_fail_1(tmpdir):
def test_redis_fail_2(tmpdir, redis_master):
# load new redis port
import pytest_gui_status.status_gui.gui_backend as gui_backend
reload(gui_backend)
reload(status_plugin)
reload_2_3(gui_backend)
reload_2_3(status_plugin)
# start Redis
redis_master.init(tmpdir.strpath)
@@ -86,7 +88,7 @@ def test_redis_fail_2(tmpdir, redis_master):
with pytest.raises(redis.exceptions.ConnectionError) as error_info:
gui_backend.Controller(fake_app_gui).redraw()
assert error_info.value.message == "Redis is running on this port, but it is not related to pytest status\n"\
assert str(error_info.value) == "Redis is running on this port, but it is not related to pytest status\n"\
+ "Stopping pytest_status_gui"
fake_app_gui.stop.assert_called_with()
@@ -100,8 +102,8 @@ def test_redis_fail_2(tmpdir, redis_master):
def test_redis_fail_3(tmpdir, redis_master):
# load new redis port
import pytest_gui_status.status_gui.gui_backend as gui_backend
reload(gui_backend)
reload(status_plugin)
reload_2_3(gui_backend)
reload_2_3(status_plugin)
# start Redis
redis_master.init(tmpdir.strpath)
@@ -116,7 +118,7 @@ def test_redis_fail_3(tmpdir, redis_master):
with pytest.raises(redis.exceptions.ConnectionError) as error_info:
gui_backend.Controller(fake_app_gui).redraw()
assert error_info.value.message == "Redis is running on this port, but it is not related to pytest status\n"\
assert str(error_info.value) == "Redis is running on this port, but it is not related to pytest status\n"\
+ "Stopping pytest_status_gui"
fake_app_gui.stop.assert_called_with()
@@ -130,8 +132,8 @@ def test_redis_fail_3(tmpdir, redis_master):
def test_redis_fail_4(tmpdir, redis_master):
# load new redis port
import pytest_gui_status.status_gui.gui_backend as gui_backend
reload(gui_backend)
reload(status_plugin)
reload_2_3(gui_backend)
reload_2_3(status_plugin)
# start Redis
redis_master.init(tmpdir.strpath)
@@ -146,7 +148,7 @@ def test_redis_fail_4(tmpdir, redis_master):
with pytest.raises(redis.exceptions.ConnectionError) as error_info:
gui_backend.Controller(fake_app_gui).redraw()
assert error_info.value.message == "dir_name = {dir_name} not found in redis db".format(
assert str(error_info.value) == "dir_name = {dir_name} not found in redis db".format(
dir_name=fake_app_gui.dir_name)
fake_app_gui.stop.assert_called_with()
+5 -57
View File
@@ -4,64 +4,12 @@ from mock import patch
import redis
import os
import tempfile
import shutil
from ..utils_test import create_temp_case, reload_2_3
REDIS_TEST_PORT = status_plugin.REDIS_PORT + 1
s = status_plugin.s
class patched_chdir(object):
"""Just like chdir, but can be used in `with` - sets back the old path at exit"""
def __init__(self, new_path):
self.old_path = os.getcwd()
self.new_path = new_path
def __enter__(self):
os.chdir(self.new_path)
def __exit__(self, type, value, traceback):
os.chdir(self.old_path)
def create_temp_case(case_name, path_tmpdir=None):
with patched_chdir(os.path.dirname(__file__)):
path_test_cases = os.path.abspath("../testcases/")
path_case = os.path.join(path_test_cases, case_name)
path_tmpdir = path_tmpdir or tempfile.mkdtemp()
path_tmpdir_case = os.path.join(path_tmpdir, case_name)
shutil.copytree(path_case, path_tmpdir_case)
return path_tmpdir_case
def reload_2_3(module_name):
'''
reload that works in all versions of Python.
Uses builtin reload in py2, imp.reload for <=py3.3,
importlib.reload for >=3.4
The module six has similar functionalities,
but would be too huge a dependency for this simple case.
'''
from sys import version_info
major_ver, minor_ver = version_info[:2]
if major_ver == 2:
reload(module_name)
elif major_ver == 3 and minor_ver <= 3:
import imp
imp.reload(module_name)
elif major_ver == 3 and minor_ver >= 4:
import importlib
importlib.reload(module_name)
else:
raise NotImplementedError("Not sure how to reload in "
"this version of Python, supported upto 3.x")
CASE_BASEDIR = __file__
@patch.dict("os.environ",
@@ -72,7 +20,7 @@ def test_whole_1(tmpdir):
'''
os.chdir(os.path.dirname(__file__))
tmpdir = str(tmpdir)
path_case = create_temp_case("case_1", tmpdir)
path_case = create_temp_case("case_1", CASE_BASEDIR, tmpdir)
os.chdir(path_case)
assert(os.environ.get("PYTEST_STATUS_PORT") == str(REDIS_TEST_PORT))
@@ -101,7 +49,7 @@ def test_whole_2(tmpdir):
'''
os.chdir(os.path.dirname(__file__))
tmpdir = str(tmpdir)
path_case = create_temp_case("case_2", tmpdir)
path_case = create_temp_case("case_2", CASE_BASEDIR, tmpdir)
os.chdir(path_case)
popen_pytest = subprocess.Popen(["py.test", "-s"], shell=True)
@@ -126,7 +74,7 @@ def test_intermediate_1(tmpdir):
'''
os.chdir(os.path.dirname(__file__))
tmpdir = str(tmpdir)
path_case = create_temp_case("case_3", tmpdir)
path_case = create_temp_case("case_3", CASE_BASEDIR, tmpdir)
os.chdir(path_case)
popen_pytest = subprocess.Popen(["py.test", "-s"], shell=True)
+57
View File
@@ -0,0 +1,57 @@
import os
import tempfile
import shutil
class patched_chdir(object):
"""Just like chdir, but can be used in `with` - sets back the old path at exit"""
def __init__(self, new_path):
self.old_path = os.getcwd()
self.new_path = new_path
def __enter__(self):
os.chdir(self.new_path)
def __exit__(self, type, value, traceback):
os.chdir(self.old_path)
def create_temp_case(case_name, case_basedir=None, path_tmpdir=None):
case_basedir = case_basedir or __file__
with patched_chdir(os.path.dirname(case_basedir)):
path_test_cases = os.path.abspath("../testcases/")
path_case = os.path.join(path_test_cases, case_name)
path_tmpdir = path_tmpdir or tempfile.mkdtemp()
path_tmpdir_case = os.path.join(path_tmpdir, case_name)
shutil.copytree(path_case, path_tmpdir_case)
return path_tmpdir_case
def reload_2_3(module_name):
'''
reload that works in all versions of Python.
Uses builtin reload in py2, imp.reload for <=py3.3,
importlib.reload for >=3.4
The module six has similar functionalities,
but would be too huge a dependency for this simple case.
'''
from sys import version_info
major_ver, minor_ver = version_info[:2]
if major_ver == 2:
reload(module_name)
elif major_ver == 3 and minor_ver <= 3:
import imp
imp.reload(module_name)
elif major_ver == 3 and minor_ver >= 4:
import importlib
importlib.reload(module_name)
else:
raise NotImplementedError("Not sure how to reload in "
"this version of Python, supported upto 3.x")