From 381601f4eecf89bfdf9754a97259c68a12215a5e Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 26 Sep 2019 07:42:29 +0300 Subject: [PATCH] dvc: cache graph/stages/pipelines (#2529) * dvc: cache graph/stages/pipelines Signed-off-by: Ruslan Kuprieiev * repo: check_dag: don't reset the cache https://github.com/iterative/dvc/pull/2529#discussion_r327711928 Signed-off-by: Ruslan Kuprieiev * repo: rename check_dag(stages) -> check_modified_graph(new_stages) https://github.com/iterative/dvc/pull/2529#discussion_r327708439 Signed-off-by: Ruslan Kuprieiev * repo: refactor "active" logic and remove it from Repo class * repo: use tree.setter to automate cache reset As noted by @Suor's , this is much less prone to errors. * repo: move graph-related staticmethods into graph.py As per @Suor's request. * dvc: remove unused from_directory https://github.com/iterative/dvc/pull/2529#discussion_r328079258 * repo: use _tree instead of __tree https://github.com/iterative/dvc/pull/2529#discussion_r328071115 * repo: add a note about cache invalidation on repo.lock release https://github.com/iterative/dvc/pull/2529#discussion_r328071635 * repo: rename reset() -> _reset() https://github.com/iterative/dvc/pull/2529#discussion_r328079899 * pipeline: use graph helpers https://github.com/iterative/dvc/pull/2529#discussion_r327502293 * repo: use list comprehension for collect() https://github.com/iterative/dvc/pull/2529#discussion_r328076651 * repo: graph: only use first cycle https://github.com/iterative/dvc/pull/2529#discussion_r328081346 * dvc: don't use G.nodes() https://github.com/iterative/dvc/pull/2529#discussion_r328089671 https://github.com/iterative/dvc/pull/2529#discussion_r328089878 --- dvc/command/pipeline.py | 15 ++-- dvc/exceptions.py | 7 -- dvc/repo/__init__.py | 143 ++++++++++++++++++------------------ dvc/repo/add.py | 35 +++++---- dvc/repo/checkout.py | 4 +- dvc/repo/destroy.py | 2 +- dvc/repo/diff.py | 4 +- dvc/repo/graph.py | 29 ++++++++ dvc/repo/imp_url.py | 2 +- dvc/repo/metrics/modify.py | 2 + dvc/repo/metrics/show.py | 4 +- dvc/repo/reproduce.py | 54 ++++++++------ dvc/repo/run.py | 2 +- dvc/repo/tag/add.py | 4 + dvc/repo/tag/list.py | 4 + dvc/repo/tag/remove.py | 3 + tests/func/test_checkout.py | 6 +- tests/func/test_diff.py | 34 ++++----- tests/func/test_pipeline.py | 8 +- tests/func/test_repo.py | 4 +- tests/func/test_repro.py | 2 +- 21 files changed, 203 insertions(+), 165 deletions(-) create mode 100644 dvc/repo/graph.py diff --git a/dvc/command/pipeline.py b/dvc/command/pipeline.py index b3b7d84b7..6dc7b9a77 100644 --- a/dvc/command/pipeline.py +++ b/dvc/command/pipeline.py @@ -9,6 +9,8 @@ import logging from dvc.exceptions import DvcException from dvc.command.base import CmdBase, fix_subparsers, append_doc_link +from dvc.repo.graph import get_pipeline + logger = logging.getLogger(__name__) @@ -19,7 +21,7 @@ class CmdPipelineShow(CmdBase): from dvc.stage import Stage stage = Stage.load(self.repo, target) - G = self.repo.graph()[0] + G = self.repo.graph stages = networkx.get_node_attributes(G, "stage") node = relpath(stage.path, self.repo.root_dir) nodes = networkx.dfs_postorder_nodes(G, node) @@ -45,16 +47,11 @@ class CmdPipelineShow(CmdBase): stage = Stage.load(self.repo, target) node = relpath(stage.path, self.repo.root_dir) - pipelines = list( - filter(lambda g: node in g.nodes(), self.repo.pipelines()) - ) - - assert len(pipelines) == 1 - G = pipelines[0] + G = get_pipeline(self.repo.pipelines, node) stages = networkx.get_node_attributes(G, "stage") nodes = [] - for n in G.nodes(): + for n in G: stage = stages[n] if commands: if stage.cmd is None: @@ -168,7 +165,7 @@ class CmdPipelineList(CmdBase): def run(self): import networkx - pipelines = self.repo.pipelines() + pipelines = self.repo.pipelines for p in pipelines: stages = networkx.get_node_attributes(p, "stage") for stage in stages: diff --git a/dvc/exceptions.py b/dvc/exceptions.py index 4ca3c9c48..822944ef5 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -244,13 +244,6 @@ class OverlappingOutputPathsError(DvcException): ) -class TargetNotDirectoryError(DvcException): - def __init__(self, path): - super(TargetNotDirectoryError, self).__init__( - "Target: {} is not a directory".format(path) - ) - - class CheckoutErrorSuggestGit(DvcException): def __init__(self, target, cause): super(CheckoutErrorSuggestGit, self).__init__( diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index a9eb15d51..fed493531 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -12,7 +12,6 @@ from dvc.config import Config from dvc.exceptions import ( NotDvcRepoError, OutputNotFoundError, - TargetNotDirectoryError, OutputFileMissingError, ) from dvc.ignore import DvcIgnoreFilter @@ -21,6 +20,9 @@ from dvc.remote.base import RemoteActionNotImplemented from dvc.utils.compat import open as _open, fspath_py35, FileNotFoundError from dvc.utils import relpath +from .graph import check_acyclic, get_pipeline, get_pipelines, get_stages + + logger = logging.getLogger(__name__) @@ -28,7 +30,10 @@ def locked(f): @wraps(f) def wrapper(repo, *args, **kwargs): with repo.lock: - return f(repo, *args, **kwargs) + ret = f(repo, *args, **kwargs) + # Our graph cache is no longer valid after we release the repo.lock + repo._reset() + return ret return wrapper @@ -102,6 +107,17 @@ class Repo(object): self._ignore() + @property + def tree(self): + return self._tree + + @tree.setter + def tree(self, tree): + self._tree = tree + # Our graph cache is no longer valid, as it was based on the previous + # tree. + self._reset() + def __repr__(self): return "Repo: '{root_dir}'".format(root_dir=self.root_dir) @@ -153,44 +169,43 @@ class Repo(object): self.scm.ignore_list(flist) - def check_dag(self, stages): + def check_modified_graph(self, new_stages): """Generate graph including the new stage to check for errors""" - self.graph(stages=stages) + self._collect_graph(self.stages + new_stages) - @staticmethod - def _check_cyclic_graph(graph): - import networkx as nx - from dvc.exceptions import CyclicGraphError - - cycles = list(nx.simple_cycles(graph)) - - if cycles: - raise CyclicGraphError(cycles[0]) - - def _get_pipeline(self, node): - pipelines = [i for i in self.pipelines() if i.has_node(node)] - assert len(pipelines) == 1 - return pipelines[0] - - def collect(self, target, with_deps=False, recursive=False): + def collect(self, target, with_deps=False, recursive=False, graph=None): import networkx as nx from dvc.stage import Stage - if not target or (recursive and os.path.isdir(target)): - return self.stages(target) + G = graph or self.graph + + if not target: + return get_stages(G) + + target = os.path.abspath(target) + + if recursive and os.path.isdir(target): + attrs = nx.get_node_attributes(G, "stage") + nodes = [node for node in nx.dfs_postorder_nodes(G)] + + ret = [] + for node in nodes: + stage = attrs[node] + if stage.path.startswith(target + os.sep): + ret.append(stage) + return ret stage = Stage.load(self, target) if not with_deps: return [stage] node = relpath(stage.path, self.root_dir) - G = self._get_pipeline(node) + pipeline = get_pipeline(get_pipelines(G), node) - ret = [] - for n in nx.dfs_postorder_nodes(G, node): - ret.append(G.node[n]["stage"]) - - return ret + return [ + pipeline.node[n]["stage"] + for n in nx.dfs_postorder_nodes(pipeline, node) + ] def used_cache( self, @@ -232,14 +247,12 @@ class Repo(object): if targets: stages = [] for target in targets: - if recursive and os.path.isdir(target): - stages.extend(self.stages(target)) - else: - stages.extend( - self.collect(target, with_deps=with_deps) - ) + collected = self.collect( + target, recursive=recursive, with_deps=with_deps + ) + stages.extend(collected) else: - stages = self.stages() + stages = self.stages for stage in stages: if stage.is_repo_import: @@ -258,7 +271,7 @@ class Repo(object): return cache - def graph(self, stages=None, from_directory=None): + def _collect_graph(self, stages=None): """Generate a graph by using the given stages on the given directory The nodes of the graph are the stage's path relative to the root. @@ -288,11 +301,8 @@ class Repo(object): (weakly connected components) Args: - stages (list): used to build a graph, if None given, use the ones - on the `from_directory`. - - from_directory (str): directory where to look at for stages, if - None is given, use the current working directory + stages (list): used to build a graph, if None given, collect stages + in the repository. Raises: OutputDuplicationError: two outputs with the same path @@ -308,8 +318,7 @@ class Repo(object): ) G = nx.DiGraph() - G_active = nx.DiGraph() - stages = stages or self.stages(from_directory, check_dag=False) + stages = stages or self.collect_stages() stages = [stage for stage in stages if stage] outs = {} @@ -336,7 +345,6 @@ class Repo(object): node = relpath(stage.path, self.root_dir) G.add_node(node, stage=stage) - G_active.add_node(node, stage=stage) for dep in stage.deps: if dep.path_info is None: @@ -352,22 +360,18 @@ class Repo(object): dep_node = relpath(dep_stage.path, self.root_dir) G.add_node(dep_node, stage=dep_stage) G.add_edge(node, dep_node) - if not stage.locked: - G_active.add_node(dep_node, stage=dep_stage) - G_active.add_edge(node, dep_node) - self._check_cyclic_graph(G) + check_acyclic(G) - return G, G_active + return G - def pipelines(self, from_directory=None): - import networkx as nx + @cached_property + def graph(self): + return self._collect_graph() - G, G_active = self.graph(from_directory=from_directory) - - return [ - G.subgraph(c).copy() for c in nx.weakly_connected_components(G) - ] + @cached_property + def pipelines(self): + return get_pipelines(self.graph) @staticmethod def _filter_out_dirs(dirs, outs, root_dir): @@ -380,7 +384,7 @@ class Repo(object): return list(filter(filter_dirs, dirs)) - def stages(self, from_directory=None, check_dag=True): + def collect_stages(self): """ Walks down the root directory looking for Dvcfiles, skipping the directories that are related with @@ -392,16 +396,11 @@ class Repo(object): """ from dvc.stage import Stage - if not from_directory: - from_directory = self.root_dir - elif not os.path.isdir(from_directory): - raise TargetNotDirectoryError(from_directory) - stages = [] outs = [] for root, dirs, files in self.tree.walk( - from_directory, dvcignore=self.dvcignore + self.root_dir, dvcignore=self.dvcignore ): for fname in files: path = os.path.join(root, fname) @@ -415,18 +414,15 @@ class Repo(object): dirs[:] = self._filter_out_dirs(dirs, outs, root) - if check_dag: - self.check_dag(stages) - return stages + @cached_property + def stages(self): + return get_stages(self.graph) + def find_outs_by_path(self, path, outs=None, recursive=False): if not outs: - # there is no `from_directory=path` argument because some data - # files might be generated to an upper level, and so it is - # needed to look at all the files (project root_dir) - stages = self.stages() - outs = [out for stage in stages for out in stage.outs] + outs = [out for stage in self.stages for out in stage.outs] abs_path = os.path.abspath(path) is_dir = self.tree.isdir(abs_path) @@ -509,3 +505,8 @@ class Repo(object): @locked def fetch(self, *args, **kwargs): return self._fetch(*args, **kwargs) + + def _reset(self): + self.__dict__.pop("graph", None) + self.__dict__.pop("stages", None) + self.__dict__.pop("pipelines", None) diff --git a/dvc/repo/add.py b/dvc/repo/add.py index 6822b8f63..a6485b905 100644 --- a/dvc/repo/add.py +++ b/dvc/repo/add.py @@ -1,3 +1,5 @@ +from __future__ import unicode_literals + import os import logging import colorama @@ -34,12 +36,19 @@ def add(repo, target, recursive=False, no_commit=False, fname=None): ) ) - stages = _create_stages(repo, targets, fname, no_commit) + with repo.state: + stages = _create_stages(repo, targets, fname) - repo.check_dag(repo.stages() + stages) + repo.check_modified_graph(stages) + + for stage in stages: + stage.save() + + if not no_commit: + stage.commit() + + stage.dump() - for stage in stages: - stage.dump() return stages @@ -56,21 +65,15 @@ def _find_all_targets(repo, target, recursive): return [target] -def _create_stages(repo, targets, fname, no_commit): +def _create_stages(repo, targets, fname): stages = [] - with repo.state: - for out in targets: - stage = Stage.create(repo, outs=[out], add=True, fname=fname) + for out in targets: + stage = Stage.create(repo, outs=[out], add=True, fname=fname) - if not stage: - continue + if not stage: + continue - stage.save() - - if not no_commit: - stage.commit() - - stages.append(stage) + stages.append(stage) return stages diff --git a/dvc/repo/checkout.py b/dvc/repo/checkout.py index 0c259df5c..4647a21ed 100644 --- a/dvc/repo/checkout.py +++ b/dvc/repo/checkout.py @@ -28,8 +28,6 @@ def _checkout( ): from dvc.stage import StageFileDoesNotExistError, StageFileBadNameError - all_stages = self.stages() - try: stages = self.collect(target, with_deps=with_deps, recursive=recursive) except (StageFileDoesNotExistError, StageFileBadNameError) as exc: @@ -38,7 +36,7 @@ def _checkout( raise CheckoutErrorSuggestGit(target, exc) with self.state: - _cleanup_unused_links(self, all_stages) + _cleanup_unused_links(self, self.stages) total = get_all_files_numbers(stages) if total == 0: logger.info("Nothing to do") diff --git a/dvc/repo/destroy.py b/dvc/repo/destroy.py index f3958bbe0..e5938e195 100644 --- a/dvc/repo/destroy.py +++ b/dvc/repo/destroy.py @@ -2,7 +2,7 @@ from dvc.utils import remove def destroy(self): - for stage in self.stages(): + for stage in self.stages: stage.remove(remove_outs=False) remove(self.dvc_dir) diff --git a/dvc/repo/diff.py b/dvc/repo/diff.py index cdfe48907..32efe9880 100644 --- a/dvc/repo/diff.py +++ b/dvc/repo/diff.py @@ -142,9 +142,9 @@ def _is_dir(path, a_outs, b_outs): def _get_diff_outs(self, diff_dct): self.tree = diff_dct[DIFF_A_TREE] - a_outs = {str(out): out for st in self.stages() for out in st.outs} + a_outs = {str(out): out for st in self.stages for out in st.outs} self.tree = diff_dct[DIFF_B_TREE] - b_outs = {str(out): out for st in self.stages() for out in st.outs} + b_outs = {str(out): out for st in self.stages for out in st.outs} outs_paths = set(a_outs.keys()) outs_paths.update(b_outs.keys()) results = {} diff --git a/dvc/repo/graph.py b/dvc/repo/graph.py new file mode 100644 index 000000000..58cbd4ea0 --- /dev/null +++ b/dvc/repo/graph.py @@ -0,0 +1,29 @@ +from __future__ import unicode_literals + + +def check_acyclic(graph): + import networkx as nx + from dvc.exceptions import CyclicGraphError + + cycle = next(nx.simple_cycles(graph), None) + + if cycle: + raise CyclicGraphError(cycle) + + +def get_pipeline(pipelines, node): + found = [i for i in pipelines if i.has_node(node)] + assert len(found) == 1 + return found[0] + + +def get_pipelines(G): + import networkx as nx + + return [G.subgraph(c).copy() for c in nx.weakly_connected_components(G)] + + +def get_stages(G): + import networkx + + return list(networkx.get_node_attributes(G, "stage").values()) diff --git a/dvc/repo/imp_url.py b/dvc/repo/imp_url.py index 366570c4d..55cc0cef6 100644 --- a/dvc/repo/imp_url.py +++ b/dvc/repo/imp_url.py @@ -19,7 +19,7 @@ def imp_url(self, url, out=None, fname=None, erepo=None, locked=True): if stage is None: return None - self.check_dag(self.stages() + [stage]) + self.check_modified_graph([stage]) with self.state: stage.run() diff --git a/dvc/repo/metrics/modify.py b/dvc/repo/metrics/modify.py index 71fe51b2a..a8baac8ae 100644 --- a/dvc/repo/metrics/modify.py +++ b/dvc/repo/metrics/modify.py @@ -1,8 +1,10 @@ from __future__ import unicode_literals +from dvc.repo import locked from dvc.exceptions import DvcException +@locked def modify(repo, path, typ=None, xpath=None, delete=False): supported_types = ["raw", "json", "csv", "tsv", "hcsv", "htsv"] outs = repo.find_outs_by_path(path) diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index cd1ab7459..d61f98241 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -8,6 +8,7 @@ import logging from jsonpath_ng.ext import parse +from dvc.repo import locked from dvc.exceptions import OutputNotFoundError, NoMetricsError from dvc.utils.compat import builtin_str, open, StringIO, csv_reader @@ -170,7 +171,7 @@ def _collect_metrics(repo, path, recursive, typ, xpath, branch): - typ: - xpath: """ - outs = [out for stage in repo.stages() for out in stage.outs] + outs = [out for stage in repo.stages for out in stage.outs] if path: try: @@ -257,6 +258,7 @@ def _read_metrics(repo, metrics, branch): return res +@locked def show( repo, targets=None, diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index 1b7b5b372..f65369a5f 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -1,6 +1,5 @@ from __future__ import unicode_literals -import os import logging from dvc.exceptions import ReproductionError @@ -8,6 +7,7 @@ from dvc.repo.scm_context import scm_context from dvc.utils import relpath from . import locked +from .graph import get_pipeline, get_pipelines logger = logging.getLogger(__name__) @@ -32,6 +32,23 @@ def _reproduce_stage(stages, node, **kwargs): return [stage] +def _get_active_graph(G): + import networkx as nx + + active = G.copy() + stages = nx.get_node_attributes(G, "stage") + for node in G: + stage = stages[node] + if not stage.locked: + continue + for n in nx.dfs_postorder_nodes(G, node): + if n == node: + continue + if n in active: + active.remove_node(n) + return active + + @locked @scm_context def reproduce( @@ -56,43 +73,38 @@ def reproduce( config.SECTION_CORE_INTERACTIVE, False ) - targets = [] - if recursive and os.path.isdir(target): - G = self.graph(from_directory=target)[1] - dir_targets = [ - os.path.join(self.root_dir, n) for n in nx.dfs_postorder_nodes(G) - ] - targets.extend(dir_targets) - elif pipeline or all_pipelines: - if pipeline: + active_graph = _get_active_graph(self.graph) + active_pipelines = get_pipelines(active_graph) + + if pipeline or all_pipelines: + if all_pipelines: + pipelines = active_pipelines + else: stage = Stage.load(self, target) node = relpath(stage.path, self.root_dir) - pipelines = [self._get_pipeline(node)] - else: - pipelines = self.pipelines() + pipelines = [get_pipeline(active_pipelines, node)] + targets = [] for G in pipelines: - for node in G.nodes(): + attrs = nx.get_node_attributes(G, "stage") + for node in G: if G.in_degree(node) == 0: - targets.append(os.path.join(self.root_dir, node)) + targets.append(attrs[node]) else: - targets.append(target) + targets = self.collect(target, recursive=recursive, graph=active_graph) ret = [] with self.state: for target in targets: - stages = _reproduce(self, target, **kwargs) + stages = _reproduce(self, active_graph, target, **kwargs) ret.extend(stages) return ret -def _reproduce(self, target, **kwargs): +def _reproduce(self, G, stage, **kwargs): import networkx as nx - from dvc.stage import Stage - stage = Stage.load(self, target) - G = self.graph()[1] stages = nx.get_node_attributes(G, "stage") node = relpath(stage.path, self.root_dir) diff --git a/dvc/repo/run.py b/dvc/repo/run.py index 0fd4728c7..65c4933dd 100644 --- a/dvc/repo/run.py +++ b/dvc/repo/run.py @@ -15,7 +15,7 @@ def run(self, no_exec=False, **kwargs): if stage is None: return None - self.check_dag(self.stages() + [stage]) + self.check_modified_graph([stage]) with self.state: if not no_exec: diff --git a/dvc/repo/tag/add.py b/dvc/repo/tag/add.py index d5d8278ac..44f02d70c 100644 --- a/dvc/repo/tag/add.py +++ b/dvc/repo/tag/add.py @@ -1,9 +1,13 @@ import logging from copy import copy +from dvc.repo import locked + + logger = logging.getLogger(__name__) +@locked def add(self, tag, target=None, with_deps=False, recursive=False): stages = self.collect(target, with_deps=with_deps, recursive=recursive) for stage in stages: diff --git a/dvc/repo/tag/list.py b/dvc/repo/tag/list.py index cad7083ad..487d5afdb 100644 --- a/dvc/repo/tag/list.py +++ b/dvc/repo/tag/list.py @@ -1,3 +1,7 @@ +from dvc.repo import locked + + +@locked def list(self, target=None, with_deps=False, recursive=False): ret = {} stages = self.collect(target, with_deps=with_deps, recursive=recursive) diff --git a/dvc/repo/tag/remove.py b/dvc/repo/tag/remove.py index 6ce540164..91c8321ce 100644 --- a/dvc/repo/tag/remove.py +++ b/dvc/repo/tag/remove.py @@ -1,9 +1,12 @@ import logging +from dvc.repo import locked + logger = logging.getLogger(__name__) +@locked def remove(self, tag, target=None, with_deps=False, recursive=False): stages = self.collect(target, with_deps=with_deps, recursive=recursive) for stage in stages: diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py index 3efee9af7..a4760194a 100644 --- a/tests/func/test_checkout.py +++ b/tests/func/test_checkout.py @@ -19,7 +19,6 @@ from dvc.remote.local import RemoteLOCAL from dvc.exceptions import ( DvcException, ConfirmRemoveError, - TargetNotDirectoryError, CheckoutErrorSuggestGit, CheckoutError, ) @@ -423,10 +422,7 @@ class TestCheckoutRecursiveNotDirectory(TestDvc): ret = main(["add", self.FOO]) self.assertEqual(0, ret) - try: - self.dvc.checkout(target=self.FOO + ".dvc", recursive=True) - except TargetNotDirectoryError: - self.fail("should not raise TargetNotDirectoryError") + self.dvc.checkout(target=self.FOO + ".dvc", recursive=True) class TestCheckoutMovedCacheDirWithSymlinks(TestDvc): diff --git a/tests/func/test_diff.py b/tests/func/test_diff.py index c40e37252..799b95c1e 100644 --- a/tests/func/test_diff.py +++ b/tests/func/test_diff.py @@ -8,22 +8,15 @@ import dvc.repo.diff as diff from tests.basic_env import TestDvcGit -def _get_checksum(repo, file_name): - outs = [out for s in repo.stages() for out in s.outs] - for out in outs: - if out.def_path == file_name: - return out.checksum - - class TestDiff(TestDvcGit): def setUp(self): super(TestDiff, self).setUp() self.new_file = "new_test_file" self.create(self.new_file, self.new_file) - self.dvc.add(self.new_file) + stage = self.dvc.add(self.new_file)[0] self.a_ref = self.git.git.rev_parse(self.git.head.commit, short=True) - self.new_checksum = _get_checksum(self.dvc, self.new_file) + self.new_checksum = stage.outs[0].checksum self.git.index.add([self.new_file + ".dvc"]) self.git.index.commit("adds new_file") self.test_dct = { @@ -67,19 +60,19 @@ class TestDiffDir(TestDvcGit): def setUp(self): super(TestDiffDir, self).setUp() - self.dvc.add(self.DATA_DIR) + stage = self.dvc.add(self.DATA_DIR)[0] self.git.index.add([self.DATA_DIR + ".dvc"]) self.git.index.commit("adds data_dir") self.a_ref = self.git.git.rev_parse( self.dvc.scm.repo.head.commit, short=True ) - self.old_checksum = _get_checksum(self.dvc, self.DATA_DIR) + self.old_checksum = stage.outs[0].checksum self.new_file = os.path.join(self.DATA_SUB_DIR, diff.DIFF_NEW_FILE) self.create(self.new_file, self.new_file) - self.dvc.add(self.DATA_DIR) + stage = self.dvc.add(self.DATA_DIR)[0] self.git.index.add([self.DATA_DIR + ".dvc"]) self.git.index.commit(message="adds data_dir with new_file") - self.new_checksum = _get_checksum(self.dvc, self.DATA_DIR) + self.new_checksum = stage.outs[0].checksum def test(self): out = self.dvc.scm.get_diff_trees(self.a_ref) @@ -128,9 +121,10 @@ class TestDiffDirRepoDeletedFile(TestDiffDir): super(TestDiffDirRepoDeletedFile, self).setUp() self.b_ref = self.a_ref + tmp = self.new_checksum self.new_checksum = self.old_checksum self.a_ref = str(self.dvc.scm.repo.head.commit) - self.old_checksum = _get_checksum(self.dvc, self.DATA_DIR) + self.old_checksum = tmp def test(self): result = self.dvc.diff( @@ -175,14 +169,14 @@ class TestDiffModifiedFile(TestDiff): def setUp(self): super(TestDiffModifiedFile, self).setUp() - self.old_checksum = _get_checksum(self.dvc, self.new_file) + self.old_checksum = self.new_checksum self.new_file_content = "new_test_file_bigger_content_123456789" self.diff_len = len(self.new_file) + len(self.new_file_content) self.create(self.new_file, self.new_file_content) - self.dvc.add(self.new_file) + stage = self.dvc.add(self.new_file)[0] self.git.index.add([self.new_file + ".dvc"]) self.git.index.commit("change new_file content to be bigger") - self.new_checksum = _get_checksum(self.dvc, self.new_file) + self.new_checksum = stage.outs[0].checksum self.b_ref = self.git.git.rev_parse(self.git.head.commit, short=True) def test(self): @@ -211,14 +205,14 @@ class TestDiffDirWithFile(TestDiffDir): super(TestDiffDirWithFile, self).setUp() self.a_ref = self.git.git.rev_parse(self.git.head.commit, short=True) - self.old_checksum = _get_checksum(self.dvc, self.DATA_DIR) + self.old_checksum = self.new_checksum self.new_file_content = "new_test_file_bigger_content_123456789" self.diff_len = len(self.new_file_content) self.create(self.new_file, self.new_file_content) - self.dvc.add(self.DATA_DIR) + stage = self.dvc.add(self.DATA_DIR)[0] self.git.index.add([self.DATA_DIR + ".dvc"]) self.git.index.commit(message="modify file in the data dir") - self.new_checksum = _get_checksum(self.dvc, self.DATA_DIR) + self.new_checksum = stage.outs[0].checksum self.b_ref = self.git.git.rev_parse(self.git.head.commit, short=True) def test(self): diff --git a/tests/func/test_pipeline.py b/tests/func/test_pipeline.py index bee789879..a9bac3780 100644 --- a/tests/func/test_pipeline.py +++ b/tests/func/test_pipeline.py @@ -186,14 +186,14 @@ class TestPipelineListSingle(TestPipelineShowDeep): class TestDvcRepoPipeline(TestDvc): def test_no_stages(self): - pipelines = self.dvc.pipelines() + pipelines = self.dvc.pipelines self.assertEqual(len(pipelines), 0) def one_pipeline(self): self.dvc.add("foo") self.dvc.run(deps=["foo"], outs=["bar"], cmd="") self.dvc.run(deps=["bar"], outs=["baz"], cmd="echo baz > baz") - pipelines = self.dvc.pipelines() + pipelines = self.dvc.pipelines self.assertEqual(len(pipelines), 1) self.assertEqual(pipelines[0].nodes, 3) @@ -206,7 +206,7 @@ class TestDvcRepoPipeline(TestDvc): self.dvc.add("code.py") - pipelines = self.dvc.pipelines() + pipelines = self.dvc.pipelines self.assertEqual(len(pipelines), 2) self.assertEqual(pipelines[0].nodes, 3) @@ -218,5 +218,5 @@ class TestDvcRepoPipeline(TestDvc): self.dvc.add("foo") self.dvc.lock_stage("foo.dvc") - pipelines = self.dvc.pipelines() + pipelines = self.dvc.pipelines self.assertEqual(len(pipelines), 0) diff --git a/tests/func/test_repo.py b/tests/func/test_repo.py index 8184721c6..f1f24d2a6 100644 --- a/tests/func/test_repo.py +++ b/tests/func/test_repo.py @@ -56,13 +56,13 @@ class TestIgnore(TestDvcGit): ret = main(["add", self.FOO, self.BAR, self.DATA, self.DATA_SUB]) self.assertEqual(0, ret) - stages = self.dvc.stages() + stages = self.dvc.stages self.assertEqual(4, len(stages)) self.create(DvcIgnore.DVCIGNORE_FILE, self.DATA_DIR) self.dvc = Repo(self.dvc.root_dir) - stages = self.dvc.stages() + stages = self.dvc.stages self.assertEqual(2, len(stages)) stagenames = [s.relpath for s in stages] diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py index f9b0f43a4..adfbc0d1a 100644 --- a/tests/func/test_repro.py +++ b/tests/func/test_repro.py @@ -162,7 +162,7 @@ class TestReproWorkingDirectoryAsOutput(TestDvc): # NOTE: os.walk() walks in a sorted order and we need dir2 subdirs to # be processed before dir1 to load error.dvc first. - with patch.object(DvcRepo, "stages") as mock_stages: + with patch.object(self.dvc, "collect_stages") as mock_stages: mock_stages.return_value = [ nested_stage, Stage.load(self.dvc, error_stage_path),