# Copyright 2022-2025 The Ramble Authors
#
# Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
# https://www.apache.org/licenses/LICENSE-2.0> or the MIT license
# <LICENSE-MIT or https://opensource.org/licenses/MIT>, at your
# option. This file may not be copied, modified, or distributed
# except according to those terms.
import ramble.cmd.common.arguments as arguments
import ramble.config
import ramble.mirror
import ramble.repository
import ramble.spec
import ramble.workspace
from ramble.error import RambleError
from ramble.util.logger import logger
import spack.util.url as url_util
import spack.util.web as web_util
from spack.util.spack_yaml import syaml_dict
description = "manage mirrors (inputs)"
section = "config"
level = "long"
[docs]
def setup_parser(subparser):
arguments.add_common_arguments(subparser, ["no_checksum"])
sp = subparser.add_subparsers(metavar="SUBCOMMAND", dest="mirror_command")
# Destroy
destroy_parser = sp.add_parser(
"destroy", help=mirror_destroy.__doc__, description=mirror_destroy.__doc__
)
destroy_target = destroy_parser.add_mutually_exclusive_group(required=True)
destroy_target.add_argument(
"-m",
"--mirror-name",
metavar="mirror_name",
type=str,
help="find mirror to destroy by name",
)
destroy_target.add_argument(
"-u", "--mirror-url", metavar="mirror_url", type=str, help="find mirror to destroy by url"
)
# used to construct scope arguments below
scopes = ramble.config.scopes()
scopes_metavar = ramble.config.scopes_metavar
# Add
add_parser = sp.add_parser("add", help=mirror_add.__doc__, description=mirror_add.__doc__)
add_parser.add_argument("name", help="mnemonic name for mirror", metavar="mirror")
add_parser.add_argument("url", help="url of mirror directory from 'ramble mirror create'")
add_parser.add_argument(
"--scope",
choices=scopes,
metavar=scopes_metavar,
default=ramble.config.default_modify_scope(),
help="configuration scope to modify",
)
# Remove
remove_parser = sp.add_parser(
"remove", aliases=["rm"], help=mirror_remove.__doc__, description=mirror_remove.__doc__
)
remove_parser.add_argument("name", help="mnemonic name for mirror", metavar="mirror")
remove_parser.add_argument(
"--scope",
choices=scopes,
metavar=scopes_metavar,
default=ramble.config.default_modify_scope(),
help="configuration scope to modify",
)
# Set-Url
set_url_parser = sp.add_parser(
"set-url", help=mirror_set_url.__doc__, description=mirror_set_url.__doc__
)
set_url_parser.add_argument("name", help="mnemonic name for mirror", metavar="mirror")
set_url_parser.add_argument("url", help="url of mirror directory from 'ramble mirror create'")
set_url_parser.add_argument(
"--push", action="store_true", help="set only the URL used for uploading new resources"
)
set_url_parser.add_argument(
"--scope",
choices=scopes,
metavar=scopes_metavar,
default=ramble.config.default_modify_scope(),
help="configuration scope to modify",
)
# List
list_parser = sp.add_parser("list", help=mirror_list.__doc__, description=mirror_list.__doc__)
list_parser.add_argument(
"--scope",
choices=scopes,
metavar=scopes_metavar,
default=ramble.config.default_list_scope(),
help="configuration scope to read from",
)
[docs]
def mirror_add(args):
"""Add a mirror to Ramble."""
url = url_util.format(args.url)
ramble.mirror.add(args.name, url, args.scope)
[docs]
def mirror_remove(args):
"""Remove a mirror by name."""
ramble.mirror.remove(args.name, args.scope)
[docs]
def mirror_set_url(args):
"""Change the URL of a mirror."""
url = url_util.format(args.url)
mirrors = ramble.config.get("mirrors", scope=args.scope)
if not mirrors:
mirrors = syaml_dict()
if args.name not in mirrors:
logger.die(f"No mirror found with name {args.name}.")
entry = mirrors[args.name]
try:
fetch_url = entry["fetch"]
push_url = entry["push"]
except TypeError:
fetch_url, push_url = entry, entry
changes_made = False
if args.push:
changes_made = changes_made or push_url != url
push_url = url
else:
changes_made = changes_made or push_url != url
fetch_url, push_url = url, url
items = [
(
(mirror_name, mirror_url)
if mirror_name != args.name
else (
(mirror_name, {"fetch": fetch_url, "push": push_url})
if fetch_url != push_url
else (mirror_name, {"fetch": fetch_url, "push": fetch_url})
)
)
for mirror_name, mirror_url in mirrors.items()
]
mirrors = syaml_dict(items)
ramble.config.set("mirrors", mirrors, scope=args.scope)
if changes_made:
logger.msg(
"Changed%s url or connection information for mirror %s."
% ((" (push)" if args.push else ""), args.name)
)
else:
logger.msg(f"No changes made to mirror {args.name}.")
[docs]
def mirror_list(args):
"""Print out available mirrors to the console."""
mirrors = ramble.mirror.MirrorCollection(scope=args.scope)
if not mirrors:
logger.msg("No mirrors configured.")
return
mirrors.display()
def _read_specs_from_file(filename):
specs = []
with open(filename) as stream:
for i, string in enumerate(stream):
try:
s = ramble.Spec(string)
s.application
specs.append(s)
except RambleError as e:
logger.debug(e)
logger.die(
"Parse error in %s, line %d:" % (filename, i + 1), ">>> " + string, str(e)
)
return specs
[docs]
def mirror_destroy(args):
"""Given a url, recursively delete everything under it."""
mirror_url = None
if args.mirror_name:
result = ramble.mirror.MirrorCollection().lookup(args.mirror_name)
mirror_url = result.push_url
elif args.mirror_url:
mirror_url = args.mirror_url
web_util.remove_url(mirror_url, recursive=True)
[docs]
def mirror(parser, args):
action = {
"destroy": mirror_destroy,
"add": mirror_add,
"remove": mirror_remove,
"rm": mirror_remove,
"set-url": mirror_set_url,
"list": mirror_list,
}
if args.no_checksum:
ramble.config.set("config:checksum", False, scope="command_line")
action[args.mirror_command](args)