import re
import sys
from pathlib import Path

import airflow
import pytest

from conductor.cli.cmd.deploy import compile_dags, deploy
from conductor.core import Conductor
from tests.cli.cmd.test_files import FIXTURE_DIR
from tests.utils import MockValidator, RegexValidator


@pytest.mark.datafiles(Path(FIXTURE_DIR, "test_project"))
@pytest.mark.no_auto_env
def test_deploy(datafiles, mocker, monkeypatch):
    monkeypatch.setenv("CONDUCTOR_ENV", "test-env")
    monkeypatch.setenv("CONDUCTOR_GIT_BRANCH", "test-branch")
    monkeypatch.setenv("CONDUCTOR_COMMIT_HASH", "test-commit-hash")

    exec_mock = mocker.patch("conductor.cli.cmd.deploy.exec_sync")
    push_dags_mock = mocker.patch("conductor.cli.cmd.deploy.push_dags")
    sys.path.append(str(datafiles))
    deploy(datafiles)
    exec_mock.assert_any_call(
        "docker push 123456789.dkr.ecr.us-west-2.amazonaws.com/test-project.test-env:test-branch-test-commit-hash"
    )

    exec_mock.assert_any_call(
        RegexValidator("^/.*node_modules/conductor-cdk/node_modules/aws-cdk/bin/cdk.*")
    )
    push_dags_mock.assert_any_call(
        MockValidator(lambda project: project.name == "test-project"),
        MockValidator(
            lambda dagfiles: re.search(
                "^.*build/serialized_dags/test-project.test-env.my-dag.test-branch.py$",
                str(dagfiles[0]),
            )
        ),
    )


def test_compile_dags(tmpdir, basic_project):
    c = Conductor(
        "test-dag",
        basic_project,
        default_args={"start_date": airflow.utils.dates.days_ago(0)},
    )
    c.operators.SageMakerTransformOperator(task_id="transform", model_name="test-model")
    compile_dags([c], tmpdir)

    exec(
        Path(
            tmpdir, "serialized_dags", "test-project.test-env.test-dag.test-branch.py"
        ).read_text()
    )
