Compare commits

...
31 Commits
Author SHA1 Message Date
Ruslan Kuprieiev f0c1ff0219 fs: make _prepare_credentails/fs_args lazy (#9848)
* tests: unit: move s3 tests to dvc-s3

* deps: bump dvc-data to 2.14.0

* tests: unit: move azure tests to dvc-azure
2023-08-16 05:51:20 +03:00
Ruslan Kuprieiev 27c548575c pyinstaller: declare dvc_ssh as hidden import (#9840)
Likely with removal of dvc-machine pyinstaller stopped picking up dvc_ssh automatically, so we need to declare it explicitly just like the rest of plugins.
2023-08-13 00:26:43 +03:00
Dave Berenbaum 79cc28501b don't unprotect outs if not cached (#9838)
* don't unprotect outs if not cached

* fix tests
2023-08-11 18:12:25 +03:00
Peter Rowlands a051b8b60f remove support for dvc machine 2023-08-11 14:42:52 +09:00
Ruslan Kuprieiev bee04c6d5f deps: bump dvc-data to 2.13.1
Fixes #9733
2023-08-11 05:53:19 +03:00
Tibor MachandTibor Mach 04e891cef9 exp: Unify dvc exp list and show output (#9808)
* use _describe in exp list

* fix pre-commit checks

* simplistic fix

* single _describe, too complex

* slightly simpler

* even simpler

* simple, but with regex

* less substitutions

* describe public

* name handling in UI (tests need updating)

* fixed old tests

* lazy imports

---------

Co-authored-by: Tibor Mach <[email protected]>
2023-08-10 19:23:27 +09:00
Peter Rowlands f5dda2d3cc deps: bump scmrepo minver to 1.2.1 2023-08-10 19:21:06 +09:00
Ruslan Kuprieiev e9545c8b55 fetch: merge config instead of updating
Update will replace sections instead of merging them together, which
is not desirable here.

Related https://github.com/iterative/studio/pull/6541
2023-08-10 04:52:08 +03:00
Ruslan Kuprieiev 09cfd911bb config: use local config from local_dvc_dir
This used to work like so but by accident, when we didn't reload config
at all when collecting other revisions.

Regression from https://github.com/iterative/dvc/pull/9758
2023-08-10 04:36:42 +03:00
Ruslan Kuprieiev 4f61a63afd config: resolve: handle ~ home dir
Fixes #9822
2023-08-10 03:28:32 +03:00
Ruslan Kuprieiev c71b07b996 index: plot_keys: relparts -> parts
`_plot_sources` is messed up and needs a rewrite hard, but this will
do for now for studio purposes.

Related https://github.com/iterative/studio/pull/6541
2023-08-08 17:14:28 +03:00
Ruslan Kuprieiev f4c0946f65 repo: cache: support some view options 2023-08-08 01:09:50 +03:00
Ruslan Kuprieiev e35fb76dac repo: add basic cache_check_missing 2023-08-07 22:23:49 +03:00
Ruslan Kuprieiev 722839dda6 fetch: add basic onerror
Related https://github.com/iterative/studio/pull/6541
2023-08-07 03:24:39 +03:00
Ruslan Kuprieiev d5533348f7 deps: bump dvc-data to 2.12.0 2023-08-07 02:42:41 +03:00
dberenbaum 5d2ca11557 exp: abstract random name gen from repo 2023-08-06 13:13:11 +02:00
009b238807 Feature/hydra config modules (#9783)
* hydra: enable config modules

Lets the user specify Hydra configs in installed Python modules as well
as inside directories. Adds `hydra.config_module` to `.dvc/config`.

Fixes #9740

old tests green

tests for config modules

* hydra: fix linter errors

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* hydra: fix mypy error

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* hydra: fix tests

* Update dvc/utils/hydra.py

Co-authored-by: David de la Iglesia Castro <[email protected]>

* address comments

* convince mypy not to worry

* split long line

* fix test

* check for raised error in test

---------

Co-authored-by: Dom Miketa <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: David de la Iglesia Castro <[email protected]>
2023-08-06 10:39:09 +02:00
Ruslan Kuprieiev 43b371c43a deps: bump dvc-data to 2.11.0 2023-08-04 15:42:25 +03:00
skshetry d1160a28e0 dvc.yaml: add matrix-do (#9725)
* dvc.yaml: add matrix-do

* remove docs

* no do in matrix

* add types
2023-08-04 13:29:15 +05:45
skshetry bc86fa0643 completion: fix broken completion generation for zsh (#9797)
`dvc commit --help` was also completely broken, which is also fixed.

This adds unit tests for completion generation for both zsh/bash, and
tests that --help works for all commands.
2023-08-04 09:21:01 +05:45
Saugat Pachhai (सौगात) 4067a86926 fix flaky test for repro --keep-going and --ignore 2023-08-04 05:46:58 +03:00
Ruslan Kuprieiev 9d1d51caa7 deps: bump dvc-data to 2.10.0 2023-08-04 03:40:36 +03:00
Peter Rowlands 761422638f output: add progress for commit 2023-08-03 18:19:19 +09:00
Peter Rowlands d01e2355d4 push/pull: generate explicit progress bars for dvc-data transfers 2023-08-03 18:19:19 +09:00
Peter Rowlands aa94bfd66b deps: bump dvc-data to 2.9.1 2023-08-03 18:19:19 +09:00
skshetry b9b9f91084 ruff: enable autofix (#9790) 2023-08-03 11:08:29 +05:45
Ruslan Kuprieiev 8a382f51fd deps: bump dvc-data to 2.9.0 2023-08-03 02:05:14 +03:00
Ruslan Kuprieiev b235487ecc ls/import/get: introduce --remote and --remote-config
Fixes #9656
2023-08-02 16:28:16 +03:00
pre-commit-ci[bot]andSaugat Pachhai 502b3df6c9 [pre-commit.ci] pre-commit autoupdate (#9756)
* [pre-commit.ci] pre-commit autoupdate

updates:
- [github.com/asottile/pyupgrade: v3.9.0 → v3.10.1](https://github.com/asottile/pyupgrade/compare/v3.9.0...v3.10.1)
- [github.com/astral-sh/ruff-pre-commit: v0.0.278 → v0.0.281](https://github.com/astral-sh/ruff-pre-commit/compare/v0.0.278...v0.0.281)
- [github.com/PyCQA/flake8: 6.0.0 → 6.1.0](https://github.com/PyCQA/flake8/compare/6.0.0...6.1.0)

* fix issues reported by ruff

- also remove redundant flake8 plugins
- remove beautysh as we don't have any scripts now
- remove pyupgrade as it is redundant due to ruff

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Saugat Pachhai (सौगात) <[email protected]>
2023-08-02 18:08:30 +05:45
dependabot[bot] 5c2fd67e6d build(deps-dev): bump pylint from 2.17.4 to 2.17.5 (#9769)
Bumps [pylint](https://github.com/pylint-dev/pylint) from 2.17.4 to 2.17.5.
- [Release notes](https://github.com/pylint-dev/pylint/releases)
- [Commits](https://github.com/pylint-dev/pylint/compare/v2.17.4...v2.17.5)

---
updated-dependencies:
- dependency-name: pylint
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2023-07-31 16:35:40 +05:45
Fabian Zills ed34c0f0bd dvc commit update description (#9760)
* Update commit.py

* fix formatting
2023-07-31 08:45:52 +00:00
90 changed files with 1096 additions and 2324 deletions
+8 -7
View File
@@ -1,19 +1,20 @@
[flake8]
ignore=
max_line_length = 88
extend-ignore=
# Whitespace before ':'
E203
# Too many leading '#' for block comment
E266
# Line break occurred before a binary operator
W503
# unindexed parameters in the str.format, see:
# https://pypi.org/project/flake8-string-format/
P1
# use `!r` instead of surrounding by quotes
B028
max_line_length = 88
max-complexity = 10
select = B,C,E,F,W,T4,B902,T,P,TYP
show_source = true
count = true
min_python_version = 3.8.0
per-file-ignores =
dvc/testing/benchmarks/conftest.py:F401
dvc/testing/conftest.py:F401
dvc/testing/plugin.py:F401
tests/conftest.py:F401
tests/func/experiments/executor/test_ssh.py:F401
+3 -16
View File
@@ -23,17 +23,6 @@ repos:
- id: mixed-line-ending
- id: sort-simple-yaml
- id: trailing-whitespace
- repo: https://github.com/lovesegfault/beautysh
rev: v6.2.1
hooks:
- id: beautysh
args: ["-i", "2"]
- repo: https://github.com/asottile/pyupgrade
rev: v3.9.0
hooks:
- id: pyupgrade
args:
- --py38-plus
- repo: https://github.com/psf/black
rev: 23.7.0
hooks:
@@ -43,22 +32,20 @@ repos:
hooks:
- id: isort
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: 'v0.0.278'
rev: 'v0.0.282'
hooks:
- id: ruff
args: [--fix, --exit-non-zero-on-fix]
- repo: https://github.com/codespell-project/codespell
rev: v2.2.5
hooks:
- id: codespell
additional_dependencies: ["tomli"]
- repo: https://github.com/PyCQA/flake8
rev: 6.0.0
rev: 6.1.0
hooks:
- id: flake8
additional_dependencies:
- flake8-bugbear
- flake8-comprehensions
- flake8-debugger
- flake8-string-format
- flake8-typing-imports
- repo: https://github.com/PyCQA/bandit
+1
View File
@@ -22,6 +22,7 @@ hiddenimports = [
"dvc_hdfs",
"dvc_oss",
"dvc_s3",
"dvc_ssh",
"dvc_webdav",
"dvc_webhdfs",
# https://github.com/pypa/setuptools/issues/1963
+1 -1
View File
@@ -67,7 +67,7 @@ class _OpenContextManager(GCM):
raise AttributeError("dvc.api.open() should be used in a with statement.")
def open( # noqa, pylint: disable=redefined-builtin
def open( # noqa: A001, pylint: disable=redefined-builtin
path: str,
repo: Optional[str] = None,
rev: Optional[str] = None,
+1 -1
View File
@@ -230,7 +230,7 @@ def main(argv=None): # noqa: C901, PLR0912, PLR0915
logger.exception("")
except DvcParserError:
ret = 254
except Exception as exc: # noqa, pylint: disable=broad-except
except Exception as exc: # noqa: BLE001, pylint: disable=broad-except
# pylint: disable=no-member
ret = _log_exceptions(exc) or 255
-2
View File
@@ -31,7 +31,6 @@ from dvc.commands import (
install,
ls,
ls_url,
machine,
metrics,
move,
params,
@@ -89,7 +88,6 @@ COMMANDS = [
stage,
experiments,
check_ignore,
machine,
data,
]
+24
View File
@@ -1,3 +1,27 @@
import argparse
class DictAction(argparse.Action):
def __call__(self, parser, args, values, option_string=None): # noqa: ARG002
d = getattr(args, self.dest) or {}
if isinstance(values, list):
kvs = values
else:
kvs = [values]
for kv in kvs:
key, value = kv.split("=")
if not value:
raise argparse.ArgumentError(
self,
f'Could not parse argument "{values}" as k1=v1 k2=v2 ... format',
)
d[key] = value
setattr(args, self.dest, d)
def fix_subparsers(subparsers):
"""Workaround for bug in Python 3. See more info at:
https://bugs.python.org/issue16308
+4 -1
View File
@@ -47,7 +47,10 @@ def add_parser(subparsers, parent_parser):
"--force",
action="store_true",
default=False,
help="Commit even if hash value for dependencies/outputs changed.",
help=(
"Commit data even if hash values for dependencies or "
"outputs did not change."
),
)
commit_parser.add_argument(
"-d",
+4 -1
View File
@@ -9,6 +9,9 @@ from dvc.ui import ui
logger = logging.getLogger(__name__)
SUPPORTED_SHELLS = ["bash", "zsh"]
class CmdCompletion(CmdBaseNoRepo):
def run(self):
import shtab
@@ -35,6 +38,6 @@ def add_parser(subparsers, parent_parser):
"--shell",
help="Shell syntax for completions.",
default="bash",
choices=["bash", "zsh"],
choices=SUPPORTED_SHELLS,
)
completion_parser.set_defaults(func=CmdCompletion)
+14 -7
View File
@@ -23,15 +23,22 @@ class CmdExperimentsList(CmdBase):
git_remote=git_remote,
)
from dvc.repo.experiments.utils import describe
from dvc.scm import Git
if name_only or sha_only:
names = {}
else:
assert isinstance(self.repo.scm, Git)
names = describe(
self.repo.scm,
(baseline for baseline in exps),
logger=logger,
)
for baseline in exps:
if not (name_only or sha_only):
tag_base = "refs/tags/"
branch_base = "refs/heads/"
name = baseline[:7]
if baseline.startswith(tag_base):
name = baseline[len(tag_base) :]
elif baseline.startswith(branch_base):
name = baseline[len(branch_base) :]
name = names.get(baseline) or baseline[:7]
ui.write(f"{name}:")
for exp_name, rev in exps[baseline]:
if name_only:
-10
View File
@@ -17,7 +17,6 @@ class CmdExperimentsRun(CmdRepro):
jobs=self.args.jobs,
params=self.args.set_param,
tmp_dir=self.args.tmp_dir,
machine=self.args.machine,
copy_paths=self.args.copy_paths,
message=self.args.message,
**self._common_kwargs,
@@ -90,15 +89,6 @@ def _add_run_common(parser):
"your workspace."
),
)
parser.add_argument(
"--machine",
default=None,
help=argparse.SUPPRESS,
# help=(
# "Run this experiment on the specified 'dvc machine' instance."
# )
# metavar="<name>",
)
parser.add_argument(
"-C",
"--copy-paths",
+18 -1
View File
@@ -3,7 +3,7 @@ import logging
from dvc.cli import completion
from dvc.cli.command import CmdBaseNoRepo
from dvc.cli.utils import append_doc_link
from dvc.cli.utils import DictAction, append_doc_link
from dvc.exceptions import DvcException
logger = logging.getLogger(__name__)
@@ -38,6 +38,8 @@ class CmdGet(CmdBaseNoRepo):
jobs=self.args.jobs,
force=self.args.force,
config=self.args.config,
remote=self.args.remote,
remote_config=self.args.remote_config,
)
return 0
except CloneError:
@@ -111,4 +113,19 @@ def add_parser(subparsers, parent_parser):
"in the target repository."
),
)
get_parser.add_argument(
"--remote",
type=str,
help="Remote name to set as a default in the target repository.",
)
get_parser.add_argument(
"--remote-config",
type=str,
nargs="*",
action=DictAction,
help=(
"Remote config options to merge with a remote's config (default or one "
"specified by '--remote') in the target repository."
),
)
get_parser.set_defaults(func=CmdGet)
+18 -1
View File
@@ -3,7 +3,7 @@ import logging
from dvc.cli import completion
from dvc.cli.command import CmdBase
from dvc.cli.utils import append_doc_link
from dvc.cli.utils import DictAction, append_doc_link
from dvc.exceptions import DvcException
logger = logging.getLogger(__name__)
@@ -23,6 +23,8 @@ class CmdImport(CmdBase):
no_download=self.args.no_download,
jobs=self.args.jobs,
config=self.args.config,
remote=self.args.remote,
remote_config=self.args.remote_config,
)
except CloneError:
logger.exception("failed to import '%s'", self.args.path)
@@ -103,4 +105,19 @@ def add_parser(subparsers, parent_parser):
"in the target repository."
),
)
import_parser.add_argument(
"--remote",
type=str,
help="Remote name to set as a default in the target repository.",
)
import_parser.add_argument(
"--remote-config",
type=str,
nargs="*",
action=DictAction,
help=(
"Remote config options to merge with a remote's config (default or one "
"specified by '--remote') in the target repository."
),
)
import_parser.set_defaults(func=CmdImport)
+18 -1
View File
@@ -3,7 +3,7 @@ import logging
from dvc.cli import completion
from dvc.cli.command import CmdBaseNoRepo
from dvc.cli.utils import append_doc_link
from dvc.cli.utils import DictAction, append_doc_link
from dvc.commands.ls.ls_colors import LsColors
from dvc.exceptions import DvcException
from dvc.ui import ui
@@ -35,6 +35,8 @@ class CmdList(CmdBaseNoRepo):
recursive=self.args.recursive,
dvc_only=self.args.dvc_only,
config=self.args.config,
remote=self.args.remote,
remote_config=self.args.remote_config,
)
if self.args.json:
ui.write_json(entries)
@@ -89,6 +91,21 @@ def add_parser(subparsers, parent_parser):
"in the target repository."
),
)
list_parser.add_argument(
"--remote",
type=str,
help="Remote name to set as a default in the target repository.",
)
list_parser.add_argument(
"--remote-config",
type=str,
nargs="*",
action=DictAction,
help=(
"Remote config options to merge with a remote's config (default or one "
"specified by '--remote') in the target repository."
),
)
list_parser.add_argument(
"path",
nargs="?",
-490
View File
@@ -1,490 +0,0 @@
import argparse
from typing import Dict, List
from dvc.cli.command import CmdBase
from dvc.cli.utils import append_doc_link, fix_subparsers
from dvc.commands.config import CmdConfig
from dvc.compare import TabularData
from dvc.config import ConfigError
from dvc.exceptions import DvcException
from dvc.ui import ui
from dvc.utils import format_link
class MachineDisabledError(ConfigError):
def __init__(self):
super().__init__("Machine feature is disabled")
class CmdMachineConfig(CmdConfig):
def __init__(self, args):
super().__init__(args)
if not self.config["feature"].get("machine", False):
raise MachineDisabledError
if getattr(self.args, "name", None):
self.args.name = self.args.name.lower()
def _check_exists(self, conf):
if self.args.name not in conf["machine"]:
raise ConfigError(f"machine '{self.args.name}' doesn't exist.")
class CmdMachineAdd(CmdMachineConfig):
def run(self):
from dvc.machine import validate_name
validate_name(self.args.name)
if self.args.default:
ui.write(f"Setting '{self.args.name}' as a default machine.")
with self.config.edit(self.args.level) as conf:
if self.args.name in conf["machine"] and not self.args.force:
raise ConfigError(
"machine '{}' already exists. Use `-f|--force` to "
"overwrite it.".format(self.args.name)
)
conf["machine"][self.args.name] = {"cloud": self.args.cloud}
if self.args.default:
conf["core"]["machine"] = self.args.name
return 0
class CmdMachineRemove(CmdMachineConfig):
def run(self):
with self.config.edit(self.args.level) as conf:
self._check_exists(conf)
del conf["machine"][self.args.name]
up_to_level = self.args.level or "repo"
# Remove core.machine refs to this machine in any shadowing configs
for level in reversed(self.config.LEVELS):
with self.config.edit(level) as conf:
if conf["core"].get("machine") == self.args.name:
del conf["core"]["machine"]
if level == up_to_level:
break
return 0
class CmdMachineList(CmdMachineConfig):
TABLE_COLUMNS = [
"name",
"cloud",
"region",
"image",
"spot",
"spot_price",
"instance_hdd_size",
"instance_type",
"ssh_private",
"startup_script",
]
PRIVATE_COLUMNS = ["ssh_private", "startup_script"]
def _hide_private(self, conf):
for machine in conf:
for column in self.PRIVATE_COLUMNS:
if column in conf[machine]:
conf[machine][column] = "***"
def _show_origin(self):
levels = [self.args.level] if self.args.level else self.config.LEVELS
for level in levels:
conf = self.config.read(level)["machine"]
if self.args.name:
conf = conf.get(self.args.name, {})
self._hide_private(conf)
prefix = self._config_file_prefix(True, self.config, level)
configs = list(self._format_config(conf, prefix))
if configs:
ui.write("\n".join(configs))
def _show_table(self):
td = TabularData(self.TABLE_COLUMNS, fill_value="-")
conf = self.config.read()["machine"]
if self.args.name:
conf = {self.args.name: conf.get(self.args.name, {})}
self._hide_private(conf)
for machine, machine_config in conf.items():
machine_config["name"] = machine
td.row_from_dict(machine_config)
td.dropna("cols", "all")
td.render()
def run(self):
if self.args.show_origin:
self._show_origin()
else:
self._show_table()
return 0
class CmdMachineModify(CmdMachineConfig):
def run(self):
from dvc.config import merge
with self.config.edit(self.args.level) as conf:
merged = self.config.load_config_to_level(self.args.level)
merge(merged, conf)
self._check_exists(merged)
if self.args.name not in conf["machine"]:
conf["machine"][self.args.name] = {}
section = conf["machine"][self.args.name]
if self.args.unset:
section.pop(self.args.option, None)
else:
section[self.args.option] = self.args.value
return 0
class CmdMachineRename(CmdBase):
def _check_exists(self, conf):
if self.args.name not in conf["machine"]:
raise ConfigError(f"machine '{self.args.name}' doesn't exist.")
def _rename_default(self, conf):
if conf["core"].get("machine") == self.args.name:
conf["core"]["machine"] = self.args.new
def _check_before_rename(self):
from dvc.machine import validate_name
validate_name(self.args.new)
all_config = self.config.load_config_to_level(None)
if self.args.new in all_config.get("machine", {}):
raise ConfigError(
f"Rename failed. Machine '{self.args.new}' already exists."
)
ui.write(f"Rename machine '{self.args.name}' to '{self.args.new}'.")
def run(self):
self._check_before_rename()
with self.config.edit(self.args.level) as conf:
self._check_exists(conf)
conf["machine"][self.args.new] = conf["machine"][self.args.name]
try:
assert self.repo.machine
self.repo.machine.rename(self.args.name, self.args.new)
except DvcException as error:
del conf["machine"][self.args.new]
raise ConfigError("terraform rename failed") from error
del conf["machine"][self.args.name]
self._rename_default(conf)
up_to_level = self.args.level or "repo"
for level in reversed(self.config.LEVELS):
if level == up_to_level:
break
with self.config.edit(level) as level_conf:
self._rename_default(level_conf)
return 0
class CmdMachineDefault(CmdMachineConfig):
def run(self):
if self.args.name is None and not self.args.unset:
conf = self.config.read(self.args.level)
try:
ui.write(conf["core"]["machine"])
except KeyError:
ui.write("No default machine set")
return 1
else:
with self.config.edit(self.args.level) as conf:
if self.args.unset:
conf["core"].pop("machine", None)
else:
merged_conf = self.config.load_config_to_level(self.args.level)
if (
self.args.name in conf["machine"]
or self.args.name in merged_conf["machine"]
):
conf["core"]["machine"] = self.args.name
else:
raise ConfigError(
"default machine must be present in machine list."
)
return 0
class CmdMachineCreate(CmdBase):
def run(self):
if self.repo.machine is None:
raise MachineDisabledError
self.repo.machine.create(self.args.name)
return 0
class CmdMachineStatus(CmdBase):
INSTANCE_FIELD = ["name", "instance", "status"]
SHOWN_FIELD = [
"cloud",
"instance_ip",
"instance_type",
"instance_hdd_size",
"instance_gpu",
]
FILL_VALUE = "-"
def _add_row(
self,
name: str,
all_status: List[Dict],
td: TabularData,
):
if not all_status:
row = [
name,
self.FILL_VALUE,
"offline",
] # back to `None` after #7167
td.append(row)
for i, status in enumerate(all_status, start=1):
row = [name, f"num_{i}", "running" if status else "offline"]
for field in self.SHOWN_FIELD:
value = str(status.get(field, ""))
row.append(value)
td.append(row)
def run(self):
if self.repo.machine is None:
raise MachineDisabledError
td = TabularData(
self.INSTANCE_FIELD + self.SHOWN_FIELD, fill_value=self.FILL_VALUE
)
if self.args.name:
all_status = list(self.repo.machine.status(self.args.name))
self._add_row(self.args.name, all_status, td)
else:
name_set = set()
for level in self.repo.config.LEVELS:
conf = self.repo.config.read(level)["machine"]
name_set.update(conf.keys())
name_list = list(name_set)
for name in sorted(name_list):
all_status = list(self.repo.machine.status(name))
self._add_row(name, all_status, td)
td.dropna("cols", "all")
td.render()
return 0
class CmdMachineDestroy(CmdBase):
def run(self):
if self.repo.machine is None:
raise MachineDisabledError
self.repo.machine.destroy(self.args.name)
return 0
class CmdMachineSsh(CmdBase):
def run(self):
if self.repo.machine is None:
raise MachineDisabledError
self.repo.machine.run_shell(self.args.name)
return 0
def add_parser(subparsers, parent_parser): # noqa: PLR0915
from dvc.commands.config import parent_config_parser
machine_HELP = "Set up and manage cloud machines."
machine_parser = subparsers.add_parser(
"machine",
parents=[parent_parser],
description=append_doc_link(machine_HELP, "machine"),
# NOTE: suppress help during development to hide command
# help=machine_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_subparsers = machine_parser.add_subparsers(
dest="cmd",
help="Use `dvc machine CMD --help` for command-specific help.",
)
fix_subparsers(machine_subparsers)
machine_ADD_HELP = "Add a new data machine."
machine_add_parser = machine_subparsers.add_parser(
"add",
parents=[parent_config_parser, parent_parser],
description=append_doc_link(machine_ADD_HELP, "machine/add"),
help=machine_ADD_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_add_parser.add_argument("name", help="Name of the machine")
machine_add_parser.add_argument(
"cloud",
help="Machine cloud. See full list of supported clouds at {}".format(
format_link(
"https://github.com/iterative/terraform-provider-iterative#machine"
)
),
)
machine_add_parser.add_argument(
"-d",
"--default",
action="store_true",
default=False,
help="Set as default machine.",
)
machine_add_parser.add_argument(
"-f",
"--force",
action="store_true",
default=False,
help="Force overwriting existing configs",
)
machine_add_parser.set_defaults(func=CmdMachineAdd)
machine_DEFAULT_HELP = "Set/unset the default machine."
machine_default_parser = machine_subparsers.add_parser(
"default",
parents=[parent_config_parser, parent_parser],
description=append_doc_link(machine_DEFAULT_HELP, "machine/default"),
help=machine_DEFAULT_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_default_parser.add_argument("name", nargs="?", help="Name of the machine")
machine_default_parser.add_argument(
"-u",
"--unset",
action="store_true",
default=False,
help="Unset default machine.",
)
machine_default_parser.set_defaults(func=CmdMachineDefault)
machine_LIST_HELP = "List the configuration of one/all machines."
machine_list_parser = machine_subparsers.add_parser(
"list",
parents=[parent_config_parser, parent_parser],
description=append_doc_link(machine_LIST_HELP, "machine/list"),
help=machine_LIST_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_list_parser.add_argument(
"--show-origin",
default=False,
action="store_true",
help="Show the source file containing each config value.",
)
machine_list_parser.add_argument(
"name",
nargs="?",
type=str,
help="name of machine to specify",
)
machine_list_parser.set_defaults(func=CmdMachineList)
machine_MODIFY_HELP = "Modify the configuration of an machine."
machine_modify_parser = machine_subparsers.add_parser(
"modify",
parents=[parent_config_parser, parent_parser],
description=append_doc_link(machine_MODIFY_HELP, "machine/modify"),
help=machine_MODIFY_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_modify_parser.add_argument("name", help="Name of the machine")
machine_modify_parser.add_argument("option", help="Name of the option to modify.")
machine_modify_parser.add_argument(
"value", nargs="?", help="(optional) Value of the option."
)
machine_modify_parser.add_argument(
"-u",
"--unset",
default=False,
action="store_true",
help="Unset option.",
)
machine_modify_parser.set_defaults(func=CmdMachineModify)
machine_RENAME_HELP = "Rename a machine "
machine_rename_parser = machine_subparsers.add_parser(
"rename",
parents=[parent_config_parser, parent_parser],
description=append_doc_link(machine_RENAME_HELP, "remote/rename"),
help=machine_RENAME_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_rename_parser.add_argument("name", help="Machine to be renamed")
machine_rename_parser.add_argument("new", help="New name of the machine")
machine_rename_parser.set_defaults(func=CmdMachineRename)
machine_REMOVE_HELP = "Remove an machine."
machine_remove_parser = machine_subparsers.add_parser(
"remove",
parents=[parent_config_parser, parent_parser],
description=append_doc_link(machine_REMOVE_HELP, "machine/remove"),
help=machine_REMOVE_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_remove_parser.add_argument("name", help="Name of the machine to remove.")
machine_remove_parser.set_defaults(func=CmdMachineRemove)
machine_CREATE_HELP = "Create and start a machine instance."
machine_create_parser = machine_subparsers.add_parser(
"create",
parents=[parent_parser],
description=append_doc_link(machine_CREATE_HELP, "machine/create"),
help=machine_CREATE_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_create_parser.add_argument("name", help="Name of the machine to create.")
machine_create_parser.set_defaults(func=CmdMachineCreate)
machine_STATUS_HELP = "List the status of running instances for one/all machines."
machine_status_parser = machine_subparsers.add_parser(
"status",
parents=[parent_parser],
description=append_doc_link(machine_STATUS_HELP, "machine/status"),
help=machine_STATUS_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_status_parser.add_argument(
"name", nargs="?", help="(optional) Name of the machine."
)
machine_status_parser.set_defaults(func=CmdMachineStatus)
machine_DESTROY_HELP = "Destroy an machine instance."
machine_destroy_parser = machine_subparsers.add_parser(
"destroy",
parents=[parent_parser],
description=append_doc_link(machine_DESTROY_HELP, "machine/destroy"),
help=machine_DESTROY_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_destroy_parser.add_argument(
"name", help="Name of the machine instance to destroy."
)
machine_destroy_parser.set_defaults(func=CmdMachineDestroy)
machine_SSH_HELP = "Connect to a machine via SSH."
machine_ssh_parser = machine_subparsers.add_parser(
"ssh",
parents=[parent_parser],
description=append_doc_link(machine_SSH_HELP, "machine/ssh"),
help=machine_SSH_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
machine_ssh_parser.add_argument(
"name", help="Name of the machine instance to connect to."
)
machine_ssh_parser.set_defaults(func=CmdMachineSsh)
+4 -5
View File
@@ -57,7 +57,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
return self._columns[name]
def items(self) -> ItemsView[str, Column]:
projection = {k: self.column(k) for k in self.keys()} # noqa: SIM118
projection = {k: self.column(k) for k in self.keys()}
return projection.items()
def keys(self) -> List[str]:
@@ -171,8 +171,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
self.add_column(key)
row: List["CellT"] = [
with_value(d.get(key), self._fill_value)
for key in self.keys() # noqa: SIM118
with_value(d.get(key), self._fill_value) for key in self.keys()
]
self.append(row)
@@ -227,7 +226,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
to_drop -= match_line
if axis == "rows":
for name in self.keys(): # noqa: SIM118
for name in self.keys():
self._columns[name] = Column(
[x for n, x in enumerate(self._columns[name]) if n not in to_drop]
)
@@ -275,7 +274,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
else:
unique_rows.append(tuple_row)
for name in self.keys(): # noqa: SIM118
for name in self.keys():
self._columns[name] = Column(
[
x
+38 -13
View File
@@ -90,6 +90,7 @@ class Config(dict):
def __init__(
self,
dvc_dir: Optional["StrPath"] = None,
local_dvc_dir: Optional["StrPath"] = None,
validate: bool = True,
fs: Optional["FileSystem"] = None,
config: Optional["DictStrAny"] = None,
@@ -105,6 +106,10 @@ class Config(dict):
if dvc_dir:
self.dvc_dir = self.fs.path.abspath(dvc_dir)
self.local_dvc_dir = local_dvc_dir
if not fs and not local_dvc_dir:
self.local_dvc_dir = dvc_dir
self.load(
validate=validate, config=config, remote=remote, remote_config=remote_config
)
@@ -140,7 +145,9 @@ class Config(dict):
if self.dvc_dir is not None:
files["repo"] = self.fs.path.join(self.dvc_dir, self.CONFIG)
files["local"] = self.fs.path.join(self.dvc_dir, self.CONFIG_LOCAL)
if self.local_dvc_dir is not None:
files["local"] = self.wfs.path.join(self.local_dvc_dir, self.CONFIG_LOCAL)
return files
@@ -158,6 +165,9 @@ class Config(dict):
with open(config_file, "w+", encoding="utf-8"):
return Config(dvc_dir)
def merge(self, config):
merge(self, config)
def load(
self,
validate: bool = True,
@@ -250,21 +260,30 @@ class Config(dict):
return conf
@staticmethod
def _load_paths(conf, filename):
abs_conf_dir = os.path.abspath(os.path.dirname(filename))
def _resolve(conf_dir, path):
from .config_schema import ExpPath, RelPath
def resolve(path):
from .config_schema import RelPath
if re.match(r"\w+://", path):
return path
if os.path.isabs(path) or re.match(r"\w+://", path):
return path
if os.path.isabs(path):
return path
# on windows convert slashes to backslashes
# to have path compatible with abs_conf_dir
if os.path.sep == "\\" and "/" in path:
path = path.replace("/", "\\")
# on windows convert slashes to backslashes
# to have path compatible with abs_conf_dir
if os.path.sep == "\\" and "/" in path:
path = path.replace("/", "\\")
return RelPath(os.path.join(abs_conf_dir, path))
expanded = os.path.expanduser(path)
if os.path.isabs(expanded):
return ExpPath(expanded, path)
return RelPath(os.path.abspath(os.path.join(conf_dir, path)))
@classmethod
def _load_paths(cls, conf, filename):
conf_dir = os.path.abspath(os.path.dirname(filename))
resolve = partial(cls._resolve, conf_dir)
return Config._map_dirs(conf, resolve)
@@ -273,11 +292,17 @@ class Config(dict):
from dvc.fs import localfs
from dvc.utils import relpath
from .config_schema import RelPath
from .config_schema import ExpPath, RelPath
if re.match(r"\w+://", path):
return path
if isinstance(path, ExpPath):
return path.def_path
if os.path.expanduser(path) != path:
return localfs.path.as_posix(path)
if isinstance(path, RelPath) or not os.path.isabs(path):
path = relpath(path, conf_dir)
return localfs.path.as_posix(path)
+14 -1
View File
@@ -8,6 +8,7 @@ from voluptuous import (
All,
Any,
Coerce,
Exclusive,
Invalid,
Lower,
Optional,
@@ -74,6 +75,17 @@ def ByUrl(mapping): # noqa: N802
return validate
class ExpPath(str):
__slots__ = ("def_path",)
def_path: str
def __new__(cls, string, def_path):
ret = super().__new__(cls, string)
ret.def_path = def_path
return ret
class RelPath(str):
__slots__ = ()
@@ -324,7 +336,8 @@ SCHEMA = {
},
"hydra": {
Optional("enabled", default=False): Bool,
"config_dir": str,
Exclusive("config_dir", "config_source"): str,
Exclusive("config_module", "config_source"): str,
"config_name": str,
},
"studio": {
+30 -19
View File
@@ -4,6 +4,7 @@ import logging
from typing import TYPE_CHECKING, Iterable, Optional, Set, Tuple
from dvc.config import NoRemoteError, RemoteConfigError
from dvc.fs.callbacks import Callback
from dvc.utils.objects import cached_property
from dvc_data.hashfile.db import get_index
from dvc_data.hashfile.transfer import TransferResult
@@ -209,15 +210,20 @@ class DataCloud:
cache = self.repo.cache.legacy
else:
cache = self.repo.cache.local
return self.transfer(
cache,
odb,
objs,
jobs=jobs,
dest_index=get_index(odb),
cache_odb=cache,
validate_status=self._log_missing,
)
with Callback.as_tqdm_callback(
desc=f"Pushing to {odb.fs.unstrip_protocol(odb.path)}",
unit="file",
) as cb:
return self.transfer(
cache,
odb,
objs,
jobs=jobs,
dest_index=get_index(odb),
cache_odb=cache,
validate_status=self._log_missing,
callback=cb,
)
def pull(
self,
@@ -263,16 +269,21 @@ class DataCloud:
cache = self.repo.cache.legacy
else:
cache = self.repo.cache.local
return self.transfer(
odb,
cache,
objs,
jobs=jobs,
src_index=get_index(odb),
cache_odb=cache,
verify=odb.verify,
validate_status=self._log_missing,
)
with Callback.as_tqdm_callback(
desc=f"Fetching from {odb.fs.unstrip_protocol(odb.path)}",
unit="file",
) as cb:
return self.transfer(
odb,
cache,
objs,
jobs=jobs,
src_index=get_index(odb),
cache_odb=cache,
verify=odb.verify,
validate_status=self._log_missing,
callback=cb,
)
def status(
self,
+17 -1
View File
@@ -1,6 +1,6 @@
from typing import TYPE_CHECKING, Dict, Optional, Union
from voluptuous import Required
from voluptuous import Any, Required
from dvc.utils import as_posix
@@ -17,6 +17,7 @@ class RepoDependency(Dependency):
PARAM_REV = "rev"
PARAM_REV_LOCK = "rev_lock"
PARAM_CONFIG = "config"
PARAM_REMOTE = "remote"
REPO_SCHEMA = {
PARAM_REPO: {
@@ -24,6 +25,7 @@ class RepoDependency(Dependency):
PARAM_REV: str,
PARAM_REV_LOCK: str,
PARAM_CONFIG: str,
PARAM_REMOTE: Any(str, dict),
}
}
@@ -76,6 +78,10 @@ class RepoDependency(Dependency):
if config:
repo[self.PARAM_CONFIG] = config
remote = self.def_repo.get(self.PARAM_REMOTE)
if remote:
repo[self.PARAM_REMOTE] = remote
return {
self.PARAM_PATH: self.def_path,
self.PARAM_REPO: repo,
@@ -99,6 +105,14 @@ class RepoDependency(Dependency):
from dvc.config import Config
from dvc.fs import DVCFileSystem
rem = self.def_repo.get("remote")
if isinstance(rem, dict):
remote = None # type: ignore[unreachable]
remote_config = rem
else:
remote = rem
remote_config = None
conf = self.def_repo.get("config")
if conf:
config = Config.load_file(conf)
@@ -113,6 +127,8 @@ class RepoDependency(Dependency):
rev=rev or self._get_rev(locked=locked),
subrepos=True,
config=config,
remote=remote,
remote_config=remote_config,
)
def _get_rev(self, locked: bool = True):
+1 -1
View File
@@ -297,7 +297,7 @@ class URLMissingError(DvcException):
super().__init__(f"The path '{url}' does not exist")
class IsADirectoryError(DvcException): # noqa,pylint:disable=redefined-builtin
class IsADirectoryError(DvcException): # noqa: A001,pylint:disable=redefined-builtin
"""Raised when a file operation is requested on a directory."""
+1 -1
View File
@@ -160,7 +160,7 @@ class LoggerHandler(logging.StreamHandler):
self.flush()
except (BrokenPipeError, RecursionError):
raise
except Exception: # noqa, pylint: disable=broad-except
except Exception: # noqa: BLE001, pylint: disable=broad-except
self.handleError(record)
-227
View File
@@ -1,227 +0,0 @@
import logging
import os
from typing import (
TYPE_CHECKING,
Any,
Dict,
Iterable,
Iterator,
Mapping,
Optional,
Tuple,
Type,
)
from dvc.exceptions import DvcException
if TYPE_CHECKING:
from dvc.repo import Repo
from dvc.types import StrPath
from .backend.base import BaseMachineBackend
BackendCls = Type[BaseMachineBackend]
logger = logging.getLogger(__name__)
RESERVED_NAMES = {"local", "localhost"}
DEFAULT_STARTUP_SCRIPT = """#!/bin/bash
sudo apt-get update
sudo apt-get install --yes python3 python3-pip
python3 -m pip install --upgrade pip
pushd /etc/apt/sources.list.d
sudo wget https://dvc.org/deb/dvc.list
sudo apt-get update
sudo apt-get install --yes dvc
popd
sudo echo "OK" > /var/log/dvc-machine-init.log
"""
def validate_name(name: str):
from dvc.exceptions import InvalidArgumentError
name = name.lower()
if name in RESERVED_NAMES:
raise InvalidArgumentError(
f"Machine name '{name}' is reserved for internal DVC use."
)
class MachineBackends(Mapping):
try:
from .backend.terraform import TerraformBackend
except ImportError:
TerraformBackend = None # type: ignore[assignment, misc]
DEFAULT: Dict[str, Optional["BackendCls"]] = {
"terraform": TerraformBackend,
}
def __getitem__(self, key: str) -> "BaseMachineBackend":
"""Lazily initialize backends and cache it afterwards"""
initialized = self.initialized.get(key)
if not initialized:
backend = self.backends[key]
initialized = backend(os.path.join(self.tmp_dir, key), **self.kwargs)
self.initialized[key] = initialized
return initialized
def __init__(
self,
selected: Optional[Iterable[str]],
tmp_dir: "StrPath",
**kwargs,
) -> None:
selected = selected or list(self.DEFAULT)
self.backends: Dict[str, "BackendCls"] = {}
for key in selected:
cls = self.DEFAULT.get(key)
if cls is None:
raise DvcException(
f"'dvc machine' backend '{key}' is missing required "
"dependencies. Install them with:\n"
f"\tpip install dvc[{key}]"
)
self.backends[key] = cls
self.initialized: Dict[str, "BaseMachineBackend"] = {}
self.tmp_dir = tmp_dir
self.kwargs = kwargs
def __iter__(self):
return iter(self.backends)
def __len__(self) -> int:
return len(self.backends)
def close_initialized(self) -> None:
for backend in self.initialized.values():
backend.close()
class MachineManager:
"""Class that manages dvc cloud machines.
Args:
repo (dvc.repo.Repo): repo instance that belongs to the repo that
we are working on.
Raises:
config.ConfigError: thrown when config has invalid format.
"""
CLOUD_BACKENDS = {
"aws": "terraform",
"azure": "terraform",
}
def __init__(
self, repo: "Repo", backends: Optional[Iterable[str]] = None, **kwargs
):
self.repo = repo
assert self.repo.tmp_dir
tmp_dir = os.path.join(self.repo.tmp_dir, "machine")
self.backends = MachineBackends(backends, tmp_dir=tmp_dir, **kwargs)
def get_config_and_backend(
self,
name: Optional[str] = None,
) -> Tuple[dict, "BaseMachineBackend"]:
from dvc.config import NoMachineError
if not name:
name = self.repo.config["core"].get("machine")
if name:
config = self._get_config(name=name)
backend = self._get_backend(config["cloud"])
return config, backend
if bool(self.repo.config["machine"]):
error_msg = (
"no machine specified. Setup default machine with\n"
" dvc machine default <name>\n"
)
else:
error_msg = (
"no machine specified. Create a default machine with\n"
" dvc machine add -d <name> <cloud>"
)
raise NoMachineError(error_msg)
def _get_config(self, **kwargs):
config = self.repo.config
name = kwargs.get("name")
if name:
try:
conf = config["machine"][name.lower()]
conf["name"] = name
except KeyError:
from dvc.config import MachineNotFoundError
raise MachineNotFoundError( # noqa: B904
f"machine '{name}' doesn't exist"
)
else:
conf = kwargs
return conf
def _get_backend(self, cloud: str) -> "BaseMachineBackend":
from dvc.config import NoMachineError
try:
backend = self.CLOUD_BACKENDS[cloud]
return self.backends[backend]
except KeyError:
raise NoMachineError( # noqa: B904
f"Machine platform '{cloud}' unsupported"
)
def create(self, name: Optional[str]):
"""Create and start the specified machine instance."""
config, backend = self.get_config_and_backend(name)
if "startup_script" in config:
with open(config["startup_script"], encoding="utf-8") as fobj:
startup_script = fobj.read()
else:
startup_script = DEFAULT_STARTUP_SCRIPT
config["startup_script"] = startup_script
config.pop("setup_script", None)
return backend.create(**config)
def destroy(self, name: Optional[str]):
"""Destroy the specified machine instance."""
config, backend = self.get_config_and_backend(name)
return backend.destroy(**config)
def get_sshfs(self, name: Optional[str]):
config, backend = self.get_config_and_backend(name)
return backend.get_sshfs(**config)
def run_shell(self, name: Optional[str]):
config, backend = self.get_config_and_backend(name)
return backend.run_shell(**config)
def status(self, name: str) -> Iterator[Dict[Any, Any]]:
config, backend = self.get_config_and_backend(name)
return backend.instances(**config)
def rename(self, name: str, new: str):
"""move configuration to a new location if the machine is running."""
config, backend = self.get_config_and_backend(name)
return backend.rename(new=new, **config)
def get_executor_kwargs(self, name: Optional[str]):
config, backend = self.get_config_and_backend(name)
return backend.get_executor_kwargs(**config)
def get_setup_script(self, name: Optional[str]) -> Optional[str]:
config, _backend = self.get_config_and_backend(name)
return config.get("setup_script")
View File
-50
View File
@@ -1,50 +0,0 @@
from abc import ABC, abstractmethod
from contextlib import contextmanager
from typing import TYPE_CHECKING, Iterator, Optional
if TYPE_CHECKING:
from dvc_ssh import SSHFileSystem
from dvc.types import StrPath
class BaseMachineBackend(ABC):
def __init__(self, tmp_dir: "StrPath", **kwargs):
self.tmp_dir = tmp_dir
@abstractmethod
def create(self, name: Optional[str] = None, **config):
"""Create and start an instance of the specified machine."""
@abstractmethod
def destroy(self, name: Optional[str] = None, **config):
"""Stop and destroy all instances of the specified machine."""
@abstractmethod
def instances(self, name: Optional[str] = None, **config) -> Iterator[dict]:
"""Iterate over status of all instances of the specified machine."""
def close(self): # noqa: B027
pass
@abstractmethod
def run_shell(self, name: Optional[str] = None, **config):
"""Spawn an interactive SSH shell for the specified machine."""
@abstractmethod
def get_executor_kwargs(self, name: str, **config) -> dict:
"""Return SSHExecutor kwargs which can be used for DVC
experiment/pipeline execution on the specified machine.
"""
@abstractmethod
@contextmanager
def get_sshfs(
self, name: Optional[str] = None, **config
) -> Iterator["SSHFileSystem"]:
"""Return an sshfs instance for the default directory on the
specified machine."""
@abstractmethod
def rename(self, name: str, new: str, **config):
"""Rename a machine instance."""
-99
View File
@@ -1,99 +0,0 @@
import os
from contextlib import contextmanager
from functools import partial, partialmethod
from typing import TYPE_CHECKING, Iterator, Optional
from dvc_ssh import DEFAULT_PORT, SSHFileSystem
from dvc.exceptions import DvcException
from .base import BaseMachineBackend
if TYPE_CHECKING:
from dvc.types import StrPath
@contextmanager
def _sshfs(resource: dict):
from tpi import TerraformProviderIterative
with TerraformProviderIterative.pemfile(resource) as pem:
fs = SSHFileSystem(
host=resource["instance_ip"],
user="ubuntu",
keyfile=pem,
)
yield fs
class TerraformBackend(BaseMachineBackend):
def __init__(self, tmp_dir: "StrPath", **kwargs):
super().__init__(tmp_dir, **kwargs)
os.makedirs(tmp_dir, exist_ok=True)
@contextmanager
def make_tpi(self, name: str):
from tpi import TPIError
from tpi.terraform import TerraformBackend as TPIBackend
try:
working_dir = os.path.join(self.tmp_dir, name)
os.makedirs(working_dir, exist_ok=True)
yield TPIBackend(working_dir=working_dir)
except TPIError as exc:
raise DvcException("TPI operation failed") from exc
def _tpi_func(self, fname, name: Optional[str] = None, **config):
from tpi import TPIError
assert name
with self.make_tpi(name) as tpi:
func = getattr(tpi, fname)
try:
return func(name=name, **config)
except TPIError as exc:
raise DvcException(f"TPI {fname} failed") from exc
create = partialmethod(_tpi_func, "create") # type: ignore[assignment]
destroy = partialmethod(_tpi_func, "destroy") # type: ignore[assignment]
instances = partialmethod(_tpi_func, "instances") # type: ignore[assignment]
run_shell = partialmethod(_tpi_func, "run_shell") # type: ignore[assignment]
def get_executor_kwargs(self, name: str, **config) -> dict:
with self.make_tpi(name) as tpi:
resource = tpi.default_resource(name)
return {
"host": resource["instance_ip"],
"port": DEFAULT_PORT,
"username": "ubuntu",
"fs_factory": partial(_sshfs, dict(resource)),
}
@contextmanager
def get_sshfs( # pylint: disable=unused-argument
self, name: Optional[str] = None, **config
) -> Iterator["SSHFileSystem"]:
assert name
with self.make_tpi(name) as tpi:
resource = tpi.default_resource(name)
with _sshfs(resource) as fs:
yield fs
def rename(self, name: str, new: str, **config):
"""rename a dvc machine instance to a new name"""
import shutil
mtype = "iterative_machine"
new_dir = os.path.join(self.tmp_dir, new)
old_dir = os.path.join(self.tmp_dir, name)
if os.path.exists(new_dir):
raise DvcException(f"rename failed: path {new_dir} already exists")
if not os.path.exists(old_dir):
return
with self.make_tpi(name) as tpi:
tpi.state_mv(f"{mtype}.{name}", f"{mtype}.{new}", **config)
shutil.move(old_dir, new_dir)
+50 -24
View File
@@ -787,13 +787,18 @@ class Output:
self.hash_name,
ignore=self.dvcignore,
)
otransfer(
staging,
self.cache,
{obj.hash_info},
shallow=False,
hardlink=hardlink,
)
with Callback.as_tqdm_callback(
desc=f"Committing {self} to cache",
unit="file",
) as cb:
otransfer(
staging,
self.cache,
{obj.hash_info},
shallow=False,
hardlink=hardlink,
callback=cb,
)
if relink:
rel = self.fs.path.relpath(filter_info or self.fs_path)
with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as cb:
@@ -822,13 +827,18 @@ class Output:
save_obj = obj.filter(prefix)
assert isinstance(save_obj, Tree)
checkout_obj = save_obj.get_obj(self.cache, prefix)
otransfer(
staging,
self.cache,
{save_obj.hash_info} | {oid for _, _, oid in save_obj},
shallow=True,
hardlink=hardlink,
)
with Callback.as_tqdm_callback(
desc=f"Committing {self} to cache",
unit="file",
) as cb:
otransfer(
staging,
self.cache,
{save_obj.hash_info} | {oid for _, _, oid in save_obj},
shallow=True,
hardlink=hardlink,
callback=cb,
)
return checkout_obj
def dumpd(self, **kwargs): # noqa: C901, PLR0912
@@ -1048,14 +1058,19 @@ class Output:
upload=upload,
no_progress_bar=no_progress_bar,
)
otransfer(
staging,
odb,
{obj.hash_info},
jobs=jobs,
hardlink=False,
shallow=False,
)
with Callback.as_tqdm_callback(
desc=f"Transferring to {odb.fs.unstrip_protocol(odb.path)}",
unit="file",
) as cb:
otransfer(
staging,
odb,
{obj.hash_info},
jobs=jobs,
hardlink=False,
shallow=False,
callback=cb,
)
self.hash_info = obj.hash_info
self.files = None
@@ -1075,7 +1090,7 @@ class Output:
return len(obj) if obj else 0
def unprotect(self):
if self.exists:
if self.exists and self.use_cache:
self.cache.unprotect(self.fs_path)
def get_dir_cache(self, **kwargs) -> Optional["Tree"]:
@@ -1418,7 +1433,18 @@ class Output:
assert staging
assert obj.hash_info
otransfer(staging, self.cache, {obj.hash_info}, hardlink=relink, shallow=False)
with Callback.as_tqdm_callback(
desc=f"Adding {self} to cache",
unit="file",
) as cb:
otransfer(
staging,
self.cache,
{obj.hash_info},
hardlink=relink,
shallow=False,
callback=cb,
)
if relink:
with CheckoutCallback(
+125 -24
View File
@@ -2,6 +2,7 @@ import logging
import os
from collections.abc import Mapping, Sequence
from copy import deepcopy
from itertools import product
from typing import (
TYPE_CHECKING,
Any,
@@ -51,6 +52,7 @@ VARS_KWD = "vars"
WDIR_KWD = "wdir"
PARAMS_KWD = "params"
FOREACH_KWD = "foreach"
MATRIX_KWD = "matrix"
DO_KWD = "do"
DEFAULT_PARAMS_FILE = "params.yaml"
@@ -105,7 +107,7 @@ def is_map_or_seq(data: Any) -> bool:
return not isinstance(data, str) and _is_map_or_seq(data)
def split_foreach_name(name: str) -> Tuple[str, Optional[str]]:
def split_group_name(name: str) -> Tuple[str, Optional[str]]:
group, *keys = name.rsplit(JOIN, maxsplit=1)
return group, first(keys)
@@ -120,13 +122,15 @@ def check_interpolations(data: "DictStrAny", where: str, path: str):
return recurse(func)(data)
Definition = Union["ForeachDefinition", "EntryDefinition"]
Definition = Union["ForeachDefinition", "EntryDefinition", "MatrixDefinition"]
def make_definition(
resolver: "DataResolver", name: str, definition: "DictStrAny", **kwargs
) -> Definition:
args = resolver, resolver.context, name, definition
if MATRIX_KWD in definition:
return MatrixDefinition(*args, **kwargs)
if FOREACH_KWD in definition:
return ForeachDefinition(*args, **kwargs)
return EntryDefinition(*args, **kwargs)
@@ -159,7 +163,8 @@ class DataResolver:
self.tracked_vars: Dict[str, Mapping] = {}
stages_data = d.get(STAGES_KWD, {})
# we wrap the definitions into ForeachDefinition and EntryDefinition,
# we wrap the definitions into:
# ForeachDefinition, MatrixDefinition, and EntryDefinition
# that helps us to optimize, cache and selectively load each one of
# them as we need, and simplify all of this DSL/parsing logic.
self.definitions: Dict[str, Definition] = {
@@ -168,12 +173,13 @@ class DataResolver:
}
def resolve_one(self, name: str):
group, key = split_foreach_name(name)
group, key = split_group_name(name)
if not self._has_group_and_key(group, key):
raise EntryNotFound(f"Could not find '{name}'")
# all of the checks for `key` not being None for `ForeachDefinition`
# all of the checks for `key` not being None for
# `ForeachDefinition`/`MatrixDefinition`
# and/or `group` not existing in the `interim`, etc. should be
# handled by the `self.has_key()` above.
definition = self.definitions[group]
@@ -190,7 +196,7 @@ class DataResolver:
return {STAGES_KWD: data}
def has_key(self, key: str):
return self._has_group_and_key(*split_foreach_name(key))
return self._has_group_and_key(*split_group_name(key))
def _has_group_and_key(self, group: str, key: Optional[str] = None):
try:
@@ -198,16 +204,14 @@ class DataResolver:
except KeyError:
return False
if key:
return isinstance(definition, ForeachDefinition) and definition.has_member(
key
)
return not isinstance(definition, ForeachDefinition)
if not isinstance(definition, (ForeachDefinition, MatrixDefinition)):
return key is None
return key is not None and definition.has_member(key)
@collecting
def get_keys(self):
for name, definition in self.definitions.items():
if isinstance(definition, ForeachDefinition):
if isinstance(definition, (ForeachDefinition, MatrixDefinition)):
yield from definition.get_generated_names()
continue
yield name
@@ -256,8 +260,9 @@ class EntryDefinition:
name = self.name
if not skip_checks:
# we can check for syntax errors as we go for interpolated entries,
# but for foreach-generated ones, once is enough, which it does
# that itself. See `ForeachDefinition.do_definition`.
# but for foreach and matrix generated ones, once is enough, which it does
# that itself. See `ForeachDefinition.template`
# and `MatrixDefinition.template`.
check_syntax_errors(self.definition, name, self.relpath)
# we need to pop vars from generated/evaluated data
@@ -331,17 +336,18 @@ class ForeachDefinition:
self.name = name
assert DO_KWD in definition
assert MATRIX_KWD not in definition
self.foreach_data = definition[FOREACH_KWD]
self._do_definition = definition[DO_KWD]
self._template = definition[DO_KWD]
self.pair = IterationPair()
self.where = where
@cached_property
def do_definition(self):
def template(self):
# optimization: check for syntax errors only once for `foreach` stages
check_syntax_errors(self._do_definition, self.name, self.relpath)
return self._do_definition
check_syntax_errors(self._template, self.name, self.relpath)
return self._template
@cached_property
def resolved_iterable(self):
@@ -444,16 +450,111 @@ class ForeachDefinition:
# i.e. quadratic complexity).
generated = self._generate_name(key)
entry = EntryDefinition(
self.resolver, self.context, generated, self.do_definition
self.resolver, self.context, generated, self.template
)
try:
# optimization: skip checking for syntax errors on each foreach
# generated stages. We do it once when accessing do_definition.
# generated stages. We do it once when accessing template.
return entry.resolve_stage(skip_checks=True)
except ContextError as exc:
format_and_raise(exc, f"stage '{generated}'", self.relpath)
# let mypy know that this state is unreachable as format_and_raise
# does not return at all (it's not able to understand it for some
# reason)
raise AssertionError("unreachable")
class MatrixDefinition:
def __init__(
self,
resolver: DataResolver,
context: Context,
name: str,
definition: "DictStrAny",
where: str = STAGES_KWD,
):
self.resolver = resolver
self.relpath = self.resolver.relpath
self.context = context
self.name = name
assert MATRIX_KWD in definition
assert DO_KWD not in definition
assert FOREACH_KWD not in definition
self._template = definition.copy()
self.matrix_data = self._template.pop(MATRIX_KWD)
self.pair = IterationPair()
self.where = where
@cached_property
def template(self) -> "DictStrAny":
# optimization: check for syntax errors only once for `matrix` stages
check_syntax_errors(self._template, self.name, self.relpath)
return self._template
@cached_property
def resolved_iterable(self) -> Dict[str, List]:
return self._resolve_matrix_data()
def _resolve_matrix_data(self) -> Dict[str, List]:
try:
iterable = self.context.resolve(self.matrix_data, unwrap=False)
except (ContextError, ParseError) as exc:
format_and_raise(exc, f"'{self.where}.{self.name}.matrix'", self.relpath)
return iterable
@cached_property
def normalized_iterable(self) -> Dict[str, "DictStrAny"]:
"""Convert sequence to Mapping with keys normalized."""
iterable = self.resolved_iterable
assert isinstance(iterable, Mapping)
ret: Dict[str, "DictStrAny"] = {}
matrix = {key: enumerate(v) for key, v in iterable.items()}
for combination in product(*matrix.values()):
d: "DictStrAny" = {}
fragments: List[str] = []
for k, (i, v) in zip(matrix.keys(), combination):
d[k] = v
fragments.append(f"{k}{i}" if is_map_or_seq(v) else to_str(v))
key = "-".join(fragments)
ret[key] = d
return ret
def has_member(self, key: str) -> bool:
return key in self.normalized_iterable
def get_generated_names(self) -> List[str]:
return list(map(self._generate_name, self.normalized_iterable))
def _generate_name(self, key: str) -> str:
return f"{self.name}{JOIN}{key}"
def resolve_all(self) -> "DictStrAny":
return join(map(self.resolve_one, self.normalized_iterable))
def resolve_one(self, key: str) -> "DictStrAny":
return self._each_iter(key)
def _each_iter(self, key: str) -> "DictStrAny":
err_message = f"Could not find '{key}' in matrix group '{self.name}'"
with reraise(KeyError, EntryNotFound(err_message)):
value = self.normalized_iterable[key]
temp_dict = {self.pair.value: value}
with self.context.set_temporarily(temp_dict, reserve=True):
# optimization: item and key can be removed on __exit__() as they
# are top-level values, and are not merged recursively.
# This helps us avoid cloning context, which is slower
# (increasing the size of the context might increase
# the no. of items to be generated which means more cloning,
# i.e. quadratic complexity).
generated = self._generate_name(key)
entry = EntryDefinition(
self.resolver, self.context, generated, self.template
)
try:
# optimization: skip checking for syntax errors on each matrix
# generated stages. We do it once when accessing template.
return entry.resolve_stage(skip_checks=True)
except ContextError as exc:
format_and_raise(exc, f"stage '{generated}'", self.relpath)
+3 -13
View File
@@ -15,7 +15,6 @@ from typing import (
from dvc.exceptions import NotDvcRepoError, OutputNotFoundError
from dvc.ignore import DvcIgnoreFilter
from dvc.utils import env2bool
from dvc.utils.fs import path_isin
from dvc.utils.objects import cached_property
@@ -24,7 +23,6 @@ if TYPE_CHECKING:
from dvc.fs.data import DataFileSystem
from dvc.fs.dvc import DVCFileSystem
from dvc.lock import LockBase
from dvc.machine import MachineManager
from dvc.scm import Git, NoSCM
from dvc.stage import Stage
from dvc.types import DictStrAny
@@ -93,6 +91,7 @@ class Repo:
from dvc.repo.status import status # type: ignore[misc]
from dvc.repo.update import update # type: ignore[misc]
from .cache import check_missing as cache_check_missing # type: ignore[misc]
from .data import status as data_status # type: ignore[misc]
ls = staticmethod(_ls)
@@ -243,6 +242,7 @@ class Repo:
return Config(
self.dvc_dir,
local_dvc_dir=self.local_dvc_dir,
fs=self.fs,
config=self._config,
remote=self._remote,
@@ -350,16 +350,6 @@ class Repo:
return Experiments(self)
@cached_property
def machine(self) -> Optional["MachineManager"]:
from dvc.machine import MachineManager
if self.tmp_dir and (
self.config["feature"].get("machine", False) or env2bool("DVC_TEST")
):
return MachineManager(self)
return None
@property
def fs(self) -> "FileSystem":
return self._fs
@@ -497,7 +487,7 @@ class Repo:
used_run_cache=None,
revs=None,
num=1,
push: bool = False, # noqa: F811
push: bool = False,
):
"""Get the stages related to the given target and collect
the `info` of its outputs.
+31
View File
@@ -0,0 +1,31 @@
import os
def check_missing(repo, rev=None, max_size=None, types=None):
from dvc_data.index import StorageKeyError
with repo.switch(rev or "workspace"):
idx = repo.index.targets_view(
None,
max_size=max_size,
types=types,
)
index = idx.data["repo"]
def onerror(_entry, _exc):
pass
index.onerror = onerror
ret = []
for _, entry in index.iteritems():
try:
fs, path = index.storage_map.get_cache(entry)
except (StorageKeyError, ValueError):
continue
if not fs.exists(path):
ret.append(os.path.join(*entry.key))
return ret
+9 -13
View File
@@ -51,20 +51,16 @@ def _build_out_changes(index, changes):
out_keys.append(key)
out_changes = {}
for objects in changes.values():
for change in objects:
for out_key in out_keys:
if (
len(out_key) > len(change.key)
or change.key[: len(out_key)] != out_key
):
continue
for key, change in changes.items():
for out_key in out_keys:
if len(out_key) > len(key) or key[: len(out_key)] != out_key:
continue
if change.key == out_key:
out_changes[out_key] = change.typ
elif not out_changes.get(out_key):
out_changes[out_key] = MODIFY
break
if key == out_key:
out_changes[out_key] = change.typ
elif not out_changes.get(out_key):
out_changes[out_key] = MODIFY
break
return out_changes
-4
View File
@@ -136,10 +136,6 @@ class Experiments:
**kwargs,
) -> "QueueEntry":
"""Queue a single experiment."""
if kwargs.pop("machine", None) is not None:
# TODO: decide how to handle queued remote execution
raise NotImplementedError
return self.new(
queue,
**kwargs,
+4 -50
View File
@@ -23,6 +23,7 @@ from dvc.scm import Git, SCMError, iter_revs
from .exceptions import InvalidExpRefError
from .refs import EXEC_BRANCH, ExpRefInfo
from .serialize import ExpRange, ExpState, SerializableError, SerializableExp
from .utils import describe
if TYPE_CHECKING:
from dvc.repo import Repo
@@ -308,7 +309,9 @@ def collect(
if sha_only:
baseline_names: Dict[str, Optional[str]] = {}
else:
baseline_names = _describe(repo.scm, baseline_revs, refs=cached_refs)
baseline_names = describe(
repo.scm, baseline_revs, refs=cached_refs, logger=logger
)
workspace_data = collect_rev(repo, "workspace", **kwargs)
result: List["ExpState"] = [workspace_data]
@@ -337,55 +340,6 @@ def collect(
return result
def _describe(
scm: "Git",
revs: Iterable[str],
refs: Optional[Iterable[str]] = None,
) -> Dict[str, Optional[str]]:
"""Describe revisions using a tag, branch.
The first matching name will be returned for each rev. Names are preferred in this
order:
- current branch (if rev matches HEAD and HEAD is a branch)
- tags
- branches
Returns:
Dict mapping revisions from revs to a name.
"""
head_rev = scm.get_rev()
head_ref = scm.get_ref("HEAD", follow=False)
if head_ref and head_ref.startswith("refs/heads/"):
head_branch = head_ref[len("refs/heads/") :]
else:
head_branch = None
tags = {}
branches = {}
ref_it = iter(refs) if refs else scm.iter_refs()
for ref in ref_it:
is_tag = ref.startswith("refs/tags/")
is_branch = ref.startswith("refs/heads/")
if not (is_tag or is_branch):
continue
rev = scm.get_ref(ref)
if not rev:
logger.debug("unresolved ref %s", ref)
continue
if is_tag and rev not in tags:
tags[rev] = ref[len("refs/tags/") :]
if is_branch and rev not in branches:
branches[rev] = ref[len("refs/heads/") :]
names: Dict[str, Optional[str]] = {}
for rev in revs:
if rev == head_rev and head_branch:
names[rev] = head_branch
else:
names[rev] = tags.get(rev) or branches.get(rev)
return names
def _sorted_ranges(exp_ranges: Iterable["ExpRange"]) -> List["ExpRange"]:
"""Return list of ExpRange sorted by (timestamp, rev)."""
-292
View File
@@ -1,292 +0,0 @@
import logging
import os
import posixpath
import sys
from contextlib import contextmanager
from typing import TYPE_CHECKING, Callable, Iterable, List, Optional
from dvc_ssh import SSHFileSystem
from funcy import first
from dvc.repo.experiments.refs import (
EXEC_BASELINE,
EXEC_BRANCH,
EXEC_HEAD,
EXEC_MERGE,
EXEC_NAMESPACE,
)
from .base import BaseExecutor, ExecutorInfo, ExecutorResult, TaskStatus
if TYPE_CHECKING:
from queue import Queue
from dvc.repo import Repo
from dvc.repo.experiments.refs import ExpRefInfo
from dvc.repo.experiments.stash import ExpStashEntry
from dvc.scm import Git
logger = logging.getLogger(__name__)
@contextmanager
def _sshfs(fs_factory, **kwargs):
if fs_factory:
with fs_factory() as fs:
yield fs
return
yield SSHFileSystem(**kwargs)
class SSHExecutor(BaseExecutor):
"""SSH experiment executor."""
WARN_UNTRACKED = True
QUIET = True
SETUP_SCRIPT_FILENAME = "exec-setup.sh"
def __init__(
self,
*args,
host: Optional[str] = None,
port: Optional[int] = None,
username: Optional[str] = None,
fs_factory: Optional[Callable] = None,
setup_script: Optional[str] = None,
**kwargs,
):
assert host
super().__init__(*args, **kwargs)
self.host: str = host
self.port = port
self.username = username
self._fs_factory = fs_factory
self._repo_abspath = ""
self._setup_script = setup_script
@classmethod
def gen_dirname(cls, name: Optional[str] = None):
from shortuuid import uuid
return "-".join([name or "dvc-exp", "executor", uuid()])
@classmethod
def from_stash_entry(
cls,
repo: "Repo",
entry: "ExpStashEntry",
**kwargs,
):
machine_name: Optional[str] = kwargs.pop("machine_name", None)
assert repo.machine
executor = cls._from_stash_entry(
repo,
entry,
cls.gen_dirname(entry.name),
location=machine_name,
**repo.machine.get_executor_kwargs(machine_name),
setup_script=repo.machine.get_setup_script(machine_name),
)
logger.debug("Init SSH executor for host '%s'", executor.host)
return executor
def sshfs(self):
return _sshfs(self._fs_factory, host=self.host, port=self.port)
@property
def git_url(self) -> str:
user = f"{self.username}@" if self.username else ""
port = f":{self.port}" if self.port is not None else ""
path = f"{self.root_dir}" if self.root_dir else ""
if path and not posixpath.isabs(path):
path = f"/~/{path}"
return f"ssh://{user}{self.host}{port}{path}"
@property
def abs_url(self) -> str:
assert self._repo_abspath
user = f"{self.username}@" if self.username else ""
port = f":{self.port}" if self.port is not None else ""
return f"ssh://{user}{self.host}{port}{self._repo_abspath}"
@staticmethod
def _git_client_args(fs):
return {
"password": fs.fs_args.get("password"),
"key_filename": first(fs.fs_args.get("client_keys", [])),
}
def init_git(
self,
repo: "Repo", # noqa: ARG002
scm: "Git",
stash_rev: str,
entry: "ExpStashEntry",
infofile: Optional[str],
branch: Optional[str] = None,
):
from dvc.repo.experiments.utils import push_refspec
self.status = TaskStatus.PREPARING
if infofile:
self.info.dump_json(infofile)
with self.sshfs() as fs:
fs.makedirs(self.root_dir)
self._ssh_cmd(fs, "git init .")
self._ssh_cmd(fs, "git config user.name dvc-exp")
self._ssh_cmd(fs, "git config user.email [email protected]")
result = self._ssh_cmd(fs, "pwd")
path = result.stdout.strip()
self._repo_abspath = path
# TODO: support multiple client key retries in git backends
# (see https://github.com/iterative/dvc/issues/6508)
kwargs = self._git_client_args(fs)
ref_dict = {
EXEC_HEAD: entry.head_rev,
EXEC_MERGE: stash_rev,
EXEC_BASELINE: entry.baseline_rev,
}
with self.set_temp_refs(scm, ref_dict):
exec_namespace = f"{EXEC_NAMESPACE}/"
refspec = [(exec_namespace, exec_namespace)]
push_refspec(scm, self.git_url, refspec, **kwargs)
if branch:
push_refspec(scm, self.git_url, [(branch, branch)], **kwargs)
self._ssh_cmd(fs, f"git symbolic-ref {EXEC_BRANCH} {branch}")
else:
self._ssh_cmd(fs, f"git symbolic-ref -d {EXEC_BRANCH}", check=False)
# checkout EXEC_HEAD and apply EXEC_MERGE on top of it without
# committing
head = EXEC_BRANCH if branch else EXEC_HEAD
self._ssh_cmd(fs, f"git checkout {head}")
merge_rev = scm.get_ref(EXEC_MERGE)
self._ssh_cmd(fs, f"git stash apply {merge_rev}")
if self._setup_script:
self._init_setup_script(fs)
@classmethod
def _setup_script_path(cls, dvc_dir: str):
return posixpath.join(
dvc_dir,
"tmp",
cls.SETUP_SCRIPT_FILENAME,
)
def _init_setup_script(self, fs: "SSHFileSystem"):
assert self._repo_abspath
script_path = self._setup_script_path(
posixpath.join(self._repo_abspath, self.dvc_dir)
)
assert self._setup_script
fs.put_file(self._setup_script, script_path)
def _ssh_cmd(self, sshfs, cmd, chdir=None, **kwargs):
working_dir = chdir or self.root_dir
return sshfs.fs.execute(f"cd {working_dir};{cmd}", **kwargs)
def init_cache(self, repo: "Repo", rev: str, run_cache: bool = True):
from dvc.repo.push import push
with self.get_odb() as odb:
push(
repo,
revs=[rev],
run_cache=run_cache,
odb=odb,
include_imports=True,
)
def collect_cache(
self, repo: "Repo", exp_ref: "ExpRefInfo", run_cache: bool = True
):
"""Collect DVC cache."""
from dvc.repo.experiments.pull import _pull_cache
_pull_cache(repo, exp_ref, run_cache=run_cache)
@contextmanager
def get_odb(self):
from dvc.cachemgr import CacheManager, get_odb
cache_path = posixpath.join(
self._repo_abspath,
self.dvc_dir,
CacheManager.CACHE_DIR,
)
with self.sshfs() as fs:
yield get_odb(fs, cache_path, **fs.config)
def fetch_exps(self, *args, **kwargs) -> Iterable[str]:
with self.sshfs() as fs:
kwargs.update(self._git_client_args(fs))
return super().fetch_exps(*args, **kwargs)
@classmethod
def reproduce(
cls,
info: "ExecutorInfo",
rev: str, # noqa: ARG003
queue: Optional["Queue"] = None, # noqa: ARG003
infofile: Optional[str] = None,
log_errors: bool = True,
log_level: Optional[int] = None,
copy_paths: Optional[List[str]] = None, # noqa: ARG003
message: Optional[str] = None, # noqa: ARG003
**kwargs,
) -> "ExecutorResult":
"""Reproduce an experiment on a remote machine over SSH.
Internally uses 'dvc exp exec-run' over SSH.
"""
import json
import time
from tempfile import TemporaryFile
from asyncssh import ProcessError
fs_factory: Optional[Callable] = kwargs.pop("fs_factory", None)
if log_errors and log_level is not None:
cls._set_log_level(log_level)
with _sshfs(fs_factory) as fs:
while not fs.exists("/var/log/dvc-machine-init.log"):
logger.info("Waiting for dvc-machine startup script to complete...")
time.sleep(5)
logger.info("Reproducing experiment on '%s'", fs.fs_args.get("host"))
with TemporaryFile(mode="w+", encoding="utf-8") as fobj:
json.dump(info.asdict(), fobj)
fobj.seek(0)
fs.put_file(fobj, infofile)
cmd = ["source ~/.profile"]
script_path = cls._setup_script_path(info.dvc_dir)
if fs.exists(posixpath.join(info.root_dir, script_path)):
cmd.extend([f"pushd {info.root_dir}", f"source {script_path}", "popd"])
exec_cmd = f"dvc exp exec-run --infofile {infofile}"
if log_level is not None:
if log_level <= logging.TRACE: # type: ignore[attr-defined]
exec_cmd += " -vv"
elif log_level <= logging.DEBUG:
exec_cmd += " -v"
cmd.append(exec_cmd)
try:
sys.stdout.flush()
sys.stderr.flush()
stdout = os.dup(sys.stdout.fileno())
stderr = os.dup(sys.stderr.fileno())
fs.fs.execute("; ".join(cmd), stdout=stdout, stderr=stderr)
with fs.open(infofile) as fobj:
result_info = ExecutorInfo.from_dict(json.load(fobj))
if result_info.result_hash:
return result_info.result
except ProcessError:
pass
return ExecutorResult(None, None, False)
+2 -10
View File
@@ -31,23 +31,15 @@ def ls(
rev = [rev]
revs = iter_revs(repo.scm, rev, num)
rev_set = set(revs.keys())
ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set, git_remote)
tags = repo.scm.describe(ref_info_dict.keys())
remained = {baseline for baseline, tag in tags.items() if tag is None}
base = "refs/heads"
ref_heads = repo.scm.describe(remained, base=base)
results = defaultdict(list)
for baseline in ref_info_dict:
name = baseline
if tags[baseline] or ref_heads[baseline]:
name = tags[baseline] or ref_heads[baseline]
for info in ref_info_dict[baseline]:
if git_remote:
exp_rev = None
else:
exp_rev = repo.scm.get_ref(str(info))
results[name].append((info.name, exp_rev))
results[baseline].append((info.name, exp_rev))
return results
+7 -3
View File
@@ -476,13 +476,17 @@ class BaseStashQueue(ABC):
hydra_output_file = ParamsDependency.DEFAULT_PARAMS_FILE
for path, overrides in params.items():
if hydra_enabled and path == hydra_output_file:
config_dir = os.path.join(
self.repo.root_dir, hydra_config.get("config_dir", "conf")
)
if (config_module := hydra_config.get("config_module")) is None:
config_dir = os.path.join(
self.repo.root_dir, hydra_config.get("config_dir", "conf")
)
else:
config_dir = None
config_name = hydra_config.get("config_name", "config")
compose_and_dump(
path,
config_dir,
config_module,
config_name,
overrides,
)
File diff suppressed because one or more lines are too long
+19 -2
View File
@@ -9,6 +9,14 @@ from . import locked
logger = logging.getLogger(__name__)
def _make_index_onerror(onerror, rev):
def _onerror(entry, exc):
if onerror:
return onerror(rev, entry, exc)
return _onerror
def _collect_indexes( # noqa: PLR0913
repo,
targets=None,
@@ -22,6 +30,7 @@ def _collect_indexes( # noqa: PLR0913
max_size=None,
types=None,
config=None,
onerror=None,
):
indexes = {}
collection_exc = None
@@ -39,7 +48,7 @@ def _collect_indexes( # noqa: PLR0913
all_commits=all_commits,
):
try:
repo.config.update(config)
repo.config.merge(config)
idx = repo.index.targets_view(
targets,
@@ -48,8 +57,14 @@ def _collect_indexes( # noqa: PLR0913
max_size=max_size,
types=types,
)
indexes[idx.data_tree.hash_info.value] = idx.data["repo"]
data = idx.data["repo"]
data.onerror = _make_index_onerror(onerror, rev)
indexes[idx.data_tree.hash_info.value] = data
except Exception as exc: # pylint: disable=broad-except
if onerror:
onerror(rev, None, exc)
collection_exc = exc
logger.exception("failed to collect '%s'", rev or "workspace")
@@ -75,6 +90,7 @@ def fetch( # noqa: C901, PLR0913
max_size=None,
types=None,
config=None,
onerror=None,
) -> int:
"""Download data items from a cloud and imported repositories
@@ -119,6 +135,7 @@ def fetch( # noqa: C901, PLR0913
max_size=max_size,
types=types,
config=config,
onerror=onerror,
)
cache_key = ("fetch", tokenize(sorted(indexes.keys())))
+13 -1
View File
@@ -19,7 +19,17 @@ class GetDVCFileError(DvcException):
)
def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
def get(
url,
path,
out=None,
rev=None,
jobs=None,
force=False,
config=None,
remote=None,
remote_config=None,
):
from dvc.config import Config
from dvc.dvcfile import is_valid_filename
from dvc.repo import Repo
@@ -38,6 +48,8 @@ def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
subrepos=True,
uninitialized=True,
config=config,
remote=remote,
remote_config=remote_config,
) as repo:
from dvc.fs import download
from dvc.fs.data import DataFileSystem
+24 -1
View File
@@ -1,4 +1,14 @@
def imp(self, url, path, out=None, rev=None, config=None, **kwargs):
def imp(
self,
url,
path,
out=None,
rev=None,
config=None,
remote=None,
remote_config=None,
**kwargs,
):
erepo = {"url": url}
if rev is not None:
erepo["rev"] = rev
@@ -6,4 +16,17 @@ def imp(self, url, path, out=None, rev=None, config=None, **kwargs):
if config is not None:
erepo["config"] = config
if remote is not None and remote_config is not None:
conf = erepo.get("config") or {}
remotes = conf.get("remote") or {}
remote_conf = remotes.get(remote) or {}
remote_conf.update(remote_config)
remotes[remote] = remote_conf
conf["remote"] = remotes
erepo["config"] = conf
elif remote is not None:
erepo["remote"] = remote
elif remote_config is not None:
erepo["remote"] = remote_config
return self.imp_url(path, out=out, erepo=erepo, frozen=True, **kwargs)
+1 -1
View File
@@ -413,7 +413,7 @@ class Index:
by_workspace[workspace].add(key)
for path in self._plot_sources:
key = self.repo.fs.path.relparts(path, self.repo.root_dir)
key = self.repo.fs.path.parts(path)
by_workspace["repo"].add(key)
return dict(by_workspace)
+12 -2
View File
@@ -14,6 +14,8 @@ def ls(
recursive: Optional[bool] = None,
dvc_only: bool = False,
config: Optional[str] = None,
remote: Optional[str] = None,
remote_config: Optional[dict] = None,
):
"""Methods for getting files and outputs for the repo.
@@ -23,7 +25,9 @@ def ls(
rev (str, optional): SHA commit, branch or tag name
recursive (bool, optional): recursively walk the repo
dvc_only (bool, optional): show only DVC-artifacts
config (bool, optional): path to config file
config (str, optional): path to config file
remote (str, optional): remote name to set as a default remote in the repo
remote_config (str, dict): remote config to merge with a remote in the repo
Returns:
list of `entry`
@@ -47,7 +51,13 @@ def ls(
config_dict = None
with Repo.open(
url, rev=rev, subrepos=True, uninitialized=True, config=config_dict
url,
rev=rev,
subrepos=True,
uninitialized=True,
config=config_dict,
remote=remote,
remote_config=remote_config,
) as repo:
path = path or ""
+1 -1
View File
@@ -483,7 +483,7 @@ def _collect_pipeline_files(repo, targets: List[str], props, onerror=None):
if isinstance(elem, str):
dvcfile_defs_dict[elem] = None
else:
k, v = list(elem.items())[0]
k, v = next(iter(elem.items()))
dvcfile_defs_dict[k] = v
resolved = _resolve_definitions(
+1 -1
View File
@@ -243,7 +243,7 @@ class StageLoad:
) -> Iterable[str]:
if not name:
return stages.keys()
if accept_group and stages.is_foreach_generated(name):
if accept_group and stages.is_foreach_or_matrix_generated(name):
return self._get_group_keys(stages, name)
if glob:
return fnmatch.filter(stages.keys(), name)
+2 -2
View File
@@ -166,7 +166,7 @@ def push_worktree(
latest_only=remote_obj.worktree,
jobs=jobs,
)
pushed += sum(len(changes) for changes in diff.changes.values())
pushed += len(diff.files_create)
except VersioningNotSupported:
logger.exception("")
raise DvcException(
@@ -427,7 +427,7 @@ def _get_update_diff_index(
meta_cmp_key=partial(_meta_checksum, remote.fs),
with_unchanged=True,
):
if change.typ == ADD or change.typ == MODIFY:
if change.typ in (ADD, MODIFY):
entry = change.new
# preserve md5's which were calculated in out.save() after
# downloading
+2 -1
View File
@@ -11,7 +11,7 @@ from dvc.output import (
META_SCHEMA,
Output,
)
from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
from dvc.parsing import DO_KWD, FOREACH_KWD, MATRIX_KWD, VARS_KWD
from dvc.stage.params import StageParams
STAGES = "stages"
@@ -77,6 +77,7 @@ PARAM_PSTAGE_NON_DEFAULT_SCHEMA = {str: [str]}
VARS_SCHEMA = [str, dict]
STAGE_DEFINITION = {
MATRIX_KWD: {str: Any(str, list)},
Required(StageParams.PARAM_CMD): Any(str, list),
Optional(StageParams.PARAM_WDIR): str,
Optional(StageParams.PARAM_DEPS): [str],
+1 -1
View File
@@ -67,7 +67,7 @@ def map_scm_exception(with_cause: bool = False) -> Iterator[None]:
into = SCMError(str(exc))
if with_cause:
raise into from exc
raise into
raise into # noqa: B904
@overload
+6 -3
View File
@@ -7,7 +7,7 @@ from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple
from funcy import get_in, lcat, once, project
from dvc import dependency, output
from dvc.parsing import FOREACH_KWD, JOIN, EntryNotFound
from dvc.parsing import FOREACH_KWD, JOIN, MATRIX_KWD, EntryNotFound
from dvc.utils.objects import cached_property
from dvc_data.hashfile.meta import Meta
@@ -167,8 +167,11 @@ class StageLoader(Mapping):
def __contains__(self, name):
return self.resolver.has_key(name) # noqa: W601
def is_foreach_generated(self, name: str):
return name in self.stages_data and FOREACH_KWD in self.stages_data[name]
def is_foreach_or_matrix_generated(self, name: str) -> bool:
return (
name in self.stages_data
and {FOREACH_KWD, MATRIX_KWD} & self.stages_data[name].keys()
)
class SingleStageLoader(Mapping):
+1 -1
View File
@@ -1 +1 @@
from .fixtures import * # noqa, pylint: disable=wildcard-import
from .fixtures import * # noqa: F403 # pylint: disable=wildcard-import
+1 -1
View File
@@ -1 +1 @@
from .fixtures import * # noqa, pylint: disable=wildcard-import
from .fixtures import * # noqa: F403, pylint: disable=wildcard-import
+2 -2
View File
@@ -1,5 +1,5 @@
from .benchmarks.fixtures import * # noqa, pylint: disable=wildcard-import
from .fixtures import * # noqa, pylint: disable=wildcard-import
from .benchmarks.fixtures import * # noqa: F403, pylint: disable=wildcard-import
from .fixtures import * # noqa: F403, pylint: disable=wildcard-import
def pytest_generate_tests(metafunc):
+16 -4
View File
@@ -1,6 +1,6 @@
import logging
from pathlib import Path
from typing import TYPE_CHECKING, List
from typing import TYPE_CHECKING, List, Optional
from dvc.exceptions import InvalidArgumentError
@@ -15,7 +15,8 @@ logger = logging.getLogger(__name__)
def compose_and_dump(
output_file: "StrPath",
config_dir: str,
config_dir: Optional[str],
config_module: Optional[str],
config_name: str,
overrides: List[str],
) -> None:
@@ -25,6 +26,8 @@ def compose_and_dump(
output_file: File where the composed config will be dumped.
config_dir: Folder containing the Hydra config files.
Must be absolute file system path.
config_module: Module containing the Hydra config files.
Ignored if `config_dir` is not `None`.
config_name: Name of the config file containing defaults,
without the .yaml extension.
overrides: List of `Hydra Override`_ patterns.
@@ -32,12 +35,21 @@ def compose_and_dump(
.. _Hydra Override:
https://hydra.cc/docs/advanced/override_grammar/basic/
"""
from hydra import compose, initialize_config_dir
from hydra import compose, initialize_config_dir, initialize_config_module
from omegaconf import OmegaConf
from .serialize import DUMPERS
with initialize_config_dir(config_dir, version_base=None):
config_source = config_dir or config_module
if not config_source:
raise ValueError("Either `config_dir` or `config_module` should be provided.")
initialize_config = (
initialize_config_dir if config_dir else initialize_config_module
)
with initialize_config( # type: ignore[attr-defined]
config_source, version_base=None
):
cfg = compose(config_name=config_name, overrides=overrides)
OmegaConf.resolve(cfg)
+5 -5
View File
@@ -1,11 +1,11 @@
from collections import defaultdict
from typing import DefaultDict
from ._common import * # noqa, pylint: disable=wildcard-import
from ._json import * # noqa, pylint: disable=wildcard-import
from ._py import * # noqa, pylint: disable=wildcard-import
from ._toml import * # noqa, pylint: disable=wildcard-import
from ._yaml import * # noqa, pylint: disable=wildcard-import
from ._common import * # noqa: F403, pylint: disable=wildcard-import
from ._json import * # noqa: F403, pylint: disable=wildcard-import
from ._py import * # noqa: F403, pylint: disable=wildcard-import
from ._toml import * # noqa: F403, pylint: disable=wildcard-import
from ._yaml import * # noqa: F403, pylint: disable=wildcard-import
LOADERS: DefaultDict[str, LoaderFn] = defaultdict( # noqa: F405
lambda: load_yaml # noqa: F405
+5 -6
View File
@@ -34,7 +34,7 @@ dependencies = [
"configobj>=5.0.6",
"distro>=1.3",
"dpath<3,>=2.1.0",
"dvc-data>=2.8.1,<2.9.0",
"dvc-data>=2.14.0,<2.15.0",
"dvc-http>=2.29.0",
"dvc-render>=0.3.1,<1",
"dvc-studio-client>=0.9.2,<1",
@@ -57,7 +57,7 @@ dependencies = [
"requests>=2.22",
"rich>=12",
"ruamel.yaml>=0.17.11",
"scmrepo>=1.0.4,<2",
"scmrepo>=1.2.1,<2",
"shortuuid>=0.5",
"shtab<2,>=1.3.4",
"tabulate>=0.8.7",
@@ -70,13 +70,13 @@ dependencies = [
[project.optional-dependencies]
all = ["dvc[azure,gdrive,gs,hdfs,oss,s3,ssh,webdav,webhdfs]"]
azure = ["dvc-azure>=2.21.2"]
dev = ["dvc[azure,gdrive,gs,hdfs,lint,oss,s3,ssh,terraform,tests,webdav,webhdfs]"]
dev = ["dvc[azure,gdrive,gs,hdfs,lint,oss,s3,ssh,tests,webdav,webhdfs]"]
gdrive = ["dvc-gdrive==2.20"]
gs = ["dvc-gs==2.22.1"]
hdfs = ["dvc-hdfs==2.19"]
lint = [
"mypy==1.4.1",
"pylint==2.17.4",
"pylint==2.17.5",
"types-colorama",
"types-psutil",
"types-pyinstaller",
@@ -89,7 +89,6 @@ oss = ["dvc-oss==2.19"]
s3 = ["dvc-s3==2.23.0"]
ssh = ["dvc-ssh>=2.22.1,<3"]
ssh_gssapi = ["dvc-ssh[gssapi]>=2.22.1,<3"]
terraform = ["tpi[ssh]>=2.1"]
testing = [
"pytest-test-utils",
"pytest-benchmark[histogram]",
@@ -300,7 +299,7 @@ ignore-variadic-names = true
known-first-party = ["dvc", "dvc_data", "dvc_objects", "dvc_render", "dvc_task", "tests"]
[tool.ruff.pep8-naming]
ignore-names = ["M", "SCM"]
extend-ignore-names = ["M", "SCM"]
[tool.ruff.per-file-ignores]
"dvc/commands/**" = ["N806"]
+4 -4
View File
@@ -7,11 +7,11 @@ import pytest
from dvc import env
from dvc.stage import PipelineStage
from dvc.testing.fixtures import * # noqa, pylint: disable=wildcard-import
from dvc.testing.fixtures import * # noqa: F403, pylint: disable=wildcard-import
from .dir_helpers import * # noqa, pylint: disable=wildcard-import
from .remotes import * # noqa, pylint: disable=wildcard-import
from .scripts import * # noqa, pylint: disable=wildcard-import
from .dir_helpers import * # noqa: F403 # pylint: disable=wildcard-import
from .remotes import * # noqa: F403 # pylint: disable=wildcard-import
from .scripts import * # noqa: F403, pylint: disable=wildcard-import
# Prevent updater and analytics from running their processes
os.environ["DVC_TEST"] = "true"
+1 -1
View File
@@ -5,7 +5,7 @@ from funcy import first, get_in
from dvc import api
from dvc.exceptions import OutputNotFoundError, PathMissingError
from dvc.testing.api_tests import TestAPI # noqa, pylint: disable=unused-import
from dvc.testing.api_tests import TestAPI # noqa: F401, pylint: disable=unused-import
from dvc.testing.tmp_dir import make_subrepo
from dvc.utils.fs import remove
+2 -1
View File
@@ -1,6 +1,7 @@
import pytest
from tests.unit.repo.experiments.conftest import ( # noqa, pylint disable=unused-argument
# pylint: disable-next=unused-argument
from tests.unit.repo.experiments.conftest import ( # noqa: F401
exp_stage,
failed_exp_stage,
session_app,
-160
View File
@@ -1,160 +0,0 @@
import posixpath
from contextlib import contextmanager
from functools import partial
from urllib.parse import urlparse
import pytest
from dvc_ssh import SSHFileSystem
from dvc_ssh.tests.cloud import TEST_SSH_KEY_PATH, TEST_SSH_USER
from dvc.repo.experiments.executor.base import ExecutorInfo, ExecutorResult
from dvc.repo.experiments.executor.ssh import SSHExecutor
from dvc.repo.experiments.refs import EXEC_HEAD, EXEC_MERGE
from tests.func.machine.conftest import * # noqa, pylint: disable=wildcard-import
@contextmanager
def _ssh_factory(cloud):
yield SSHFileSystem(
host=cloud.host,
port=cloud.port,
user=TEST_SSH_USER,
keyfile=TEST_SSH_KEY_PATH,
)
def test_init_from_stash(tmp_dir, scm, dvc, machine_instance, mocker):
mock = mocker.patch.object(SSHExecutor, "_from_stash_entry")
mock_entry = mocker.Mock()
mock_entry.name = ""
SSHExecutor.from_stash_entry(
dvc,
mock_entry,
machine_name="foo",
)
_args, kwargs = mock.call_args
assert kwargs["host"] == machine_instance["instance_ip"]
@pytest.mark.needs_internet
@pytest.mark.parametrize("cloud", [pytest.lazy_fixture("git_ssh")])
def test_init_git(tmp_dir, dvc, scm, cloud, mocker):
tmp_dir.scm_gen({"foo": "foo", "dir": {"bar": "bar"}}, commit="init")
baseline_rev = scm.get_rev()
tmp_dir.gen("foo", "stashed")
scm.gitpython.git.stash()
rev = scm.resolve_rev("stash@{0}")
mock = mocker.Mock(baseline_rev=baseline_rev, head_rev=baseline_rev)
root_url = cloud / SSHExecutor.gen_dirname()
executor = SSHExecutor(
root_dir=root_url.path,
dvc_dir=".dvc",
baseline_rev=baseline_rev,
host=root_url.host,
port=root_url.port,
username=TEST_SSH_USER,
fs_factory=partial(_ssh_factory, cloud),
)
infofile = str((root_url / "foo.run").path)
executor.init_git(dvc, scm, rev, mock, infofile=infofile)
assert root_url.path == executor._repo_abspath
fs = cloud._ssh
assert fs.exists(posixpath.join(executor._repo_abspath, "foo"))
assert fs.exists(posixpath.join(executor._repo_abspath, "dir"))
assert fs.exists(posixpath.join(executor._repo_abspath, "dir", "bar"))
@pytest.mark.needs_internet
@pytest.mark.parametrize("cloud", [pytest.lazy_fixture("git_ssh")])
def test_init_cache(tmp_dir, dvc, scm, cloud):
foo = tmp_dir.dvc_gen("foo", "foo", commit="init")[0].outs[0]
rev = scm.get_rev()
scm.set_ref(EXEC_HEAD, rev)
scm.set_ref(EXEC_MERGE, rev)
root_url = cloud / SSHExecutor.gen_dirname()
executor = SSHExecutor(
root_dir=root_url.path,
dvc_dir=".dvc",
baseline_rev=rev,
host=root_url.host,
port=root_url.port,
username=TEST_SSH_USER,
fs_factory=partial(_ssh_factory, cloud),
)
executor.init_cache(dvc, rev)
fs = cloud._ssh
foo_hash = foo.hash_info.value
assert fs.exists(
posixpath.join(
executor._repo_abspath, ".dvc", "cache", foo_hash[:2], foo_hash[2:]
)
)
@pytest.mark.needs_internet
@pytest.mark.parametrize("cloud", [pytest.lazy_fixture("git_ssh")])
def test_reproduce(tmp_dir, scm, dvc, cloud, exp_stage, mocker):
from sshfs import SSHFileSystem as _sshfs # noqa: N813
rev = scm.get_rev()
root_url = cloud / SSHExecutor.gen_dirname()
mocker.patch.object(SSHFileSystem, "exists", return_value=True)
mock_execute = mocker.patch.object(_sshfs, "execute")
info = ExecutorInfo(
str(root_url),
rev,
"machine-foo",
str(root_url.path),
".dvc",
)
infofile = str((root_url / "foo.run").path)
SSHExecutor.reproduce(
info,
rev,
fs_factory=partial(_ssh_factory, cloud),
)
mock_execute.assert_called_once()
_name, args, _kwargs = mock_execute.mock_calls[0]
assert f"dvc exp exec-run --infofile {infofile}" in args[0]
@pytest.mark.needs_internet
@pytest.mark.parametrize("cloud", [pytest.lazy_fixture("git_ssh")])
def test_run_machine(tmp_dir, scm, dvc, cloud, exp_stage, mocker):
baseline = scm.get_rev()
factory = partial(_ssh_factory, cloud)
mocker.patch.object(
dvc.machine,
"get_executor_kwargs",
return_value={
"host": cloud.host,
"port": cloud.port,
"username": TEST_SSH_USER,
"fs_factory": factory,
},
)
mocker.patch.object(dvc.machine, "get_setup_script", return_value=None)
mock_repro = mocker.patch.object(
SSHExecutor,
"reproduce",
return_value=ExecutorResult("abc123", None, False),
)
tmp_dir.gen("params.yaml", "foo: 2")
dvc.experiments.run(exp_stage.addressing, machine="foo")
mock_repro.assert_called_once()
_name, _args, kwargs = mock_repro.mock_calls[0]
info = kwargs["info"]
url = urlparse(info.git_url)
assert url.scheme == "ssh"
assert url.hostname == cloud.host
assert url.port == cloud.port
assert info.baseline_rev == baseline
assert kwargs["infofile"] is not None
assert kwargs["fs_factory"] is not None
+14 -11
View File
@@ -348,7 +348,8 @@ def test_packed_args_exists(tmp_dir, scm, dvc, exp_stage, caplog):
def test_list(tmp_dir, scm, dvc, exp_stage):
baseline_a = scm.get_rev()
baseline_old = scm.get_rev()
results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
exp_a = first(results)
ref_info_a = first(exp_refs_by_rev(scm, exp_a))
@@ -358,34 +359,36 @@ def test_list(tmp_dir, scm, dvc, exp_stage):
ref_info_b = first(exp_refs_by_rev(scm, exp_b))
tmp_dir.scm_gen("new", "new", commit="new")
baseline_new = scm.get_rev()
results = dvc.experiments.run(exp_stage.addressing, params=["foo=4"])
exp_c = first(results)
ref_info_c = first(exp_refs_by_rev(scm, exp_c))
assert dvc.experiments.ls() == {"refs/heads/master": [(ref_info_c.name, exp_c)]}
assert dvc.experiments.ls() == {baseline_new: [(ref_info_c.name, exp_c)]}
exp_list = dvc.experiments.ls(rev=ref_info_a.baseline_sha)
assert {key: set(val) for key, val in exp_list.items()} == {
baseline_a: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)}
baseline_old: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)}
}
exp_list = dvc.experiments.ls(rev=[baseline_a, scm.get_rev()])
exp_list = dvc.experiments.ls(rev=[baseline_old, baseline_new])
assert {key: set(val) for key, val in exp_list.items()} == {
baseline_a: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)},
"refs/heads/master": {(ref_info_c.name, exp_c)},
baseline_old: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)},
baseline_new: {(ref_info_c.name, exp_c)},
}
exp_list = dvc.experiments.ls(all_commits=True)
assert {key: set(val) for key, val in exp_list.items()} == {
baseline_a: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)},
"refs/heads/master": {(ref_info_c.name, exp_c)},
baseline_old: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)},
baseline_new: {(ref_info_c.name, exp_c)},
}
scm.checkout("branch", True)
exp_list = dvc.experiments.ls(all_commits=True)
assert {key: set(val) for key, val in exp_list.items()} == {
baseline_a: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)},
"refs/heads/branch": {(ref_info_c.name, exp_c)},
baseline_old: {(ref_info_a.name, exp_a), (ref_info_b.name, exp_b)},
baseline_new: {(ref_info_c.name, exp_c)},
}
@@ -666,7 +669,7 @@ def test_experiment_unchanged(tmp_dir, scm, dvc, exp_stage):
dvc.experiments.run(exp_stage.addressing)
dvc.experiments.run(exp_stage.addressing)
assert len(dvc.experiments.ls()["refs/heads/master"]) == 2
assert len(dvc.experiments.ls()[scm.get_rev()]) == 2
def test_experiment_run_dry(tmp_dir, scm, dvc, exp_stage):
+7 -5
View File
@@ -139,7 +139,7 @@ def test_push_ambiguous_name(tmp_dir, scm, dvc, git_upstream, exp_stage):
@pytest.mark.parametrize("use_url", [True, False])
def test_list_remote(tmp_dir, scm, dvc, git_downstream, exp_stage, use_url):
baseline_a = scm.get_rev()
baseline_old = scm.get_rev()
results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
exp_a = first(results)
ref_info_a = first(exp_refs_by_rev(scm, exp_a))
@@ -149,6 +149,8 @@ def test_list_remote(tmp_dir, scm, dvc, git_downstream, exp_stage, use_url):
ref_info_b = first(exp_refs_by_rev(scm, exp_b))
tmp_dir.scm_gen("new", "new", commit="new")
baseline_new = scm.get_rev()
results = dvc.experiments.run(exp_stage.addressing, params=["foo=4"])
exp_c = first(results)
ref_info_c = first(exp_refs_by_rev(scm, exp_c))
@@ -160,15 +162,15 @@ def test_list_remote(tmp_dir, scm, dvc, git_downstream, exp_stage, use_url):
assert downstream_exp.ls(git_remote=remote) == {}
git_downstream.tmp_dir.scm.fetch_refspecs(remote, ["master:master"])
exp_list = downstream_exp.ls(rev=baseline_a, git_remote=remote)
exp_list = downstream_exp.ls(rev=baseline_old, git_remote=remote)
assert {key: set(val) for key, val in exp_list.items()} == {
baseline_a: {(ref_info_a.name, None), (ref_info_b.name, None)}
baseline_old: {(ref_info_a.name, None), (ref_info_b.name, None)}
}
exp_list = downstream_exp.ls(all_commits=True, git_remote=remote)
assert {key: set(val) for key, val in exp_list.items()} == {
baseline_a: {(ref_info_a.name, None), (ref_info_b.name, None)},
"refs/heads/master": {(ref_info_c.name, None)},
baseline_old: {(ref_info_a.name, None), (ref_info_b.name, None)},
baseline_new: {(ref_info_c.name, None)},
}
+2 -1
View File
@@ -67,11 +67,12 @@ def test_exp_save_after_commit(tmp_dir, dvc, scm):
dvc.experiments.save(name="exp-1", force=True)
tmp_dir.scm_gen({"new_file": "new_file"}, commit="new baseline")
baseline_new = scm.get_rev()
dvc.experiments.save(name="exp-2", force=True)
all_exps = dvc.experiments.ls(all_commits=True)
assert all_exps[baseline][0][0] == "exp-1"
assert all_exps["refs/heads/master"][0][0] == "exp-2"
assert all_exps[baseline_new][0][0] == "exp-2"
def test_exp_save_with_staged_changes(tmp_dir, dvc, scm):
View File
-55
View File
@@ -1,55 +0,0 @@
import textwrap
import pytest
BASIC_CONFIG = textwrap.dedent(
"""\
[feature]
machine = true
['machine "foo"']
cloud = aws
"""
)
TEST_INSTANCE = {
"aws_security_group": None,
"cloud": "aws",
"id": "iterative-2jyhw8j9ieov6",
"image": "ubuntu-bionic-18.04-amd64-server-20210818",
"instance_gpu": None,
"instance_hdd_size": 35,
"instance_ip": "123.123.123.123",
"instance_launch_time": "2021-08-25T07:13:03Z",
"instance_type": "m",
"name": "test-resource",
"region": "us-west",
"spot": False,
"spot_price": -1,
"ssh_name": None,
"ssh_private": "-----BEGIN RSA PRIVATE KEY-----\\n",
"startup_script": "IyEvYmluL2Jhc2g=",
"timeouts": None,
}
@pytest.fixture
def machine_config(tmp_dir):
(tmp_dir / ".dvc" / "config").write_text(BASIC_CONFIG)
return BASIC_CONFIG
@pytest.fixture
def machine_instance(tmp_dir, dvc, mocker):
with dvc.config.edit() as conf:
conf["machine"]["foo"] = {"cloud": "aws"}
def mock_instances(name=None, **kwargs):
if name == "foo":
return iter([TEST_INSTANCE])
return iter([])
mocker.patch(
"tpi.terraform.TerraformBackend.instances",
mocker.MagicMock(side_effect=mock_instances),
)
return TEST_INSTANCE
-163
View File
@@ -1,163 +0,0 @@
import os
import textwrap
import pytest
from dvc.cli import main
from dvc.ui import ui
from tests.utils import console_width
from .conftest import BASIC_CONFIG
@pytest.mark.parametrize(
"slot,value",
[
("region", "us-west"),
("image", "iterative-cml"),
("spot", "True"),
("spot_price", "1.2345"),
("spot_price", "12345"),
("instance_hdd_size", "10"),
("instance_type", "l"),
("instance_gpu", "tesla"),
("ssh_private", "secret"),
],
)
def test_machine_modify_susccess(tmp_dir, dvc, machine_config, slot, value):
assert main(["machine", "modify", "foo", slot, value]) == 0
assert (
tmp_dir / ".dvc" / "config"
).read_text() == machine_config + f" {slot} = {value}\n"
assert main(["machine", "modify", "--unset", "foo", slot]) == 0
assert (tmp_dir / ".dvc" / "config").read_text() == machine_config
def test_machine_modify_startup_script(tmp_dir, dvc, machine_config):
slot, value = "startup_script", "start.sh"
assert main(["machine", "modify", "foo", slot, value]) == 0
assert (
tmp_dir / ".dvc" / "config"
).read_text() == machine_config + f" {slot} = ../{value}\n"
assert main(["machine", "modify", "--unset", "foo", slot]) == 0
assert (tmp_dir / ".dvc" / "config").read_text() == machine_config
@pytest.mark.parametrize(
"slot,value,msg",
[
(
"region",
"other-west",
"expected one of us-west, us-east, eu-west, eu-north",
),
("spot_price", "NUM", "expected float"),
("instance_hdd_size", "BIG", "expected int"),
],
)
def test_machine_modify_fail(tmp_dir, dvc, machine_config, caplog, slot, value, msg):
assert main(["machine", "modify", "foo", slot, value]) == 251
assert (tmp_dir / ".dvc" / "config").read_text() == machine_config
assert msg in caplog.text
FULL_CONFIG_TEXT = textwrap.dedent(
"""\
[feature]
machine = true
['machine \"bar\"']
cloud = azure
['machine \"foo\"']
cloud = aws
region = us-west
image = iterative-cml
spot = True
spot_price = 1.2345
instance_hdd_size = 10
instance_type = l
instance_gpu = tesla
ssh_private = secret
startup_script = {}
""".format(
os.path.join("..", "start.sh")
)
)
def test_machine_list(tmp_dir, dvc, capsys):
from dvc.commands.machine import CmdMachineList
(tmp_dir / ".dvc" / "config").write_text(FULL_CONFIG_TEXT)
with console_width(ui.rich_console, 255):
assert main(["machine", "list"]) == 0
out, _ = capsys.readouterr()
for key in CmdMachineList.TABLE_COLUMNS:
assert f"{key}" in out
assert "bar azure - -" in out
assert "foo aws us-west iterative-cml True 1.2345" in out
assert "10 l *** ***" in out
assert "tesla" in out
with console_width(ui.rich_console, 255):
assert main(["machine", "list", "bar"]) == 0
out, _ = capsys.readouterr()
assert "foo" not in out
assert "name cloud" in out
assert "bar azure" in out
def test_machine_rename_success(tmp_dir, scm, dvc, machine_config, capsys, mocker):
import tpi
config_file = tmp_dir / ".dvc" / "config"
mocker.patch.object(
tpi.terraform.TerraformBackend,
"state_mv",
autospec=True,
return_value=True,
)
os.makedirs(tmp_dir / ".dvc" / "tmp" / "machine" / "terraform" / "foo")
assert main(["machine", "rename", "foo", "bar"]) == 0
cap = capsys.readouterr()
assert "Rename machine 'foo' to 'bar'." in cap.out
assert config_file.read_text() == machine_config.replace("foo", "bar")
assert not (tmp_dir / ".dvc" / "tmp" / "machine" / "terraform" / "foo").exists()
assert (tmp_dir / ".dvc" / "tmp" / "machine" / "terraform" / "bar").exists()
def test_machine_rename_none_exist(tmp_dir, scm, dvc, caplog):
config_alice = BASIC_CONFIG.replace("foo", "alice")
config_file = tmp_dir / ".dvc" / "config"
config_file.write_text(config_alice)
assert main(["machine", "rename", "foo", "bar"]) == 251
assert config_file.read_text() == config_alice
assert "machine 'foo' doesn't exist." in caplog.text
def test_machine_rename_exist(tmp_dir, scm, dvc, caplog):
config_bar = BASIC_CONFIG + "['machine \"bar\"']\n cloud = aws"
config_file = tmp_dir / ".dvc" / "config"
config_file.write_text(config_bar)
assert main(["machine", "rename", "foo", "bar"]) == 251
assert config_file.read_text() == config_bar
assert "Machine 'bar' already exists." in caplog.text
def test_machine_rename_error(tmp_dir, scm, dvc, machine_config, caplog, mocker):
import tpi
config_file = tmp_dir / ".dvc" / "config"
os.makedirs(tmp_dir / ".dvc" / "tmp" / "machine" / "terraform" / "foo")
def cmd_error(self, source, destination, **kwargs):
raise tpi.TPIError("test error")
mocker.patch.object(tpi.terraform.TerraformBackend, "state_mv", cmd_error)
assert main(["machine", "rename", "foo", "bar"]) == 251
assert config_file.read_text() == machine_config
assert "rename failed" in caplog.text
-22
View File
@@ -1,22 +0,0 @@
from dvc.cli import main
from dvc.ui import ui
from tests.utils import console_width
def test_status(tmp_dir, scm, dvc, machine_config, machine_instance, capsys):
assert main(["machine", "add", "bar", "aws"]) == 0
with console_width(ui.rich_console, 255):
assert main(["machine", "status"]) == 0
cap = capsys.readouterr()
assert (
"name instance status cloud instance_ip "
"instance_type instance_hdd_size instance_gpu" in cap.out
)
assert (
"bar - offline - - "
"- - -" in cap.out
)
assert (
"foo num_1 running aws 123.123.123.123 "
"m 35 None" in cap.out
)
+52
View File
@@ -0,0 +1,52 @@
import pytest
from dvc.parsing import DataResolver, MatrixDefinition
MATRIX_DATA = {
"os": ["win", "linux"],
"pyv": [3.7, 3.8],
"dict": [{"arg1": 1}, {"arg2": 2}],
"list": [["out1", "out11"], ["out2", "out22"]],
}
@pytest.mark.parametrize(
"matrix",
[
MATRIX_DATA,
{
"os": "${os}",
"pyv": "${pyv}",
"dict": "${dict}",
"list": "${list}",
},
],
)
def test_matrix_interpolated(tmp_dir, dvc, matrix):
(tmp_dir / "params.yaml").dump(MATRIX_DATA)
resolver = DataResolver(dvc, tmp_dir.fs_path, {})
data = {
"matrix": matrix,
"cmd": "echo ${item.os} ${item.pyv} ${item.dict}"
" -- ${item.list.0} ${item.list.1}",
}
definition = MatrixDefinition(resolver, resolver.context, "build", data)
assert definition.resolve_all() == {
"[email protected]": {"cmd": "echo win 3.7 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.7 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo win 3.7 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.7 --arg2 2 -- out2 out22"},
"[email protected]": {"cmd": "echo win 3.8 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.8 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo win 3.8 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo win 3.8 --arg2 2 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.7 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.7 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.7 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.7 --arg2 2 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.8 --arg1 1 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.8 --arg1 1 -- out2 out22"},
"[email protected]": {"cmd": "echo linux 3.8 --arg2 2 -- out1 out11"},
"[email protected]": {"cmd": "echo linux 3.8 --arg2 2 -- out2 out22"},
}
+49 -17
View File
@@ -1326,14 +1326,18 @@ def test_repro_keep_going(mocker, tmp_dir, dvc, copy_script):
with pytest.raises(ReproductionError):
dvc.reproduce(on_error="keep-going", repro_fn=spy)
assert spy.call_args_list == [
mocker.call(bar_stage, upstream=[], force=False, interactive=False),
mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
mocker.call(foo_stage, upstream=[], force=False, interactive=False),
]
bar_call = mocker.call(bar_stage, upstream=[], force=False, interactive=False)
stage1_call = mocker.call(
stage1, upstream=[bar_stage], force=False, interactive=False
)
foo_call = mocker.call(foo_stage, upstream=[], force=False, interactive=False)
assert spy.call_args_list in (
[bar_call, stage1_call, foo_call],
[foo_call, bar_call, stage1_call],
)
def test_repro_ignore_errors(M, mocker, tmp_dir, dvc, copy_script):
def test_repro_ignore_errors(mocker, tmp_dir, dvc, copy_script):
from dvc.repo import reproduce
(bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
@@ -1347,14 +1351,42 @@ def test_repro_ignore_errors(M, mocker, tmp_dir, dvc, copy_script):
spy = mocker.spy(reproduce, "_reproduce_stage")
dvc.reproduce(on_error="ignore", repro_fn=spy)
assert spy.call_args_list == [
mocker.call(bar_stage, upstream=[], force=False, interactive=False),
mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
mocker.call(foo_stage, upstream=[], force=False, interactive=False),
mocker.call(
stage2,
upstream=[foo_stage, stage1],
force=False,
interactive=False,
),
]
bar_call = mocker.call(bar_stage, upstream=[], force=False, interactive=False)
foo_call = mocker.call(foo_stage, upstream=[], force=False, interactive=False)
stage1_call = mocker.call(
stage1, upstream=[bar_stage], force=False, interactive=False
)
stage2_call = mocker.call(
stage2,
upstream=[foo_stage, stage1],
force=False,
interactive=False,
)
assert spy.call_args_list in (
[bar_call, stage1_call, foo_call, stage2_call],
[foo_call, bar_call, stage1_call, stage2_call],
)
@pytest.mark.parametrize("persist", [True, False])
def test_repro_external_outputs(tmp_dir, dvc, local_workspace, persist):
local_workspace.gen("foo", "foo")
foo_path = str(local_workspace / "foo")
bar_path = str(local_workspace / "bar")
outs = {"outs_no_cache": [bar_path]}
if persist:
outs = {"outs_persist_no_cache": [bar_path]}
dvc.run(
name="mystage",
cmd=f"cp {foo_path} {bar_path}",
deps=[foo_path],
no_exec=True,
**outs,
)
dvc.reproduce()
dvc.reproduce(force=True)
assert (local_workspace / "foo").read_text() == "foo"
assert (local_workspace / "bar").read_text() == "foo"
assert not (local_workspace / "cache").exists()
+2 -2
View File
@@ -452,9 +452,9 @@ def temporary_windows_drive(tmp_path_factory):
if len(s) > 0
]
new_drive_name = [
new_drive_name = next(
letter for letter in string.ascii_uppercase if letter not in drives
][0]
)
new_drive = f"{new_drive_name}:"
target_path = tmp_path_factory.mktemp("tmp_windows_drive")
+7 -7
View File
@@ -218,7 +218,7 @@ def test_list_bad_args(tmp_dir, dvc, caplog, args):
def test_set_invalid_key(dvc):
with pytest.raises(ConfigError, match=r"extra keys not allowed"): # noqa: PT012
with pytest.raises(ConfigError, match=r"extra keys not allowed"):
with dvc.config.edit() as conf:
conf["core"]["invalid_key"] = "value"
@@ -227,9 +227,7 @@ def test_merging_two_levels(dvc):
with dvc.config.edit() as conf:
conf["remote"]["test"] = {"url": "ssh://example.com"}
with pytest.raises( # noqa: PT012
ConfigError, match=r"expected 'url' for dictionary value"
):
with pytest.raises(ConfigError, match=r"expected 'url' for dictionary value"):
with dvc.config.edit("global") as conf:
conf["remote"]["test"] = {"password": "1"}
@@ -267,14 +265,16 @@ def test_load_relative_paths(dvc, field, remote_url):
# check if written paths are correct
dvc_dir = dvc.config.dvc_dir
assert dvc.config["remote"]["test"][field] == os.path.join(
dvc_dir, "..", "file.txt"
assert dvc.config["remote"]["test"][field] == os.path.abspath(
os.path.join(dvc_dir, "..", "file.txt")
)
# load config and check that it contains what we expect
# (relative paths are evaluated correctly)
cfg = Config(dvc_dir)
assert cfg["remote"]["test"][field] == os.path.join(dvc_dir, "..", "file.txt")
assert cfg["remote"]["test"][field] == os.path.abspath(
os.path.join(dvc_dir, "..", "file.txt")
)
def test_config_gdrive_fields(tmp_dir, dvc):
+3 -1
View File
@@ -10,7 +10,9 @@ from dvc.cli import main
from dvc.exceptions import CheckoutError
from dvc.repo.open_repo import clean_repos
from dvc.stage.exceptions import StageNotFound
from dvc.testing.remote_tests import TestRemote # noqa, pylint: disable=unused-import
from dvc.testing.remote_tests import ( # noqa: F401, pylint: disable=unused-import
TestRemote,
)
from dvc.utils.fs import remove
from dvc_data.hashfile.db import HashFileDB
from dvc_data.hashfile.db.local import LocalHashFileDB
+16
View File
@@ -64,6 +64,22 @@ def test_get_repo_dir(tmp_dir, erepo_dir):
assert (tmp_dir / "dir_imported").read_text() == {"file": "contents"}
def test_get_repo_broken_dir(tmp_dir, erepo_dir):
import shutil
from dvc_data.index import DataIndexDirError
with erepo_dir.chdir():
erepo_dir.dvc_gen({"broken": {"file": "contents"}})
erepo_dir.dvc.cache.local.clear()
shutil.rmtree(erepo_dir / "broken")
with pytest.raises(DataIndexDirError):
Repo.get(os.fspath(erepo_dir), "broken", "out")
assert not (tmp_dir / "out").exists()
@pytest.mark.parametrize(
"erepo", [pytest.lazy_fixture("git_dir"), pytest.lazy_fixture("erepo_dir")]
)
+16 -1
View File
@@ -12,6 +12,7 @@ from dvc.scm import Git
from dvc.stage.exceptions import StagePathNotFoundError
from dvc.testing.tmp_dir import make_subrepo
from dvc.utils.fs import remove
from dvc_data.index.index import DataIndexDirError
def test_import(tmp_dir, scm, dvc, erepo_dir):
@@ -96,6 +97,19 @@ def test_import_dir(tmp_dir, scm, dvc, erepo_dir):
}
def test_import_broken_dir(tmp_dir, scm, dvc, erepo_dir):
with erepo_dir.chdir():
erepo_dir.dvc_gen({"dir": {"foo": "foo content"}}, commit="create dir")
erepo_dir.dvc.cache.local.clear()
remove(erepo_dir / "dir")
with pytest.raises(DataIndexDirError):
dvc.imp(os.fspath(erepo_dir), "dir", "dir_imported")
assert not (tmp_dir / "dir_imported").exists()
assert not (tmp_dir / "dir_imported.dvc").exists()
def test_import_file_from_dir(tmp_dir, scm, dvc, erepo_dir):
with erepo_dir.chdir():
erepo_dir.dvc_gen(
@@ -346,7 +360,8 @@ def test_push_wildcard_from_bare_git_repo(
with dvc_repo.chdir():
dvc_repo.dvc.imp(os.fspath(tmp_dir), "dirextra")
dvc_repo.dvc.imp(os.fspath(tmp_dir), "dir123")
with pytest.raises(DataIndexDirError):
dvc_repo.dvc.imp(os.fspath(tmp_dir), "dir123")
@pytest.mark.parametrize("dname", [".", "dir", "dir/subdir"])
+37
View File
@@ -583,3 +583,40 @@ def test_broken_symlink(tmp_dir, dvc):
"path": "link",
},
]
def test_ls_broken_dir(tmp_dir, dvc):
from dvc_data.index import DataIndexDirError
tmp_dir.dvc_gen(
{
"broken": {"baz": "baz"},
}
)
shutil.rmtree(tmp_dir / "broken")
dvc.cache.local.clear()
tmp_dir.dvc_gen(
{
"foo": "foo",
"dir": {"bar": "bar"},
}
)
entries = Repo.ls(os.fspath(tmp_dir))
assert entries == [
{"isdir": False, "isexec": False, "isout": False, "path": ".dvcignore"},
{"isdir": True, "isexec": False, "isout": True, "path": "broken"},
{"isdir": False, "isexec": False, "isout": False, "path": "broken.dvc"},
{"isdir": True, "isexec": False, "isout": True, "path": "dir"},
{"isdir": False, "isexec": False, "isout": False, "path": "dir.dvc"},
{"isdir": False, "isexec": False, "isout": True, "path": "foo"},
{"isdir": False, "isexec": False, "isout": False, "path": "foo.dvc"},
]
with pytest.raises(DataIndexDirError):
Repo.ls(os.fspath(tmp_dir), "broken")
with pytest.raises(DataIndexDirError):
Repo.ls(os.fspath(tmp_dir), recursive=True)
+9 -5
View File
@@ -84,7 +84,7 @@ def test_default_wdir_ignored_in_checksum(tmp_dir, dvc):
stage = dvc.run(cmd="cp bar foo", deps=["bar"], outs=["foo"], name="copy-foo-bar")
d = stage.dumpd()
assert Stage.PARAM_WDIR not in d.keys()
assert Stage.PARAM_WDIR not in d
d = load_yaml("dvc.yaml")
assert Stage.PARAM_WDIR not in d["stages"]["copy-foo-bar"]
@@ -246,13 +246,17 @@ def test_collect_symlink(tmp_dir, dvc, with_deps):
with pytest.raises(StageNotFoundError):
dvc.stage.collect(target=str(data_link / "foo.dvc"), with_deps=with_deps)
else:
stage = list(
dvc.stage.collect(target=str(data_link / "foo.dvc"), with_deps=with_deps)
)[0]
stage = next(
iter(
dvc.stage.collect(
target=str(data_link / "foo.dvc"), with_deps=with_deps
)
)
)
assert stage.addressing == os.path.join("data_link", "foo.dvc")
stage = list(dvc.stage.collect(target=f"{foo_path}.dvc", with_deps=with_deps))[0]
stage = next(iter(dvc.stage.collect(target=f"{foo_path}.dvc", with_deps=with_deps)))
assert stage.addressing == os.path.join("data", "foo.dvc")
+60 -4
View File
@@ -1,3 +1,5 @@
from contextlib import nullcontext as does_not_raise
import pytest
from dvc.exceptions import InvalidArgumentError
@@ -167,16 +169,70 @@ def hydra_setup(tmp_dir, config_dir, config_name):
),
],
)
def test_compose_and_dump(tmp_dir, suffix, overrides, expected):
def test_compose_and_dump_overrides(tmp_dir, suffix, overrides, expected):
from dvc.utils.hydra import compose_and_dump
config_name = "config"
output_file = tmp_dir / f"params.{suffix}"
config_dir = hydra_setup(tmp_dir, "conf", "config")
compose_and_dump(output_file, config_dir, config_name, overrides)
config_module = None
compose_and_dump(output_file, config_dir, config_module, config_name, overrides)
assert output_file.parse() == expected
def hydra_setup_dir_basic(tmp_dir, config_subdir, config_name, config_content):
if config_subdir is None:
return None
config_dir = tmp_dir / config_subdir
config_dir.mkdir()
(config_dir / f"{config_name}.yaml").dump(config_content)
return str(config_dir)
@pytest.mark.parametrize(
"config_subdir,config_module,config_content,error_context",
[
("conf", None, {"normal_yaml_config": False}, does_not_raise()),
(
None,
"hydra.test_utils.configs",
{"normal_yaml_config": True},
does_not_raise(),
),
(
"conf",
"hydra.test_utils.configs",
{"normal_yaml_config": False},
does_not_raise(),
),
(
None,
None,
None,
pytest.raises(
ValueError,
match="Either `config_dir` or `config_module` should be provided.",
),
),
],
)
def test_compose_and_dump_dir_module(
tmp_dir, config_subdir, config_module, config_content, error_context
):
from dvc.utils.hydra import compose_and_dump
output_file = tmp_dir / "params.yaml"
config_name = "config"
config_dir = hydra_setup_dir_basic(
tmp_dir, config_subdir, config_name, config_content
)
with error_context:
compose_and_dump(output_file, config_dir, config_module, config_name, [])
assert output_file.parse() == config_content
def test_compose_and_dump_yaml_handles_string(tmp_dir):
"""Regression test for https://github.com/iterative/dvc/issues/8583"""
from dvc.utils.hydra import compose_and_dump
@@ -185,7 +241,7 @@ def test_compose_and_dump_yaml_handles_string(tmp_dir):
config.parent.mkdir()
config.write_text("foo: 'no'\n")
output_file = tmp_dir / "params.yaml"
compose_and_dump(output_file, str(config.parent), "config", [])
compose_and_dump(output_file, str(config.parent), None, "config", [])
assert output_file.read_text() == "foo: 'no'\n"
@@ -197,7 +253,7 @@ def test_compose_and_dump_resolves_interpolation(tmp_dir):
config.parent.mkdir()
config.dump({"data": {"root": "path/to/root", "raw": "${.root}/raw"}})
output_file = tmp_dir / "params.yaml"
compose_and_dump(output_file, str(config.parent), "config", [])
compose_and_dump(output_file, str(config.parent), None, "config", [])
assert output_file.parse() == {
"data": {"root": "path/to/root", "raw": "path/to/root/raw"}
}
+1 -1
View File
@@ -1,4 +1,4 @@
from dvc_ssh.tests.fixtures import ( # noqa, pylint: disable=unused-import
from dvc_ssh.tests.fixtures import ( # noqa: F401, pylint: disable=unused-import
make_ssh,
ssh,
ssh_server,
+60 -7
View File
@@ -19,7 +19,14 @@ def test_list(mocker):
url = "local_dir"
m = _test_cli(mocker, url)
m.assert_called_once_with(
url, None, recursive=False, rev=None, dvc_only=False, config=None
url,
None,
recursive=False,
rev=None,
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
@@ -27,7 +34,14 @@ def test_list_recursive(mocker):
url = "local_dir"
m = _test_cli(mocker, url, "-R")
m.assert_called_once_with(
url, None, recursive=True, rev=None, dvc_only=False, config=None
url,
None,
recursive=True,
rev=None,
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
@@ -35,7 +49,14 @@ def test_list_git_ssh_rev(mocker):
url = "[email protected]:repo"
m = _test_cli(mocker, url, "--rev", "123")
m.assert_called_once_with(
url, None, recursive=False, rev="123", dvc_only=False, config=None
url,
None,
recursive=False,
rev="123",
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
@@ -44,7 +65,14 @@ def test_list_targets(mocker):
target = "subdir"
m = _test_cli(mocker, url, target)
m.assert_called_once_with(
url, target, recursive=False, rev=None, dvc_only=False, config=None
url,
target,
recursive=False,
rev=None,
dvc_only=False,
config=None,
remote=None,
remote_config=None,
)
@@ -52,15 +80,40 @@ def test_list_outputs_only(mocker):
url = "local_dir"
m = _test_cli(mocker, url, None, "--dvc-only")
m.assert_called_once_with(
url, None, recursive=False, rev=None, dvc_only=True, config=None
url,
None,
recursive=False,
rev=None,
dvc_only=True,
config=None,
remote=None,
remote_config=None,
)
def test_list_config(mocker):
url = "local_dir"
m = _test_cli(mocker, url, None, "--config", "myconfig")
m = _test_cli(
mocker,
url,
None,
"--config",
"myconfig",
"--remote",
"myremote",
"--remote-config",
"k1=v1",
"k2=v2",
)
m.assert_called_once_with(
url, None, recursive=False, rev=None, dvc_only=False, config="myconfig"
url,
None,
recursive=False,
rev=None,
dvc_only=False,
config="myconfig",
remote="myremote",
remote_config={"k1": "v1", "k2": "v2"},
)
+17
View File
@@ -0,0 +1,17 @@
import logging
import pytest
from dvc.cli import main
from dvc.commands.completion import SUPPORTED_SHELLS
@pytest.mark.parametrize("shell", SUPPORTED_SHELLS)
def test_completion(caplog, capsys, shell):
with caplog.at_level(logging.INFO):
assert main(["completion", "-s", shell]) == 0
assert not caplog.text
out, err = capsys.readouterr()
assert not err
assert out
+2 -3
View File
@@ -125,7 +125,6 @@ def test_experiments_run(dvc, scm, mocker):
"run_all": False,
"jobs": 1,
"tmp_dir": False,
"machine": None,
"copy_paths": [],
"message": None,
}
@@ -189,7 +188,7 @@ def test_experiments_list(dvc, scm, mocker):
@pytest.mark.parametrize(
"args,expected",
[
([], "main:\n\tsha-a [exp-a]\n"),
([], "master:\n\tsha-a [exp-a]\n"),
(["--name-only"], "exp-a\n"),
(["--sha-only"], "sha-a\n"),
],
@@ -198,7 +197,7 @@ def test_experiments_list_format(mocker, capsys, args, expected, dvc, scm):
mocker.patch(
"dvc.repo.experiments.ls.ls",
return_value={
"refs/heads/main": [
scm.get_rev(): [
("exp-a", "sha-a"),
]
},
+7
View File
@@ -16,6 +16,11 @@ def test_get(mocker):
"4",
"--config",
"myconfig",
"--remote",
"myremote",
"--remote-config",
"k1=v1",
"k2=v2",
]
)
assert cli_args.func == CmdGet
@@ -33,6 +38,8 @@ def test_get(mocker):
jobs=4,
config="myconfig",
force=False,
remote="myremote",
remote_config={"k1": "v1", "k2": "v2"},
)
+48
View File
@@ -0,0 +1,48 @@
import logging
from argparse import SUPPRESS, ArgumentParser
from typing import Tuple
import pytest
import shtab
from dvc.cli import main
from dvc.cli.parser import get_main_parser
def command_tuples():
root: Tuple[str, ...] = ()
commands = [root]
def recurse_parser(parser: ArgumentParser, parents: Tuple[str, ...] = root) -> None:
for positional in parser._get_positional_actions():
if positional.help != SUPPRESS and isinstance(positional.choices, dict):
public_cmds = shtab.get_public_subcommands(positional)
for subcmd, subparser in positional.choices.items():
cmd = (*parents, subcmd)
if subcmd in public_cmds:
commands.append(cmd)
recurse_parser(subparser, cmd)
main_parser = get_main_parser()
recurse_parser(main_parser)
return sorted(commands)
def pytest_generate_tests(metafunc):
def ids(values):
if values:
return "-".join(values)
return "dvc"
metafunc.parametrize("command_tuples", command_tuples(), ids=ids)
def test_help(caplog, capsys, command_tuples):
with caplog.at_level(logging.INFO), pytest.raises(SystemExit) as e:
main([*command_tuples, "--help"])
assert e.value.code == 0
assert not caplog.text
out, err = capsys.readouterr()
assert not err
assert out
+11
View File
@@ -16,6 +16,11 @@ def test_import(mocker, dvc):
"3",
"--config",
"myconfig",
"--remote",
"myremote",
"--remote-config",
"k1=v1",
"k2=v2",
]
)
assert cli_args.func == CmdImport
@@ -34,6 +39,8 @@ def test_import(mocker, dvc):
no_download=False,
jobs=3,
config="myconfig",
remote="myremote",
remote_config={"k1": "v1", "k2": "v2"},
)
@@ -65,6 +72,8 @@ def test_import_no_exec(mocker, dvc):
no_download=False,
jobs=None,
config=None,
remote=None,
remote_config=None,
)
@@ -96,4 +105,6 @@ def test_import_no_download(mocker, dvc):
no_download=True,
jobs=None,
config=None,
remote=None,
remote_config=None,
)
-167
View File
@@ -1,167 +0,0 @@
import os
import configobj
import pytest
from dvc.cli import parse_args
from dvc.commands.machine import (
CmdMachineAdd,
CmdMachineCreate,
CmdMachineDestroy,
CmdMachineList,
CmdMachineModify,
CmdMachineRemove,
CmdMachineRename,
CmdMachineSsh,
CmdMachineStatus,
)
DATA = {
".dvc": {
"config": (
"[feature]\n"
" machine = true\n"
"['machine \"foo\"']\n"
" cloud = aws\n"
"['machine \"myaws\"']\n"
" cloud = aws"
)
}
}
def test_add(tmp_dir):
tmp_dir.gen({".dvc": {"config": "[feature]\n machine = true"}})
cli_args = parse_args(["machine", "add", "foo", "aws"])
assert cli_args.func == CmdMachineAdd
cmd = cli_args.func(cli_args)
assert cmd.run() == 0
config = configobj.ConfigObj(str(tmp_dir / ".dvc" / "config"))
assert config['machine "foo"']["cloud"] == "aws"
def test_remove(tmp_dir):
tmp_dir.gen(DATA)
cli_args = parse_args(["machine", "remove", "foo"])
assert cli_args.func == CmdMachineRemove
cmd = cli_args.func(cli_args)
assert cmd.run() == 0
config = configobj.ConfigObj(str(tmp_dir / ".dvc" / "config"))
assert list(config.keys()) == ["feature", 'machine "myaws"']
def test_create(tmp_dir, dvc, mocker):
cli_args = parse_args(["machine", "create", "foo"])
assert cli_args.func == CmdMachineCreate
cmd = cli_args.func(cli_args)
m = mocker.patch.object(cmd.repo.machine, "create", autospec=True, return_value=0)
assert cmd.run() == 0
m.assert_called_once_with("foo")
def test_status(tmp_dir, scm, dvc, mocker):
tmp_dir.gen(DATA)
cli_args = parse_args(["machine", "status", "foo"])
assert cli_args.func == CmdMachineStatus
cmd = cli_args.func(cli_args)
m = mocker.patch.object(cmd.repo.machine, "status", autospec=True, return_value=[])
assert cmd.run() == 0
m.assert_called_once_with("foo")
cli_args = parse_args(["machine", "status"])
cmd = cli_args.func(cli_args)
m = mocker.patch.object(cmd.repo.machine, "status", autospec=True, return_value=[])
assert cmd.run() == 0
assert m.call_count == 2
m.assert_has_calls([mocker.call("foo"), mocker.call("myaws")])
def test_destroy(tmp_dir, dvc, mocker):
cli_args = parse_args(["machine", "destroy", "foo"])
assert cli_args.func == CmdMachineDestroy
cmd = cli_args.func(cli_args)
m = mocker.patch.object(cmd.repo.machine, "destroy", autospec=True, return_value=0)
assert cmd.run() == 0
m.assert_called_once_with("foo")
def test_ssh(tmp_dir, dvc, mocker):
cli_args = parse_args(["machine", "ssh", "foo"])
assert cli_args.func == CmdMachineSsh
cmd = cli_args.func(cli_args)
m = mocker.patch.object(
cmd.repo.machine, "run_shell", autospec=True, return_value=0
)
assert cmd.run() == 0
m.assert_called_once_with("foo")
@pytest.mark.parametrize("show_origin", [["--show-origin"], []])
def test_list(tmp_dir, mocker, show_origin):
from dvc.compare import TabularData
from dvc.ui import ui
tmp_dir.gen(DATA)
cli_args = parse_args(["machine", "list", *show_origin, "foo"])
assert cli_args.func == CmdMachineList
cmd = cli_args.func(cli_args)
if show_origin:
m = mocker.patch.object(ui, "write", autospec=True)
else:
m = mocker.patch.object(TabularData, "render", autospec=True)
assert cmd.run() == 0
if show_origin:
m.assert_called_once_with(f".dvc{os.sep}config cloud=aws")
else:
m.assert_called_once()
def test_modified(tmp_dir):
tmp_dir.gen(DATA)
cli_args = parse_args(["machine", "modify", "foo", "cloud", "azure"])
assert cli_args.func == CmdMachineModify
cmd = cli_args.func(cli_args)
assert cmd.run() == 0
config = configobj.ConfigObj(str(tmp_dir / ".dvc" / "config"))
assert config['machine "foo"']["cloud"] == "azure"
def test_rename(tmp_dir, scm, dvc):
tmp_dir.gen(DATA)
cli_args = parse_args(["machine", "rename", "foo", "bar"])
assert cli_args.func == CmdMachineRename
cmd = cli_args.func(cli_args)
assert cmd.run() == 0
config = configobj.ConfigObj(str(tmp_dir / ".dvc" / "config"))
assert config['machine "bar"']["cloud"] == "aws"
@pytest.mark.parametrize(
"cmd,use_config",
[
("add", True),
("default", True),
("remove", True),
("list", True),
("modify", True),
("create", False),
("destroy", False),
("rename", True),
("status", False),
("ssh", False),
],
)
def test_help_message(tmp_dir, scm, dvc, cmd, use_config, capsys):
try:
parse_args(["machine", cmd, "--help"])
except SystemExit:
pass
cap = capsys.readouterr()
assert ("--global" in cap.out) is use_config
-84
View File
@@ -1,84 +0,0 @@
import pytest
from dvc_azure import AzureAuthError, AzureFileSystem
container_name = "container-name"
connection_string = (
"DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;"
"AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsu"
"Fq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;"
"BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;"
)
def test_strip_protocol_env_var(monkeypatch, dvc):
monkeypatch.setenv("AZURE_STORAGE_CONTAINER_NAME", container_name)
monkeypatch.setenv("AZURE_STORAGE_CONNECTION_STRING", connection_string)
assert AzureFileSystem._strip_protocol("azure://") == container_name
def test_strip_protocol(dvc):
assert (
AzureFileSystem._strip_protocol(f"azure://{container_name}") == container_name
)
def test_init(dvc):
config = {"connection_string": connection_string}
fs = AzureFileSystem(**config)
assert fs.fs_args["connection_string"] == connection_string
def test_azure_login_methods():
def get_login_method(config):
fs = AzureFileSystem(**config)
return fs.login_method
with pytest.raises(AzureAuthError):
get_login_method({})
assert get_login_method({"connection_string": "test"}) == "connection string"
assert get_login_method({"account_name": "test"}).startswith("default credentials")
assert (
get_login_method({"account_name": "test", "allow_anonymous_login": True})
== "anonymous login"
)
with pytest.raises(AzureAuthError):
get_login_method(
{"tenant_id": "test", "client_id": "test", "client_secret": "test"}
)
assert (
get_login_method(
{
"account_name": "test",
"tenant_id": "test",
"client_id": "test",
"client_secret": "test",
}
)
== "AD service principal"
)
assert (
get_login_method({"account_name": "test", "account_key": "test"})
== "account key"
)
assert (
get_login_method({"account_name": "test", "sas_token": "test"}) == "SAS token"
)
assert (
get_login_method(
{
"connection_string": "test",
"account_name": "test",
"sas_token": "test",
}
)
== "connection string"
)
assert (
get_login_method({"connection_string": "test", "sas_token": "test"})
== "connection string"
)
-137
View File
@@ -1,137 +0,0 @@
import os
import pytest
from dvc_s3 import S3FileSystem
from dvc.fs import ConfigError
bucket_name = "bucket-name"
prefix = "some/prefix"
url = f"s3://{bucket_name}/{prefix}"
key_id = "key-id"
key_secret = "key-secret"
session_token = "session-token"
@pytest.fixture(autouse=True)
def grants():
return {
"grant_read": "id=read-permission-id,id=other-read-permission-id",
"grant_read_acp": "id=read-acp-permission-id",
"grant_write_acp": "id=write-acp-permission-id",
"grant_full_control": "id=full-control-permission-id",
}
def test_verify_ssl_default_param(dvc):
config = {"url": url}
fs = S3FileSystem(**config)
assert "client_kwargs" not in fs.fs_args
config = {
"url": url,
"endpointurl": "https://my.custom.s3:1234",
}
fs = S3FileSystem(**config)
assert "verify" not in fs.fs_args["client_kwargs"]
def test_s3_config_credentialpath(dvc, monkeypatch):
environment = {}
monkeypatch.setattr(os, "environ", environment)
config = {"url": url, "credentialpath": "somewhere"}
S3FileSystem(**config)
assert environment["AWS_SHARED_CREDENTIALS_FILE"] == "somewhere"
environment.clear()
config = {"url": url, "configpath": "somewhere"}
S3FileSystem(**config)
assert environment["AWS_CONFIG_FILE"] == "somewhere"
environment.clear()
config = {
"url": url,
"credentialpath": "somewhere",
"configpath": "elsewhere",
}
S3FileSystem(**config)
assert environment["AWS_SHARED_CREDENTIALS_FILE"] == "somewhere"
assert environment["AWS_CONFIG_FILE"] == "elsewhere"
environment.clear()
def test_ssl_verify_bool_param(dvc):
config = {"url": url, "ssl_verify": False}
fs = S3FileSystem(**config)
assert fs.fs_args["client_kwargs"]["verify"] == config["ssl_verify"]
def test_ssl_verify_path_param(dvc):
config = {"url": url, "ssl_verify": "/path/to/custom/cabundle.pem"}
fs = S3FileSystem(**config)
assert fs.fs_args["client_kwargs"]["verify"] == config["ssl_verify"]
def test_ssl_verify_none_param(dvc):
config = {"url": url, "ssl_verify": None}
fs = S3FileSystem(**config)
assert "client_kwargs" not in fs.fs_args
config = {
"url": url,
"endpointurl": "https://my.custom.s3:1234",
"ssl_verify": None,
}
fs = S3FileSystem(**config)
assert "verify" not in fs.fs_args["client_kwargs"]
def test_grants(dvc):
config = {
"url": url,
"grant_read": "id=read-permission-id,id=other-read-permission-id",
"grant_read_acp": "id=read-acp-permission-id",
"grant_write_acp": "id=write-acp-permission-id",
"grant_full_control": "id=full-control-permission-id",
}
fs = S3FileSystem(**config)
extra_args = fs.fs_args["s3_additional_kwargs"]
assert (
extra_args["GrantRead"] == "id=read-permission-id,id=other-read-permission-id"
)
assert extra_args["GrantReadACP"] == "id=read-acp-permission-id"
assert extra_args["GrantWriteACP"] == "id=write-acp-permission-id"
assert extra_args["GrantFullControl"] == "id=full-control-permission-id"
def test_grants_mutually_exclusive_acl_error(dvc, grants):
for grant_option, grant_value in grants.items():
config = {"url": url, "acl": "public-read", grant_option: grant_value}
with pytest.raises(ConfigError):
S3FileSystem(**config)
def test_sse_kms_key_id(dvc):
fs = S3FileSystem(url=url, sse_kms_key_id="key")
assert fs.fs_args["s3_additional_kwargs"]["SSEKMSKeyId"] == "key"
def test_key_id_and_secret(dvc):
fs = S3FileSystem(
url=url,
access_key_id=key_id,
secret_access_key=key_secret,
session_token=session_token,
)
assert fs.fs_args["key"] == key_id
assert fs.fs_args["secret"] == key_secret
assert fs.fs_args["token"] == session_token
View File
-36
View File
@@ -1,36 +0,0 @@
import pytest
def test_validate_name():
from dvc.exceptions import InvalidArgumentError
from dvc.machine import RESERVED_NAMES, validate_name
for name in RESERVED_NAMES:
with pytest.raises(InvalidArgumentError):
validate_name(name)
@pytest.mark.parametrize("cloud", ["aws", "azure"])
def test_get_config_and_backend(tmp_dir, dvc, cloud):
from dvc.machine.backend.terraform import TerraformBackend
name = "foo"
with dvc.config.edit() as conf:
conf["machine"][name] = {"cloud": cloud}
config, backend = dvc.machine.get_config_and_backend(name)
assert config == {"cloud": cloud, "name": name}
assert isinstance(backend, TerraformBackend)
def test_get_config_and_backend_nonexistent(tmp_dir, dvc):
from dvc.config import MachineNotFoundError
with pytest.raises(MachineNotFoundError):
dvc.machine.get_config_and_backend("foo")
def test_get_config_and_backend_default(tmp_dir, dvc):
from dvc.config import NoMachineError
with pytest.raises(NoMachineError):
dvc.machine.get_config_and_backend()
+12 -1
View File
@@ -1,8 +1,15 @@
import re
import pytest
from dvc.exceptions import InvalidArgumentError
from dvc.repo.experiments.refs import EXPS_NAMESPACE, ExpRefInfo
from dvc.repo.experiments.utils import check_ref_format, resolve_name, to_studio_params
from dvc.repo.experiments.utils import (
check_ref_format,
gen_random_name,
resolve_name,
to_studio_params,
)
def commit_exp_ref(tmp_dir, scm, file="foo", contents="foo", name="foo"):
@@ -74,3 +81,7 @@ def test_run_check_ref_format(scm, name, result):
)
def test_to_studio_params(params, expected):
assert to_studio_params(params) == expected
def test_gen_random_name():
assert re.match("[a-zA-Z]+-[a-zA-Z]+", gen_random_name())
+12
View File
@@ -76,6 +76,12 @@ def test_iter_revs(
commit_time_offset=0,
message="dummy",
parents=["dummy"],
committer_name="dummy",
committer_email="dummy",
author_name="dummy",
author_email="dummy",
author_time=datetime(2022, 6, 28).timestamp(),
author_time_offset=0,
)
if rev == rev_old:
raise SCMError
@@ -85,6 +91,12 @@ def test_iter_revs(
commit_time_offset=0,
message="dummy",
parents=["dummy"],
committer_name="dummy",
committer_email="dummy",
author_name="dummy",
author_email="dummy",
author_time=datetime(2022, 6, 28).timestamp(),
author_time_offset=0,
)
mocker.patch(
+23
View File
@@ -21,6 +21,29 @@ def test_to_relpath(path, expected):
assert Config._to_relpath(os.path.join(".", "config"), path) == expected
@pytest.mark.parametrize(
"path, expected",
[
("cache", os.path.abspath(os.path.join("conf_dir", "cache"))),
("dir/cache", os.path.abspath(os.path.join("conf_dir", "dir", "cache"))),
("../cache", os.path.abspath("cache")),
(os.getcwd(), os.getcwd()),
("ssh://some/path", "ssh://some/path"),
],
)
def test_resolve(path, expected):
conf_dir = os.path.abspath("conf_dir")
assert Config._resolve(conf_dir, path) == expected
def test_resolve_homedir():
# NOTE: our test suit patches $HOME, but that only works within the
# test itself, so we can't use expanduser in @parametrize here.
conf_dir = os.path.abspath("conf_dir")
expected = os.path.expanduser(os.path.join("~", "cache"))
assert Config._resolve(conf_dir, "~/cache") == expected
def test_get_fs(tmp_dir, scm):
tmp_dir.scm_gen("foo", "foo", commit="add foo")