From 8a7ccab82fb26be7c6c6256d978c3ffa1b683a6f Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 25 May 2021 19:25:09 +0300 Subject: [PATCH] fs: deps/outs: get rid of types (#6054) Pre-requisite for removing fs.path_info and switching to regular strings as paths in filesystems. With new filesystems we no longer need the zoo of different dependency and output types, as most of them are now just handled by the underlying filesystem. Though deps/outs are still due for a refactor in other aspects. --- dvc/command/stage.py | 6 +- dvc/dependency/__init__.py | 51 +--- dvc/dependency/azure.py | 8 - dvc/dependency/base.py | 3 +- dvc/dependency/gs.py | 8 - dvc/dependency/hdfs.py | 6 - dvc/dependency/http.py | 8 - dvc/dependency/https.py | 6 - dvc/dependency/local.py | 6 - dvc/dependency/param.py | 6 +- dvc/dependency/repo.py | 6 +- dvc/dependency/s3.py | 6 - dvc/dependency/ssh.py | 6 - dvc/dependency/webdav.py | 8 - dvc/dependency/webdavs.py | 6 - dvc/dependency/webhdfs.py | 6 - dvc/fs/__init__.py | 11 +- dvc/fs/_metadata.py | 4 +- dvc/fs/dvc.py | 4 +- dvc/{output/base.py => output.py} | 307 ++++++++++++++++++++--- dvc/output/__init__.py | 241 ------------------ dvc/output/hdfs.py | 7 - dvc/output/local.py | 111 -------- dvc/output/s3.py | 7 - dvc/output/ssh.py | 7 - dvc/output/webhdfs.py | 7 - dvc/repo/add.py | 2 +- dvc/repo/collect.py | 6 +- dvc/repo/live.py | 8 +- dvc/repo/metrics/show.py | 6 +- dvc/repo/params/show.py | 6 +- dvc/repo/plots/__init__.py | 6 +- dvc/schema.py | 30 +-- dvc/stage/__init__.py | 10 +- dvc/stage/monitor.py | 6 +- dvc/stage/serialize.py | 16 +- dvc/stage/utils.py | 18 +- tests/func/test_add.py | 2 +- tests/func/test_commit.py | 2 +- tests/func/test_dependency.py | 21 -- tests/func/test_ignore.py | 2 +- tests/func/test_output.py | 30 --- tests/func/test_repro.py | 6 +- tests/func/test_run_cache.py | 4 +- tests/func/test_run_single_stage.py | 5 +- tests/func/test_stage.py | 12 +- tests/unit/dependency/test_azure.py | 7 - tests/unit/dependency/test_dependency.py | 12 + tests/unit/dependency/test_gs.py | 7 - tests/unit/dependency/test_hdfs.py | 10 - tests/unit/dependency/test_http.py | 7 - tests/unit/dependency/test_local.py | 20 -- tests/unit/dependency/test_s3.py | 7 - tests/unit/dependency/test_ssh.py | 7 - tests/unit/dependency/test_webdav.py | 7 - tests/unit/fs/test_fs.py | 31 +++ tests/unit/output/test_hdfs.py | 10 - tests/unit/output/test_load.py | 19 +- tests/unit/output/test_local.py | 31 +-- tests/unit/output/test_output.py | 26 +- tests/unit/output/test_s3.py | 7 - tests/unit/output/test_ssh.py | 7 - tests/unit/stage/test_cache.py | 8 +- 63 files changed, 466 insertions(+), 807 deletions(-) delete mode 100644 dvc/dependency/azure.py delete mode 100644 dvc/dependency/gs.py delete mode 100644 dvc/dependency/hdfs.py delete mode 100644 dvc/dependency/http.py delete mode 100644 dvc/dependency/https.py delete mode 100644 dvc/dependency/local.py delete mode 100644 dvc/dependency/s3.py delete mode 100644 dvc/dependency/ssh.py delete mode 100644 dvc/dependency/webdav.py delete mode 100644 dvc/dependency/webdavs.py delete mode 100644 dvc/dependency/webhdfs.py rename dvc/{output/base.py => output.py} (72%) delete mode 100644 dvc/output/__init__.py delete mode 100644 dvc/output/hdfs.py delete mode 100644 dvc/output/local.py delete mode 100644 dvc/output/s3.py delete mode 100644 dvc/output/ssh.py delete mode 100644 dvc/output/webhdfs.py delete mode 100644 tests/func/test_dependency.py delete mode 100644 tests/func/test_output.py delete mode 100644 tests/unit/dependency/test_azure.py create mode 100644 tests/unit/dependency/test_dependency.py delete mode 100644 tests/unit/dependency/test_gs.py delete mode 100644 tests/unit/dependency/test_hdfs.py delete mode 100644 tests/unit/dependency/test_http.py delete mode 100644 tests/unit/dependency/test_local.py delete mode 100644 tests/unit/dependency/test_s3.py delete mode 100644 tests/unit/dependency/test_ssh.py delete mode 100644 tests/unit/dependency/test_webdav.py create mode 100644 tests/unit/fs/test_fs.py delete mode 100644 tests/unit/output/test_hdfs.py delete mode 100644 tests/unit/output/test_s3.py delete mode 100644 tests/unit/output/test_ssh.py 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()