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_job_client_list(mocker):
mocker.patch.object(JobClient, "_get", return_value=[JOB_SUMMARY])
schema_mock = mocker.patch("faculty.clients.job._JobSummarySchema")
client = JobClient(mocker.Mock())
assert client.list(PROJECT_ID) == [JOB_SUMMARY]
schema_mock.assert_called_once_with(many=True)
JobClient._get.assert_called_once_with(
"/project/{}/job".format(PROJECT_ID), schema_mock.return_value
)
def test_job_client_update_metadata(mocker):
mocker.patch.object(JobClient, "_put_raw")
client = JobClient(mocker.Mock())
client.update_metadata(PROJECT_ID, JOB_ID, "A name", "A desc")
JobClient._put_raw.assert_called_once_with(
"/project/{}/job/{}/meta".format(PROJECT_ID, JOB_ID),
json={"name": "A name", "description": "A desc"},
)
def test_job_client_get(mocker):
mocker.patch.object(JobClient, "_get", return_value=JOB)
schema_mock = mocker.patch("faculty.clients.job._JobSchema")
client = JobClient(mocker.Mock())
assert client.get(PROJECT_ID, JOB_ID) == JOB
schema_mock.assert_called_once_with()
JobClient._get.assert_called_once_with(
"/project/{}/job/{}".format(PROJECT_ID, JOB_ID),
schema_mock.return_value,
)
def test_job_client_create_run(mocker):
mocker.patch.object(JobClient, "_post", return_value=RUN_ID)
schema_mock = mocker.patch("faculty.clients.job._RunIdSchema")
client = JobClient(mocker.Mock())
assert (
client.create_run(
PROJECT_ID,
JOB_ID,
[{"param": "one", "other": "two"}, {"param": "three"}],
)
== RUN_ID
)
schema_mock.assert_called_once_with()
last_call_args, last_call_kwargs = JobClient._post.call_args
assert last_call_args == (
"/project/{}/job/{}/run".format(PROJECT_ID, JOB_ID),
schema_mock.return_value,
)
def test_job_client_get_run(mocker, run_identifier):
mocker.patch.object(JobClient, "_get", return_value=RUN)
schema_mock = mocker.patch("faculty.clients.job._RunSchema")
client = JobClient(mocker.Mock())
assert client.get_run(PROJECT_ID, JOB_ID, run_identifier) == RUN
schema_mock.assert_called_once_with()
JobClient._get.assert_called_once_with(
"/project/{}/job/{}/run/{}".format(PROJECT_ID, JOB_ID, run_identifier),
schema_mock.return_value,
)
def test_job_client_cancel_run(mocker, run_identifier):
mocker.patch.object(JobClient, "_delete_raw")
client = JobClient(mocker.Mock())
client.cancel_run(PROJECT_ID, JOB_ID, run_identifier)
JobClient._delete_raw.assert_called_once_with(
"/project/{}/job/{}/run/{}".format(PROJECT_ID, JOB_ID, run_identifier)
)
def test_job_client_create(mocker):
mocker.patch.object(JobClient, "_post", return_value=JOB_ID)
response_schema_mock = mocker.patch("faculty.clients.job._JobIdSchema")
mocker.patch.object(_JobDefinitionSchema, "dump")
client = JobClient(mocker.Mock())
assert (
client.create(
PROJECT_ID,
JOB_METADATA.name,
JOB_METADATA.description,
JOB_DEFINITION,
)
== JOB_ID
)
response_schema_mock.assert_called_once_with()
_JobDefinitionSchema.dump.assert_called_once_with(JOB_DEFINITION)
JobClient._post.assert_called_once_with(
"/project/{}/job".format(PROJECT_ID),
response_schema_mock.return_value,
json={
def test_job_client_get_subrun(mocker, run_identifier, subrun_identifier):
mocker.patch.object(JobClient, "_get", return_value=SUBRUN)
schema_mock = mocker.patch("faculty.clients.job._SubrunSchema")
client = JobClient(mocker.Mock())
assert (
client.get_subrun(
PROJECT_ID, JOB_ID, run_identifier, subrun_identifier
)
== SUBRUN
)
schema_mock.assert_called_once_with()
JobClient._get.assert_called_once_with(
"/project/{}/job/{}/run/{}/subrun/{}".format(
PROJECT_ID, JOB_ID, run_identifier, subrun_identifier
),
schema_mock.return_value,
)
def test_job_client_list(mocker):
mocker.patch.object(JobClient, "_get", return_value=[JOB_SUMMARY])
schema_mock = mocker.patch("faculty.clients.job._JobSummarySchema")
client = JobClient(mocker.Mock())
assert client.list(PROJECT_ID) == [JOB_SUMMARY]
schema_mock.assert_called_once_with(many=True)
JobClient._get.assert_called_once_with(
"/project/{}/job".format(PROJECT_ID), schema_mock.return_value
)
from faculty.clients.model import ModelClient
from faculty.clients.object import ObjectClient
from faculty.clients.project import ProjectClient
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: