Mirrors/modelscope
1
0
Fork 0
mirror of https://github.com/modelscope/modelscope.git synced 2026-07-09 11:59:58 +02:00
Code Issues Packages Projects Releases Wiki Activity

[Feature] Add studio module (#1727)

Browse Source
This commit is contained in:
Xingjun.Wang
2026-06-05 14:33:25 +08:00
committed by GitHub
parent 63ff6ec6e9
commit 1804ac6bf8
13 changed files with 1452 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
modelscope/cli/cli.py
View File

@@ -14,6 +14,7 @@ from modelscope.cli.plugins import PluginsCMD
from modelscope.cli.scancache import ScanCacheCMD
from modelscope.cli.server import ServerCMD
from modelscope.cli.skills import SkillsCMD
from modelscope.cli.studio import StudioCMD
from modelscope.cli.upload import UploadCMD
from modelscope.hub.constants import MODELSCOPE_ASCII
from modelscope.utils.logger import get_logger
@@ -47,6 +48,7 @@ def run_cmd():
LoginCMD.define_args(subparsers)
LlamafileCMD.define_args(subparsers)
ScanCacheCMD.define_args(subparsers)
StudioCMD.define_args(subparsers)
args = parser.parse_args()

42
modelscope/cli/create.py
View File

@@ -7,7 +7,9 @@ from modelscope.hub.constants import (GatedMode, Licenses, ModelVisibility,
Visibility, VisibilityMap)
from modelscope.hub.utils.aigc import AigcModel
from modelscope.hub.utils.utils import resolve_endpoint
from modelscope.utils.constant import REPO_TYPE_MODEL, REPO_TYPE_SUPPORT
from modelscope.utils.constant import (REPO_TYPE_MODEL, REPO_TYPE_STUDIO,
REPO_TYPE_SUPPORT, StudioHardware,
StudioSDKType)
from modelscope.utils.logger import get_logger
logger = get_logger()
@@ -107,6 +109,35 @@ class CreateCMD(CLICommand):
'then defaults to https://www.modelscope.cn.',
)
# Studio specific arguments (only meaningful when --repo_type studio)
studio_group = parser.add_argument_group(
'Studio Repo Creation',
'Optional arguments used only when `--repo_type studio` is set.')
studio_group.add_argument(
'--sdk-type',
dest='sdk_type',
choices=StudioSDKType.SUPPORTED,
default=None,
help='Studio SDK type (only for studio repo-type).')
studio_group.add_argument(
'--sdk-version',
dest='sdk_version',
type=str,
default=None,
help='Studio SDK version (only for gradio).')
studio_group.add_argument(
'--base-image',
dest='base_image',
type=str,
default=None,
help='Studio base image (only for gradio/streamlit).')
studio_group.add_argument(
'--hardware',
dest='hardware',
choices=StudioHardware.SUPPORTED,
default=None,
help='Studio hardware configuration.')
# AIGC specific arguments
aigc_group = parser.add_argument_group(
'AIGC Model Creation',
@@ -179,6 +210,14 @@ class CreateCMD(CLICommand):
endpoint = resolve_endpoint(self.args.endpoint)
api = HubApi(endpoint=endpoint)
extra_kwargs = {}
if self.args.repo_type == REPO_TYPE_STUDIO:
# Pass studio-specific fields only when creating a studio repo.
for field in ('sdk_type', 'sdk_version', 'base_image', 'hardware'):
value = getattr(self.args, field, None)
if value is not None:
extra_kwargs[field] = value
# Create repo
api.create_repo(
repo_id=self.args.repo_id,
@@ -191,6 +230,7 @@ class CreateCMD(CLICommand):
create_default_config=True,
endpoint=endpoint,
gated_mode=self.args.gated_mode,
**extra_kwargs,
)
def _create_aigc_model(self):

14
modelscope/cli/download.py
View File

@@ -11,7 +11,9 @@ from modelscope.hub.file_download import (dataset_file_download,
from modelscope.hub.snapshot_download import (dataset_snapshot_download,
snapshot_download)
from modelscope.hub.utils.utils import convert_patterns, resolve_endpoint
from modelscope.utils.constant import DEFAULT_DATASET_REVISION
from modelscope.utils.constant import (DEFAULT_DATASET_REVISION,
REPO_TYPE_DATASET, REPO_TYPE_MODEL,
REPO_TYPE_STUDIO, REPO_TYPE_SUPPORT)
from modelscope.utils.logger import get_logger
logger = get_logger(log_level=logging.WARNING)
@@ -60,8 +62,8 @@ class DownloadCMD(CLICommand):
)
parser.add_argument(
'--repo-type',
choices=['model', 'dataset'],
default='model',
choices=REPO_TYPE_SUPPORT,
default=REPO_TYPE_MODEL,
help="Type of repo to download from (defaults to 'model').",
)
parser.add_argument(
@@ -135,9 +137,11 @@ class DownloadCMD(CLICommand):
self.args.files = [self.args.repo_id]
else:
if self.args.repo_id is not None:
if self.args.repo_type == 'model':
if self.args.repo_type in (REPO_TYPE_MODEL, REPO_TYPE_STUDIO):
# studio repos share the same snapshot_download path
# as model repos.
self.args.model = self.args.repo_id
elif self.args.repo_type == 'dataset':
elif self.args.repo_type == REPO_TYPE_DATASET:
self.args.dataset = self.args.repo_id
else:
raise Exception('Not support repo-type: %s'

384
modelscope/cli/studio.py Normal file
View File

@@ -0,0 +1,384 @@
# Copyright (c) Alibaba, Inc. and its affiliates.
"""ModelScope Studio runtime management CLI.
This module exposes a ``modelscope studio`` command group that wraps the
Studio OpenAPI methods on :class:`~modelscope.hub.api.HubApi`. Each CLI
subcommand is a thin adapter that parses arguments, delegates to a single
``HubApi`` method and renders a concise human-readable result.
Subcommand layout::
modelscope studio deploy <studio_id>
modelscope studio stop <studio_id>
modelscope studio logs <studio_id> [--type ...] [--keyword ...] ...
modelscope studio settings <studio_id> [--display-name ...] ...
modelscope studio secret list <studio_id>
modelscope studio secret add <studio_id> <key> <value>
modelscope studio secret update <studio_id> <key> <value>
modelscope studio secret delete <studio_id> <key>
"""
from argparse import ArgumentParser, _SubParsersAction
import json
from modelscope.cli.base import CLICommand
from modelscope.hub.api import HubApi
from modelscope.hub.utils.utils import resolve_endpoint
from modelscope.utils.constant import StudioSDKType
from modelscope.utils.logger import get_logger
logger = get_logger()
# Common log type choices for `studio logs --type ...`.
_LOG_TYPES = ('run', 'build')
def subparser_func(args):
""" Function which will be called for a specific sub parser.
"""
return StudioCMD(args)
class StudioCMD(CLICommand):
"""Studio runtime management commands.
Dispatches to a per-subcommand handler based on ``args.studio_action``
(and, for ``secret``, ``args.secret_action``).
"""
name = 'studio'
def __init__(self, args):
self.args = args
# ------------------------------------------------------------------
# Argument parsing
# ------------------------------------------------------------------
@staticmethod
def define_args(parsers: _SubParsersAction):
parser: ArgumentParser = parsers.add_parser(
StudioCMD.name, help='Manage ModelScope studios at runtime.')
# Common args available to every studio subcommand.
parser.add_argument(
'--token',
type=str,
default=None,
help='Optional access token used for authentication.')
parser.add_argument(
'--endpoint',
type=str,
default=None,
help='ModelScope server endpoint. Falls back to env '
'MODELSCOPE_DOMAIN, then https://www.modelscope.cn.')
action_subparsers = parser.add_subparsers(
dest='studio_action', help='Studio subcommands.')
StudioCMD._add_simple_id_parser(
action_subparsers,
name='deploy',
help_text='Deploy (re-pull and rebuild) a studio.')
StudioCMD._add_simple_id_parser(
action_subparsers, name='stop', help_text='Stop a running studio.')
StudioCMD._add_logs_parser(action_subparsers)
StudioCMD._add_settings_parser(action_subparsers)
StudioCMD._add_secret_parser(action_subparsers)
parser.set_defaults(func=subparser_func)
@staticmethod
def _add_simple_id_parser(action_subparsers, name, help_text):
sub = action_subparsers.add_parser(name, help=help_text)
sub.add_argument(
'studio_id',
type=str,
help='Studio ID in the format `owner/repo_name`.')
@staticmethod
def _add_logs_parser(action_subparsers):
sub = action_subparsers.add_parser(
'logs', help='Fetch studio runtime or build logs.')
sub.add_argument(
'studio_id',
type=str,
help='Studio ID in the format `owner/repo_name`.')
sub.add_argument(
'--type',
dest='log_type',
choices=_LOG_TYPES,
default='run',
help="Log type to fetch (defaults to 'run').")
sub.add_argument(
'--keyword',
type=str,
default=None,
help='Optional keyword to filter log lines.')
sub.add_argument(
'--page-num',
dest='page_num',
type=int,
default=1,
help='Page number, starting from 1.')
sub.add_argument(
'--page-size',
dest='page_size',
type=int,
default=100,
help='Number of log entries per page.')
sub.add_argument(
'--start-timestamp',
dest='start_timestamp',
type=int,
default=None,
help='Optional start timestamp (seconds since epoch).')
sub.add_argument(
'--end-timestamp',
dest='end_timestamp',
type=int,
default=None,
help='Optional end timestamp (seconds since epoch).')
@staticmethod
def _add_settings_parser(action_subparsers):
sub = action_subparsers.add_parser(
'settings',
help='Update studio settings (only specified fields are modified).'
)
sub.add_argument(
'studio_id',
type=str,
help='Studio ID in the format `owner/repo_name`.')
sub.add_argument(
'--display-name',
dest='display_name',
type=str,
default=None,
help='Studio display name (Chinese name).')
sub.add_argument(
'--description',
type=str,
default=None,
help='Studio description.')
sub.add_argument(
'--license', type=str, default=None, help='Studio license.')
sub.add_argument(
'--cover-image',
dest='cover_image',
type=str,
default=None,
help='Studio cover image URL.')
sub.add_argument(
'--sdk-type',
dest='sdk_type',
choices=StudioSDKType.SUPPORTED,
default=None,
help='Studio SDK type (requires redeployment).')
sub.add_argument(
'--sdk-version',
dest='sdk_version',
type=str,
default=None,
help='Studio SDK version (requires redeployment).')
sub.add_argument(
'--base-image',
dest='base_image',
type=str,
default=None,
help='Studio base image (requires redeployment).')
sub.add_argument(
'--hardware',
type=str,
default=None,
help='Studio hardware configuration (requires redeployment).')
visibility_group = sub.add_mutually_exclusive_group()
visibility_group.add_argument(
'--private',
dest='private',
action='store_const',
const=True,
default=None,
help='Mark the studio as private.')
visibility_group.add_argument(
'--public',
dest='private',
action='store_const',
const=False,
help='Mark the studio as public.')
@staticmethod
def _add_secret_parser(action_subparsers):
secret_parser = action_subparsers.add_parser(
'secret', help='Manage studio environment variables (secrets).')
secret_subparsers = secret_parser.add_subparsers(
dest='secret_action', help='Secret subcommands.')
list_sub = secret_subparsers.add_parser(
'list', help='List secret keys (values are not returned).')
list_sub.add_argument('studio_id', type=str)
add_sub = secret_subparsers.add_parser('add', help='Add a secret.')
add_sub.add_argument('studio_id', type=str)
add_sub.add_argument('key', type=str, help='Secret name.')
add_sub.add_argument('value', type=str, help='Secret value.')
upd_sub = secret_subparsers.add_parser(
'update', help='Update an existing secret.')
upd_sub.add_argument('studio_id', type=str)
upd_sub.add_argument('key', type=str, help='Secret name.')
upd_sub.add_argument('value', type=str, help='New secret value.')
del_sub = secret_subparsers.add_parser(
'delete', help='Delete a secret.')
del_sub.add_argument('studio_id', type=str)
del_sub.add_argument('key', type=str, help='Secret name to delete.')
# ------------------------------------------------------------------
# Execution
# ------------------------------------------------------------------
def execute(self):
action = getattr(self.args, 'studio_action', None)
if action is None:
raise SystemExit(
'No studio subcommand specified. '
"Run 'modelscope studio --help' to see available subcommands.")
endpoint = resolve_endpoint(self.args.endpoint)
api = HubApi(endpoint=endpoint)
handlers = {
'deploy': self._do_deploy,
'stop': self._do_stop,
'logs': self._do_logs,
'settings': self._do_settings,
'secret': self._do_secret,
}
handler = handlers.get(action)
if handler is None:
raise SystemExit(f'Unknown studio subcommand: {action}')
handler(api, endpoint)
# -- deploy / stop --------------------------------------------------
def _do_deploy(self, api, endpoint):
data = api.deploy_studio(
self.args.studio_id, token=self.args.token, endpoint=endpoint)
print(f'Deploy triggered for studio {self.args.studio_id}.')
self._print_status(data)
def _do_stop(self, api, endpoint):
data = api.stop_studio(
self.args.studio_id, token=self.args.token, endpoint=endpoint)
print(f'Stop triggered for studio {self.args.studio_id}.')
self._print_status(data)
@staticmethod
def _print_status(data):
if not data:
return
status = data.get('status') if isinstance(data, dict) else None
if status:
print(f'Status: {status}')
else:
print(json.dumps(data, ensure_ascii=False, indent=2))
# -- logs -----------------------------------------------------------
def _do_logs(self, api, endpoint):
data = api.get_studio_logs(
self.args.studio_id,
log_type=self.args.log_type,
page_num=self.args.page_num,
page_size=self.args.page_size,
keyword=self.args.keyword,
start_timestamp=self.args.start_timestamp,
end_timestamp=self.args.end_timestamp,
token=self.args.token,
endpoint=endpoint)
if not isinstance(data, dict):
print(data)
return
# Server response shape may be {'logs': [...], 'total': N} or similar.
logs = data.get('logs')
if logs is None:
# Fall back to dumping the raw payload so users can still see it.
print(json.dumps(data, ensure_ascii=False, indent=2))
return
for entry in logs:
if isinstance(entry, dict):
ts = entry.get('timestamp') or entry.get('time') or ''
msg = entry.get('content') or entry.get('message') or ''
line = f'[{ts}] {msg}' if ts else str(msg)
else:
line = str(entry)
print(line)
total = data.get('total')
if total is not None:
print(
f'-- page {self.args.page_num} (size {self.args.page_size}), '
f'total {total} --')
# -- settings -------------------------------------------------------
def _do_settings(self, api, endpoint):
fields = [
'display_name', 'description', 'license', 'cover_image',
'sdk_type', 'sdk_version', 'base_image', 'hardware', 'private'
]
settings = {
f: getattr(self.args, f)
for f in fields if getattr(self.args, f, None) is not None
}
if not settings:
raise SystemExit(
'No setting specified. Provide at least one of: '
'--display-name, --description, --license, --cover-image, '
'--sdk-type, --sdk-version, --base-image, --hardware, '
'--private/--public.')
data = api.update_studio_settings(
self.args.studio_id,
token=self.args.token,
endpoint=endpoint,
**settings)
print(f'Updated settings for studio {self.args.studio_id}: '
f"{', '.join(sorted(settings))}.")
if data:
print(json.dumps(data, ensure_ascii=False, indent=2))
# -- secret ---------------------------------------------------------
def _do_secret(self, api, endpoint):
action = getattr(self.args, 'secret_action', None)
if action is None:
raise SystemExit(
'No secret subcommand specified. '
"Run 'modelscope studio secret --help' for usage.")
if action == 'list':
secrets = api.list_studio_secrets(
self.args.studio_id, token=self.args.token, endpoint=endpoint)
if not secrets:
print('(no secrets)')
return
for item in secrets:
key = item.get('key') if isinstance(item, dict) else item
print(key)
elif action == 'add':
api.add_studio_secret(
self.args.studio_id,
self.args.key,
self.args.value,
token=self.args.token,
endpoint=endpoint)
print(f'Secret {self.args.key!r} added.')
elif action == 'update':
api.update_studio_secret(
self.args.studio_id,
self.args.key,
self.args.value,
token=self.args.token,
endpoint=endpoint)
print(f'Secret {self.args.key!r} updated.')
elif action == 'delete':
api.delete_studio_secret(
self.args.studio_id,
self.args.key,
token=self.args.token,
endpoint=endpoint)
print(f'Secret {self.args.key!r} deleted.')
else:
raise SystemExit(f'Unknown secret subcommand: {action}')

306
modelscope/hub/api.py
View File

@@ -100,11 +100,11 @@ from modelscope.utils.constant import (DEFAULT_DATASET_REVISION,
DEFAULT_REPOSITORY_REVISION,
MASTER_MODEL_BRANCH, META_FILES_FORMAT,
REPO_TYPE_DATASET, REPO_TYPE_MODEL,
REPO_TYPE_SUPPORT, ConfigFields,
DatasetFormations, DatasetMetaFormats,
DownloadChannel, DownloadMode,
Frameworks, ModelFile, Tasks,
VirgoDatasetConfig)
REPO_TYPE_STUDIO, REPO_TYPE_SUPPORT,
ConfigFields, DatasetFormations,
DatasetMetaFormats, DownloadChannel,
DownloadMode, Frameworks, ModelFile,
Tasks, VirgoDatasetConfig)
from modelscope.utils.file_utils import (compute_file_hash, get_file_size,
is_relative_path)
from modelscope.utils.logger import get_logger
@@ -782,6 +782,18 @@ class HubApi:
if repo_type == REPO_TYPE_DATASET:
return self.dataset_info(repo_id=repo_id, revision=revision, endpoint=endpoint)
if repo_type == REPO_TYPE_STUDIO:
if (repo_id is None) or repo_id.count('/') != 1:
raise InvalidParameter(
f'Invalid repo_id: {repo_id}, must be of format owner/repo_name')
_endpoint = endpoint or self.endpoint
owner, name = repo_id.split('/', 1)
path = f'{_endpoint}/openapi/v1/studios/{owner}/{name}'
headers = self._build_bearer_headers(token=None, token_required=False)
r = self.session.get(path, headers=headers)
handle_http_response(r, logger, None, repo_id)
return r.json().get('data', {})
raise InvalidParameter(
f'Arg repo_type {repo_type} not supported. Please choose from {REPO_TYPE_SUPPORT}.')
@@ -803,7 +815,7 @@ class HubApi:
by a `/`.
repo_type (`str`, *optional*):
`None` or `"model"` if getting repository info from a model. Default is `None`.
TODO: support studio
Supported values are `"model"`, `"dataset"` and `"studio"`.
endpoint(`str`):
None or specific endpoint to use, when None, use the default endpoint
set in HubApi class (self.endpoint)
@@ -822,13 +834,18 @@ class HubApi:
cookies = self.get_cookies(access_token=token, cookies_required=False)
owner_or_group, name = model_id_to_group_owner_name(repo_id)
if (repo_type is not None) and repo_type.lower() == REPO_TYPE_DATASET:
if (repo_type is not None) and repo_type.lower() == REPO_TYPE_STUDIO:
path = f'{endpoint}/openapi/v1/studios/{owner_or_group}/{name}'
headers = self._build_bearer_headers(token=token, token_required=False)
r = self.session.get(path, headers=headers)
elif (repo_type is not None) and repo_type.lower() == REPO_TYPE_DATASET:
path = f'{endpoint}/api/v1/datasets/{owner_or_group}/{name}'
r = self.session.get(path, cookies=cookies,
headers=self.builder_headers(self.headers))
else:
path = f'{endpoint}/api/v1/models/{owner_or_group}/{name}'
r = self.session.get(path, cookies=cookies,
headers=self.builder_headers(self.headers))
r = self.session.get(path, cookies=cookies,
headers=self.builder_headers(self.headers))
code = handle_http_response(r, logger, cookies, repo_id, False)
if code == 200:
return True
@@ -858,19 +875,12 @@ class HubApi:
A namespace (user or an organization) and a repo name separated
by a `/`.
repo_type (`str`):
The type of the repository. Supported types are `model` and `dataset`.
The type of the repository. Supported types are `model`, `dataset` and `studio`.
endpoint(`str`):
The endpoint to use. If not provided, the default endpoint is `https://www.modelscope.cn`
Could be set to `https://ai.modelscope.ai` for international version.
token (str): Access token of the ModelScope.
"""
warnings.warn(
'This function is deprecated due to security reasons, '
'and will be recovered in future versions with proper token authentication. ',
DeprecationWarning,
stacklevel=2
)
if not endpoint:
endpoint = self.endpoint
@@ -885,6 +895,14 @@ class HubApi:
model_id=repo_id,
endpoint=endpoint,
token=token)
elif repo_type == REPO_TYPE_STUDIO:
logger.warning(
f'Deleting an entire studio repo ({repo_id}) is not supported '
f'via the OpenAPI for security reasons. '
f'To delete studio environment variables, use '
f'HubApi.delete_studio_secret(studio_id, key). '
f'To delete the studio itself, please use the web console.')
return
else:
raise Exception(f'Arg repo_type {repo_type} not supported.')
@@ -2319,11 +2337,263 @@ class HubApi:
)
print(f'New dataset created successfully at {repo_url}.', flush=True)
elif repo_type == REPO_TYPE_STUDIO:
repo_url = self._create_studio_repo(
owner=namespace,
repo_name=repo_name,
visibility=visibility,
license=license,
chinese_name=chinese_name,
token=token,
endpoint=endpoint,
**kwargs,
)
print(f'New studio created successfully at {repo_url}.', flush=True)
else:
raise ValueError(f'Invalid repo type: {repo_type}, supported repos: {REPO_TYPE_SUPPORT}')
return repo_url
# --- Studio Operations ---
@staticmethod
def _parse_studio_id(studio_id: str):
"""Parse a studio_id of the form ``owner/repo_name`` into (owner, name)."""
if not studio_id or studio_id.count('/') != 1:
raise InvalidParameter(
f'Invalid studio_id: {studio_id}, must be of format owner/repo_name')
owner, name = studio_id.split('/', 1)
if not owner or not name:
raise InvalidParameter(
f'Invalid studio_id: {studio_id}, must be of format owner/repo_name')
return owner, name
# Map Licenses display names to SPDX identifiers expected by the
# Studio OpenAPI endpoint.
_LICENSE_TO_SPDX = {
'Apache License 2.0': 'apache-2.0',
'GPL-2.0': 'gpl-2.0',
'GPL-3.0': 'gpl-3.0',
'LGPL-2.1': 'lgpl-2.1',
'LGPL-3.0': 'lgpl-3.0',
'AFL-3.0': 'afl-3.0',
'ECL-2.0': 'ecl-2.0',
'MIT': 'mit',
}
def _create_studio_repo(self,
owner: str,
repo_name: str,
visibility: Optional[str] = Visibility.PUBLIC,
license: Optional[str] = None,
chinese_name: Optional[str] = None,
token: Optional[str] = None,
endpoint: Optional[str] = None,
**kwargs) -> str:
"""Create a studio repo via the OpenAPI ``/openapi/v1/studios`` endpoint.
Supported optional studio fields in ``kwargs``:
description, sdk_type, sdk_version, base_image, hardware, cover_image.
"""
endpoint = endpoint or self.endpoint
path = f'{endpoint}/openapi/v1/studios'
headers = self._build_bearer_headers(token=token, token_required=True)
is_private = visibility is not None and visibility != Visibility.PUBLIC
# Convert license display name to SPDX identifier if needed.
license_spdx = self._LICENSE_TO_SPDX.get(license, license) if license else None
body = {
'repo_name': repo_name,
'owner': owner,
'private': is_private,
'license': license_spdx,
'display_name': chinese_name,
'description': kwargs.get('description'),
'sdk_type': kwargs.get('sdk_type'),
'sdk_version': kwargs.get('sdk_version'),
'base_image': kwargs.get('base_image'),
'hardware': kwargs.get('hardware'),
'cover_image': kwargs.get('cover_image'),
}
body = {k: v for k, v in body.items() if v is not None}
r = self.session.post(path, json=body, headers=headers)
handle_http_response(r, logger, None, f'{owner}/{repo_name}')
return f'{endpoint}/studios/{owner}/{repo_name}'
def deploy_studio(self, studio_id, token=None, endpoint=None):
"""Deploy a studio (re-pull code and rebuild).
Args:
studio_id: Studio ID in format ``owner/repo_name``.
token: Optional access token.
endpoint: Optional API endpoint.
Returns:
dict: Runtime status info including status and active_config.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/deploy'
headers = self._build_bearer_headers(token=token, token_required=True)
r = self.session.post(path, headers=headers)
handle_http_response(r, logger, None, studio_id)
return self._parse_openapi_response(r)
def stop_studio(self, studio_id, token=None, endpoint=None):
"""Stop a running studio.
Args:
studio_id: Studio ID in format ``owner/repo_name``.
token: Optional access token.
endpoint: Optional API endpoint.
Returns:
dict: Runtime status info.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/stop'
headers = self._build_bearer_headers(token=token, token_required=True)
r = self.session.post(path, headers=headers)
handle_http_response(r, logger, None, studio_id)
return self._parse_openapi_response(r)
def get_studio_logs(self, studio_id, log_type='run', page_num=1,
page_size=100, keyword=None, start_timestamp=None,
end_timestamp=None, token=None, endpoint=None):
"""Get studio build or runtime logs.
Args:
studio_id: Studio ID in format ``owner/repo_name``.
log_type: Log type, ``'run'`` or ``'build'``.
page_num: Page number, starting from 1.
page_size: Number of log entries per page.
keyword: Optional keyword filter.
start_timestamp: Optional start timestamp in seconds.
end_timestamp: Optional end timestamp in seconds.
token: Optional access token.
endpoint: Optional API endpoint.
Returns:
dict: Logs data with pagination info.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/logs/{log_type}'
headers = self._build_bearer_headers(token=token, token_required=True)
params = {'page_num': page_num, 'page_size': page_size}
if keyword:
params['keyword'] = keyword
if start_timestamp is not None:
params['start_timestamp'] = start_timestamp
if end_timestamp is not None:
params['end_timestamp'] = end_timestamp
r = self.session.get(path, params=params, headers=headers)
handle_http_response(r, logger, None, studio_id)
return self._parse_openapi_response(r)
def update_studio_settings(self, studio_id, token=None, endpoint=None, **settings):
"""Update studio settings (PATCH, only specified fields are modified).
Args:
studio_id: Studio ID in format ``owner/repo_name``.
token: Optional access token.
endpoint: Optional API endpoint.
**settings: Fields to update. Supported: ``display_name``, ``license``,
``private``, ``description``, ``cover_image``, ``sdk_type``,
``sdk_version``, ``base_image``, ``hardware``. Note:
``sdk_type``/``sdk_version``/``base_image``/``hardware`` changes
require redeployment.
Returns:
dict: Updated studio info.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/settings'
headers = self._build_bearer_headers(token=token, token_required=True)
body = {k: v for k, v in settings.items() if v is not None}
r = self.session.patch(path, json=body, headers=headers)
handle_http_response(r, logger, None, studio_id)
return self._parse_openapi_response(r)
def list_studio_secrets(self, studio_id, token=None, endpoint=None):
"""List studio environment variable keys (values not returned for security).
Args:
studio_id: Studio ID in format ``owner/repo_name``.
token: Optional access token.
endpoint: Optional API endpoint.
Returns:
list: List of secret key dicts, e.g. ``[{'key': 'API_KEY'}, ...]``.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/secrets'
headers = self._build_bearer_headers(token=token, token_required=True)
r = self.session.get(path, headers=headers)
handle_http_response(r, logger, None, studio_id)
data = self._parse_openapi_response(r)
return data.get('secrets', []) if isinstance(data, dict) else []
def add_studio_secret(self, studio_id, key, value, token=None, endpoint=None):
"""Add an environment variable to a studio.
Args:
studio_id: Studio ID in format ``owner/repo_name``.
key: Secret name (max 128 chars).
value: Secret value (max 4096 chars).
token: Optional access token.
endpoint: Optional API endpoint.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/secrets'
headers = self._build_bearer_headers(token=token, token_required=True)
r = self.session.post(
path, json={'key': key, 'value': value}, headers=headers)
handle_http_response(r, logger, None, studio_id)
def update_studio_secret(self, studio_id, key, value, token=None, endpoint=None):
"""Update an existing environment variable in a studio.
Args:
studio_id: Studio ID in format ``owner/repo_name``.
key: Secret name (max 128 chars).
value: New secret value (max 4096 chars).
token: Optional access token.
endpoint: Optional API endpoint.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/secrets'
headers = self._build_bearer_headers(token=token, token_required=True)
r = self.session.put(
path, json={'key': key, 'value': value}, headers=headers)
handle_http_response(r, logger, None, studio_id)
def delete_studio_secret(self, studio_id, key, token=None, endpoint=None):
"""Delete an environment variable from a studio.
Args:
studio_id: Studio ID in format ``owner/repo_name``.
key: Secret name to delete.
token: Optional access token.
endpoint: Optional API endpoint.
"""
endpoint = endpoint or self.endpoint
owner, name = self._parse_studio_id(studio_id)
path = f'{endpoint}/openapi/v1/studios/{owner}/{name}/secrets'
headers = self._build_bearer_headers(token=token, token_required=True)
r = self.session.delete(path, json={'key': key}, headers=headers)
handle_http_response(r, logger, None, studio_id)
# --- End Studio Operations ---
def create_commit(
self,
repo_id: str,

41
modelscope/hub/errors.py
View File

@@ -116,23 +116,48 @@ def handle_http_response(response: requests.Response,
else:
reason = response.reason
request_id = get_request_id(response)
# Try to extract server-side error detail from JSON response body.
server_message = ''
if response.status_code >= 400:
try:
resp_json = response.json()
# OpenAPI envelope: {"success": false, "code": "...", "message": "..."}
msg = resp_json.get('message') or resp_json.get('Message') or ''
code = resp_json.get('code') or ''
if msg:
server_message = f' | Server message: [{code}] {msg}' if code else f' | Server message: {msg}'
except (ValueError, AttributeError):
# Not JSON or unexpected structure; try raw text (truncated)
body_text = response.text[:500] if response.text else ''
if body_text:
server_message = f' | Response body: {body_text}'
if 404 == response.status_code:
http_error_msg = 'The request model: %s does not exist!' % (model_id)
http_error_msg = (
u'404 Not Found: %s does not exist or is not accessible, '
u'Request id: %s for url: %s%s' %
(model_id, request_id, response.url, server_message))
elif 403 == response.status_code:
if cookies is None:
http_error_msg = (
f'Authentication token does not exist, failed to access model {model_id} '
'which may not exist or may be private. Please login first.')
f'Authentication token does not exist, failed to access {model_id} '
f'which may not exist or may be private. Please login first.{server_message}'
)
else:
http_error_msg = f'The authentication token is invalid, failed to access model {model_id}.'
http_error_msg = (
f'The authentication token is invalid, failed to access {model_id}.{server_message}'
)
elif 400 <= response.status_code < 500:
http_error_msg = u'%s Client Error: %s, Request id: %s for url: %s' % (
response.status_code, reason, request_id, response.url)
http_error_msg = u'%s Client Error: %s, Request id: %s for url: %s%s' % (
response.status_code, reason, request_id, response.url,
server_message)
elif 500 <= response.status_code < 600:
http_error_msg = u'%s Server Error: %s, Request id: %s, for url: %s' % (
response.status_code, reason, request_id, response.url)
http_error_msg = u'%s Server Error: %s, Request id: %s, for url: %s%s' % (
response.status_code, reason, request_id, response.url,
server_message)
if http_error_msg and raise_on_error: # there is error.
logger.error(http_error_msg)
raise HTTPError(http_error_msg, response=response)

19
modelscope/hub/snapshot_download.py
View File

@@ -17,7 +17,7 @@ from modelscope.utils.constant import (DEFAULT_DATASET_REVISION,
DEFAULT_MODEL_REVISION,
INTRA_CLOUD_ACCELERATION,
REPO_TYPE_DATASET, REPO_TYPE_MODEL,
REPO_TYPE_SUPPORT)
REPO_TYPE_STUDIO, REPO_TYPE_SUPPORT)
from modelscope.utils.file_utils import get_modelscope_cache_dir
from modelscope.utils.logger import get_logger
from modelscope.utils.thread_utils import thread_executor
@@ -71,7 +71,7 @@ def snapshot_download(
repo_id (str): A user or an organization name and a repo name separated by a `/`.
model_id (str): A user or an organization name and a model name separated by a `/`.
if `repo_id` is provided, `model_id` will be ignored.
repo_type (str, optional): The type of the repo, either 'model' or 'dataset'.
repo_type (str, optional): The type of the repo, one of 'model', 'dataset' or 'studio'.
revision (str, optional): An optional Git revision id which can be a branch name, a tag, or a
commit hash. NOTE: currently only branch and tag name is supported
cache_dir (str, Path, optional): Path to the folder where cached files are stored, model will
@@ -324,16 +324,21 @@ def _snapshot_download(
repo_id=repo_id, repo_type=repo_type, token=token)
if cookies is None:
cookies = _api.get_cookies()
if repo_type == REPO_TYPE_MODEL:
# Studio repos are git-backed and share the model file/listing protocol,
# so they reuse the model code path with a distinct cache subdirectory.
if repo_type in (REPO_TYPE_MODEL, REPO_TYPE_STUDIO):
if local_dir:
directory = os.path.abspath(local_dir)
elif cache_dir:
directory = os.path.join(system_cache, *repo_id.split('/'))
else:
directory = os.path.join(system_cache, 'models',
subdir = 'studios' if repo_type == REPO_TYPE_STUDIO else 'models'
directory = os.path.join(system_cache, subdir,
*repo_id.split('/'))
repo_label = 'Studio' if repo_type == REPO_TYPE_STUDIO else 'Model'
print(
f'Downloading Model from {endpoint} to directory: {directory}')
f'Downloading {repo_label} from {endpoint} to directory: {directory}'
)
revision_detail = _api.get_valid_revision_detail(
repo_id, revision=revision, cookies=cookies, endpoint=endpoint)
revision = revision_detail['Revision']
@@ -1001,7 +1006,9 @@ def _download_file_lists(
@thread_executor(
max_workers=max_workers, disable_tqdm=False, fault_tolerant=True)
def _download_single_file(repo_file):
if repo_type == REPO_TYPE_MODEL:
# Studio shares the model download URL template since both are
# single git-backed repos with the same file-fetch protocol.
if repo_type in (REPO_TYPE_MODEL, REPO_TYPE_STUDIO):
url = get_file_download_url(
model_id=repo_id,
file_path=repo_file['Path'],

37
modelscope/utils/constant.py
View File

@@ -506,7 +506,8 @@ class Frameworks(object):
REPO_TYPE_MODEL = 'model'
REPO_TYPE_DATASET = 'dataset'
REPO_TYPE_SUPPORT = [REPO_TYPE_MODEL, REPO_TYPE_DATASET]
REPO_TYPE_STUDIO = 'studio'
REPO_TYPE_SUPPORT = [REPO_TYPE_MODEL, REPO_TYPE_DATASET, REPO_TYPE_STUDIO]
DEFAULT_MODEL_REVISION = 'master'
MASTER_MODEL_BRANCH = 'master'
DEFAULT_REPOSITORY_REVISION = 'master'
@@ -635,3 +636,37 @@ class MaxComputeEnvs:
PROJECT_NAME = 'ODPS_PROJECT_NAME'
ENDPOINT = 'ODPS_ENDPOINT'
class StudioSDKType(object):
"""Studio SDK type enumeration."""
GRADIO = 'gradio'
STREAMLIT = 'streamlit'
DOCKER = 'docker'
STATIC = 'static'
SUPPORTED = [GRADIO, STREAMLIT, DOCKER, STATIC]
class StudioHardware(object):
"""Studio hardware configuration enumeration."""
CPU_2V_16G = 'platform/2v-cpu-16g-mem'
GPU_16G = 'xgpu/8v-cpu-32g-mem-16g'
GPU_48G = 'xgpu/8v-cpu-64g-mem-48g'
SUPPORTED = [CPU_2V_16G, GPU_16G, GPU_48G]
DEFAULT = CPU_2V_16G
class StudioStatus(object):
"""Studio runtime status enumeration."""
INITIALIZED = 'Initialized'
BUILDING = 'Building'
BUILD_FAILED = 'BuildFailed'
DEPLOYING = 'Deploying'
DEPLOY_FAILED = 'DeployFailed'
RUNNING = 'Running'
STOPPING = 'Stopping'
STOPPED = 'Stopped'
DUPLICATING = 'Duplicating'
SLEEPING = 'Sleeping'

7
tests/run_analysis.py
View File

@@ -94,8 +94,12 @@ def get_models_info(groups: list) -> dict:
def gather_test_suites_files(test_dir='./tests',
pattern='test_*.py',
is_full_path=True):
# Directories excluded from CI (manual-only test suites)
_CI_EXCLUDED_DIRS = {'studios'}
case_file_list = []
for dirpath, dirnames, filenames in os.walk(test_dir):
# Skip excluded directories
dirnames[:] = [d for d in dirnames if d not in _CI_EXCLUDED_DIRS]
for file in filenames:
if fnmatch(file, pattern):
if is_full_path:
@@ -183,7 +187,8 @@ def analysis_diff():
get_file_register_modules(modified_file))
elif ((modified_file.startswith('./tests')
or modified_file.startswith('tests'))
and os.path.basename(modified_file).startswith('test_')):
and os.path.basename(modified_file).startswith('test_')
and '/studios/' not in modified_file):
modified_cases.append(modified_file)
return modified_register_modules, modified_cases

31
tests/studios/.env.example Normal file
View File

@@ -0,0 +1,31 @@
# ModelScope Studio CLI Tests Configuration
# Copy this file to .env and fill in your values.
# All variables are optional for the mock-based CLI tests; defaults are
# applied when missing.
# Access token used as a fallback placeholder when assembling mock requests.
MODELSCOPE_API_TOKEN=your_access_token_here
# Owner used to compose the mock studio_id (owner/name) in CLI tests.
TEST_STUDIO_OWNER=your_username
# Studio visibility: private (default) or public.
TEST_STUDIO_VISIBILITY=private
# Studio ID for real API call tests (format: owner/name).
TEST_STUDIO_ID=your_owner/your_studio_name
# Whether to delete temporary remote resources (studios, secrets) after tests.
# Set to false to keep them for debugging. Default: true.
TEST_CLEANUP_REMOTE=true
# Studio creation parameters (required by OpenAPI).
# These have sensible defaults; override only if needed.
# TEST_STUDIO_LICENSE=apache-2.0
# TEST_STUDIO_SDK_TYPE=gradio
# TEST_STUDIO_SDK_VERSION=6.2.0
# TEST_STUDIO_BASE_IMAGE=ubuntu22.04-py311-torch2.9.1-modelscope1.35.0
# TEST_STUDIO_HARDWARE=platform/2v-cpu-16g-mem
# API endpoint (optional, defaults to https://modelscope.cn).
# MODELSCOPE_ENDPOINT=https://modelscope.cn

0
tests/studios/__init__.py Normal file
View File

133
tests/studios/conftest_env.py Normal file
View File

@@ -0,0 +1,133 @@
# Copyright (c) Alibaba, Inc. and its affiliates.
"""Helper module for loading Studio test environment from a local .env file.
The ``.env`` file (sibling of this module) is optional and not committed to
version control. ``.env.example`` documents the supported variables.
"""
import os
def load_test_env():
"""Load test environment variables from a local ``.env`` file if present.
Existing environment variables take precedence (``setdefault`` semantics)
so that CI overrides remain authoritative.
"""
env_file = os.path.join(os.path.dirname(__file__), '.env')
if not os.path.exists(env_file):
return
with open(env_file, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#') or '=' not in line:
continue
key, value = line.split('=', 1)
os.environ.setdefault(key.strip(), value.strip())
def get_test_config():
"""Return the merged Studio test configuration dictionary."""
load_test_env()
cleanup_raw = os.environ.get('TEST_CLEANUP_REMOTE', 'true').strip().lower()
return {
'token':
os.environ.get('MODELSCOPE_API_TOKEN'),
'owner':
os.environ.get('TEST_STUDIO_OWNER', 'test_user'),
'visibility':
os.environ.get('TEST_STUDIO_VISIBILITY', 'private'),
'endpoint':
os.environ.get('MODELSCOPE_ENDPOINT', 'https://modelscope.cn'),
'studio_id':
os.environ.get('TEST_STUDIO_ID'),
'cleanup':
cleanup_raw not in ('false', '0', 'no'),
# Studio creation defaults (required by OpenAPI)
'license':
os.environ.get('TEST_STUDIO_LICENSE', 'apache-2.0'),
'sdk_type':
os.environ.get('TEST_STUDIO_SDK_TYPE', 'gradio'),
'sdk_version':
os.environ.get('TEST_STUDIO_SDK_VERSION', '6.2.0'),
'base_image':
os.environ.get('TEST_STUDIO_BASE_IMAGE',
'ubuntu22.04-py311-torch2.9.1-modelscope1.35.0'),
'hardware':
os.environ.get('TEST_STUDIO_HARDWARE', 'platform/2v-cpu-16g-mem'),
}
def create_temp_studio(config, name_prefix='ut_test_cli'):
"""Create a temporary studio for testing purposes.
Args:
config: dict from get_test_config()
name_prefix: prefix for the generated studio name
Returns:
str: The studio_id (owner/name) of the created studio.
Raises:
unittest.SkipTest: If token or owner are not configured.
"""
import unittest
from uuid import uuid4
from modelscope.hub.api import HubApi
token = config['token']
owner = config['owner']
if not token or not owner:
raise unittest.SkipTest(
'MODELSCOPE_API_TOKEN and TEST_STUDIO_OWNER required')
name = f'{name_prefix}_{uuid4().hex[:8]}'
repo_id = f'{owner}/{name}'
visibility = config.get('visibility', 'private')
api = HubApi()
api.create_repo(
repo_id,
repo_type='studio',
visibility=visibility,
license=config.get('license', 'apache-2.0'),
token=token,
endpoint=config.get('endpoint'),
sdk_type=config.get('sdk_type', 'gradio'),
sdk_version=config.get('sdk_version', '6.2.0'),
base_image=config.get('base_image',
'ubuntu22.04-py311-torch2.9.1-modelscope1.35.0'),
hardware=config.get('hardware', 'platform/2v-cpu-16g-mem'),
create_default_config=False,
)
return repo_id
class TestResultMixin:
"""Mixin that prints test pass/fail status after each test method.
Must be placed BEFORE ``unittest.TestCase`` in the MRO so that this
``tearDown`` runs and still chains to ``TestCase.tearDown`` via ``super``.
"""
def tearDown(self):
try:
super().tearDown()
finally:
status = 'PASSED'
outcome = getattr(self, '_outcome', None)
if outcome is not None:
# Python <= 3.10 exposes ``result`` directly.
result = getattr(outcome, 'result', None)
if result is not None and hasattr(result, 'failures'):
test_failed = any(test is self
for test, _ in (result.failures
+ result.errors))
status = 'FAILED' if test_failed else 'PASSED'
elif hasattr(outcome, 'errors'):
# Python 3.11+: inspect ``errors`` list of (ctx, exc_info).
errors = getattr(outcome, 'errors', [])
test_failed = any(exc_info is not None
for _, exc_info in errors)
status = 'FAILED' if test_failed else 'PASSED'
# else: running under pytest or other runner — cannot detect
print(f'[TEST {status}] {self.id()}')

476
tests/studios/test_studio_cli.py Normal file
View File

@@ -0,0 +1,476 @@
# Copyright (c) Alibaba, Inc. and its affiliates.
"""Tests for the ``modelscope studio`` CLI surface.
Two layers of coverage:
1. ``--help`` smoke tests via ``subprocess`` — they verify argument plumbing
without touching the network.
2. Direct calls into :class:`~modelscope.cli.studio.StudioCMD` with
:class:`~argparse.Namespace` instances and mocked :class:`HubApi` methods,
which exercise the dispatch logic that ``subprocess`` cannot mock.
"""
import argparse
import os
import shutil
import subprocess
import sys
import time
import unittest
from unittest.mock import ANY, patch
from uuid import uuid4
from tests.studios.conftest_env import (TestResultMixin, create_temp_studio,
get_test_config)
from modelscope.cli.studio import StudioCMD
from modelscope.hub.api import HubApi
def _cli_invocation():
"""Return a shell-friendly invocation prefix for the modelscope CLI.
Prefer the installed ``modelscope`` console script when available;
fall back to ``python -c '...run_cmd()...'`` so the tests still run in
editable installs without the entry point shim.
"""
exe = shutil.which('modelscope')
if exe:
return exe
py = sys.executable
return (f"{py} -c 'from modelscope.cli.cli import run_cmd; run_cmd()'")
CLI_PREFIX = _cli_invocation()
class TestStudioCLIHelp(TestResultMixin, unittest.TestCase):
"""Smoke-test the help output of every studio subcommand."""
def _run_help(self, *cli_args):
cmd = ' '.join([CLI_PREFIX, *cli_args, '--help'])
stat, output = subprocess.getstatusoutput(cmd)
return stat, output
def test_studio_help(self):
stat, output = self._run_help('studio')
self.assertEqual(stat, 0, output)
for sub in ('deploy', 'stop', 'logs', 'settings', 'secret'):
self.assertIn(sub, output)
def test_studio_deploy_help(self):
stat, output = self._run_help('studio', 'deploy')
self.assertEqual(stat, 0, output)
self.assertIn('studio_id', output)
def test_studio_stop_help(self):
stat, output = self._run_help('studio', 'stop')
self.assertEqual(stat, 0, output)
self.assertIn('studio_id', output)
def test_studio_logs_help(self):
stat, output = self._run_help('studio', 'logs')
self.assertEqual(stat, 0, output)
self.assertIn('--type', output)
self.assertIn('--keyword', output)
self.assertIn('--page-num', output)
self.assertIn('--page-size', output)
def test_studio_settings_help(self):
stat, output = self._run_help('studio', 'settings')
self.assertEqual(stat, 0, output)
for flag in ('--sdk-type', '--hardware', '--private', '--public',
'--display-name'):
self.assertIn(flag, output)
def test_studio_secret_help(self):
stat, output = self._run_help('studio', 'secret')
self.assertEqual(stat, 0, output)
for sub in ('list', 'add', 'update', 'delete'):
self.assertIn(sub, output)
def test_download_repo_type_includes_studio(self):
stat, output = self._run_help('download')
self.assertEqual(stat, 0, output)
self.assertIn('studio', output)
def test_create_includes_studio_args(self):
stat, output = self._run_help('create')
self.assertEqual(stat, 0, output)
self.assertIn('--sdk-type', output)
self.assertIn('--hardware', output)
self.assertIn('studio', output)
class TestStudioCLIErrors(TestResultMixin, unittest.TestCase):
"""Validate user-facing error paths that don't need the network."""
def setUp(self):
config = get_test_config()
self.owner = config['owner']
# Mock tests use a fixed repo name to keep assertions deterministic.
self.name = 'mock-studio'
self.studio_id = f'{self.owner}/{self.name}'
# Token from .env (MODELSCOPE_API_TOKEN); fallback for pure-mock runs.
self.token = config['token'] or 'test-token-placeholder'
def test_studio_deploy_missing_id(self):
cmd = f'{CLI_PREFIX} studio deploy'
stat, output = subprocess.getstatusoutput(cmd)
self.assertNotEqual(stat, 0)
def test_studio_no_subcommand_raises(self):
args = argparse.Namespace(
studio_action=None, token=None, endpoint=None)
cmd = StudioCMD(args)
with self.assertRaises(SystemExit):
cmd.execute()
def test_studio_settings_no_field_raises(self):
args = argparse.Namespace(
studio_action='settings',
studio_id=self.studio_id,
token=self.token,
endpoint=None,
display_name=None,
description=None,
license=None,
cover_image=None,
sdk_type=None,
sdk_version=None,
base_image=None,
hardware=None,
private=None,
)
with patch.object(
HubApi,
'_build_bearer_headers',
return_value={'Authorization': f'Bearer {self.token}'}):
cmd = StudioCMD(args)
with self.assertRaises(SystemExit):
cmd.execute()
def test_secret_no_action_raises(self):
args = argparse.Namespace(
studio_action='secret',
secret_action=None,
token=None,
endpoint=None,
)
with patch.object(
HubApi,
'_build_bearer_headers',
return_value={'Authorization': f'Bearer {self.token}'}):
cmd = StudioCMD(args)
with self.assertRaises(SystemExit):
cmd.execute()
class TestStudioCreate(TestResultMixin, unittest.TestCase):
"""Test studio creation via HubApi.create_repo(repo_type='studio')."""
@classmethod
def setUpClass(cls):
config = get_test_config()
cls.token = config['token']
cls.owner = config['owner']
cls.config = config
cls._cleanup = config.get('cleanup', True)
if not cls.token or not cls.owner:
raise unittest.SkipTest(
'MODELSCOPE_API_TOKEN and TEST_STUDIO_OWNER required')
cls._created_studios = []
@classmethod
def tearDownClass(cls):
"""Studio deletion is not supported via OpenAPI. Log created studios for manual cleanup."""
if cls._created_studios:
import logging
logging.getLogger('modelscope').info(
f'Test studios created (manual cleanup needed): '
f'{", ".join(cls._created_studios)}')
def _create_and_track(self, **kwargs):
"""Create a studio with unique name, track for cleanup."""
name = f'ut_test_create_{uuid4().hex[:8]}'
repo_id = f'{self.owner}/{name}'
api = HubApi()
# Merge studio creation defaults from config with caller overrides
create_kwargs = {
'sdk_type':
self.config.get('sdk_type', 'gradio'),
'sdk_version':
self.config.get('sdk_version', '6.2.0'),
'base_image':
self.config.get('base_image',
'ubuntu22.04-py311-torch2.9.1-modelscope1.35.0'),
'hardware':
self.config.get('hardware', 'platform/2v-cpu-16g-mem'),
}
create_kwargs.update(kwargs)
print(f'Creating studio {repo_id} with config: {self.config}')
url = api.create_repo(
repo_id,
repo_type='studio',
visibility=self.config.get('visibility', 'private'),
license=self.config.get('license', 'apache-2.0'),
token=self.token,
endpoint=self.config.get('endpoint'),
create_default_config=False,
**create_kwargs,
)
self._created_studios.append(repo_id)
print(f'Created studio {repo_id} at {url}')
return repo_id, url
def test_create_studio_basic(self):
"""Create a basic studio and verify it exists."""
repo_id, url = self._create_and_track(sdk_type='gradio')
self.assertIn(repo_id, url)
# Verify it actually exists
api = HubApi()
exists = api.repo_exists(repo_id, repo_type='studio', token=self.token)
self.assertTrue(exists,
f'Studio {repo_id} should exist after creation')
def test_create_studio_exist_ok(self):
"""Creating an existing studio with exist_ok=True should not raise."""
repo_id, _ = self._create_and_track(sdk_type='gradio')
api = HubApi()
# Create again with exist_ok=True
url = api.create_repo(
repo_id,
repo_type='studio',
visibility=self.config.get('visibility', 'private'),
token=self.token,
endpoint=self.config.get('endpoint'),
exist_ok=True,
create_default_config=False,
)
self.assertIn(repo_id, url)
def test_create_studio_exist_raises(self):
"""Creating an existing studio without exist_ok should raise ValueError."""
repo_id, _ = self._create_and_track(sdk_type='gradio')
api = HubApi()
with self.assertRaises(ValueError):
api.create_repo(
repo_id,
repo_type='studio',
visibility=self.config.get('visibility', 'private'),
token=self.token,
endpoint=self.config.get('endpoint'),
exist_ok=False,
create_default_config=False,
)
class TestStudioCLIDirectCall(TestResultMixin, unittest.TestCase):
"""Drive ``StudioCMD.execute`` with real API calls (no HubApi mocks)."""
@classmethod
def setUpClass(cls):
config = get_test_config()
cls.token = config['token']
cls.studio_id = config['studio_id']
cls._auto_created_studio = False
cls._cleanup = config.get('cleanup', True)
if not cls.token:
raise unittest.SkipTest(
'MODELSCOPE_API_TOKEN required for real API tests')
# If no TEST_STUDIO_ID configured, create a temporary one
if not cls.studio_id:
cls.studio_id = create_temp_studio(config)
cls._auto_created_studio = True
@classmethod
def tearDownClass(cls):
"""Studio deletion is not supported via OpenAPI. Log for manual cleanup."""
if cls._auto_created_studio and cls.studio_id:
import logging
logging.getLogger('modelscope').info(
f'Auto-created test studio (manual cleanup needed): '
f'{cls.studio_id}')
def setUp(self):
# Track secret keys created during a test for cleanup.
self._secrets_to_cleanup = []
def tearDown(self):
# Best-effort cleanup of any secrets created during the test.
if not self._cleanup:
return
api = HubApi()
for key in self._secrets_to_cleanup:
try:
api.delete_studio_secret(
self.studio_id, key, token=self.token, endpoint=None)
except Exception:
pass
def _unique_secret_key(self):
"""Generate a unique secret key with test prefix."""
key = f'_TEST_CLI_SECRET_{uuid4().hex[:8].upper()}'
self._secrets_to_cleanup.append(key)
return key
def test_deploy_direct(self):
args = argparse.Namespace(
studio_action='deploy',
studio_id=self.studio_id,
token=self.token,
endpoint=None,
)
with patch('builtins.print') as mock_print:
StudioCMD(args).execute()
printed = ' '.join(
str(c.args[0]) for c in mock_print.call_args_list if c.args)
self.assertIn('Deploy triggered', printed)
def test_stop_direct(self):
args = argparse.Namespace(
studio_action='stop',
studio_id=self.studio_id,
token=self.token,
endpoint=None,
)
with patch('builtins.print') as mock_print:
StudioCMD(args).execute()
printed = ' '.join(
str(c.args[0]) for c in mock_print.call_args_list if c.args)
self.assertIn('Stop triggered', printed)
def test_logs_direct(self):
args = argparse.Namespace(
studio_action='logs',
studio_id=self.studio_id,
token=self.token,
endpoint=None,
log_type='run',
keyword=None,
page_num=1,
page_size=100,
start_timestamp=None,
end_timestamp=None,
)
# Studio may not be deployed yet, so 404 (no logs) is acceptable.
from requests.exceptions import HTTPError
with patch('builtins.print'):
try:
StudioCMD(args).execute()
except HTTPError as e:
# 404 is expected for a non-deployed studio (no logs available)
if e.response is not None and e.response.status_code == 404:
pass
else:
raise
def test_settings_direct(self):
display_name = f'CLI Test {int(time.time())}'
args = argparse.Namespace(
studio_action='settings',
studio_id=self.studio_id,
token=self.token,
endpoint=None,
display_name=display_name,
description=None,
license=None,
cover_image=None,
sdk_type=None,
sdk_version=None,
base_image=None,
hardware=None,
private=None,
)
with patch('builtins.print') as mock_print:
StudioCMD(args).execute()
printed = ' '.join(
str(c.args[0]) for c in mock_print.call_args_list if c.args)
self.assertIn('Updated settings', printed)
def test_secret_list_direct(self):
args = argparse.Namespace(
studio_action='secret',
secret_action='list',
studio_id=self.studio_id,
token=self.token,
endpoint=None,
)
# Should not raise regardless of whether secrets exist.
with patch('builtins.print'):
StudioCMD(args).execute()
def test_secret_add_direct(self):
key = self._unique_secret_key()
args = argparse.Namespace(
studio_action='secret',
secret_action='add',
studio_id=self.studio_id,
key=key,
value='test_value',
token=self.token,
endpoint=None,
)
with patch('builtins.print') as mock_print:
StudioCMD(args).execute()
printed = ' '.join(
str(c.args[0]) for c in mock_print.call_args_list if c.args)
self.assertIn('added', printed.lower())
def test_secret_update_direct(self):
key = self._unique_secret_key()
# Pre-add the secret so we can update it.
api = HubApi()
api.add_studio_secret(
self.studio_id,
key,
'initial_value',
token=self.token,
endpoint=None)
args = argparse.Namespace(
studio_action='secret',
secret_action='update',
studio_id=self.studio_id,
key=key,
value='updated_value',
token=self.token,
endpoint=None,
)
with patch('builtins.print') as mock_print:
StudioCMD(args).execute()
printed = ' '.join(
str(c.args[0]) for c in mock_print.call_args_list if c.args)
self.assertIn('updated', printed.lower())
def test_secret_delete_direct(self):
key = self._unique_secret_key()
# Pre-add the secret so we can delete it.
api = HubApi()
api.add_studio_secret(
self.studio_id,
key,
'to_be_deleted',
token=self.token,
endpoint=None)
# Remove from cleanup list since we expect CLI to delete it.
self._secrets_to_cleanup.remove(key)
args = argparse.Namespace(
studio_action='secret',
secret_action='delete',
studio_id=self.studio_id,
key=key,
token=self.token,
endpoint=None,
)
with patch('builtins.print') as mock_print:
StudioCMD(args).execute()
printed = ' '.join(
str(c.args[0]) for c in mock_print.call_args_list if c.args)
self.assertIn('deleted', printed.lower())
if __name__ == '__main__':
unittest.main()