diff --git a/dvc/command/stage.py b/dvc/command/stage.py index 8a1bd066b..06cc03bbb 100644 --- a/dvc/command/stage.py +++ b/dvc/command/stage.py @@ -9,7 +9,7 @@ from dvc.utils.cli_parse import parse_params from dvc.utils.humanize import truncate_text if TYPE_CHECKING: - from dvc.output.base import BaseOutput + from dvc.output import Output from dvc.stage import Stage logger = logging.getLogger(__name__) @@ -18,7 +18,7 @@ MAX_TEXT_LENGTH = 80 def generate_description(stage: "Stage") -> str: - def part_desc(outs: Iterable["BaseOutput"]) -> str: + def part_desc(outs: Iterable["Output"]) -> str: return ", ".join(out.def_path for out in outs) if not stage.deps and not stage.outs: @@ -27,7 +27,7 @@ def generate_description(stage: "Stage") -> str: if not stage.outs and stage.deps: return "Depends on " + part_desc(stage.deps) - def is_plot_or_metric(out: "BaseOutput"): + def is_plot_or_metric(out: "Output"): return bool(out.plot) or bool(out.metric) desc: List[str] = [] diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index c03cd2d5e..f3578c0fa 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -1,59 +1,25 @@ from collections import defaultdict -from urllib.parse import urlparse import dvc.output as output -from dvc.dependency.azure import AzureDependency -from dvc.dependency.gs import GSDependency -from dvc.dependency.hdfs import HDFSDependency -from dvc.dependency.http import HTTPDependency -from dvc.dependency.https import HTTPSDependency -from dvc.dependency.local import LocalDependency -from dvc.dependency.param import ParamsDependency -from dvc.dependency.s3 import S3Dependency -from dvc.dependency.ssh import SSHDependency -from dvc.dependency.webdav import WebDAVDependency -from dvc.dependency.webdavs import WebDAVSDependency -from dvc.dependency.webhdfs import WebHDFSDependency -from dvc.output.base import BaseOutput -from dvc.scheme import Schemes +from dvc.output import Output -from ..fs import get_cloud_fs +from .base import Dependency +from .param import ParamsDependency from .repo import RepoDependency -DEP_MAP = { - Schemes.LOCAL: LocalDependency, - Schemes.SSH: SSHDependency, - Schemes.S3: S3Dependency, - Schemes.AZURE: AzureDependency, - Schemes.GS: GSDependency, - Schemes.HDFS: HDFSDependency, - Schemes.HTTP: HTTPDependency, - Schemes.HTTPS: HTTPSDependency, - Schemes.WEBDAV: WebDAVDependency, - Schemes.WEBDAVS: WebDAVSDependency, - Schemes.WEBHDFS: WebHDFSDependency, -} - - # NOTE: schema for dependencies is basically the same as for outputs, but # without output-specific entries like 'cache' (whether or not output is # cached, see -o and -O flags for `dvc run`) and 'metric' (whether or not # output is a metrics file and how to parse it, see `-M` flag for `dvc run`). SCHEMA = output.SCHEMA.copy() -del SCHEMA[BaseOutput.PARAM_CACHE] -del SCHEMA[BaseOutput.PARAM_METRIC] -del SCHEMA[BaseOutput.PARAM_DESC] +del SCHEMA[Output.PARAM_CACHE] +del SCHEMA[Output.PARAM_METRIC] +del SCHEMA[Output.PARAM_DESC] SCHEMA.update(RepoDependency.REPO_SCHEMA) SCHEMA.update(ParamsDependency.PARAM_SCHEMA) def _get(stage, p, info): - parsed = urlparse(p) if p else None - if parsed and parsed.scheme == "remote": - cls, config = get_cloud_fs(stage.repo, name=parsed.netloc) - fs = cls(**config) - return DEP_MAP[fs.scheme](stage, p, info, fs=fs) - if info and info.get(RepoDependency.PARAM_REPO): repo = info.pop(RepoDependency.PARAM_REPO) return RepoDependency(repo, stage, p, info) @@ -62,14 +28,13 @@ def _get(stage, p, info): params = info.pop(ParamsDependency.PARAM_PARAMS) return ParamsDependency(stage, p, params) - dep_cls = DEP_MAP.get(parsed.scheme, LocalDependency) - return dep_cls(stage, p, info) + return Dependency(stage, p, info) def loadd_from(stage, d_list): ret = [] for d in d_list: - p = d.pop(BaseOutput.PARAM_PATH, None) + p = d.pop(Output.PARAM_PATH, None) ret.append(_get(stage, p, d)) return ret diff --git a/dvc/dependency/azure.py b/dvc/dependency/azure.py deleted file mode 100644 index 063ea867f..000000000 --- a/dvc/dependency/azure.py +++ /dev/null @@ -1,8 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.base import BaseOutput - -from ..fs.azure import AzureFileSystem - - -class AzureDependency(BaseDependency, BaseOutput): - FS_CLS = AzureFileSystem diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 68ac5079d..b08435653 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -1,6 +1,7 @@ from typing import Type from dvc.exceptions import DvcException +from dvc.output import Output class DependencyDoesNotExistError(DvcException): @@ -20,7 +21,7 @@ class DependencyIsStageFileError(DvcException): super().__init__(f"DVC file '{path}' cannot be a dependency.") -class BaseDependency: +class Dependency(Output): IS_DEPENDENCY = True DoesNotExistError = DependencyDoesNotExistError # type: Type[DvcException] diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py deleted file mode 100644 index ca75749af..000000000 --- a/dvc/dependency/gs.py +++ /dev/null @@ -1,8 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.base import BaseOutput - -from ..fs.gs import GSFileSystem - - -class GSDependency(BaseDependency, BaseOutput): - FS_CLS = GSFileSystem diff --git a/dvc/dependency/hdfs.py b/dvc/dependency/hdfs.py deleted file mode 100644 index 61a6475a1..000000000 --- a/dvc/dependency/hdfs.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.hdfs import HDFSOutput - - -class HDFSDependency(BaseDependency, HDFSOutput): - pass diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py deleted file mode 100644 index 521bff00a..000000000 --- a/dvc/dependency/http.py +++ /dev/null @@ -1,8 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.base import BaseOutput - -from ..fs.http import HTTPFileSystem - - -class HTTPDependency(BaseDependency, BaseOutput): - FS_CLS = HTTPFileSystem diff --git a/dvc/dependency/https.py b/dvc/dependency/https.py deleted file mode 100644 index a906bceeb..000000000 --- a/dvc/dependency/https.py +++ /dev/null @@ -1,6 +0,0 @@ -from ..fs.https import HTTPSFileSystem -from .http import HTTPDependency - - -class HTTPSDependency(HTTPDependency): - FS_CLS = HTTPSFileSystem diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py deleted file mode 100644 index 54df58f82..000000000 --- a/dvc/dependency/local.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.local import LocalOutput - - -class LocalDependency(BaseDependency, LocalOutput): - pass diff --git a/dvc/dependency/param.py b/dvc/dependency/param.py index 741d7278d..f9c0059e5 100644 --- a/dvc/dependency/param.py +++ b/dvc/dependency/param.py @@ -4,11 +4,12 @@ from collections import defaultdict import dpath.util from voluptuous import Any -from dvc.dependency.local import LocalDependency from dvc.exceptions import DvcException from dvc.hash_info import HashInfo from dvc.utils.serialize import LOADERS, ParseError +from .base import Dependency + class MissingParamsError(DvcException): pass @@ -18,7 +19,7 @@ class BadParamFileError(DvcException): pass -class ParamsDependency(LocalDependency): +class ParamsDependency(Dependency): PARAM_PARAMS = "params" PARAM_SCHEMA = {PARAM_PARAMS: Any(dict, list, None)} DEFAULT_PARAMS_FILE = "params.yaml" @@ -39,7 +40,6 @@ class ParamsDependency(LocalDependency): path or os.path.join(stage.repo.root_dir, self.DEFAULT_PARAMS_FILE), info=info, - fs=stage.repo.fs, ) def dumpd(self): diff --git a/dvc/dependency/repo.py b/dvc/dependency/repo.py index 6d94d5f0e..fd5f6f708 100644 --- a/dvc/dependency/repo.py +++ b/dvc/dependency/repo.py @@ -2,10 +2,10 @@ from voluptuous import Required from dvc.path_info import PathInfo -from .local import LocalDependency +from .base import Dependency -class RepoDependency(LocalDependency): +class RepoDependency(Dependency): PARAM_REPO = "repo" PARAM_URL = "url" PARAM_REV = "rev" @@ -23,7 +23,7 @@ class RepoDependency(LocalDependency): self.def_repo = def_repo super().__init__(stage, *args, **kwargs) - def _parse_path(self, fs, path): + def _parse_path(self, fs, path_info): return None @property diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py deleted file mode 100644 index 733a2316a..000000000 --- a/dvc/dependency/s3.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.s3 import S3Output - - -class S3Dependency(BaseDependency, S3Output): - pass diff --git a/dvc/dependency/ssh.py b/dvc/dependency/ssh.py deleted file mode 100644 index 00c0bebe8..000000000 --- a/dvc/dependency/ssh.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.ssh import SSHOutput - - -class SSHDependency(BaseDependency, SSHOutput): - pass diff --git a/dvc/dependency/webdav.py b/dvc/dependency/webdav.py deleted file mode 100644 index bb428bb2a..000000000 --- a/dvc/dependency/webdav.py +++ /dev/null @@ -1,8 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.base import BaseOutput - -from ..fs.webdav import WebDAVFileSystem - - -class WebDAVDependency(BaseDependency, BaseOutput): - FS_CLS = WebDAVFileSystem diff --git a/dvc/dependency/webdavs.py b/dvc/dependency/webdavs.py deleted file mode 100644 index acdd164ed..000000000 --- a/dvc/dependency/webdavs.py +++ /dev/null @@ -1,6 +0,0 @@ -from ..fs.webdavs import WebDAVSFileSystem -from .webdav import WebDAVDependency - - -class WebDAVSDependency(WebDAVDependency): - FS_CLS = WebDAVSFileSystem diff --git a/dvc/dependency/webhdfs.py b/dvc/dependency/webhdfs.py deleted file mode 100644 index e3d5f1deb..000000000 --- a/dvc/dependency/webhdfs.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.dependency.base import BaseDependency -from dvc.output.webhdfs import WebHDFSOutput - - -class WebHDFSDependency(BaseDependency, WebHDFSOutput): - pass diff --git a/dvc/fs/__init__.py b/dvc/fs/__init__.py index a7b547660..6e3696e30 100644 --- a/dvc/fs/__init__.py +++ b/dvc/fs/__init__.py @@ -82,25 +82,28 @@ def get_cloud_fs(repo, **kwargs): from dvc.config import ConfigError from dvc.config_schema import SCHEMA, Invalid - remote_conf = get_fs_config(repo.config, **kwargs) + repo_config = repo.config if repo else {} + core_config = repo_config.get("core", {}) + + remote_conf = get_fs_config(repo_config, **kwargs) try: remote_conf = SCHEMA["remote"][str](remote_conf) except Invalid as exc: raise ConfigError(str(exc)) from None if "jobs" not in remote_conf: - jobs = repo.config["core"].get("jobs") + jobs = core_config.get("jobs") if jobs: remote_conf["jobs"] = jobs if "checksum_jobs" not in remote_conf: - checksum_jobs = repo.config["core"].get("checksum_jobs") + checksum_jobs = core_config.get("checksum_jobs") if checksum_jobs: remote_conf["checksum_jobs"] = checksum_jobs cls = get_fs_cls(remote_conf) - if cls == GDriveFileSystem: + if cls == GDriveFileSystem and repo: remote_conf["gdrive_credentials_tmp_dir"] = repo.tmp_dir return cls, remote_conf diff --git a/dvc/fs/_metadata.py b/dvc/fs/_metadata.py index 6ea8822a2..a61db9484 100644 --- a/dvc/fs/_metadata.py +++ b/dvc/fs/_metadata.py @@ -1,7 +1,7 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, List -from dvc.output import BaseOutput +from dvc.output import Output from dvc.path_info import PathInfo if TYPE_CHECKING: @@ -33,7 +33,7 @@ class Metadata: # optional fields isdir: bool = False # is it a directory? is_exec: bool = False # is it an executable? - outs: List[BaseOutput] = field(default_factory=list) # list of outputs + outs: List[Output] = field(default_factory=list) # list of outputs def __post_init__(self): self.output_exists = bool(self.outs) diff --git a/dvc/fs/dvc.py b/dvc/fs/dvc.py index 8a5238553..b6a651edc 100644 --- a/dvc/fs/dvc.py +++ b/dvc/fs/dvc.py @@ -10,7 +10,7 @@ from ._metadata import Metadata from .base import BaseFileSystem if typing.TYPE_CHECKING: - from dvc.output.base import BaseOutput + from dvc.output import Output logger = logging.getLogger(__name__) @@ -43,7 +43,7 @@ class DvcFileSystem(BaseFileSystem): # pylint:disable=abstract-method return outs def _get_granular_hash( - self, path_info: PathInfo, out: "BaseOutput", remote=None + self, path_info: PathInfo, out: "Output", remote=None ): assert isinstance(path_info, PathInfo) # NOTE: use string paths here for performance reasons diff --git a/dvc/output/base.py b/dvc/output.py similarity index 72% rename from dvc/output/base.py rename to dvc/output.py index 3f3544b39..ad6c00d2e 100644 --- a/dvc/output/base.py +++ b/dvc/output.py @@ -1,10 +1,12 @@ import logging import os +from collections import defaultdict from copy import copy from typing import Type from urllib.parse import urlparse -from voluptuous import Any +from funcy import collecting, project +from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo import dvc.objects as objects import dvc.prompt as prompt @@ -16,18 +18,179 @@ from dvc.exceptions import ( MergeError, RemoteCacheRequiredError, ) -from dvc.hash_info import HashInfo -from dvc.objects import save as osave -from dvc.objects.db import NamedCache -from dvc.objects.errors import ObjectFormatError -from dvc.objects.stage import stage as ostage -from dvc.scheme import Schemes -from ..fs.base import BaseFileSystem +from .fs import get_cloud_fs +from .fs.hdfs import HDFSFileSystem +from .fs.local import LocalFileSystem +from .fs.s3 import S3FileSystem +from .hash_info import HashInfo +from .istextfile import istextfile +from .objects import save as osave +from .objects.db import NamedCache +from .objects.errors import ObjectFormatError +from .objects.stage import stage as ostage +from .scheme import Schemes +from .utils import relpath +from .utils.fs import path_isin logger = logging.getLogger(__name__) +CHECKSUM_SCHEMA = Any( + None, + And(str, Length(max=0), SetTo(None)), + And(Any(str, And(int, Coerce(str))), Length(min=3), Lower), +) + +# NOTE: currently there are only 3 possible checksum names: +# +# 1) md5 (LOCAL, SSH); +# 2) etag (S3); +# 3) checksum (HDFS); +# +# so when a few types of outputs share the same name, we only need +# specify it once. +CHECKSUMS_SCHEMA = { + LocalFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA, + S3FileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA, + HDFSFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA, +} + + +def _get(stage, path, **kwargs): + return Output(stage, path, **kwargs) + + +def loadd_from(stage, d_list): + ret = [] + for d in d_list: + p = d.pop(Output.PARAM_PATH) + cache = d.pop(Output.PARAM_CACHE, True) + metric = d.pop(Output.PARAM_METRIC, False) + plot = d.pop(Output.PARAM_PLOT, False) + persist = d.pop(Output.PARAM_PERSIST, False) + checkpoint = d.pop(Output.PARAM_CHECKPOINT, False) + desc = d.pop(Output.PARAM_DESC, False) + isexec = d.pop(Output.PARAM_ISEXEC, False) + live = d.pop(Output.PARAM_LIVE, False) + ret.append( + _get( + stage, + p, + info=d, + cache=cache, + metric=metric, + plot=plot, + persist=persist, + checkpoint=checkpoint, + desc=desc, + isexec=isexec, + live=live, + ) + ) + return ret + + +def loads_from( + stage, + s_list, + use_cache=True, + metric=False, + plot=False, + persist=False, + checkpoint=False, + isexec=False, + live=False, +): + return [ + _get( + stage, + s, + info={}, + cache=use_cache, + metric=metric, + plot=plot, + persist=persist, + checkpoint=checkpoint, + isexec=isexec, + live=live, + ) + for s in s_list + ] + + +def _split_dict(d, keys): + return project(d, keys), project(d, d.keys() - keys) + + +def _merge_data(s_list): + d = defaultdict(dict) + for key in s_list: + if isinstance(key, str): + d[key].update({}) + continue + if not isinstance(key, dict): + raise ValueError(f"'{type(key).__name__}' not supported.") + + for k, flags in key.items(): + if not isinstance(flags, dict): + raise ValueError( + f"Expected dict for '{k}', got: '{type(flags).__name__}'" + ) + d[k].update(flags) + return d + + +@collecting +def load_from_pipeline(stage, data, typ="outs"): + if typ not in ( + stage.PARAM_OUTS, + stage.PARAM_METRICS, + stage.PARAM_PLOTS, + stage.PARAM_LIVE, + ): + raise ValueError(f"'{typ}' key is not allowed for pipeline files.") + + metric = typ == stage.PARAM_METRICS + plot = typ == stage.PARAM_PLOTS + live = typ == stage.PARAM_LIVE + + if live: + # `live` is single object + data = [data] + + d = _merge_data(data) + + for path, flags in d.items(): + plt_d, live_d = {}, {} + if plot: + from dvc.schema import PLOT_PROPS + + plt_d, flags = _split_dict(flags, keys=PLOT_PROPS.keys()) + if live: + from dvc.schema import LIVE_PROPS + + live_d, flags = _split_dict(flags, keys=LIVE_PROPS.keys()) + extra = project( + flags, + [ + Output.PARAM_CACHE, + Output.PARAM_PERSIST, + Output.PARAM_CHECKPOINT, + ], + ) + + yield _get( + stage, + path, + info={}, + plot=plt_d or plot, + metric=metric, + live=live_d or live, + **extra, + ) + + class OutputDoesNotExistError(DvcException): def __init__(self, path): msg = f"output '{path}' does not exist" @@ -61,11 +224,9 @@ class OutputIsIgnoredError(DvcException): super().__init__(f"Path '{match.file}' is ignored by\n{lines}") -class BaseOutput: +class Output: IS_DEPENDENCY = False - FS_CLS = BaseFileSystem - PARAM_PATH = "path" PARAM_CACHE = "cache" PARAM_CHECKPOINT = "checkpoint" @@ -108,7 +269,6 @@ class BaseOutput: stage, path, info=None, - fs=None, cache=True, metric=False, plot=False, @@ -119,10 +279,20 @@ class BaseOutput: isexec=False, ): self.repo = stage.repo if stage else None - if fs: - self.fs = fs + + fs_cls, fs_config = get_cloud_fs(self.repo, url=path) + self.fs = fs_cls(**fs_config) + path_info = self.fs.path_info + + if ( + self.fs.scheme == "local" + and stage + and path_isin(path, stage.repo.root_dir) + ): + self.def_path = relpath(path, stage.wdir) else: - self.fs = self.FS_CLS() + self.def_path = path + self._validate_output_path(path, stage) # This output (and dependency) objects have too many paths/urls # here is a list and comments: @@ -135,7 +305,6 @@ class BaseOutput: # By resolved path, which contains actual location, # should be absolute and don't contain remote:// refs. self.stage = stage - self.def_path = path self.hash_info = HashInfo.from_dict(info) self.use_cache = False if self.IS_DEPENDENCY else cache self.metric = False if self.IS_DEPENDENCY else metric @@ -145,18 +314,30 @@ class BaseOutput: self.live = live self.desc = desc - self.path_info = self._parse_path(fs, path) + self.path_info = self._parse_path(self.fs, path_info) if self.use_cache and self.odb is None: raise RemoteCacheRequiredError(self.path_info) self.obj = None self.isexec = False if self.IS_DEPENDENCY else isexec - def _parse_path(self, fs, path): - if fs: - parsed = urlparse(path) - return fs.path_info / parsed.path.lstrip("/") - return self.FS_CLS.PATH_CLS(path) + def _parse_path(self, fs, path_info): + if fs.scheme != "local": + return path_info + + parsed = urlparse(self.def_path) + if parsed.scheme != "remote": + # NOTE: we can path either from command line or .dvc file, + # so we should expect both posix and windows style paths. + # PathInfo accepts both, i.e. / works everywhere, \ only on win. + # + # FIXME: if we have Windows path containing / or posix one with \ + # then we have #2059 bug and can't really handle that. + if self.stage and not path_info.is_absolute(): + path_info = self.stage.wdir / path_info + + abs_p = os.path.abspath(os.path.normpath(path_info)) + return fs.PATH_CLS(abs_p) def __repr__(self): return "{class_name}: '{def_path}'".format( @@ -164,15 +345,40 @@ class BaseOutput: ) def __str__(self): - return self.def_path + if self.fs.scheme != "local": + return self.def_path + + if ( + not self.repo + or urlparse(self.def_path).scheme == "remote" + or os.path.isabs(self.def_path) + ): + return str(self.def_path) + + cur_dir = os.getcwd() + if path_isin(cur_dir, self.repo.root_dir): + return relpath(self.path_info, cur_dir) + + return relpath(self.path_info, self.repo.root_dir) @property def scheme(self): - return self.FS_CLS.scheme + return self.fs.scheme @property def is_in_repo(self): - return False + if self.fs.scheme != "local": + return False + + if urlparse(self.def_path).scheme == "remote": + return False + + if os.path.isabs(self.def_path): + return False + + return self.repo and path_isin( + os.path.realpath(self.path_info), self.repo.root_dir + ) @property def use_scm_ignore(self): @@ -265,6 +471,8 @@ class BaseOutput: @property def dvcignore(self): + if self.fs.scheme == "local": + return self.repo.dvcignore return None @property @@ -371,7 +579,13 @@ class BaseOutput: def dumpd(self): ret = copy(self.hash_info.to_dict()) - ret[self.PARAM_PATH] = self.def_path + + if self.is_in_repo: + path = self.path_info.relpath(self.stage.wdir).as_posix() + else: + path = self.def_path + + ret[self.PARAM_PATH] = path if self.IS_DEPENDENCY: return ret @@ -410,7 +624,27 @@ class BaseOutput: return ret def verify_metric(self): - raise DvcException(f"verify metric is not supported for {self.scheme}") + if self.fs.scheme != "local": + raise DvcException( + f"verify metric is not supported for {self.scheme}" + ) + + if not self.metric or self.plot: + return + + path = os.fspath(self.path_info) + if not os.path.exists(path): + return + + name = "metrics" if self.metric else "plot" + if os.path.isdir(path): + msg = "directory '%s' cannot be used as %s." + logger.debug(msg, str(self.path_info), name) + return + + if not istextfile(path, self.fs): + msg = "binary file '{}' cannot be used as {}." + raise DvcException(msg.format(self.path_info, name)) def download(self, to, jobs=None): self.fs.download(self.path_info, to.path_info, jobs=jobs) @@ -505,8 +739,6 @@ class BaseOutput: def transfer( self, source, odb=None, jobs=None, update=False, no_progress_bar=False ): - from dvc.fs import get_cloud_fs - if odb is None: odb = self.odb @@ -738,3 +970,20 @@ class BaseOutput: self.hash_info = merge( self.odb, ancestor_info, self.hash_info, other.hash_info ) + + @property + def fspath(self): + return self.path_info.fspath + + +SCHEMA = CHECKSUMS_SCHEMA.copy() +SCHEMA[Required(Output.PARAM_PATH)] = str +SCHEMA[Output.PARAM_CACHE] = bool +SCHEMA[Output.PARAM_METRIC] = Output.METRIC_SCHEMA +SCHEMA[Output.PARAM_PLOT] = bool +SCHEMA[Output.PARAM_PERSIST] = bool +SCHEMA[Output.PARAM_CHECKPOINT] = bool +SCHEMA[HashInfo.PARAM_SIZE] = int +SCHEMA[HashInfo.PARAM_NFILES] = int +SCHEMA[Output.PARAM_DESC] = str +SCHEMA[Output.PARAM_ISEXEC] = bool diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py deleted file mode 100644 index f292d23a0..000000000 --- a/dvc/output/__init__.py +++ /dev/null @@ -1,241 +0,0 @@ -from collections import defaultdict -from urllib.parse import urlparse - -from funcy import collecting, project -from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo - -from dvc.hash_info import HashInfo -from dvc.output.base import BaseOutput -from dvc.output.hdfs import HDFSOutput -from dvc.output.local import LocalOutput -from dvc.output.s3 import S3Output -from dvc.output.ssh import SSHOutput -from dvc.output.webhdfs import WebHDFSOutput -from dvc.scheme import Schemes - -from ..fs import get_cloud_fs -from ..fs.hdfs import HDFSFileSystem -from ..fs.local import LocalFileSystem -from ..fs.s3 import S3FileSystem -from ..fs.webhdfs import WebHDFSFileSystem - -OUTS_MAP = { - Schemes.HDFS: HDFSOutput, - Schemes.S3: S3Output, - Schemes.SSH: SSHOutput, - Schemes.LOCAL: LocalOutput, - Schemes.WEBHDFS: WebHDFSOutput, -} - -CHECKSUM_SCHEMA = Any( - None, - And(str, Length(max=0), SetTo(None)), - And(Any(str, And(int, Coerce(str))), Length(min=3), Lower), -) - -# NOTE: currently there are only 3 possible checksum names: -# -# 1) md5 (LOCAL, SSH); -# 2) etag (S3); -# 3) checksum (HDFS); -# -# so when a few types of outputs share the same name, we only need -# specify it once. -CHECKSUMS_SCHEMA = { - LocalFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA, - S3FileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA, - HDFSFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA, - WebHDFSFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA, -} - -SCHEMA = CHECKSUMS_SCHEMA.copy() -SCHEMA[Required(BaseOutput.PARAM_PATH)] = str -SCHEMA[BaseOutput.PARAM_CACHE] = bool -SCHEMA[BaseOutput.PARAM_METRIC] = BaseOutput.METRIC_SCHEMA -SCHEMA[BaseOutput.PARAM_PLOT] = bool -SCHEMA[BaseOutput.PARAM_PERSIST] = bool -SCHEMA[BaseOutput.PARAM_CHECKPOINT] = bool -SCHEMA[HashInfo.PARAM_SIZE] = int -SCHEMA[HashInfo.PARAM_NFILES] = int -SCHEMA[BaseOutput.PARAM_DESC] = str -SCHEMA[BaseOutput.PARAM_ISEXEC] = bool - - -def _get( - stage, - p, - info=None, - cache=True, - metric=False, - plot=False, - persist=False, - checkpoint=False, - live=False, - desc=None, - isexec=False, -): - parsed = urlparse(p) - - if parsed.scheme == "remote": - cls, config = get_cloud_fs(stage.repo, name=parsed.netloc) - fs = cls(**config) - return OUTS_MAP[fs.scheme]( - stage, - p, - info, - cache=cache, - fs=fs, - metric=metric, - plot=plot, - persist=persist, - checkpoint=checkpoint, - live=live, - desc=desc, - isexec=isexec, - ) - - out_cls = OUTS_MAP.get(parsed.scheme, LocalOutput) - return out_cls( - stage, - p, - info, - cache=cache, - fs=None, - metric=metric, - plot=plot, - persist=persist, - checkpoint=checkpoint, - live=live, - desc=desc, - isexec=isexec, - ) - - -def loadd_from(stage, d_list): - ret = [] - for d in d_list: - p = d.pop(BaseOutput.PARAM_PATH) - cache = d.pop(BaseOutput.PARAM_CACHE, True) - metric = d.pop(BaseOutput.PARAM_METRIC, False) - plot = d.pop(BaseOutput.PARAM_PLOT, False) - persist = d.pop(BaseOutput.PARAM_PERSIST, False) - checkpoint = d.pop(BaseOutput.PARAM_CHECKPOINT, False) - desc = d.pop(BaseOutput.PARAM_DESC, False) - isexec = d.pop(BaseOutput.PARAM_ISEXEC, False) - live = d.pop(BaseOutput.PARAM_LIVE, False) - ret.append( - _get( - stage, - p, - info=d, - cache=cache, - metric=metric, - plot=plot, - persist=persist, - checkpoint=checkpoint, - desc=desc, - isexec=isexec, - live=live, - ) - ) - return ret - - -def loads_from( - stage, - s_list, - use_cache=True, - metric=False, - plot=False, - persist=False, - checkpoint=False, - isexec=False, - live=False, -): - return [ - _get( - stage, - s, - info={}, - cache=use_cache, - metric=metric, - plot=plot, - persist=persist, - checkpoint=checkpoint, - isexec=isexec, - live=live, - ) - for s in s_list - ] - - -def _split_dict(d, keys): - return project(d, keys), project(d, d.keys() - keys) - - -def _merge_data(s_list): - d = defaultdict(dict) - for key in s_list: - if isinstance(key, str): - d[key].update({}) - continue - if not isinstance(key, dict): - raise ValueError(f"'{type(key).__name__}' not supported.") - - for k, flags in key.items(): - if not isinstance(flags, dict): - raise ValueError( - f"Expected dict for '{k}', got: '{type(flags).__name__}'" - ) - d[k].update(flags) - return d - - -@collecting -def load_from_pipeline(stage, data, typ="outs"): - if typ not in ( - stage.PARAM_OUTS, - stage.PARAM_METRICS, - stage.PARAM_PLOTS, - stage.PARAM_LIVE, - ): - raise ValueError(f"'{typ}' key is not allowed for pipeline files.") - - metric = typ == stage.PARAM_METRICS - plot = typ == stage.PARAM_PLOTS - live = typ == stage.PARAM_LIVE - - if live: - # `live` is single object - data = [data] - - d = _merge_data(data) - - for path, flags in d.items(): - plt_d, live_d = {}, {} - if plot: - from dvc.schema import PLOT_PROPS - - plt_d, flags = _split_dict(flags, keys=PLOT_PROPS.keys()) - if live: - from dvc.schema import LIVE_PROPS - - live_d, flags = _split_dict(flags, keys=LIVE_PROPS.keys()) - extra = project( - flags, - [ - BaseOutput.PARAM_CACHE, - BaseOutput.PARAM_PERSIST, - BaseOutput.PARAM_CHECKPOINT, - ], - ) - - yield _get( - stage, - path, - {}, - plot=plt_d or plot, - metric=metric, - live=live_d or live, - **extra, - ) diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py deleted file mode 100644 index 6a4f16b14..000000000 --- a/dvc/output/hdfs.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.output.base import BaseOutput - -from ..fs.hdfs import HDFSFileSystem - - -class HDFSOutput(BaseOutput): - FS_CLS = HDFSFileSystem diff --git a/dvc/output/local.py b/dvc/output/local.py deleted file mode 100644 index e06d0990a..000000000 --- a/dvc/output/local.py +++ /dev/null @@ -1,111 +0,0 @@ -import logging -import os -from urllib.parse import urlparse - -from dvc.exceptions import DvcException -from dvc.istextfile import istextfile -from dvc.output.base import BaseOutput -from dvc.utils import relpath -from dvc.utils.fs import path_isin - -from ..fs.local import LocalFileSystem - -logger = logging.getLogger(__name__) - - -class LocalOutput(BaseOutput): - FS_CLS = LocalFileSystem - sep = os.sep - - def __init__(self, stage, path, *args, **kwargs): - if stage and path_isin(path, stage.repo.root_dir): - path = relpath(path, stage.wdir) - - super().__init__(stage, path, *args, **kwargs) - if ( - self.is_in_repo - and self.repo - and isinstance(self.repo.fs, LocalFileSystem) - ): - self.fs = self.repo.fs - - def _parse_path(self, fs, path): - parsed = urlparse(path) - if parsed.scheme == "remote": - p = fs.path_info / parsed.path.lstrip("/") - else: - # NOTE: we can path either from command line or .dvc file, - # so we should expect both posix and windows style paths. - # PathInfo accepts both, i.e. / works everywhere, \ only on win. - # - # FIXME: if we have Windows path containing / or posix one with \ - # then we have #2059 bug and can't really handle that. - p = self.FS_CLS.PATH_CLS(path) - if self.stage and not p.is_absolute(): - p = self.stage.wdir / p - - abs_p = os.path.abspath(os.path.normpath(p)) - return self.FS_CLS.PATH_CLS(abs_p) - - def __str__(self): - if ( - not self.repo - or urlparse(self.def_path).scheme == "remote" - or os.path.isabs(self.def_path) - ): - return str(self.def_path) - - cur_dir = os.getcwd() - if path_isin(cur_dir, self.repo.root_dir): - return relpath(self.path_info, cur_dir) - - return relpath(self.path_info, self.repo.root_dir) - - @property - def fspath(self): - return self.path_info.fspath - - @property - def is_in_repo(self): - if urlparse(self.def_path).scheme == "remote": - return False - - if os.path.isabs(self.def_path): - return False - - return self.repo and path_isin( - os.path.realpath(self.path_info), self.repo.root_dir - ) - - def dumpd(self): - ret = super().dumpd() - if self.is_in_repo: - path = self.path_info.relpath(self.stage.wdir).as_posix() - else: - path = self.def_path - - ret[self.PARAM_PATH] = path - - return ret - - def verify_metric(self): - if not self.metric or self.plot: - return - - path = os.fspath(self.path_info) - if not os.path.exists(path): - return - - name = "metrics" if self.metric else "plot" - if os.path.isdir(path): - msg = "directory '%s' cannot be used as %s." - logger.debug(msg, str(self.path_info), name) - return - - if not istextfile(path, self.fs): - msg = "binary file '{}' cannot be used as {}." - raise DvcException(msg.format(self.path_info, name)) - - @property - def dvcignore(self): - return self.repo.dvcignore diff --git a/dvc/output/s3.py b/dvc/output/s3.py deleted file mode 100644 index c7d31352a..000000000 --- a/dvc/output/s3.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.output.base import BaseOutput - -from ..fs.s3 import S3FileSystem - - -class S3Output(BaseOutput): - FS_CLS = S3FileSystem diff --git a/dvc/output/ssh.py b/dvc/output/ssh.py deleted file mode 100644 index 1bc460f39..000000000 --- a/dvc/output/ssh.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.output.base import BaseOutput - -from ..fs.ssh import SSHFileSystem - - -class SSHOutput(BaseOutput): - FS_CLS = SSHFileSystem diff --git a/dvc/output/webhdfs.py b/dvc/output/webhdfs.py deleted file mode 100644 index ee52a1bf9..000000000 --- a/dvc/output/webhdfs.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.output.base import BaseOutput - -from ..fs.webhdfs import WebHDFSFileSystem - - -class WebHDFSOutput(BaseOutput): - FS_CLS = WebHDFSFileSystem diff --git a/dvc/repo/add.py b/dvc/repo/add.py index 73fc48f9e..b7f09658b 100644 --- a/dvc/repo/add.py +++ b/dvc/repo/add.py @@ -156,7 +156,7 @@ def _process_stages( link_failures = [] from dvc.dvcfile import Dvcfile - from ..output.base import OutputDoesNotExistError + from ..output import OutputDoesNotExistError if to_remote or to_cache: # Already verified in the add() diff --git a/dvc/repo/collect.py b/dvc/repo/collect.py index 1d22cd92a..bb68dd4fd 100644 --- a/dvc/repo/collect.py +++ b/dvc/repo/collect.py @@ -6,14 +6,14 @@ from dvc.path_info import PathInfo from dvc.types import DvcPath if TYPE_CHECKING: - from dvc.output.base import BaseOutput + from dvc.output import Output from dvc.repo import Repo logger = logging.getLogger(__name__) -FilterFn = Callable[["BaseOutput"], bool] -Outputs = List["BaseOutput"] +FilterFn = Callable[["Output"], bool] +Outputs = List["Output"] DvcPaths = List[DvcPath] diff --git a/dvc/repo/live.py b/dvc/repo/live.py index 73c51c83d..6d71719a3 100644 --- a/dvc/repo/live.py +++ b/dvc/repo/live.py @@ -8,7 +8,7 @@ from dvc.exceptions import MetricDoesNotExistError, MetricsError logger = logging.getLogger(__name__) if TYPE_CHECKING: - from dvc.output import BaseOutput + from dvc.output import Output from dvc.path_info import PathInfo from dvc.repo import Repo @@ -24,13 +24,13 @@ def create_summary(out): logger.info(f"\nfile://{os.path.abspath(html_path)}") -def summary_path_info(out: "BaseOutput") -> Optional["PathInfo"]: - from dvc.output import BaseOutput +def summary_path_info(out: "Output") -> Optional["PathInfo"]: + from dvc.output import Output assert out.live has_summary = True if isinstance(out.live, dict): - has_summary = out.live.get(BaseOutput.PARAM_LIVE_SUMMARY, True) + has_summary = out.live.get(Output.PARAM_LIVE_SUMMARY, True) if has_summary: return out.path_info.with_suffix(".json") return None diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index c03379a5f..5ff8bb11a 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -7,7 +7,7 @@ from dvc.exceptions import ( NoMetricsParsedError, ) from dvc.fs.repo import RepoFileSystem -from dvc.output import BaseOutput +from dvc.output import Output from dvc.path_info import PathInfo from dvc.repo import locked from dvc.repo.collect import collect @@ -18,11 +18,11 @@ from dvc.utils.serialize import YAMLFileCorruptedError, load_yaml logger = logging.getLogger(__name__) -def _is_metric(out: BaseOutput) -> bool: +def _is_metric(out: Output) -> bool: return bool(out.metric) or bool(out.live) -def _to_path_infos(metrics: List[BaseOutput]) -> List[PathInfo]: +def _to_path_infos(metrics: List[Output]) -> List[PathInfo]: result = [] for out in metrics: if out.metric: diff --git a/dvc/repo/params/show.py b/dvc/repo/params/show.py index 0e1138d0a..65b5d90f0 100644 --- a/dvc/repo/params/show.py +++ b/dvc/repo/params/show.py @@ -13,7 +13,7 @@ from dvc.stage import PipelineStage from dvc.utils.serialize import LOADERS, ParseError if TYPE_CHECKING: - from dvc.output.base import BaseOutput + from dvc.output import Output from dvc.repo import Repo from dvc.types import DvcPath @@ -24,13 +24,13 @@ class NoParamsError(DvcException): pass -def _is_params(dep: "BaseOutput"): +def _is_params(dep: "Output"): return isinstance(dep, ParamsDependency) def _collect_configs( repo: "Repo", rev, targets=None -) -> Tuple[List["BaseOutput"], List["DvcPath"]]: +) -> Tuple[List["Output"], List["DvcPath"]]: params, path_infos = collect( repo, targets=targets or [], diff --git a/dvc/repo/plots/__init__.py b/dvc/repo/plots/__init__.py index 4dca98210..6f76cfb3a 100644 --- a/dvc/repo/plots/__init__.py +++ b/dvc/repo/plots/__init__.py @@ -14,7 +14,7 @@ from dvc.types import StrPath from dvc.utils import relpath if TYPE_CHECKING: - from dvc.output import BaseOutput + from dvc.output import Output from dvc.repo import Repo logger = logging.getLogger(__name__) @@ -254,7 +254,7 @@ class Plots: write(path, plots, metrics, html_template_path) -def _is_plot(out: "BaseOutput") -> bool: +def _is_plot(out: "Output") -> bool: return bool(out.plot) or bool(out.live) @@ -278,7 +278,7 @@ def _collect_plots( return result -def _plot_props(out: "BaseOutput") -> Dict: +def _plot_props(out: "Output") -> Dict: from dvc.schema import PLOT_PROPS if not (out.plot or out.live): diff --git a/dvc/schema.py b/dvc/schema.py index 7a2c690e3..0d6e910f7 100644 --- a/dvc/schema.py +++ b/dvc/schema.py @@ -2,7 +2,7 @@ from voluptuous import Any, Optional, Required, Schema from dvc import dependency, output from dvc.hash_info import HashInfo -from dvc.output import CHECKSUMS_SCHEMA, BaseOutput +from dvc.output import CHECKSUMS_SCHEMA, Output from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD from dvc.parsing.versions import SCHEMA_KWD, lockfile_version_schema from dvc.stage.params import StageParams @@ -26,7 +26,7 @@ DATA_SCHEMA = { Required("path"): str, HashInfo.PARAM_SIZE: int, HashInfo.PARAM_NFILES: int, - BaseOutput.PARAM_ISEXEC: bool, + Output.PARAM_ISEXEC: bool, } LOCK_FILE_STAGE_SCHEMA = { Required(StageParams.PARAM_CMD): Any(str, list), @@ -45,28 +45,28 @@ LOCKFILE_V2_SCHEMA = { OUT_PSTAGE_DETAILED_SCHEMA = { str: { - BaseOutput.PARAM_CACHE: bool, - BaseOutput.PARAM_PERSIST: bool, - BaseOutput.PARAM_CHECKPOINT: bool, - BaseOutput.PARAM_DESC: str, + Output.PARAM_CACHE: bool, + Output.PARAM_PERSIST: bool, + Output.PARAM_CHECKPOINT: bool, + Output.PARAM_DESC: str, } } PLOT_PROPS = { - BaseOutput.PARAM_PLOT_TEMPLATE: str, - BaseOutput.PARAM_PLOT_X: str, - BaseOutput.PARAM_PLOT_Y: str, - BaseOutput.PARAM_PLOT_X_LABEL: str, - BaseOutput.PARAM_PLOT_Y_LABEL: str, - BaseOutput.PARAM_PLOT_TITLE: str, - BaseOutput.PARAM_PLOT_HEADER: bool, + Output.PARAM_PLOT_TEMPLATE: str, + Output.PARAM_PLOT_X: str, + Output.PARAM_PLOT_Y: str, + Output.PARAM_PLOT_X_LABEL: str, + Output.PARAM_PLOT_Y_LABEL: str, + Output.PARAM_PLOT_TITLE: str, + Output.PARAM_PLOT_HEADER: bool, } PLOT_PROPS_SCHEMA = {**OUT_PSTAGE_DETAILED_SCHEMA[str], **PLOT_PROPS} PLOT_PSTAGE_SCHEMA = {str: Any(PLOT_PROPS_SCHEMA, [PLOT_PROPS_SCHEMA])} LIVE_PROPS = { - BaseOutput.PARAM_LIVE_SUMMARY: bool, - BaseOutput.PARAM_LIVE_HTML: bool, + Output.PARAM_LIVE_SUMMARY: bool, + Output.PARAM_LIVE_HTML: bool, } LIVE_PROPS_SCHEMA = {**PLOT_PROPS_SCHEMA, **LIVE_PROPS} LIVE_PSTAGE_SCHEMA = {str: LIVE_PROPS_SCHEMA} diff --git a/dvc/stage/__init__.py b/dvc/stage/__init__.py index ae6a6e10b..6ab42e794 100644 --- a/dvc/stage/__init__.py +++ b/dvc/stage/__init__.py @@ -268,7 +268,7 @@ class Stage(params.StageParams): env: Env = {} if out.live: from dvc.env import DVCLIVE_HTML, DVCLIVE_PATH, DVCLIVE_SUMMARY - from dvc.output import BaseOutput + from dvc.output import Output from dvc.schema import LIVE_PROPS env[DVCLIVE_PATH] = str(out.path_info) @@ -276,10 +276,10 @@ class Stage(params.StageParams): config = project(out.live, LIVE_PROPS) env[DVCLIVE_SUMMARY] = str( - int(config.get(BaseOutput.PARAM_LIVE_SUMMARY, True)) + int(config.get(Output.PARAM_LIVE_SUMMARY, True)) ) env[DVCLIVE_HTML] = str( - int(config.get(BaseOutput.PARAM_LIVE_HTML, True)) + int(config.get(Output.PARAM_LIVE_HTML, True)) ) elif out.checkpoint and checkpoint_func: from dvc.env import DVC_CHECKPOINT @@ -453,7 +453,7 @@ class Stage(params.StageParams): raise def save_outs(self, allow_missing=False): - from dvc.output.base import OutputDoesNotExistError + from dvc.output import OutputDoesNotExistError for out in self.outs: try: @@ -484,7 +484,7 @@ class Stage(params.StageParams): @rwlocked(write=["outs"]) def commit(self, allow_missing=False, filter_info=None): - from dvc.output.base import OutputDoesNotExistError + from dvc.output import OutputDoesNotExistError link_failures = [] for out in self.filter_outs(filter_info): diff --git a/dvc/stage/monitor.py b/dvc/stage/monitor.py index 77af843c3..db6e16d5b 100644 --- a/dvc/stage/monitor.py +++ b/dvc/stage/monitor.py @@ -11,7 +11,7 @@ from dvc.stage.decorators import relock_repo from dvc.stage.exceptions import StageCmdFailedError if TYPE_CHECKING: - from dvc.output import BaseOutput + from dvc.output import Output from dvc.stage import Stage @@ -84,9 +84,7 @@ class LiveTask(MonitorTask): SIGNAL_FILE = "DVC_LIVE" error_cls = LiveKilledError - def __init__( - self, stage: "Stage", out: "BaseOutput", proc: subprocess.Popen - ): + def __init__(self, stage: "Stage", out: "Output", proc: subprocess.Popen): super().__init__(stage, functools.partial(create_summary, out), proc) def after_run(self): diff --git a/dvc/stage/serialize.py b/dvc/stage/serialize.py index f726cc8a8..19f92319f 100644 --- a/dvc/stage/serialize.py +++ b/dvc/stage/serialize.py @@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, List, no_type_check from funcy import post_processing from dvc.dependency import ParamsDependency -from dvc.output import BaseOutput +from dvc.output import Output from dvc.utils.collections import apply_diff from dvc.utils.serialize import parse_yaml_for_update @@ -22,12 +22,12 @@ PARAM_PATH = ParamsDependency.PARAM_PATH PARAM_DEPS = StageParams.PARAM_DEPS PARAM_OUTS = StageParams.PARAM_OUTS -PARAM_CACHE = BaseOutput.PARAM_CACHE -PARAM_METRIC = BaseOutput.PARAM_METRIC -PARAM_PLOT = BaseOutput.PARAM_PLOT -PARAM_PERSIST = BaseOutput.PARAM_PERSIST -PARAM_CHECKPOINT = BaseOutput.PARAM_CHECKPOINT -PARAM_DESC = BaseOutput.PARAM_DESC +PARAM_CACHE = Output.PARAM_CACHE +PARAM_METRIC = Output.PARAM_METRIC +PARAM_PLOT = Output.PARAM_PLOT +PARAM_PERSIST = Output.PARAM_PERSIST +PARAM_CHECKPOINT = Output.PARAM_CHECKPOINT +PARAM_DESC = Output.PARAM_DESC DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE @@ -60,7 +60,7 @@ def _serialize_out(out): @no_type_check -def _serialize_outs(outputs: List[BaseOutput]): +def _serialize_outs(outputs: List[Output]): outs, metrics, plots, live = [], [], [], None for out in sort_by_path(outputs): bucket = outs diff --git a/dvc/stage/utils.py b/dvc/stage/utils.py index c0976d403..62f5299b3 100644 --- a/dvc/stage/utils.py +++ b/dvc/stage/utils.py @@ -84,7 +84,7 @@ def _load_live_output( live_html=False, **kwargs, ): - from dvc.output import BaseOutput, loads_from + from dvc.output import Output, loads_from outs = [] if live or live_no_cache: @@ -96,8 +96,8 @@ def _load_live_output( [path], use_cache=not bool(live_no_cache), live={ - BaseOutput.PARAM_LIVE_SUMMARY: live_summary, - BaseOutput.PARAM_LIVE_HTML: live_html, + Output.PARAM_LIVE_SUMMARY: live_summary, + Output.PARAM_LIVE_HTML: live_html, }, ) @@ -170,7 +170,7 @@ def check_missing_outputs(stage): def compute_md5(stage): - from dvc.output.base import BaseOutput + from dvc.output import Output from ..utils import dict_md5 @@ -193,11 +193,11 @@ def compute_md5(stage): exclude=[ stage.PARAM_LOCKED, # backward compatibility stage.PARAM_FROZEN, - BaseOutput.PARAM_DESC, - BaseOutput.PARAM_METRIC, - BaseOutput.PARAM_PERSIST, - BaseOutput.PARAM_CHECKPOINT, - BaseOutput.PARAM_ISEXEC, + Output.PARAM_DESC, + Output.PARAM_METRIC, + Output.PARAM_PERSIST, + Output.PARAM_CHECKPOINT, + Output.PARAM_ISEXEC, HashInfo.PARAM_SIZE, HashInfo.PARAM_NFILES, ], diff --git a/tests/func/test_add.py b/tests/func/test_add.py index 14d86e5fd..ece3cc33b 100644 --- a/tests/func/test_add.py +++ b/tests/func/test_add.py @@ -24,7 +24,7 @@ from dvc.fs.local import LocalFileSystem from dvc.hash_info import HashInfo from dvc.main import main from dvc.objects.db import ODBManager -from dvc.output.base import OutputAlreadyTrackedError, OutputIsStageFileError +from dvc.output import OutputAlreadyTrackedError, OutputIsStageFileError from dvc.stage import Stage from dvc.stage.exceptions import ( StageExternalOutputsError, diff --git a/tests/func/test_commit.py b/tests/func/test_commit.py index fb4d4c687..958c1983c 100644 --- a/tests/func/test_commit.py +++ b/tests/func/test_commit.py @@ -4,7 +4,7 @@ import pytest from dvc.dependency.base import DependencyDoesNotExistError from dvc.dvcfile import PIPELINE_FILE -from dvc.output.base import OutputDoesNotExistError +from dvc.output import OutputDoesNotExistError from dvc.stage.exceptions import StageCommitError from dvc.utils.serialize import dump_yaml, load_yaml diff --git a/tests/func/test_dependency.py b/tests/func/test_dependency.py deleted file mode 100644 index 7bc92295a..000000000 --- a/tests/func/test_dependency.py +++ /dev/null @@ -1,21 +0,0 @@ -import copy - -import pytest - -from dvc.dependency import DEP_MAP, _get -from dvc.stage import Stage -from tests.func.test_output import TESTS as OUT_TESTS - -TESTS = copy.copy(OUT_TESTS) -TESTS.append(("http://example.com/path/to/file", "http")) -TESTS.append(("https://example.com/path/to/file", "https")) - - -def _get_dep(dvc, path): - return _get(Stage(dvc), path, None) - - -@pytest.mark.parametrize("url,scheme", TESTS) -def test_scheme(dvc, url, scheme): - # pylint: disable=unidiomatic-typecheck - assert type(_get_dep(dvc, url)) == DEP_MAP[scheme] diff --git a/tests/func/test_ignore.py b/tests/func/test_ignore.py index 3ef3088a1..c0f8d8e75 100644 --- a/tests/func/test_ignore.py +++ b/tests/func/test_ignore.py @@ -6,7 +6,7 @@ import pytest from dvc.exceptions import DvcIgnoreInCollectedDirError from dvc.ignore import DvcIgnore, DvcIgnorePatterns -from dvc.output.base import OutputIsIgnoredError +from dvc.output import OutputIsIgnoredError from dvc.pathspec_math import PatternInfo, merge_patterns from dvc.repo import Repo from dvc.types import List diff --git a/tests/func/test_output.py b/tests/func/test_output.py deleted file mode 100644 index 1ca7da832..000000000 --- a/tests/func/test_output.py +++ /dev/null @@ -1,30 +0,0 @@ -import pytest - -from dvc.output import OUTS_MAP, _get -from dvc.stage import Stage - -MARKERS = [pytest.mark.hdfs] -TESTS = [ - ("s3://bucket/path", "s3"), - ("ssh://example.com:/dir/path", "ssh"), - ("path/to/file", "local"), - ("path\\to\\file", "local"), - ("file", "local"), - ("./file", "local"), - (".\\file", "local"), - ("../file", "local"), - ("..\\file", "local"), - ("unknown://path", "local"), - pytest.param("hdfs://example.com/dir/path", "hdfs", marks=MARKERS), - pytest.param("hdfs://example.com/dir/path", "hdfs", marks=MARKERS), -] - - -def _get_out(dvc, path): - return _get(Stage(dvc), path, None, None, None, None) - - -@pytest.mark.parametrize("url,scheme", TESTS) -def test_scheme(dvc, url, scheme): - # pylint: disable=unidiomatic-typecheck - assert type(_get_out(dvc, url)) == OUTS_MAP[scheme] diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py index b9616febd..73dd2f35c 100644 --- a/tests/func/test_repro.py +++ b/tests/func/test_repro.py @@ -15,7 +15,7 @@ from dvc.exceptions import ( ) from dvc.fs.local import LocalFileSystem from dvc.main import main -from dvc.output.base import BaseOutput +from dvc.output import Output from dvc.stage import Stage from dvc.stage.exceptions import StageFileDoesNotExistError from dvc.system import System @@ -883,9 +883,9 @@ class TestReproAlreadyCached(TestRepro): ) patch_checkout = patch.object( - BaseOutput, + Output, "checkout", - side_effect=BaseOutput.checkout, + side_effect=Output.checkout, autospec=True, ) diff --git a/tests/func/test_run_cache.py b/tests/func/test_run_cache.py index bf84550bd..5061268ed 100644 --- a/tests/func/test_run_cache.py +++ b/tests/func/test_run_cache.py @@ -150,7 +150,7 @@ def test_memory_runs_of_multiple_stages(tmp_dir, dvc, run_copy, mocker): def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote): - from dvc.output import base + import dvc.output as dvc_output tmp_dir.gen("foo", "foo") stage = run_copy("foo", "bar", name="copy-foo-bar") @@ -159,7 +159,7 @@ def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote): mock_restore = mocker.spy(dvc.stage_cache, "restore") mock_run = mocker.patch("dvc.stage.run.cmd_run") - mock_checkout = mocker.spy(base, "checkout") + mock_checkout = mocker.spy(dvc_output, "checkout") # removing any information that `dvc` could use to re-generate from (tmp_dir / "bar").unlink() diff --git a/tests/func/test_run_single_stage.py b/tests/func/test_run_single_stage.py index cf032091d..a13fd825f 100644 --- a/tests/func/test_run_single_stage.py +++ b/tests/func/test_run_single_stage.py @@ -19,8 +19,7 @@ from dvc.exceptions import ( StagePathAsOutputError, ) from dvc.main import main -from dvc.output import BaseOutput -from dvc.output.base import OutputIsStageFileError +from dvc.output import Output, OutputIsStageFileError from dvc.repo import Repo as DvcRepo from dvc.stage import Stage from dvc.stage.exceptions import ( @@ -763,7 +762,7 @@ class TestRunPersist(TestDvc): def stage_should_contain_persist_flag(self, stage_file): stage_file_content = load_yaml(stage_file) self.assertEqual( - True, stage_file_content["outs"][0][BaseOutput.PARAM_PERSIST] + True, stage_file_content["outs"][0][Output.PARAM_PERSIST] ) def should_append_upon_repro(self, file, stage_file): diff --git a/tests/func/test_stage.py b/tests/func/test_stage.py index 783a1dad9..f91f46d50 100644 --- a/tests/func/test_stage.py +++ b/tests/func/test_stage.py @@ -6,7 +6,7 @@ import pytest from dvc.dvcfile import SingleStageFile from dvc.fs.local import LocalFileSystem from dvc.main import main -from dvc.output.local import LocalOutput +from dvc.output import Output from dvc.repo import Repo, lock_repo from dvc.stage import PipelineStage, Stage from dvc.stage.exceptions import StageFileFormatError @@ -55,15 +55,15 @@ def test_empty_list(): def test_list(): lst = [ - {LocalOutput.PARAM_PATH: "foo", LocalFileSystem.PARAM_CHECKSUM: "123"}, - {LocalOutput.PARAM_PATH: "bar", LocalFileSystem.PARAM_CHECKSUM: None}, - {LocalOutput.PARAM_PATH: "baz"}, + {Output.PARAM_PATH: "foo", LocalFileSystem.PARAM_CHECKSUM: "123"}, + {Output.PARAM_PATH: "bar", LocalFileSystem.PARAM_CHECKSUM: None}, + {Output.PARAM_PATH: "baz"}, ] d = {Stage.PARAM_DEPS: lst} SingleStageFile.validate(d) - lst[0][LocalOutput.PARAM_CACHE] = True - lst[1][LocalOutput.PARAM_CACHE] = False + lst[0][Output.PARAM_CACHE] = True + lst[1][Output.PARAM_CACHE] = False d = {Stage.PARAM_OUTS: lst} SingleStageFile.validate(d) diff --git a/tests/unit/dependency/test_azure.py b/tests/unit/dependency/test_azure.py deleted file mode 100644 index fff791061..000000000 --- a/tests/unit/dependency/test_azure.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.dependency.azure import AzureDependency -from tests.unit.dependency.test_local import TestLocalDependency - - -class TestAzureDependency(TestLocalDependency): - def _get_cls(self): - return AzureDependency diff --git a/tests/unit/dependency/test_dependency.py b/tests/unit/dependency/test_dependency.py new file mode 100644 index 000000000..6f7501bb6 --- /dev/null +++ b/tests/unit/dependency/test_dependency.py @@ -0,0 +1,12 @@ +import pytest + +from dvc.dependency import Dependency +from dvc.stage import Stage + + +def test_save_missing(dvc, mocker): + stage = Stage(dvc) + dep = Dependency(stage, "path") + with mocker.patch.object(dep.fs, "exists", return_value=False): + with pytest.raises(dep.DoesNotExistError): + dep.save() diff --git a/tests/unit/dependency/test_gs.py b/tests/unit/dependency/test_gs.py deleted file mode 100644 index 67daf0a61..000000000 --- a/tests/unit/dependency/test_gs.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.dependency.gs import GSDependency -from tests.unit.dependency.test_local import TestLocalDependency - - -class TestGSDependency(TestLocalDependency): - def _get_cls(self): - return GSDependency diff --git a/tests/unit/dependency/test_hdfs.py b/tests/unit/dependency/test_hdfs.py deleted file mode 100644 index 0057f8f3e..000000000 --- a/tests/unit/dependency/test_hdfs.py +++ /dev/null @@ -1,10 +0,0 @@ -import pytest - -from dvc.dependency.hdfs import HDFSDependency -from tests.unit.dependency.test_local import TestLocalDependency - - -@pytest.mark.hdfs -class TestHDFSDependency(TestLocalDependency): - def _get_cls(self): - return HDFSDependency diff --git a/tests/unit/dependency/test_http.py b/tests/unit/dependency/test_http.py deleted file mode 100644 index f28bed782..000000000 --- a/tests/unit/dependency/test_http.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.dependency.http import HTTPDependency -from tests.unit.dependency.test_local import TestLocalDependency - - -class TestHTTPDependency(TestLocalDependency): - def _get_cls(self): - return HTTPDependency diff --git a/tests/unit/dependency/test_local.py b/tests/unit/dependency/test_local.py deleted file mode 100644 index a4505fc07..000000000 --- a/tests/unit/dependency/test_local.py +++ /dev/null @@ -1,20 +0,0 @@ -import mock - -from dvc.dependency import LocalDependency -from dvc.stage import Stage -from tests.basic_env import TestDvc - - -class TestLocalDependency(TestDvc): - def _get_cls(self): - return LocalDependency - - def _get_dependency(self): - stage = Stage(self.dvc) - return self._get_cls()(stage, "path") - - def test_save_missing(self): - d = self._get_dependency() - with mock.patch.object(d.fs, "exists", return_value=False): - with self.assertRaises(d.DoesNotExistError): - d.save() diff --git a/tests/unit/dependency/test_s3.py b/tests/unit/dependency/test_s3.py deleted file mode 100644 index 7f9baad3c..000000000 --- a/tests/unit/dependency/test_s3.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.dependency.s3 import S3Dependency -from tests.unit.dependency.test_local import TestLocalDependency - - -class TestS3Dependency(TestLocalDependency): - def _get_cls(self): - return S3Dependency diff --git a/tests/unit/dependency/test_ssh.py b/tests/unit/dependency/test_ssh.py deleted file mode 100644 index ef09e7096..000000000 --- a/tests/unit/dependency/test_ssh.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.dependency.ssh import SSHDependency -from tests.unit.dependency.test_local import TestLocalDependency - - -class TestSSHDependency(TestLocalDependency): - def _get_cls(self): - return SSHDependency diff --git a/tests/unit/dependency/test_webdav.py b/tests/unit/dependency/test_webdav.py deleted file mode 100644 index 6e8e07994..000000000 --- a/tests/unit/dependency/test_webdav.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.dependency.webdav import WebDAVDependency -from tests.unit.dependency.test_local import TestLocalDependency - - -class TestWebDAVDependency(TestLocalDependency): - def _get_cls(self): - return WebDAVDependency diff --git a/tests/unit/fs/test_fs.py b/tests/unit/fs/test_fs.py new file mode 100644 index 000000000..511486ed1 --- /dev/null +++ b/tests/unit/fs/test_fs.py @@ -0,0 +1,31 @@ +import pytest + +from dvc.fs import get_fs_cls +from dvc.fs.hdfs import HDFSFileSystem +from dvc.fs.http import HTTPFileSystem +from dvc.fs.https import HTTPSFileSystem +from dvc.fs.local import LocalFileSystem +from dvc.fs.s3 import S3FileSystem +from dvc.fs.ssh import SSHFileSystem + + +@pytest.mark.parametrize( + "url, cls", + [ + ("s3://bucket/path", S3FileSystem), + ("ssh://example.com:/dir/path", SSHFileSystem), + ("hdfs://example.com/dir/path", HDFSFileSystem), + ("http://example.com/path/to/file", HTTPFileSystem), + ("https://example.com/path/to/file", HTTPSFileSystem), + ("path/to/file", LocalFileSystem), + ("path\\to\\file", LocalFileSystem), + ("file", LocalFileSystem), + ("./file", LocalFileSystem), + (".\\file", LocalFileSystem), + ("../file", LocalFileSystem), + ("..\\file", LocalFileSystem), + ("unknown://path", LocalFileSystem), + ], +) +def test_get_fs_cls(url, cls): + assert get_fs_cls({"url": url}) == cls diff --git a/tests/unit/output/test_hdfs.py b/tests/unit/output/test_hdfs.py deleted file mode 100644 index 33a1a3115..000000000 --- a/tests/unit/output/test_hdfs.py +++ /dev/null @@ -1,10 +0,0 @@ -import pytest - -from dvc.output.hdfs import HDFSOutput -from tests.unit.output.test_local import TestLocalOutput - - -@pytest.mark.hdfs -class TestHDFSOutput(TestLocalOutput): - def _get_cls(self): - return HDFSOutput diff --git a/tests/unit/output/test_load.py b/tests/unit/output/test_load.py index 922172f09..41e07d361 100644 --- a/tests/unit/output/test_load.py +++ b/tests/unit/output/test_load.py @@ -3,7 +3,9 @@ import pytest from dvc import output -from dvc.output import LocalOutput, S3Output +from dvc.fs.local import LocalFileSystem +from dvc.fs.s3 import S3FileSystem +from dvc.output import Output from dvc.stage import Stage @@ -35,7 +37,8 @@ def test_load_from_pipeline(dvc, out_type, type_test_func): assert len(outs) == 6 for i, out in enumerate(outs, start=1): - assert isinstance(out, LocalOutput) + assert isinstance(out, Output) + assert isinstance(out.fs, LocalFileSystem) assert out.def_path == f"file{i}" assert out.use_cache == (out.def_path in cached_outs) assert out.persist == (out.def_path in persisted_outs) @@ -55,7 +58,8 @@ def test_load_from_pipeline_accumulates_flag(dvc): "outs", ) for out in outs: - assert isinstance(out, LocalOutput) + assert isinstance(out, Output) + assert isinstance(out.fs, LocalFileSystem) assert not out.plot and not out.metric assert not out.hash_info @@ -68,7 +72,8 @@ def test_load_remote_files_from_pipeline(dvc): (out,) = output.load_from_pipeline( stage, [{"s3://dvc-test/file.txt": {"cache": False}}], typ="metrics" ) - assert isinstance(out, S3Output) + assert isinstance(out, Output) + assert isinstance(out.fs, S3FileSystem) assert not out.plot and out.metric assert not out.persist assert not out.hash_info @@ -105,12 +110,14 @@ def test_plots_load_from_pipeline(dvc): ], "plots", ) - assert isinstance(outs[0], LocalOutput) + assert isinstance(outs[0], Output) + assert isinstance(outs[0].fs, LocalFileSystem) assert outs[0].use_cache assert outs[0].plot is True and not outs[0].metric assert not outs[0].persist - assert isinstance(outs[1], LocalOutput) + assert isinstance(outs[1], Output) + assert isinstance(outs[1].fs, LocalFileSystem) assert not outs[1].use_cache assert outs[1].plot == {"x": 3} and not outs[1].metric assert outs[1].persist diff --git a/tests/unit/output/test_local.py b/tests/unit/output/test_local.py index d758fa42a..39f551e40 100644 --- a/tests/unit/output/test_local.py +++ b/tests/unit/output/test_local.py @@ -3,42 +3,27 @@ import os from mock import patch from dvc.hash_info import HashInfo -from dvc.output import LocalOutput +from dvc.output import Output from dvc.stage import Stage from dvc.utils import relpath from tests.basic_env import TestDvc -class TestLocalOutput(TestDvc): - def _get_cls(self): - return LocalOutput - - def _get_output(self): - stage = Stage(self.dvc) - return self._get_cls()(stage, "path", cache=False) - - def test_save_missing(self): - o = self._get_output() - with patch.object(o.fs, "exists", return_value=False): - with self.assertRaises(o.DoesNotExistError): - o.save() - - def test_str_workdir_outside_repo(tmp_dir, erepo_dir): stage = Stage(erepo_dir.dvc) - output = LocalOutput(stage, "path", cache=False) + output = Output(stage, "path", cache=False) assert relpath("path", erepo_dir.dvc.root_dir) == str(output) def test_str_workdir_inside_repo(dvc): stage = Stage(dvc) - output = LocalOutput(stage, "path", cache=False) + output = Output(stage, "path", cache=False) assert "path" == str(output) stage = Stage(dvc, wdir="some_folder") - output = LocalOutput(stage, "path", cache=False) + output = Output(stage, "path", cache=False) assert os.path.join("some_folder", "path") == str(output) @@ -48,7 +33,7 @@ def test_str_on_local_absolute_path(dvc): rel_path = os.path.join("path", "to", "file") abs_path = os.path.abspath(rel_path) - output = LocalOutput(stage, abs_path, cache=False) + output = Output(stage, abs_path, cache=False) assert output.def_path == rel_path assert output.path_info.fspath == abs_path @@ -60,7 +45,7 @@ def test_str_on_external_absolute_path(dvc): rel_path = os.path.join("..", "path", "to", "file") abs_path = os.path.abspath(rel_path) - output = LocalOutput(stage, abs_path, cache=False) + output = Output(stage, abs_path, cache=False) assert output.def_path == abs_path assert output.path_info.fspath == abs_path @@ -70,7 +55,7 @@ def test_str_on_external_absolute_path(dvc): class TestGetFilesNumber(TestDvc): def _get_output(self): stage = Stage(self.dvc) - return LocalOutput(stage, "path") + return Output(stage, "path") def test_return_0_on_no_cache(self): o = self._get_output() @@ -82,7 +67,7 @@ class TestGetFilesNumber(TestDvc): o.hash_info = HashInfo("md5", "12345678.dir", nfiles=2) self.assertEqual(2, o.get_files_number()) - @patch.object(LocalOutput, "is_dir_checksum", False) + @patch.object(Output, "is_dir_checksum", False) def test_return_1_on_single_file_cache(self): o = self._get_output() o.hash_info = HashInfo("md5", "12345678") diff --git a/tests/unit/output/test_output.py b/tests/unit/output/test_output.py index a60201a7c..667453912 100644 --- a/tests/unit/output/test_output.py +++ b/tests/unit/output/test_output.py @@ -1,4 +1,5 @@ import logging +import os import pytest from funcy import first @@ -6,7 +7,17 @@ from voluptuous import MultipleInvalid, Schema from dvc.ignore import _no_match from dvc.objects.db import NamedCache -from dvc.output import CHECKSUM_SCHEMA, BaseOutput +from dvc.output import CHECKSUM_SCHEMA, Output +from dvc.path_info import PathInfo +from dvc.stage import Stage + + +def test_save_missing(dvc, mocker): + stage = Stage(dvc) + out = Output(stage, "path", cache=False) + with mocker.patch.object(out.fs, "exists", return_value=False): + with pytest.raises(out.DoesNotExistError): + out.save() @pytest.mark.parametrize( @@ -73,20 +84,19 @@ def test_get_used_cache(exists, expected_message, mocker, caplog): stage = mocker.MagicMock() mocker.patch.object(stage, "__str__", return_value="stage: 'stage.dvc'") mocker.patch.object(stage, "addressing", "stage.dvc") - mocker.patch.object(stage, "wdir", ".") + mocker.patch.object(stage, "wdir", PathInfo(os.getcwd())) + mocker.patch.object(stage.repo, "root_dir", os.getcwd()) + mocker.patch.object(stage.repo.dvcignore, "is_ignored", return_value=False) mocker.patch.object( - stage.repo.fs.dvcignore, "is_ignored", return_value=False - ) - mocker.patch.object( - stage.repo.fs.dvcignore, "check_ignore", return_value=_no_match("path") + stage.repo.dvcignore, "check_ignore", return_value=_no_match("path") ) - output = BaseOutput(stage, "path") + output = Output(stage, "path") mocker.patch.object(output, "use_cache", True) mocker.patch.object(stage, "is_repo_import", False) mocker.patch.object( - BaseOutput, "exists", new_callable=mocker.PropertyMock + Output, "exists", new_callable=mocker.PropertyMock ).return_value = exists with caplog.at_level(logging.WARNING, logger="dvc"): diff --git a/tests/unit/output/test_s3.py b/tests/unit/output/test_s3.py deleted file mode 100644 index 34b308018..000000000 --- a/tests/unit/output/test_s3.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.output.s3 import S3Output -from tests.unit.output.test_local import TestLocalOutput - - -class TestS3Output(TestLocalOutput): - def _get_cls(self): - return S3Output diff --git a/tests/unit/output/test_ssh.py b/tests/unit/output/test_ssh.py deleted file mode 100644 index 0f829c53c..000000000 --- a/tests/unit/output/test_ssh.py +++ /dev/null @@ -1,7 +0,0 @@ -from dvc.output.ssh import SSHOutput -from tests.unit.output.test_local import TestLocalOutput - - -class TestSSHOutput(TestLocalOutput): - def _get_cls(self): - return SSHOutput diff --git a/tests/unit/stage/test_cache.py b/tests/unit/stage/test_cache.py index 99d9f124e..233cf7b30 100644 --- a/tests/unit/stage/test_cache.py +++ b/tests/unit/stage/test_cache.py @@ -2,7 +2,7 @@ import os import pytest -from dvc.output import base +import dvc.output as dvc_output def test_stage_cache(tmp_dir, dvc, mocker): @@ -44,7 +44,7 @@ def test_stage_cache(tmp_dir, dvc, mocker): assert os.path.isfile(cache_file) run_spy = mocker.patch("dvc.stage.run.cmd_run") - checkout_spy = mocker.spy(base, "checkout") + checkout_spy = mocker.spy(dvc_output, "checkout") with dvc.lock: stage.run() @@ -97,7 +97,7 @@ def test_stage_cache_params(tmp_dir, dvc, mocker): assert os.path.isfile(cache_file) run_spy = mocker.patch("dvc.stage.run.cmd_run") - checkout_spy = mocker.spy(base, "checkout") + checkout_spy = mocker.spy(dvc_output, "checkout") with dvc.lock: stage.run() @@ -151,7 +151,7 @@ def test_stage_cache_wdir(tmp_dir, dvc, mocker): assert os.path.isfile(cache_file) run_spy = mocker.patch("dvc.stage.run.cmd_run") - checkout_spy = mocker.spy(base, "checkout") + checkout_spy = mocker.spy(dvc_output, "checkout") with dvc.lock: stage.run()