puppet/puppetboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #326 from mterzo/better_testing

Browse Source 
Flask testing.
This commit is contained in:
Tim Meusel
2017-01-24 06:58:22 +01:00
committed by GitHub
parent 333347d113 caadaa0b35
commit 38b1e9fe06
11 changed files with 633 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
puppetboard/app.py
View File

@@ -121,7 +121,7 @@ def bad_request(e):
@app.errorhandler(403)
def forbidden(e):
envs = environments()
return render_template('403.html', envs=envs), 400
return render_template('403.html', envs=envs), 403
@app.errorhandler(404)

12
puppetboard/templates/412.html
View File

@@ -1,11 +1,5 @@
{% extends 'layout.html' %}
{% block row_fluid %}
<div class="container" style="margin-bottom:55px;">
<div class="row">
<div class="span12">
<h2>Feature unavailable</h2>
<p>You've configured Puppetboard with an API version that does not support this feature.</p>
</div>
</div>
</div>
{% block content %}
<h1>Feature unavailable</h1>
<p>You've configured Puppetboard with an API version that does not support this feature.</p>
{% endblock %}

6
puppetboard/templates/layout.html
View File

@@ -13,10 +13,10 @@
}
</style>
{% else %}
<link href='//fonts.googleapis.com/css?family=Open+Sans' rel='stylesheet' type='text/css'>
<link href='//fonts.googleapis.com/css?family=Open+Sans' rel='stylesheet' type='text/css' />
{% endif %}
<link href="{{ url_for('static', filename='Semantic-UI-2.1.8/semantic.min.css') }}" rel="stylesheet">
<link href="{{ url_for('static', filename='css/puppetboard.css') }}" rel="stylesheet">
<link href="{{ url_for('static', filename='Semantic-UI-2.1.8/semantic.min.css') }}" rel="stylesheet" />
<link href="{{ url_for('static', filename='css/puppetboard.css') }}" rel="stylesheet" />
{% block head %} {% endblock head %}
</head>

2
puppetboard/templates/node.html
View File

@@ -2,7 +2,7 @@
{% import '_macros.html' as macros %}
{% block head %}
{% if config.DAILY_REPORTS_CHART_ENABLED %}
<link href="{{ url_for('static', filename='css/c3.min.css') }}" rel="stylesheet">
<link href="{{ url_for('static', filename='css/c3.min.css') }}" rel="stylesheet" />
{% endif %}
{% endblock head %}
{% block content %}

12
puppetboard/templates/radiator.html
View File

@@ -15,7 +15,7 @@
</head>
<body class='radiator_controller radiator_index_action no-sidebar'>
<table class='node_summary'>
<tr class='failed '>
<tr class='failed'>
<td class='count_column'>
<span class='count'>{{stats['failed']}}</span>
</td>
@@ -30,7 +30,7 @@
</div>
</td>
</tr>
<tr class='unreported '>
<tr class='unreported'>
<td class='count_column'>
<span class='count'>{{stats['unreported']}}</span>
</td>
@@ -45,7 +45,7 @@
</div>
</td>
</tr>
<tr class='noop '>
<tr class='noop'>
<td class='count_column'>
<span class='count'>{{stats['noop']}}</span>
</td>
@@ -60,7 +60,7 @@
</div>
</td>
</tr>
<tr class='changed '>
<tr class='changed'>
<td class='count_column'>
<span class='count'>{{stats['changed']}}</span>
</td>
@@ -75,7 +75,7 @@
</div>
</td>
</tr>
<tr class='unchanged '>
<tr class='unchanged'>
<td class='count_column'>
<span class='count'>{{stats['unchanged']}}</span>
</td>
@@ -90,7 +90,7 @@
</div>
</td>
</tr>
<tr class='total '>
<tr class='total'>
<td class='count_column'>
<span class='count'>{{total}}</span>
</td>

2
requirements-test.txt
View File

@@ -4,6 +4,8 @@ mock==1.3.0
pytest==3.0.1
pytest-pep8==1.0.5
pytest-cov==2.2.1
pytest-mock==1.5.0
cov-core==1.15.0
unittest2==1.1.0; python_version < '2.7'
bandit
beautifulsoup4==4.5.3

0
test/__init__.py Normal file
View File

333
test/test_app.py
View File

@@ -1,19 +1,328 @@
import os
import pytest
import json
from puppetboard import app
import unittest
import tempfile
from pypuppetdb.types import Node
from puppetboard import default_settings
from bs4 import BeautifulSoup
class AppTestCase(unittest.TestCase):
def setUp(self):
pass
class MockDbQuery(object):
def __init__(self, responses):
self.responses = responses
def tearDown(self):
pass
def get(self, method, **kws):
resp = None
if method in self.responses:
resp = self.responses[method].pop(0)
def test_first_test(self):
self.assertTrue(True)
if 'validate' in resp:
checks = resp['validate']['checks']
resp = resp['validate']['data']
for check in checks:
assert check in kws
expected_value = checks[check]
assert expected_value == kws[check]
return resp
if __name__ == '__main__':
unittest.main()
@pytest.fixture
def mock_puppetdb_environments(mocker):
environemnts = [
{'name': 'production'},
{'name': 'staging'}
]
return mocker.patch.object(app.puppetdb, 'environments',
return_value=environemnts)
@pytest.fixture
def mock_puppetdb_default_nodes(mocker):
node_list = [
Node('_', 'node-unreported',
report_timestamp='2013-08-01T09:57:00.000Z',
latest_report_hash='1234567',
catalog_timestamp='2013-08-01T09:57:00.000Z',
facts_timestamp='2013-08-01T09:57:00.000Z',
status='unreported'),
Node('_', 'node-changed',
report_timestamp='2013-08-01T09:57:00.000Z',
latest_report_hash='1234567',
catalog_timestamp='2013-08-01T09:57:00.000Z',
facts_timestamp='2013-08-01T09:57:00.000Z',
status='changed'),
Node('_', 'node-failed',
report_timestamp='2013-08-01T09:57:00.000Z',
latest_report_hash='1234567',
catalog_timestamp='2013-08-01T09:57:00.000Z',
facts_timestamp='2013-08-01T09:57:00.000Z',
status='failed'),
Node('_', 'node-noop',
report_timestamp='2013-08-01T09:57:00.000Z',
latest_report_hash='1234567',
catalog_timestamp='2013-08-01T09:57:00.000Z',
facts_timestamp='2013-08-01T09:57:00.000Z',
status='noop'),
Node('_', 'node-unchanged',
report_timestamp='2013-08-01T09:57:00.000Z',
latest_report_hash='1234567',
catalog_timestamp='2013-08-01T09:57:00.000Z',
facts_timestamp='2013-08-01T09:57:00.000Z',
status='unchanged'),
Node('_', 'node-skipped',
report_timestamp='2013-08-01T09:57:00.000Z',
latest_report_hash='1234567',
catalog_timestamp='2013-08-01T09:57:00.000Z',
facts_timestamp='2013-08-01T09:57:00.000Z',
status='skipped')
]
return mocker.patch.object(app.puppetdb, 'nodes',
return_value=iter(node_list))
@pytest.fixture
def client():
client = app.app.test_client()
return client
def test_first_test():
assert app is not None, ("%s" % reg.app)
def test_no_env(client, mock_puppetdb_environments):
rv = client.get('/nonexsistenv/')
assert rv.status_code == 404
def test_get_index(client, mocker,
mock_puppetdb_environments,
mock_puppetdb_default_nodes):
query_data = {
'nodes': [[{'count': 10}]],
'resources': [[{'count': 40}]],
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/')
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
assert rv.status_code == 200
def test_index_all(client, mocker,
mock_puppetdb_environments,
mock_puppetdb_default_nodes):
base_str = 'puppetlabs.puppetdb.population:'
query_data = {
'mbean': [
{
'validate': {
'data': {'Value': '50'},
'checks': {
'path': '%sname=num-nodes' % base_str
}
}
},
{
'validate': {
'data': {'Value': '60'},
'checks': {
'path': '%sname=num-resources' % base_str
}
}
},
{
'validate': {
'data': {'Value': 60.3},
'checks': {
'path': '%sname=avg-resources-per-node' % base_str
}
}
}
]
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/%2A/')
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
vals = soup.find_all('h1',
{"class": "ui header darkblue no-margin-bottom"})
assert len(vals) == 3
assert vals[0].string == '50'
assert vals[1].string == '60'
assert vals[2].string == ' 60'
assert rv.status_code == 200
def test_index_division_by_zero(client, mocker):
mock_puppetdb_environments(mocker)
mock_puppetdb_default_nodes(mocker)
query_data = {
'nodes': [[{'count': 0}]],
'resources': [[{'count': 40}]],
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/')
assert rv.status_code == 200
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
vals = soup.find_all('h1',
{"class": "ui header darkblue no-margin-bottom"})
assert len(vals) == 3
assert vals[2].string == '0'
def test_offline_mode(client, mocker):
app.app.config['OFFLINE_MODE'] = True
mock_puppetdb_environments(mocker)
mock_puppetdb_default_nodes(mocker)
query_data = {
'nodes': [[{'count': 10}]],
'resources': [[{'count': 40}]],
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/')
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
for link in soup.find_all('link'):
assert "//" not in link['href']
for script in soup.find_all('script'):
if "src" in script.attrs:
assert "//" not in script['src']
assert rv.status_code == 200
def test_default_node_view(client, mocker,
mock_puppetdb_environments,
mock_puppetdb_default_nodes):
rv = client.get('/nodes')
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
for label in ['failed', 'changed', 'unreported', 'noop']:
vals = soup.find_all('a',
{"class": "ui %s label status" % label})
assert len(vals) == 1
assert 'node-%s' % (label) in vals[0].attrs['href']
assert rv.status_code == 200
def test_radiator_view(client, mocker,
mock_puppetdb_environments,
mock_puppetdb_default_nodes):
query_data = {
'nodes': [[{'count': 10}]],
'resources': [[{'count': 40}]],
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/radiator')
assert rv.status_code == 200
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
assert soup.h1 != 'Not Found'
total = soup.find(class_='total')
assert '10' in total.text
def test_radiator_view_json(client, mocker,
mock_puppetdb_environments,
mock_puppetdb_default_nodes):
query_data = {
'nodes': [[{'count': 10}]],
'resources': [[{'count': 40}]],
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/radiator', headers={'Accept': 'application/json'})
assert rv.status_code == 200
json_data = json.loads(rv.data.decode('utf-8'))
assert json_data['unreported'] == 1
assert json_data['noop'] == 1
assert json_data['failed'] == 1
assert json_data['changed'] == 1
assert json_data['skipped'] == 1
assert json_data['unchanged'] == 1
def test_radiator_view_bad_env(client, mocker):
mock_puppetdb_environments(mocker)
mock_puppetdb_default_nodes(mocker)
query_data = {
'nodes': [[{'count': 10}]],
'resources': [[{'count': 40}]],
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/nothere/radiator')
assert rv.status_code == 404
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
assert soup.h1.text == 'Not Found'
def test_radiator_view_division_by_zero(client, mocker):
mock_puppetdb_environments(mocker)
mock_puppetdb_default_nodes(mocker)
query_data = {
'nodes': [[{'count': 0}]],
'resources': [[{'count': 40}]],
}
dbquery = MockDbQuery(query_data)
mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get)
rv = client.get('/radiator')
assert rv.status_code == 200
soup = BeautifulSoup(rv.data, 'html.parser')
assert soup.title.contents[0] == 'Puppetboard'
total = soup.find(class_='total')
assert '0' in total.text

73
test/test_app_error.py Normal file
View File

@@ -0,0 +1,73 @@
import pytest
from flask import Flask, current_app
from puppetboard import app
from bs4 import BeautifulSoup
@pytest.fixture
def mock_puppetdb_environments(mocker):
environemnts = [
{'name': 'production'},
{'name': 'staging'}
]
return mocker.patch.object(app.puppetdb, 'environments',
return_value=environemnts)
def test_error_no_content():
result = app.no_content(None)
assert result[0] == ''
assert result[1] == 204
def test_error_bad_request(mock_puppetdb_environments):
with app.app.test_request_context():
(output, error_code) = app.bad_request(None)
soup = BeautifulSoup(output, 'html.parser')
assert 'The request sent to PuppetDB was invalid' in soup.p.text
assert error_code == 400
def test_error_forbidden(mock_puppetdb_environments):
with app.app.test_request_context():
(output, error_code) = app.forbidden(None)
soup = BeautifulSoup(output, 'html.parser')
long_string = "%s %s" % ('What you were looking for has',
'been disabled by the administrator')
assert long_string in soup.p.text
assert error_code == 403
def test_error_not_found(mock_puppetdb_environments):
with app.app.test_request_context():
(output, error_code) = app.not_found(None)
soup = BeautifulSoup(output, 'html.parser')
long_string = "%s %s" % ('What you were looking for could not',
'be found in PuppetDB.')
assert long_string in soup.p.text
assert error_code == 404
def test_error_precond(mock_puppetdb_environments):
with app.app.test_request_context():
(output, error_code) = app.precond_failed(None)
soup = BeautifulSoup(output, 'html.parser')
long_string = "%s %s" % ('You\'ve configured Puppetboard with an API',
'version that does not support this feature.')
assert long_string in soup.p.text
assert error_code == 412
def test_error_server(mock_puppetdb_environments):
with app.app.test_request_context():
(output, error_code) = app.server_error(None)
soup = BeautifulSoup(output, 'html.parser')
assert 'Internal Server Error' in soup.h2.text
assert error_code == 500

94
test/test_docker_settings.py
View File

@@ -1,7 +1,8 @@
import pytest
import os
from puppetboard import docker_settings
import unittest
import tempfile
from puppetboard import app
try:
import future.utils
except:
@@ -13,8 +14,8 @@ except:
pass
class DockerTestCase(unittest.TestCase):
def setUp(self):
@pytest.fixture(scope='function')
def cleanUpEnv(request):
for env_var in dir(docker_settings):
if (env_var.startswith('__') or env_var.startswith('_') or
env_var.islower()):
@@ -23,70 +24,85 @@ class DockerTestCase(unittest.TestCase):
if env_var in os.environ:
del os.environ[env_var]
reload(docker_settings)
return
def test_default_host_port(self):
self.assertEqual(docker_settings.PUPPETDB_HOST, 'puppetdb')
self.assertEqual(docker_settings.PUPPETDB_PORT, 8080)
def test_set_host_port(self):
os.environ['PUPPETDB_HOST'] = 'puppetdb'
def test_default_host_port(cleanUpEnv):
assert docker_settings.PUPPETDB_HOST == 'puppetdb'
assert docker_settings.PUPPETDB_PORT == 8080
def test_set_host_port(cleanUpEnv):
os.environ['PUPPETDB_HOST'] = 'puppetdb2'
os.environ['PUPPETDB_PORT'] = '9081'
reload(docker_settings)
self.assertEqual(docker_settings.PUPPETDB_HOST, 'puppetdb')
self.assertEqual(docker_settings.PUPPETDB_PORT, 9081)
assert docker_settings.PUPPETDB_HOST == 'puppetdb2'
assert docker_settings.PUPPETDB_PORT == 9081
def test_cert_true_test(self):
def test_cert_true_test(cleanUpEnv):
os.environ['PUPPETDB_SSL_VERIFY'] = 'True'
reload(docker_settings)
self.assertTrue(docker_settings.PUPPETDB_SSL_VERIFY)
assert docker_settings.PUPPETDB_SSL_VERIFY is True
os.environ['PUPPETDB_SSL_VERIFY'] = 'true'
reload(docker_settings)
self.assertTrue(docker_settings.PUPPETDB_SSL_VERIFY)
assert docker_settings.PUPPETDB_SSL_VERIFY is True
def test_cert_false_test(self):
def test_cert_false_test(cleanUpEnv):
os.environ['PUPPETDB_SSL_VERIFY'] = 'False'
reload(docker_settings)
self.assertFalse(docker_settings.PUPPETDB_SSL_VERIFY)
assert docker_settings.PUPPETDB_SSL_VERIFY is False
os.environ['PUPPETDB_SSL_VERIFY'] = 'false'
reload(docker_settings)
self.assertFalse(docker_settings.PUPPETDB_SSL_VERIFY)
assert docker_settings.PUPPETDB_SSL_VERIFY is False
def test_cert_path(self):
def test_cert_path(cleanUpEnv):
ca_file = '/usr/ssl/path/ca.pem'
os.environ['PUPPETDB_SSL_VERIFY'] = ca_file
reload(docker_settings)
self.assertEqual(docker_settings.PUPPETDB_SSL_VERIFY, ca_file)
assert docker_settings.PUPPETDB_SSL_VERIFY == ca_file
def validate_facts(self, facts):
self.assertEqual(type(facts), type([]))
self.assertTrue(len(facts) > 0)
def validate_facts(facts):
assert isinstance(facts, list)
assert len(facts) > 0
for map in facts:
self.assertEqual(type(map), type(()))
self.assertTrue(len(map) == 2)
assert isinstance(map, tuple)
assert len(map) == 2
def test_inventory_facts_default(self):
self.validate_facts(docker_settings.INVENTORY_FACTS)
def test_invtory_facts_custom(self):
def test_inventory_facts_default(cleanUpEnv):
validate_facts(docker_settings.INVENTORY_FACTS)
def test_invtory_facts_custom(cleanUpEnv):
os.environ['INVENTORY_FACTS'] = "A, B, C, D"
reload(docker_settings)
self.validate_facts(docker_settings.INVENTORY_FACTS)
validate_facts(docker_settings.INVENTORY_FACTS)
def test_graph_facts_defautl(self):
def test_graph_facts_defautl(cleanUpEnv):
facts = docker_settings.GRAPH_FACTS
self.assertEqual(type(facts), type([]))
self.assertTrue('puppetversion' in facts)
assert isinstance(facts, list)
assert 'puppetversion' in facts
def test_graph_facts_custom(self):
def test_graph_facts_custom(cleanUpEnv):
os.environ['GRAPH_FACTS'] = "architecture, puppetversion, extra"
reload(docker_settings)
facts = docker_settings.GRAPH_FACTS
self.assertEqual(type(facts), type([]))
self.assertEqual(len(facts), 3)
self.assertTrue('puppetversion' in facts)
self.assertTrue('architecture' in facts)
self.assertTrue('extra' in facts)
assert isinstance(facts, list)
assert len(facts) == 3
assert 'puppetversion' in facts
assert 'architecture' in facts
assert 'extra' in facts
if __name__ == '__main__':
unittest.main()
def test_bad_log_value(cleanUpEnv):
os.environ['LOGLEVEL'] = 'g'
os.environ['PUPPETBOARD_SETTINGS'] = '../puppetboard/docker_settings.py'
reload(docker_settings)
with pytest.raises(ValueError) as error:
reload(app)

112
test/test_utils.py
View File

@@ -1,8 +1,4 @@
try:
import unittest2 as unittest
except ImportError:
import unittest
import pytest
import sys
import json
import mock
@@ -18,56 +14,61 @@ from puppetboard import utils
from puppetboard import app
from puppetboard.app import NoContent
from bs4 import BeautifulSoup
import logging
class UtilsTestCase(unittest.TestCase):
def setUp(self):
pass
def teadDown(self):
pass
def test_json_format(self):
def test_json_format():
demo = [{'foo': 'bar'}, {'bar': 'foo'}]
sample = json.dumps(demo, indent=2, separators=(',', ': '))
self.assertEqual(sample, utils.jsonprint(demo),
"Json formatting has changed")
assert sample == utils.jsonprint(demo), "Json formatting has changed"
def test_format_val_str(self):
def test_format_val_str():
x = "some string"
self.assertEqual(x, utils.formatvalue(x),
"Should return same value")
assert x == utils.formatvalue(x), "Should return same value"
def test_format_val_array(self):
def test_format_val_array():
x = ['a', 'b', 'c']
self.assertEqual("a, b, c", utils.formatvalue(x),
"Should return comma seperated string")
assert "a, b, c" == utils.formatvalue(x)
def test_format_val_dict_one_layer(self):
def test_format_val_dict_one_layer():
x = {'a': 'b'}
self.assertEqual("a => b,<br/>", utils.formatvalue(x),
"Should return stringified value")
assert "a => b,<br/>" == utils.formatvalue(x)
def test_format_val_tuple(self):
def test_format_val_tuple():
x = ('a', 'b')
self.assertEqual(str(x), utils.formatvalue(x))
assert str(x) == utils.formatvalue(x)
@mock.patch('logging.log')
class GetOrAbortTesting(unittest.TestCase):
def test_get(self, mock_log):
def test_get():
x = "hello world"
def test_get_or_abort():
return x
self.assertEqual(x, utils.get_or_abort(test_get_or_abort))
assert x == utils.get_or_abort(test_get_or_abort)
def test_http_error(self, mock_log):
def test_pretty_print():
test_data = [{'hello': 'world'}]
html = utils.prettyprint(test_data)
soup = BeautifulSoup(html, 'html.parser')
assert soup.th.text == 'hello'
@pytest.fixture
def mock_log(mocker):
return mocker.patch('logging.log')
def test_http_error(mock_log):
err = "NotFound"
def raise_http_error():
@@ -76,11 +77,12 @@ class GetOrAbortTesting(unittest.TestCase):
x.reason = err
raise HTTPError(err, response=x)
with self.assertRaises(NotFound) as error:
with pytest.raises(NotFound):
utils.get_or_abort(raise_http_error)
mock_log.error.assert_called_with(err)
mock_log.error.assert_called_once_with(err)
def test_http_connection_error(self, mock_log):
def test_http_connection_error(mock_log):
err = "ConnectionError"
def connection_error():
@@ -89,26 +91,25 @@ class GetOrAbortTesting(unittest.TestCase):
x.reason = err
raise ConnectionError(err, response=x)
with self.assertRaises(InternalServerError) as error:
with pytest.raises(InternalServerError):
utils.get_or_abort(connection_error)
mock_log.error.assert_called_with(err)
@mock.patch('flask.abort')
def test_http_empty(self, mock_log, flask_abort):
def test_http_empty(mock_log, mocker):
err = "Empty Response"
def connection_error():
raise EmptyResponseError(err)
with self.assertRaises(NoContent) as error:
flask_abort = mocker.patch('flask.abort')
with pytest.raises(NoContent):
utils.get_or_abort(connection_error)
mock_log.error.assert_called_with(err)
flask_abort.assert_called_with('204')
class yieldOrStop(unittest.TestCase):
def test_iter(self):
def test_iter():
test_list = (0, 1, 2, 3)
def my_generator():
@@ -116,14 +117,15 @@ class yieldOrStop(unittest.TestCase):
yield i
gen = utils.yield_or_stop(my_generator())
self.assertIsInstance(gen, GeneratorType)
assert isinstance(gen, GeneratorType)
i = 0
for val in gen:
self.assertEqual(i, val)
assert i == val
i = i + 1
def test_stop_empty(self):
def test_stop_empty():
def my_generator():
yield 1
raise EmptyResponseError
@@ -131,28 +133,24 @@ class yieldOrStop(unittest.TestCase):
gen = utils.yield_or_stop(my_generator())
for val in gen:
self.assertEqual(1, val)
assert 1 == val
def test_stop_conn_error(self):
def test_stop_conn_error():
def my_generator():
yield 1
raise ConnectionError
yield 2
gen = utils.yield_or_stop(my_generator())
for val in gen:
self.assertEqual(1, val)
assert 1 == val
def test_stop_http_error(self):
def test_stop_http_error():
def my_generator():
yield 1
raise HTTPError
yield 2
gen = utils.yield_or_stop(my_generator())
for val in gen:
self.assertEqual(1, val)
if __name__ == '__main__':
unittest.main()
assert 1 == val