def test_v1_or_both_fail(self):
"""Cases when registry is not available."""
registry = 'qwerty.fgh:5000'
url, ping_url_v1, ping_url_v2 = self._get_urls(registry)
def check():
try:
images.check_registry_status(url)
except APIError as e:
self.assertEqual(e.message, message)
else:
self.fail('APIError is not raised')
responses.reset()
for exception, message in self._get_err_cases(registry):
responses.add(responses.GET, ping_url_v1, body=exception)
check()
# v2 registry api is supported by the host, but both v1 and v2 are
# not available now
responses.add(responses.GET, ping_url_v1, status=404,
adding_headers=self.v2_is_supported)
responses.add(responses.GET, ping_url_v2, body=exception)
check()
python类reset()的实例源码
def setUp(self):
responses.reset()
responses.add(
responses.GET, BasePage.scrape_url,
"""<html><body>
<h1 class="foo" data-foo=1>Heading</h1>
<h1 id="the-id">15</h1>
<span class=float>3.14</span>
<span class=int>42</span>
<span class=bool></span>
<span class=date>2016-04-23</span>
<a href="/very-fake">link</a>
<table>
<tr>test123
<th>key</th> testmore
<td>value</th></tr>
<tr><th>key2<td>value2</tr>
</table
""")
responses.start()
self.addCleanup(responses.stop)
def mock_urls(request):
data = _DATA.tostring()
checksum = _CHECKSUM
url_data = _URL
url_checksum = _URL + '.md5'
if not request.param[0]:
# Data URL is corrupted.
url_data = url_data[:-1]
if not request.param[1]:
# Data is corrupted.
data = data[:-1]
if not request.param[2]:
# Checksum URL is corrupted.
url_checksum = url_checksum[:-1]
if not request.param[3]:
# Checksum is corrupted.
checksum = checksum[:-1]
_add_mock_response(url_data, data)
_add_mock_response(url_checksum, checksum)
yield request.param, url_data, url_checksum
responses.reset()
def tearDown(self):
responses.reset()
self.timer_patcher.stop()
def tearDown(self):
responses.reset()
self.timer_patcher.stop()
self.get_token_patcher.stop()
def _post_teardown(self):
"""
Disable all mocks after the test.
"""
if self.mock_requests:
responses.reset()
responses.stop()
patch.stopall()
def test_with_return_create_status(self):
class ResourceMethod(ResourceMethodBase):
method = 'put'
return_create_status = True
responses.add(
'PUT',
self.base_uri,
json='world',
status=201
)
instance = self.get_instance(ResourceMethod)
assert_that(
instance(),
equal_to({
'created': True,
'result': 'world'
})
)
# test when not created
responses.reset()
responses.add(
'PUT',
self.base_uri,
json='world',
status=200
)
assert_that(
instance(),
equal_to({
'created': False,
'result': 'world'
})
)
def test_v1_fail_v2_ok(self):
"""Cases when registry v1 api is not available, but v2 is."""
url, ping_url_v1, ping_url_v2 = self._get_urls('qwerty.fgh:5000')
for status in (200, 401):
responses.reset()
responses.add(responses.GET, ping_url_v1, status=404,
adding_headers=self.v2_is_supported)
responses.add(responses.GET, ping_url_v2, body='{}', status=status,
content_type='application/json; charset=utf-8')
images.check_registry_status(url)
self.assertEqual(len(responses.calls), 2)
self.assertEqual(responses.calls[0].request.url, ping_url_v1)
self.assertEqual(responses.calls[1].request.url, ping_url_v2)
def teardown_method(self, method):
responses.reset()
responses.stop()
test_language_pairs.py 文件源码
项目：research-recommendation-api
作者: wikimedia
项目源码
文件源码
阅读 17
收藏 0
点赞 0
评论 0
def test_language_pairs_when_fetch_is_invalid(json_value):
responses.reset()
responses.add(responses.GET, configuration.get_config_value('endpoints', 'language_pairs'),
json=json_value, status=200)
assert None is language_pairs.get_language_pairs()
assert True is language_pairs.is_valid_language_pair('any', 'combination')
def mock_url():
_add_mock_response(_URL, _DATA.tostring())
_add_mock_response(_URL + '.md5', _CHECKSUM + ' ' + op.basename(_URL))
yield
responses.reset()
def test_download_sample_data(tempdir):
name = 'hybrid_10sec.dat'
url = _BASE_URL['cortexlab'] + name
_add_mock_response(url, _DATA.tostring())
_add_mock_response(url + '.md5', _CHECKSUM)
download_sample_data(name, tempdir)
with open(op.join(tempdir, name), 'rb') as f:
data = f.read()
ae(np.fromstring(data, np.float32), _DATA)
responses.reset()
def test_dat_file(tempdir):
data = np.random.randint(size=(20000, 4),
low=-100, high=100).astype(np.int16)
fn = 'test-4ch-1s.dat'
_add_mock_response(_BASE_URL['github'] + 'test/' + fn,
data.tostring())
path = download_test_data(fn, tempdir)
with open(path, 'rb') as f:
arr = np.fromfile(f, dtype=np.int16).reshape((-1, 4))
assert arr.shape == (20000, 4)
responses.reset()
