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_object_client_presign_download(mocker):
mocker.patch.object(
ObjectClient, "_post", return_value=SIMPLE_PRESIGN_RESPONSE
)
schema_mock = mocker.patch(
"faculty.clients.object._SimplePresignResponseSchema"
)
client = ObjectClient(mocker.Mock())
returned = client.presign_download(
PROJECT_ID,
"/path",
response_content_disposition="attachement; filename=other",
)
assert returned == SIMPLE_PRESIGN_RESPONSE.url
schema_mock.assert_called_once_with()
ObjectClient._post.assert_called_once_with(
"/project/{}/presign/download".format(PROJECT_ID),
schema_mock.return_value,
json={
"path": "/path",
"responseContentDisposition": "attachement; filename=other",
},
def test_object_client_delete_default(mocker):
path = "test-path"
mocker.patch.object(ObjectClient, "_delete_raw")
client = ObjectClient(mocker.Mock())
client.delete(PROJECT_ID, path)
ObjectClient._delete_raw.assert_called_once_with(
"/project/{}/object/{}".format(PROJECT_ID, path),
params={"recursive": 0},
)
def test_object_client_presign_download_defaults(mocker):
mocker.patch.object(
ObjectClient, "_post", return_value=SIMPLE_PRESIGN_RESPONSE
)
schema_mock = mocker.patch(
"faculty.clients.object._SimplePresignResponseSchema"
)
client = ObjectClient(mocker.Mock())
returned = client.presign_download(PROJECT_ID, "/path")
assert returned == SIMPLE_PRESIGN_RESPONSE.url
schema_mock.assert_called_once_with()
ObjectClient._post.assert_called_once_with(
"/project/{}/presign/download".format(PROJECT_ID),
schema_mock.return_value,
json={"path": "/path"},
)
def test_object_client_create_directory(mocker, parents, expected_parent):
mocker.patch.object(ObjectClient, "_put_raw")
client = ObjectClient(mocker.Mock())
client.create_directory(PROJECT_ID, "test-path", parents=parents)
ObjectClient._put_raw.assert_called_once_with(
"/project/{}/directory/{}".format(PROJECT_ID, "test-path"),
params={"parents": expected_parent},
)
def test_object_client_copy_source_not_found(mocker):
error_code = "source_path_not_found"
exception = NotFound(mocker.Mock(), mocker.Mock(), error_code)
mocker.patch.object(ObjectClient, "_put_raw", side_effect=exception)
client = ObjectClient(mocker.Mock())
with pytest.raises(PathNotFound, match="'source' cannot be found"):
client.copy(PROJECT_ID, "source", "destination")
def test_object_client_copy_url_encoding(mocker):
mocker.patch.object(ObjectClient, "_put_raw")
client = ObjectClient(mocker.Mock())
client.copy(PROJECT_ID, "source", "/[1]/")
ObjectClient._put_raw.assert_called_once_with(
"/project/{}/object/%5B1%5D/".format(PROJECT_ID),
params={"sourcePath": "source", "recursive": 0},
)
The path in the project datasets.
project_id : str, optional
The project to get files from. You need to have access to this project
for it to work. Defaults to the project set by FACULTY_PROJECT_ID in
your environment.
object_client : faculty.clients.object.ObjectClient, optional
Advanced - can be used to benefit from caching in chain interactions
with datasets.
Returns
-------
str
"""
project_id = project_id or get_context().project_id
object_client = object_client or ObjectClient(get_session())
object = object_client.get(project_id, project_path)
return object.etag
from faculty.clients.report import ReportClient
from faculty.clients.secret import SecretClient
from faculty.clients.server import ServerClient
from faculty.clients.user import UserClient
from faculty.clients.workspace import WorkspaceClient
CLIENT_FOR_RESOURCE = {
"account": AccountClient,
"cluster": ClusterClient,
"environment": EnvironmentClient,
"experiment": ExperimentClient,
"job": JobClient,
"log": LogClient,
"model": ModelClient,
"object": ObjectClient,
"project": ProjectClient,
"report": ReportClient,
"secret": SecretClient,
"server": ServerClient,
"user": UserClient,
"workspace": WorkspaceClient,
}
def for_resource(resource):
try:
return CLIENT_FOR_RESOURCE[resource]
except KeyError:
raise ValueError(
"unsupported resource {}, choose one of {}".format(
resource, set(CLIENT_FOR_RESOURCE.keys())