Use hostname for Chains #3873
GitHub Actions / JUnit Test Report
failed
Feb 3, 2025 in 0s
463 tests run, 437 passed, 2 skipped, 24 failed.
Annotations
Check failure on line 190 in .venv/lib/python3.9/site-packages/numpy/core/tests/test_api.py
github-actions / JUnit Test Report
test_api.test_create_model_from_truss
KeyError: 'model_version'
Raw output
mock_post = <MagicMock name='post' id='139827602177664'>
baseten_api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268d5ee0>
@mock.patch("requests.post", return_value=mock_create_model_response())
def test_create_model_from_truss(mock_post, baseten_api):
> baseten_api.create_model_from_truss(
"model_name",
"s3key",
"config_str",
"semver_bump",
"client_version",
is_trusted=False,
deployment_name="deployment_name",
)
truss/tests/remote/baseten/test_api.py:190:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268d5ee0>
model_name = 'model_name', s3_key = 's3key', config = 'config_str'
semver_bump = 'semver_bump', client_version = 'client_version'
is_trusted = False, allow_truss_download = True
deployment_name = 'deployment_name', origin = None
def create_model_from_truss(
self,
model_name: str,
s3_key: str,
config: str,
semver_bump: str,
client_version: str,
is_trusted: bool,
allow_truss_download: bool = True,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
):
query_string = f"""
mutation {{
create_model_from_truss(
name: "{model_name}"
s3_key: "{s3_key}"
config: "{config}"
semver_bump: "{semver_bump}"
client_version: "{client_version}"
is_trusted: {"true" if is_trusted else "false"}
allow_truss_download: {"true" if allow_truss_download else "false"}
{f'version_name: "{deployment_name}"' if deployment_name else ""}
{f"model_origin: {origin.value}" if origin else ""}
) {{
model_version {{
id
invoke_base_url
oracle {{
id
name
}}
}}
}}
}}
"""
resp = self._post_graphql_query(query_string)
> return resp["data"]["create_model_from_truss"]["model_version"]
E KeyError: 'model_version'
truss/remote/baseten/api.py:166: KeyError
Check failure on line 214 in .venv/lib/python3.9/site-packages/numpy/core/tests/test_api.py
github-actions / JUnit Test Report
test_api.test_create_model_from_truss_does_not_send_deployment_name_if_not_specified
KeyError: 'model_version'
Raw output
mock_post = <MagicMock name='post' id='139827602152944'>
baseten_api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268eb190>
@mock.patch("requests.post", return_value=mock_create_model_response())
def test_create_model_from_truss_does_not_send_deployment_name_if_not_specified(
mock_post, baseten_api
):
> baseten_api.create_model_from_truss(
"model_name",
"s3key",
"config_str",
"semver_bump",
"client_version",
is_trusted=True,
deployment_name=None,
)
truss/tests/remote/baseten/test_api.py:214:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268eb190>
model_name = 'model_name', s3_key = 's3key', config = 'config_str'
semver_bump = 'semver_bump', client_version = 'client_version'
is_trusted = True, allow_truss_download = True, deployment_name = None
origin = None
def create_model_from_truss(
self,
model_name: str,
s3_key: str,
config: str,
semver_bump: str,
client_version: str,
is_trusted: bool,
allow_truss_download: bool = True,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
):
query_string = f"""
mutation {{
create_model_from_truss(
name: "{model_name}"
s3_key: "{s3_key}"
config: "{config}"
semver_bump: "{semver_bump}"
client_version: "{client_version}"
is_trusted: {"true" if is_trusted else "false"}
allow_truss_download: {"true" if allow_truss_download else "false"}
{f'version_name: "{deployment_name}"' if deployment_name else ""}
{f"model_origin: {origin.value}" if origin else ""}
) {{
model_version {{
id
invoke_base_url
oracle {{
id
name
}}
}}
}}
}}
"""
resp = self._post_graphql_query(query_string)
> return resp["data"]["create_model_from_truss"]["model_version"]
E KeyError: 'model_version'
truss/remote/baseten/api.py:166: KeyError
Check failure on line 236 in .venv/lib/python3.9/site-packages/numpy/core/tests/test_api.py
github-actions / JUnit Test Report
test_api.test_create_model_from_truss_with_allow_truss_download
KeyError: 'model_version'
Raw output
mock_post = <MagicMock name='post' id='139827601753952'>
baseten_api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c26884160>
@mock.patch("requests.post", return_value=mock_create_model_response())
def test_create_model_from_truss_with_allow_truss_download(mock_post, baseten_api):
> baseten_api.create_model_from_truss(
"model_name",
"s3key",
"config_str",
"semver_bump",
"client_version",
is_trusted=True,
allow_truss_download=False,
)
truss/tests/remote/baseten/test_api.py:236:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <truss.remote.baseten.api.BasetenApi object at 0x7f2c26884160>
model_name = 'model_name', s3_key = 's3key', config = 'config_str'
semver_bump = 'semver_bump', client_version = 'client_version'
is_trusted = True, allow_truss_download = False, deployment_name = None
origin = None
def create_model_from_truss(
self,
model_name: str,
s3_key: str,
config: str,
semver_bump: str,
client_version: str,
is_trusted: bool,
allow_truss_download: bool = True,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
):
query_string = f"""
mutation {{
create_model_from_truss(
name: "{model_name}"
s3_key: "{s3_key}"
config: "{config}"
semver_bump: "{semver_bump}"
client_version: "{client_version}"
is_trusted: {"true" if is_trusted else "false"}
allow_truss_download: {"true" if allow_truss_download else "false"}
{f'version_name: "{deployment_name}"' if deployment_name else ""}
{f"model_origin: {origin.value}" if origin else ""}
) {{
model_version {{
id
invoke_base_url
oracle {{
id
name
}}
}}
}}
}}
"""
resp = self._post_graphql_query(query_string)
> return resp["data"]["create_model_from_truss"]["model_version"]
E KeyError: 'model_version'
truss/remote/baseten/api.py:166: KeyError
Check failure on line 260 in .venv/lib/python3.9/site-packages/numpy/core/tests/test_api.py
github-actions / JUnit Test Report
test_api.test_create_development_model_from_truss_with_allow_truss_download
KeyError: 'model_version'
Raw output
mock_post = <MagicMock name='post' id='139827602426752'>
baseten_api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c26928cd0>
@mock.patch("requests.post", return_value=mock_create_development_model_response())
def test_create_development_model_from_truss_with_allow_truss_download(
mock_post, baseten_api
):
> baseten_api.create_development_model_from_truss(
"model_name",
"s3key",
"config_str",
"client_version",
is_trusted=True,
allow_truss_download=False,
)
truss/tests/remote/baseten/test_api.py:260:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <truss.remote.baseten.api.BasetenApi object at 0x7f2c26928cd0>
model_name = 'model_name', s3_key = 's3key', config = 'config_str'
client_version = 'client_version', is_trusted = True
allow_truss_download = False, origin = None
def create_development_model_from_truss(
self,
model_name,
s3_key,
config,
client_version,
is_trusted=False,
allow_truss_download=True,
origin: Optional[b10_types.ModelOrigin] = None,
):
query_string = f"""
mutation {{
deploy_draft_truss(name: "{model_name}"
s3_key: "{s3_key}"
config: "{config}"
client_version: "{client_version}"
is_trusted: {"true" if is_trusted else "false"}
allow_truss_download: {"true" if allow_truss_download else "false"}
{f"model_origin: {origin.value}" if origin else ""}
) {{
model_version {{
id
invoke_base_url
oracle {{
id
name
}}
}}
}}
}}
"""
resp = self._post_graphql_query(query_string)
> return resp["data"]["deploy_draft_truss"]["model_version"]
E KeyError: 'model_version'
truss/remote/baseten/api.py:235: KeyError
Check failure on line 91 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_eligible_environment_values[None]
KeyError: 'invoke_base_url'
Raw output
environment = None
@pytest.mark.parametrize("environment", [None, PRODUCTION_ENVIRONMENT_NAME])
def test_create_truss_service_handles_eligible_environment_values(environment):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_model_from_truss.return_value = return_value
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=False,
model_id=None,
deployment_name="deployment_name",
environment=environment,
)
truss/tests/remote/baseten/test_core.py:91:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602122064'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = False, model_id = None, deployment_name = 'deployment_name'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
if model_id is None:
if environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
model_version_json = api.create_model_from_truss(
model_name=model_name,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
deployment_name=deployment_name,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:397: KeyError
Check failure on line 91 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_eligible_environment_values[production]
KeyError: 'invoke_base_url'
Raw output
environment = 'production'
@pytest.mark.parametrize("environment", [None, PRODUCTION_ENVIRONMENT_NAME])
def test_create_truss_service_handles_eligible_environment_values(environment):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_model_from_truss.return_value = return_value
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=False,
model_id=None,
deployment_name="deployment_name",
environment=environment,
)
truss/tests/remote/baseten/test_core.py:91:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602031424'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = False, model_id = None, deployment_name = 'deployment_name'
origin = None, environment = 'production'
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
if model_id is None:
if environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
model_version_json = api.create_model_from_truss(
model_name=model_name,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
deployment_name=deployment_name,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:397: KeyError
Check failure on line 113 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_services_handles_is_draft[some_model_id]
KeyError: 'invoke_base_url'
Raw output
model_id = 'some_model_id'
@pytest.mark.parametrize("model_id", ["some_model_id", None])
def test_create_truss_services_handles_is_draft(model_id):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_development_model_from_truss.return_value = return_value
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=True,
model_id=model_id,
deployment_name="deployment_name",
)
truss/tests/remote/baseten/test_core.py:113:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602053104'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = True, model_id = 'some_model_id', deployment_name = 'deployment_name'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:375: KeyError
Check failure on line 113 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_services_handles_is_draft[None]
KeyError: 'invoke_base_url'
Raw output
model_id = None
@pytest.mark.parametrize("model_id", ["some_model_id", None])
def test_create_truss_services_handles_is_draft(model_id):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_development_model_from_truss.return_value = return_value
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=True,
model_id=model_id,
deployment_name="deployment_name",
)
truss/tests/remote/baseten/test_core.py:113:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602172368'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = True, model_id = None, deployment_name = 'deployment_name'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:375: KeyError
Check failure on line 156 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_existing_model[inputs0]
KeyError: 'invoke_base_url'
Raw output
inputs = {'deployment_name': 'some deployment', 'environment': None, 'is_trusted': True, 'preserve_previous_prod_deployment': False}
@pytest.mark.parametrize(
"inputs",
[
{
"environment": None,
"deployment_name": "some deployment",
"is_trusted": True,
"preserve_previous_prod_deployment": False,
},
{
"environment": PRODUCTION_ENVIRONMENT_NAME,
"deployment_name": None,
"is_trusted": True,
"preserve_previous_prod_deployment": False,
},
{
"environment": "staging",
"deployment_name": "some_deployment_name",
"is_trusted": False,
"preserve_previous_prod_deployment": True,
},
],
)
def test_create_truss_service_handles_existing_model(inputs):
api = MagicMock()
return_value = {"id": "model_version_id"}
api.create_model_version_from_truss.return_value = return_value
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_draft=False,
model_id="model_id",
**inputs,
)
truss/tests/remote/baseten/test_core.py:156:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602078448'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = True
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = False, model_id = 'model_id', deployment_name = 'some deployment'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
if model_id is None:
if environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
model_version_json = api.create_model_from_truss(
model_name=model_name,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
deployment_name=deployment_name,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
try:
model_version_json = api.create_model_version_from_truss(
model_id=model_id,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
preserve_previous_prod_deployment=preserve_previous_prod_deployment,
deployment_name=deployment_name,
environment=environment,
)
except ApiError as e:
if (
e.graphql_error_code
== BasetenApi.GraphQLErrorCodes.RESOURCE_NOT_FOUND.value
):
raise ValueError(
f'Environment "{environment}" does not exist. You can create environments in the Baseten UI.'
) from e
raise e
return ModelVersionHandle(
id=model_version_json["id"],
model_id=model_id,
> invoke_base_url=model_version_json["invoke_base_url"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:426: KeyError
Check failure on line 156 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_existing_model[inputs1]
KeyError: 'invoke_base_url'
Raw output
inputs = {'deployment_name': None, 'environment': 'production', 'is_trusted': True, 'preserve_previous_prod_deployment': False}
@pytest.mark.parametrize(
"inputs",
[
{
"environment": None,
"deployment_name": "some deployment",
"is_trusted": True,
"preserve_previous_prod_deployment": False,
},
{
"environment": PRODUCTION_ENVIRONMENT_NAME,
"deployment_name": None,
"is_trusted": True,
"preserve_previous_prod_deployment": False,
},
{
"environment": "staging",
"deployment_name": "some_deployment_name",
"is_trusted": False,
"preserve_previous_prod_deployment": True,
},
],
)
def test_create_truss_service_handles_existing_model(inputs):
api = MagicMock()
return_value = {"id": "model_version_id"}
api.create_model_version_from_truss.return_value = return_value
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_draft=False,
model_id="model_id",
**inputs,
)
truss/tests/remote/baseten/test_core.py:156:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602478704'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = True
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = False, model_id = 'model_id', deployment_name = None, origin = None
environment = 'production'
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
if model_id is None:
if environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
model_version_json = api.create_model_from_truss(
model_name=model_name,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
deployment_name=deployment_name,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
try:
model_version_json = api.create_model_version_from_truss(
model_id=model_id,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
preserve_previous_prod_deployment=preserve_previous_prod_deployment,
deployment_name=deployment_name,
environment=environment,
)
except ApiError as e:
if (
e.graphql_error_code
== BasetenApi.GraphQLErrorCodes.RESOURCE_NOT_FOUND.value
):
raise ValueError(
f'Environment "{environment}" does not exist. You can create environments in the Baseten UI.'
) from e
raise e
return ModelVersionHandle(
id=model_version_json["id"],
model_id=model_id,
> invoke_base_url=model_version_json["invoke_base_url"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:426: KeyError
Check failure on line 156 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_existing_model[inputs2]
KeyError: 'invoke_base_url'
Raw output
inputs = {'deployment_name': 'some_deployment_name', 'environment': 'staging', 'is_trusted': False, 'preserve_previous_prod_deployment': True}
@pytest.mark.parametrize(
"inputs",
[
{
"environment": None,
"deployment_name": "some deployment",
"is_trusted": True,
"preserve_previous_prod_deployment": False,
},
{
"environment": PRODUCTION_ENVIRONMENT_NAME,
"deployment_name": None,
"is_trusted": True,
"preserve_previous_prod_deployment": False,
},
{
"environment": "staging",
"deployment_name": "some_deployment_name",
"is_trusted": False,
"preserve_previous_prod_deployment": True,
},
],
)
def test_create_truss_service_handles_existing_model(inputs):
api = MagicMock()
return_value = {"id": "model_version_id"}
api.create_model_version_from_truss.return_value = return_value
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_draft=False,
model_id="model_id",
**inputs,
)
truss/tests/remote/baseten/test_core.py:156:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602256224'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = True, allow_truss_download = False
is_draft = False, model_id = 'model_id'
deployment_name = 'some_deployment_name', origin = None, environment = 'staging'
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
if model_id is None:
if environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
model_version_json = api.create_model_from_truss(
model_name=model_name,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
deployment_name=deployment_name,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
try:
model_version_json = api.create_model_version_from_truss(
model_id=model_id,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
preserve_previous_prod_deployment=preserve_previous_prod_deployment,
deployment_name=deployment_name,
environment=environment,
)
except ApiError as e:
if (
e.graphql_error_code
== BasetenApi.GraphQLErrorCodes.RESOURCE_NOT_FOUND.value
):
raise ValueError(
f'Environment "{environment}" does not exist. You can create environments in the Baseten UI.'
) from e
raise e
return ModelVersionHandle(
id=model_version_json["id"],
model_id=model_id,
> invoke_base_url=model_version_json["invoke_base_url"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:426: KeyError
Check failure on line 185 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_allow_truss_download_for_new_models[True-True]
KeyError: 'invoke_base_url'
Raw output
is_draft = True, allow_truss_download = True
@pytest.mark.parametrize("allow_truss_download", [True, False])
@pytest.mark.parametrize("is_draft", [True, False])
def test_create_truss_service_handles_allow_truss_download_for_new_models(
is_draft, allow_truss_download
):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_model_from_truss.return_value = return_value
api.create_development_model_from_truss.return_value = return_value
model_id = None
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=is_draft,
model_id=model_id,
deployment_name="deployment_name",
allow_truss_download=allow_truss_download,
)
truss/tests/remote/baseten/test_core.py:185:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602424160'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = True
is_draft = True, model_id = None, deployment_name = 'deployment_name'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:375: KeyError
Check failure on line 185 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_allow_truss_download_for_new_models[True-False]
KeyError: 'invoke_base_url'
Raw output
is_draft = True, allow_truss_download = False
@pytest.mark.parametrize("allow_truss_download", [True, False])
@pytest.mark.parametrize("is_draft", [True, False])
def test_create_truss_service_handles_allow_truss_download_for_new_models(
is_draft, allow_truss_download
):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_model_from_truss.return_value = return_value
api.create_development_model_from_truss.return_value = return_value
model_id = None
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=is_draft,
model_id=model_id,
deployment_name="deployment_name",
allow_truss_download=allow_truss_download,
)
truss/tests/remote/baseten/test_core.py:185:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602439760'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = True, model_id = None, deployment_name = 'deployment_name'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:375: KeyError
Check failure on line 185 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_allow_truss_download_for_new_models[False-True]
KeyError: 'invoke_base_url'
Raw output
is_draft = False, allow_truss_download = True
@pytest.mark.parametrize("allow_truss_download", [True, False])
@pytest.mark.parametrize("is_draft", [True, False])
def test_create_truss_service_handles_allow_truss_download_for_new_models(
is_draft, allow_truss_download
):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_model_from_truss.return_value = return_value
api.create_development_model_from_truss.return_value = return_value
model_id = None
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=is_draft,
model_id=model_id,
deployment_name="deployment_name",
allow_truss_download=allow_truss_download,
)
truss/tests/remote/baseten/test_core.py:185:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827602177664'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = True
is_draft = False, model_id = None, deployment_name = 'deployment_name'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
if model_id is None:
if environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
model_version_json = api.create_model_from_truss(
model_name=model_name,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
deployment_name=deployment_name,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:397: KeyError
Check failure on line 185 in .venv/lib/python3.9/site-packages/numpy/ma/tests/test_core.py
github-actions / JUnit Test Report
test_core.test_create_truss_service_handles_allow_truss_download_for_new_models[False-False]
KeyError: 'invoke_base_url'
Raw output
is_draft = False, allow_truss_download = False
@pytest.mark.parametrize("allow_truss_download", [True, False])
@pytest.mark.parametrize("is_draft", [True, False])
def test_create_truss_service_handles_allow_truss_download_for_new_models(
is_draft, allow_truss_download
):
api = MagicMock()
return_value = {"id": "id", "version_id": "model_version_id"}
api.create_model_from_truss.return_value = return_value
api.create_development_model_from_truss.return_value = return_value
model_id = None
> model_id, model_version_id = create_truss_service(
api,
"model_name",
"s3_key",
"config",
is_trusted=False,
preserve_previous_prod_deployment=False,
is_draft=is_draft,
model_id=model_id,
deployment_name="deployment_name",
allow_truss_download=allow_truss_download,
)
truss/tests/remote/baseten/test_core.py:185:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <MagicMock id='139827601023760'>, model_name = 'model_name'
s3_key = 's3_key', config = 'config', semver_bump = 'MINOR', is_trusted = False
preserve_previous_prod_deployment = False, allow_truss_download = False
is_draft = False, model_id = None, deployment_name = 'deployment_name'
origin = None, environment = None
def create_truss_service(
api: BasetenApi,
model_name: str,
s3_key: str,
config: str,
semver_bump: str = "MINOR",
is_trusted: bool = False,
preserve_previous_prod_deployment: bool = False,
allow_truss_download: bool = False,
is_draft: Optional[bool] = False,
model_id: Optional[str] = None,
deployment_name: Optional[str] = None,
origin: Optional[b10_types.ModelOrigin] = None,
environment: Optional[str] = None,
) -> ModelVersionHandle:
"""
Create a model in the Baseten remote.
Args:
api: BasetenApi instance.
model_name: Name of the model to create.
s3_key: S3 key of the uploaded TrussHandle.
config: Base64 encoded JSON string of the Truss config.
semver_bump: Semver bump type, defaults to "MINOR".
is_trusted: Whether the model is trusted, defaults to False.
promote: Whether to promote the model after deploy, defaults to False.
preserve_previous_prod_deployment: Whether to scale old production deployment
to zero.
deployment_name: Name to apply to the created deployment. Not applied to
development model.
Returns:
A Model Version handle.
"""
if is_draft:
model_version_json = api.create_development_model_from_truss(
model_name,
s3_key,
config,
f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
if model_id is None:
if environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
model_version_json = api.create_model_from_truss(
model_name=model_name,
s3_key=s3_key,
config=config,
semver_bump=semver_bump,
client_version=f"truss=={truss.version()}",
is_trusted=is_trusted,
allow_truss_download=allow_truss_download,
deployment_name=deployment_name,
origin=origin,
)
return ModelVersionHandle(
id=model_version_json["id"],
> invoke_base_url=model_version_json["invoke_base_url"],
model_id=model_version_json["oracle"]["id"],
)
E KeyError: 'invoke_base_url'
truss/remote/baseten/core.py:397: KeyError
Check failure on line 40 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_get_service_by_version_id
KeyError: 'invoke_base_url'
Raw output
def test_get_service_by_version_id():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
version = {"id": "version_id", "oracle": {"id": "model_id"}}
model_version_response = {"data": {"model_version": version}}
with requests_mock.Mocker() as m:
m.post(_TEST_REMOTE_GRAPHQL_PATH, json=model_version_response)
> service = remote.get_service(model_identifier=ModelVersionId("version_id"))
truss/tests/remote/baseten/test_remote.py:40:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
truss/remote/baseten/remote.py:381: in get_service
self._get_service_url_path_and_model_ids(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268f5a90>
model_identifier = <truss.remote.baseten.core.ModelVersionId object at 0x7f2c268f5a30>
published = False
@staticmethod
def _get_service_url_path_and_model_ids(
api: BasetenApi, model_identifier: ModelIdentifier, published: bool
) -> Tuple[str, ModelVersionHandle]:
if isinstance(model_identifier, ModelVersionId):
try:
model_version = api.get_model_version_by_id(model_identifier.value)
except ApiError:
raise RemoteError(f"Model version {model_identifier.value} not found.")
model_version_id = model_version["model_version"]["id"]
> invoke_base_url = model_version["model_version"]["invoke_base_url"]
E KeyError: 'invoke_base_url'
truss/remote/baseten/remote.py:335: KeyError
Check failure on line 73 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_get_service_by_model_name
KeyError: 'invoke_base_url'
Raw output
def test_get_service_by_model_name():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
versions = [
{"id": "1", "is_draft": False, "is_primary": False},
{"id": "2", "is_draft": False, "is_primary": True},
{"id": "3", "is_draft": True, "is_primary": False},
]
model_response = {
"data": {
"model": {"name": "model_name", "id": "model_id", "versions": versions}
}
}
with requests_mock.Mocker() as m:
m.post(_TEST_REMOTE_GRAPHQL_PATH, json=model_response)
# Check that the production version is returned when published is True.
> service = remote.get_service(
model_identifier=ModelName("model_name"), published=True
)
truss/tests/remote/baseten/test_remote.py:73:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
truss/remote/baseten/remote.py:381: in get_service
self._get_service_url_path_and_model_ids(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c26915ee0>
model_identifier = <truss.remote.baseten.core.ModelName object at 0x7f2c26915610>
published = True
@staticmethod
def _get_service_url_path_and_model_ids(
api: BasetenApi, model_identifier: ModelIdentifier, published: bool
) -> Tuple[str, ModelVersionHandle]:
if isinstance(model_identifier, ModelVersionId):
try:
model_version = api.get_model_version_by_id(model_identifier.value)
except ApiError:
raise RemoteError(f"Model version {model_identifier.value} not found.")
model_version_id = model_version["model_version"]["id"]
invoke_base_url = model_version["model_version"]["invoke_base_url"]
model_id = model_version["model_version"]["oracle"]["id"]
service_url_path = f"/model_versions/{model_version_id}"
return service_url_path, ModelVersionHandle(
id=model_version_id, model_id=model_id, invoke_base_url=invoke_base_url
)
if isinstance(model_identifier, ModelName):
model_id, model_versions = get_model_versions(api, model_identifier)
model_version = BasetenRemote._get_matching_version(
model_versions, published
)
model_version_id = model_version["id"]
> invoke_base_url = model_version["invoke_base_url"]
E KeyError: 'invoke_base_url'
truss/remote/baseten/remote.py:349: KeyError
Check failure on line 101 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_get_service_by_model_name_no_dev_version
KeyError: 'invoke_base_url'
Raw output
def test_get_service_by_model_name_no_dev_version():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
versions = [{"id": "1", "is_draft": False, "is_primary": True}]
model_response = {
"data": {
"model": {"name": "model_name", "id": "model_id", "versions": versions}
}
}
with requests_mock.Mocker() as m:
m.post(_TEST_REMOTE_GRAPHQL_PATH, json=model_response)
# Check that the production version is returned when published is True.
> service = remote.get_service(
model_identifier=ModelName("model_name"), published=True
)
truss/tests/remote/baseten/test_remote.py:101:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
truss/remote/baseten/remote.py:381: in get_service
self._get_service_url_path_and_model_ids(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268e5a90>
model_identifier = <truss.remote.baseten.core.ModelName object at 0x7f2c268e5520>
published = True
@staticmethod
def _get_service_url_path_and_model_ids(
api: BasetenApi, model_identifier: ModelIdentifier, published: bool
) -> Tuple[str, ModelVersionHandle]:
if isinstance(model_identifier, ModelVersionId):
try:
model_version = api.get_model_version_by_id(model_identifier.value)
except ApiError:
raise RemoteError(f"Model version {model_identifier.value} not found.")
model_version_id = model_version["model_version"]["id"]
invoke_base_url = model_version["model_version"]["invoke_base_url"]
model_id = model_version["model_version"]["oracle"]["id"]
service_url_path = f"/model_versions/{model_version_id}"
return service_url_path, ModelVersionHandle(
id=model_version_id, model_id=model_id, invoke_base_url=invoke_base_url
)
if isinstance(model_identifier, ModelName):
model_id, model_versions = get_model_versions(api, model_identifier)
model_version = BasetenRemote._get_matching_version(
model_versions, published
)
model_version_id = model_version["id"]
> invoke_base_url = model_version["invoke_base_url"]
E KeyError: 'invoke_base_url'
truss/remote/baseten/remote.py:349: KeyError
Check failure on line 134 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_get_service_by_model_name_no_prod_version
KeyError: 'invoke_base_url'
Raw output
def test_get_service_by_model_name_no_prod_version():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
versions = [{"id": "1", "is_draft": True, "is_primary": False}]
model_response = {
"data": {
"model": {"name": "model_name", "id": "model_id", "versions": versions}
}
}
with requests_mock.Mocker() as m:
m.post(_TEST_REMOTE_GRAPHQL_PATH, json=model_response)
# Since no production version exists, calling get_service with
# published=True should raise an error.
with pytest.raises(RemoteError):
remote.get_service(model_identifier=ModelName("model_name"), published=True)
# Check that the development version is returned when published is False.
> service = remote.get_service(
model_identifier=ModelName("model_name"), published=False
)
truss/tests/remote/baseten/test_remote.py:134:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
truss/remote/baseten/remote.py:381: in get_service
self._get_service_url_path_and_model_ids(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c26827bb0>
model_identifier = <truss.remote.baseten.core.ModelName object at 0x7f2c268275e0>
published = False
@staticmethod
def _get_service_url_path_and_model_ids(
api: BasetenApi, model_identifier: ModelIdentifier, published: bool
) -> Tuple[str, ModelVersionHandle]:
if isinstance(model_identifier, ModelVersionId):
try:
model_version = api.get_model_version_by_id(model_identifier.value)
except ApiError:
raise RemoteError(f"Model version {model_identifier.value} not found.")
model_version_id = model_version["model_version"]["id"]
invoke_base_url = model_version["model_version"]["invoke_base_url"]
model_id = model_version["model_version"]["oracle"]["id"]
service_url_path = f"/model_versions/{model_version_id}"
return service_url_path, ModelVersionHandle(
id=model_version_id, model_id=model_id, invoke_base_url=invoke_base_url
)
if isinstance(model_identifier, ModelName):
model_id, model_versions = get_model_versions(api, model_identifier)
model_version = BasetenRemote._get_matching_version(
model_versions, published
)
model_version_id = model_version["id"]
> invoke_base_url = model_version["invoke_base_url"]
E KeyError: 'invoke_base_url'
truss/remote/baseten/remote.py:349: KeyError
Check failure on line 157 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_get_service_by_model_id
KeyError: 'invoke_base_url'
Raw output
def test_get_service_by_model_id():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
model_response = {
"data": {
"model": {
"name": "model_name",
"id": "model_id",
"primary_version": {"id": "version_id"},
}
}
}
with requests_mock.Mocker() as m:
m.post(_TEST_REMOTE_GRAPHQL_PATH, json=model_response)
> service = remote.get_service(model_identifier=ModelId("model_id"))
truss/tests/remote/baseten/test_remote.py:157:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
truss/remote/baseten/remote.py:381: in get_service
self._get_service_url_path_and_model_ids(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268e5f40>
model_identifier = <truss.remote.baseten.core.ModelId object at 0x7f2c268e50d0>
published = False
@staticmethod
def _get_service_url_path_and_model_ids(
api: BasetenApi, model_identifier: ModelIdentifier, published: bool
) -> Tuple[str, ModelVersionHandle]:
if isinstance(model_identifier, ModelVersionId):
try:
model_version = api.get_model_version_by_id(model_identifier.value)
except ApiError:
raise RemoteError(f"Model version {model_identifier.value} not found.")
model_version_id = model_version["model_version"]["id"]
invoke_base_url = model_version["model_version"]["invoke_base_url"]
model_id = model_version["model_version"]["oracle"]["id"]
service_url_path = f"/model_versions/{model_version_id}"
return service_url_path, ModelVersionHandle(
id=model_version_id, model_id=model_id, invoke_base_url=invoke_base_url
)
if isinstance(model_identifier, ModelName):
model_id, model_versions = get_model_versions(api, model_identifier)
model_version = BasetenRemote._get_matching_version(
model_versions, published
)
model_version_id = model_version["id"]
invoke_base_url = model_version["invoke_base_url"]
service_url_path = f"/model_versions/{model_version_id}"
elif isinstance(model_identifier, ModelId):
# TODO(helen): consider making this consistent with getting the
# service via model_name / respect --published in service_url_path.
try:
model = api.get_model_by_id(model_identifier.value)
except ApiError:
raise RemoteError(f"Model {model_identifier.value} not found.")
model_id = model["model"]["id"]
model_version_id = model["model"]["primary_version"]["id"]
> invoke_base_url = model["model"]["primary_version"]["invoke_base_url"]
E KeyError: 'invoke_base_url'
truss/remote/baseten/remote.py:360: KeyError
Check failure on line 289 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_create_chain_with_no_publish
KeyError: 'hostname'
Raw output
def test_create_chain_with_no_publish():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
with requests_mock.Mocker() as m:
m.post(
_TEST_REMOTE_GRAPHQL_PATH,
[
{"json": {"data": {"chains": []}}},
{
"json": {
"data": {
"deploy_chain_atomic": {
"chain_deployment": {
"id": "new-chain-deployment-id",
"chain": {"id": "new-chain-id"},
}
}
}
}
},
],
)
> deployment_handle = create_chain_atomic(
api=remote.api,
chain_name="draft_chain",
entrypoint=ChainletDataAtomic(
name="chainlet-1",
oracle=OracleData(
model_name="model-1",
s3_key="s3-key-1",
encoded_config_str="encoded-config-str-1",
is_trusted=True,
),
),
dependencies=[],
is_draft=True,
environment=None,
)
truss/tests/remote/baseten/test_remote.py:289:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c267b56a0>
chain_name = 'draft_chain'
entrypoint = ChainletDataAtomic(name='chainlet-1', oracle=OracleData(model_name='model-1', s3_key='s3-key-1', encoded_config_str='encoded-config-str-1', semver_bump='MINOR', is_trusted=True, version_name=None))
dependencies = [], is_draft = True, environment = None
def create_chain_atomic(
api: BasetenApi,
chain_name: str,
entrypoint: b10_types.ChainletDataAtomic,
dependencies: List[b10_types.ChainletDataAtomic],
is_draft: bool,
environment: Optional[str],
) -> ChainDeploymentHandleAtomic:
if environment and is_draft:
logging.info(
f"Automatically publishing Chain `{chain_name}` based on "
"environment setting."
)
is_draft = False
chain_id = get_chain_id_by_name(api, chain_name)
# TODO(Tyron): Refactor for better readability:
# 1. Prepare all arguments for `deploy_chain_atomic`.
# 2. Validate argument combinations.
# 3. Make a single invocation to `deploy_chain_atomic`.
if is_draft:
res = api.deploy_chain_atomic(
chain_name=chain_name,
is_draft=True,
entrypoint=entrypoint,
dependencies=dependencies,
)
elif chain_id:
# This is the only case where promote has relevance, since
# if there is no chain already, the first deployment will
# already be production, and only published deployments can
# be promoted.
try:
res = api.deploy_chain_atomic(
chain_id=chain_id,
environment=environment,
entrypoint=entrypoint,
dependencies=dependencies,
)
except ApiError as e:
if (
e.graphql_error_code
== BasetenApi.GraphQLErrorCodes.RESOURCE_NOT_FOUND.value
):
raise ValueError(
f"Environment `{environment}` does not exist. You can "
f"create environments in the Chains UI."
) from e
raise e
elif environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
else:
res = api.deploy_chain_atomic(
chain_name=chain_name, entrypoint=entrypoint, dependencies=dependencies
)
return ChainDeploymentHandleAtomic(
chain_deployment_id=res["chain_deployment"]["id"],
chain_id=res["chain_deployment"]["chain"]["id"],
> hostname=res["chain_deployment"]["chain"]["hostname"],
is_draft=is_draft,
)
E KeyError: 'hostname'
truss/remote/baseten/core.py:170: KeyError
Check failure on line 387 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_create_chain_no_existing_chain
KeyError: 'hostname'
Raw output
def test_create_chain_no_existing_chain():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
with requests_mock.Mocker() as m:
m.post(
_TEST_REMOTE_GRAPHQL_PATH,
[
{"json": {"data": {"chains": []}}},
{
"json": {
"data": {
"deploy_chain_atomic": {
"chain_deployment": {
"id": "new-chain-deployment-id",
"chain": {"id": "new-chain-id"},
}
}
}
}
},
],
)
> deployment_handle = create_chain_atomic(
api=remote.api,
chain_name="new_chain",
entrypoint=ChainletDataAtomic(
name="chainlet-1",
oracle=OracleData(
model_name="model-1",
s3_key="s3-key-1",
encoded_config_str="encoded-config-str-1",
is_trusted=True,
),
),
dependencies=[],
is_draft=False,
environment=None,
)
truss/tests/remote/baseten/test_remote.py:387:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c267b5c70>
chain_name = 'new_chain'
entrypoint = ChainletDataAtomic(name='chainlet-1', oracle=OracleData(model_name='model-1', s3_key='s3-key-1', encoded_config_str='encoded-config-str-1', semver_bump='MINOR', is_trusted=True, version_name=None))
dependencies = [], is_draft = False, environment = None
def create_chain_atomic(
api: BasetenApi,
chain_name: str,
entrypoint: b10_types.ChainletDataAtomic,
dependencies: List[b10_types.ChainletDataAtomic],
is_draft: bool,
environment: Optional[str],
) -> ChainDeploymentHandleAtomic:
if environment and is_draft:
logging.info(
f"Automatically publishing Chain `{chain_name}` based on "
"environment setting."
)
is_draft = False
chain_id = get_chain_id_by_name(api, chain_name)
# TODO(Tyron): Refactor for better readability:
# 1. Prepare all arguments for `deploy_chain_atomic`.
# 2. Validate argument combinations.
# 3. Make a single invocation to `deploy_chain_atomic`.
if is_draft:
res = api.deploy_chain_atomic(
chain_name=chain_name,
is_draft=True,
entrypoint=entrypoint,
dependencies=dependencies,
)
elif chain_id:
# This is the only case where promote has relevance, since
# if there is no chain already, the first deployment will
# already be production, and only published deployments can
# be promoted.
try:
res = api.deploy_chain_atomic(
chain_id=chain_id,
environment=environment,
entrypoint=entrypoint,
dependencies=dependencies,
)
except ApiError as e:
if (
e.graphql_error_code
== BasetenApi.GraphQLErrorCodes.RESOURCE_NOT_FOUND.value
):
raise ValueError(
f"Environment `{environment}` does not exist. You can "
f"create environments in the Chains UI."
) from e
raise e
elif environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
else:
res = api.deploy_chain_atomic(
chain_name=chain_name, entrypoint=entrypoint, dependencies=dependencies
)
return ChainDeploymentHandleAtomic(
chain_deployment_id=res["chain_deployment"]["id"],
chain_id=res["chain_deployment"]["chain"]["id"],
> hostname=res["chain_deployment"]["chain"]["hostname"],
is_draft=is_draft,
)
E KeyError: 'hostname'
truss/remote/baseten/core.py:170: KeyError
Check failure on line 489 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_create_chain_with_existing_chain_promote_to_environment_publish_false
KeyError: 'hostname'
Raw output
def test_create_chain_with_existing_chain_promote_to_environment_publish_false():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
with requests_mock.Mocker() as m:
m.post(
_TEST_REMOTE_GRAPHQL_PATH,
[
{
"json": {
"data": {
"chains": [{"id": "old-chain-id", "name": "old_chain"}]
}
}
},
{
"json": {
"data": {
"deploy_chain_atomic": {
"chain_deployment": {
"id": "new-chain-deployment-id",
"chain": {"id": "new-chain-id"},
}
}
}
}
},
],
)
> deployment_handle = create_chain_atomic(
api=remote.api,
chain_name="old_chain",
entrypoint=ChainletDataAtomic(
name="chainlet-1",
oracle=OracleData(
model_name="model-1",
s3_key="s3-key-1",
encoded_config_str="encoded-config-str-1",
is_trusted=True,
),
),
dependencies=[],
is_draft=True,
environment="production",
)
truss/tests/remote/baseten/test_remote.py:489:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c267b5220>
chain_name = 'old_chain'
entrypoint = ChainletDataAtomic(name='chainlet-1', oracle=OracleData(model_name='model-1', s3_key='s3-key-1', encoded_config_str='encoded-config-str-1', semver_bump='MINOR', is_trusted=True, version_name=None))
dependencies = [], is_draft = False, environment = 'production'
def create_chain_atomic(
api: BasetenApi,
chain_name: str,
entrypoint: b10_types.ChainletDataAtomic,
dependencies: List[b10_types.ChainletDataAtomic],
is_draft: bool,
environment: Optional[str],
) -> ChainDeploymentHandleAtomic:
if environment and is_draft:
logging.info(
f"Automatically publishing Chain `{chain_name}` based on "
"environment setting."
)
is_draft = False
chain_id = get_chain_id_by_name(api, chain_name)
# TODO(Tyron): Refactor for better readability:
# 1. Prepare all arguments for `deploy_chain_atomic`.
# 2. Validate argument combinations.
# 3. Make a single invocation to `deploy_chain_atomic`.
if is_draft:
res = api.deploy_chain_atomic(
chain_name=chain_name,
is_draft=True,
entrypoint=entrypoint,
dependencies=dependencies,
)
elif chain_id:
# This is the only case where promote has relevance, since
# if there is no chain already, the first deployment will
# already be production, and only published deployments can
# be promoted.
try:
res = api.deploy_chain_atomic(
chain_id=chain_id,
environment=environment,
entrypoint=entrypoint,
dependencies=dependencies,
)
except ApiError as e:
if (
e.graphql_error_code
== BasetenApi.GraphQLErrorCodes.RESOURCE_NOT_FOUND.value
):
raise ValueError(
f"Environment `{environment}` does not exist. You can "
f"create environments in the Chains UI."
) from e
raise e
elif environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
else:
res = api.deploy_chain_atomic(
chain_name=chain_name, entrypoint=entrypoint, dependencies=dependencies
)
return ChainDeploymentHandleAtomic(
chain_deployment_id=res["chain_deployment"]["id"],
chain_id=res["chain_deployment"]["chain"]["id"],
> hostname=res["chain_deployment"]["chain"]["hostname"],
is_draft=is_draft,
)
E KeyError: 'hostname'
truss/remote/baseten/core.py:170: KeyError
Check failure on line 594 in truss/tests/remote/baseten/test_remote.py
github-actions / JUnit Test Report
test_remote.test_create_chain_existing_chain_publish_true_no_promotion
KeyError: 'hostname'
Raw output
def test_create_chain_existing_chain_publish_true_no_promotion():
remote = BasetenRemote(_TEST_REMOTE_URL, "api_key")
with requests_mock.Mocker() as m:
m.post(
_TEST_REMOTE_GRAPHQL_PATH,
[
{
"json": {
"data": {
"chains": [{"id": "old-chain-id", "name": "old_chain"}]
}
}
},
{
"json": {
"data": {
"deploy_chain_atomic": {
"chain_deployment": {
"id": "new-chain-deployment-id",
"chain": {"id": "new-chain-id"},
}
}
}
}
},
],
)
> deployment_handle = create_chain_atomic(
api=remote.api,
chain_name="old_chain",
entrypoint=ChainletDataAtomic(
name="chainlet-1",
oracle=OracleData(
model_name="model-1",
s3_key="s3-key-1",
encoded_config_str="encoded-config-str-1",
is_trusted=True,
),
),
dependencies=[],
is_draft=False,
environment=None,
)
truss/tests/remote/baseten/test_remote.py:594:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api = <truss.remote.baseten.api.BasetenApi object at 0x7f2c268c04f0>
chain_name = 'old_chain'
entrypoint = ChainletDataAtomic(name='chainlet-1', oracle=OracleData(model_name='model-1', s3_key='s3-key-1', encoded_config_str='encoded-config-str-1', semver_bump='MINOR', is_trusted=True, version_name=None))
dependencies = [], is_draft = False, environment = None
def create_chain_atomic(
api: BasetenApi,
chain_name: str,
entrypoint: b10_types.ChainletDataAtomic,
dependencies: List[b10_types.ChainletDataAtomic],
is_draft: bool,
environment: Optional[str],
) -> ChainDeploymentHandleAtomic:
if environment and is_draft:
logging.info(
f"Automatically publishing Chain `{chain_name}` based on "
"environment setting."
)
is_draft = False
chain_id = get_chain_id_by_name(api, chain_name)
# TODO(Tyron): Refactor for better readability:
# 1. Prepare all arguments for `deploy_chain_atomic`.
# 2. Validate argument combinations.
# 3. Make a single invocation to `deploy_chain_atomic`.
if is_draft:
res = api.deploy_chain_atomic(
chain_name=chain_name,
is_draft=True,
entrypoint=entrypoint,
dependencies=dependencies,
)
elif chain_id:
# This is the only case where promote has relevance, since
# if there is no chain already, the first deployment will
# already be production, and only published deployments can
# be promoted.
try:
res = api.deploy_chain_atomic(
chain_id=chain_id,
environment=environment,
entrypoint=entrypoint,
dependencies=dependencies,
)
except ApiError as e:
if (
e.graphql_error_code
== BasetenApi.GraphQLErrorCodes.RESOURCE_NOT_FOUND.value
):
raise ValueError(
f"Environment `{environment}` does not exist. You can "
f"create environments in the Chains UI."
) from e
raise e
elif environment and environment != PRODUCTION_ENVIRONMENT_NAME:
raise ValueError(NO_ENVIRONMENTS_EXIST_ERROR_MESSAGING)
else:
res = api.deploy_chain_atomic(
chain_name=chain_name, entrypoint=entrypoint, dependencies=dependencies
)
return ChainDeploymentHandleAtomic(
chain_deployment_id=res["chain_deployment"]["id"],
chain_id=res["chain_deployment"]["chain"]["id"],
> hostname=res["chain_deployment"]["chain"]["hostname"],
is_draft=is_draft,
)
E KeyError: 'hostname'
truss/remote/baseten/core.py:170: KeyError
Loading