Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-07 15:28:28 +01:00
Code Issues Packages Projects Releases Wiki Activity

dvc: cache graph/stages/pipelines (#2529)

Browse Source 
* dvc: cache graph/stages/pipelines

Signed-off-by: Ruslan Kuprieiev <[email protected]>

* repo: check_dag: don't reset the cache

https://github.com/iterative/dvc/pull/2529#discussion_r327711928

Signed-off-by: Ruslan Kuprieiev <[email protected]>

* repo: rename check_dag(stages) -> check_modified_graph(new_stages)

https://github.com/iterative/dvc/pull/2529#discussion_r327708439

Signed-off-by: Ruslan Kuprieiev <[email protected]>

* 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
This commit is contained in:
Ruslan Kuprieiev
2019-09-26 07:42:29 +03:00
committed by GitHub
parent fea05b3924
commit 381601f4ee
21 changed files with 203 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/pipeline.py
+6 -9
View File
@@ -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:
dvc/exceptions.py
-7
View File
@@ -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__(
dvc/repo/__init__.py
+72 -71
View File
@@ -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)
dvc/repo/add.py
+19 -16
View File
@@ -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
dvc/repo/checkout.py
+1 -3
View File
@@ -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")
dvc/repo/destroy.py
+1 -1
View File
@@ -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)
dvc/repo/diff.py
+2 -2
View File
@@ -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 = {}
dvc/repo/graph.py
+29
View File
@@ -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())
dvc/repo/imp_url.py
+1 -1
View File
@@ -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()
dvc/repo/metrics/modify.py
+2
View File
@@ -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)
dvc/repo/metrics/show.py
+3 -1
View File
@@ -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,
dvc/repo/reproduce.py
+33 -21
View File
@@ -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)
dvc/repo/run.py
+1 -1
View File
@@ -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:
dvc/repo/tag/add.py
+4
View File
@@ -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:
dvc/repo/tag/list.py
+4
View File
@@ -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)
dvc/repo/tag/remove.py
+3
View File
@@ -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:
tests/func/test_checkout.py
+1 -5
View File
@@ -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):
tests/func/test_diff.py
+14 -20
View File
@@ -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):
tests/func/test_pipeline.py
+4 -4
View File
@@ -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)
tests/func/test_repo.py
+2 -2
View File
@@ -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]
tests/func/test_repro.py
+1 -1
View File
@@ -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),