import os

import pytest

from dvc.dependency.base import DependencyDoesNotExistError
from dvc.dvcfile import PIPELINE_FILE
from dvc.output import OutputDoesNotExistError
from dvc.stage.exceptions import StageCommitError
from tests.utils import clean_staging


def test_commit_recursive(tmp_dir, dvc):
    tmp_dir.gen({"dir": {"file": "text1", "subdir": {"file2": "text2"}}})
    stages = dvc.add("dir", recursive=True, no_commit=True)

    assert len(stages) == 2
    assert dvc.status() != {}

    dvc.commit("dir", recursive=True)
    assert dvc.status() == {}


def test_commit_force(tmp_dir, dvc):
    tmp_dir.gen({"dir": {"file": "text1", "file2": "text2"}})
    (stage,) = dvc.add("dir", no_commit=True)

    assert stage.outs[0].changed_cache()

    tmp_dir.gen("dir/file", "file content modified")

    assert stage.outs[0].changed_cache()

    with pytest.raises(StageCommitError):
        dvc.commit(stage.path)

    assert stage.outs[0].changed_cache()

    dvc.commit(stage.path, force=True)
    assert dvc.status([stage.path]) == {}


@pytest.mark.parametrize("run_kw", [{"single_stage": True}, {"name": "copy"}])
def test_commit_with_deps(tmp_dir, dvc, run_copy, run_kw):
    tmp_dir.gen("foo", "foo")
    (foo_stage,) = dvc.add("foo", no_commit=True)
    assert foo_stage is not None
    assert len(foo_stage.outs) == 1

    stage = run_copy("foo", "file", no_commit=True, **run_kw)
    assert stage is not None
    assert len(stage.outs) == 1

    assert foo_stage.outs[0].changed_cache()
    assert stage.outs[0].changed_cache()

    clean_staging()

    dvc.commit(stage.path, with_deps=True)
    assert not foo_stage.outs[0].changed_cache()
    assert not stage.outs[0].changed_cache()


def test_commit_changed_md5(tmp_dir, dvc):
    tmp_dir.gen({"file": "file content"})
    (stage,) = dvc.add("file", no_commit=True)

    stage_file_content = (tmp_dir / stage.path).parse()
    stage_file_content["md5"] = "1111111111"
    (tmp_dir / stage.path).dump(stage_file_content)

    clean_staging()

    with pytest.raises(StageCommitError):
        dvc.commit(stage.path)

    dvc.commit(stage.path, force=True)
    assert "md5" not in (tmp_dir / stage.path).parse()


def test_commit_no_exec(tmp_dir, dvc):
    tmp_dir.gen({"dep": "dep", "out": "out"})
    stage = dvc.run(
        name="my", cmd="mycmd", deps=["dep"], outs=["out"], no_exec=True
    )
    clean_staging()

    assert dvc.status(stage.path)
    dvc.commit(stage.path, force=True)
    assert dvc.status(stage.path) == {}


def test_commit_granular_output(tmp_dir, dvc):
    dvc.run(
        name="mystage",
        cmd=["echo foo>foo", "echo bar>bar"],
        outs=["foo", "bar"],
        no_commit=True,
    )
    clean_staging()

    cache = tmp_dir / ".dvc" / "cache"
    assert not list(cache.glob("*/*"))

    dvc.commit("foo")
    assert list(cache.glob("*/*")) == [
        cache / "d3" / "b07384d113edec49eaa6238ad5ff00"
    ]


def test_commit_granular_output_file(tmp_dir, dvc):
    tmp_dir.gen("foo", "foo")
    dvc.add("foo", no_commit=True)
    clean_staging()
    dvc.commit("foo")
    assert dvc.status() == {}


def test_commit_granular_output_dir(tmp_dir, dvc):
    tmp_dir.gen(
        {
            "data": {
                "foo": "foo",
                "bar": "bar",
                "subdir": {"subfoo": "subfoo", "subbar": "subbar"},
            }
        }
    )
    dvc.add("data", no_commit=True)
    clean_staging()
    dvc.commit("data")
    assert dvc.status() == {}