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_url(self):
responses.add(responses.POST, self.client.url,
adding_headers={"location": 'http://master.tus.io/files/foo'})
self.assertEqual(self.uploader.get_url(), 'http://master.tus.io/files/foo')
# test for stored urls
responses.add(responses.HEAD, 'http://master.tus.io/files/foo_bar',
adding_headers={"upload-offset": "0"})
storage_path = '{}/storage_file'.format(os.path.dirname(os.path.abspath(__file__)))
resumable_uploader = self.client.uploader(
file_path='./LICENSE', store_url=True, url_storage=filestorage.FileStorage(storage_path)
)
self.assertEqual(resumable_uploader.get_url(), 'http://master.tus.io/files/foo_bar')
def test_download(tmp_path, mock_zst):
url = "https://community-tc.services.mozilla.com/api/index/v1/task/project.relman.bugbug.data_commits.latest/artifacts/public/prova.json.zst"
url_version = "https://community-tc.services.mozilla.com/api/index/v1/task/project.relman.bugbug.data_commits.latest/artifacts/public/prova.json.version"
db_path = tmp_path / "prova.json"
db.register(db_path, url, 1)
responses.add(responses.GET, url_version, status=200, body="1")
responses.add(
responses.HEAD,
url,
status=200,
headers={
"ETag": "123",
"Accept-Encoding": "zstd",
"Last-Modified": "2019-04-16",
},
)
tmp_zst_path = tmp_path / "prova_tmp.zst"
mock_zst(tmp_zst_path)
with open(tmp_zst_path, "rb") as content:
responses.add(responses.GET, url, status=200, body=content.read())
assert db.download(db_path)
def test_get_offset(self):
responses.add(responses.HEAD, self.uploader.url,
adding_headers={"upload-offset": "300"})
self.assertEqual(self.uploader.get_offset(), 300)
def test_download_missing(tmp_path, mock_zst):
url = "https://community-tc.services.mozilla.com/api/index/v1/task/project.relman.bugbug.data_commits.latest/artifacts/public/prova.json.zst"
url_version = "https://community-tc.services.mozilla.com/api/index/v1/task/project.relman.bugbug.data_commits.latest/artifacts/public/prova.json.version"
db_path = tmp_path / "prova.json"
db.register(db_path, url, 1)
responses.add(
responses.HEAD,
url,
status=404,
headers={"ETag": "123", "Accept-Encoding": "zstd"},
)
responses.add(
responses.GET, url, status=404, body=requests.exceptions.HTTPError("HTTP error")
)
responses.add(responses.GET, url_version, status=404)
assert not db.download(db_path)
assert not os.path.exists(db_path)
with pytest.raises(Exception, match="Last-Modified is not available"):
db.last_modified(db_path)
@pytest.mark.parametrize(('registry', 'insecure'), [
('https://example.com', False),
('example.com', True),
('example.com', False),
])
@pytest.mark.parametrize(('method', 'responses_method'), [
(RegistrySession.get, responses.GET),
(RegistrySession.head, responses.HEAD),
(RegistrySession.put, responses.PUT),
(RegistrySession.delete, responses.DELETE),
])
@pytest.mark.parametrize(('config_content'), [
({'username': 'john.doe', 'password': 'letmein'}),
({'auth': b64encode(b'john.doe:letmein').decode('utf-8')}),
])
@responses.activate
def test_registry_session(tmpdir, registry, insecure, method, responses_method, config_content):
temp_dir = mkdtemp(dir=str(tmpdir))
with open(os.path.join(temp_dir, '.dockercfg'), 'w+') as dockerconfig:
dockerconfig.write(json.dumps({
registry_hostname(registry): config_content
}))
session = RegistrySession(registry, insecure=insecure, dockercfg_path=temp_dir)
def test_is_not_active_201(pydrill_instance, pydrill_url):
"""
:type pydrill_instance: pydrill.client.PyDrill
"""
responses.add(**{
'method': responses.HEAD,
'url': pydrill_url,
'content_type': 'application/json',
'status': 201,
})
assert pydrill_instance.is_active() is False
def expect_response(cls, url, timeout, times):
latency = cls.latency_by_url[url]
# TODO(John Sirois): Switch to a homomorphic response in each branch once
# https://github.com/getsentry/responses/issues/234 is fixed.
response = requests.exceptions.ConnectTimeout() if latency >= timeout else (200, {}, '')
def callback(_):
if latency < timeout:
times.append(latency)
else:
# We raise a timeout exception.
pass
return response
responses.add_callback(responses.HEAD, url, callback)
for key, value in backend.urls.items():
HTTPretty.register_uri(
method=method,
uri=re.compile(key),
body=convert_httpretty_response(value),
)
def disable_patching(self):
HTTPretty.disable()
HTTPretty.reset()
RESPONSES_METHODS = [
responses.GET,
responses.DELETE,
responses.HEAD,
responses.OPTIONS,
responses.PATCH,
responses.POST,
responses.PUT,
]
class CallbackResponse(responses.CallbackResponse):
"""
Need to subclass so we can change a couple things
"""
def get_response(self, request):
"""
Need to override this so we can pass decode_content=False
"""