# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os
from pathlib import Path
import pytest

from cylc.flow.async_util import pipe
from cylc.flow.exceptions import UserInputError, WorkflowFilesError
from cylc.flow.id import detokenise, tokenise, Tokens
from cylc.flow.id_cli import (
    _expand_workflow_tokens,
    _parse_src_path,
    _validate_constraint,
    _validate_workflow_ids,
    _validate_number,
    parse_ids_async,
)
from cylc.flow.pathutil import get_cylc_run_dir
from cylc.flow.workflow_files import WorkflowFiles


@pytest.fixture(scope='module')
def abc_src_dir(tmp_path_factory):
    """Src dir containing three workflows, a, b & c."""
    cwd_before = Path.cwd()
    tmp_path = tmp_path_factory.getbasetemp()
    os.chdir(tmp_path)
    for name in ('a', 'b', 'c'):
        Path(tmp_path, name).mkdir()
        Path(tmp_path, name, WorkflowFiles.FLOW_FILE).touch()
    yield tmp_path
    os.chdir(cwd_before)


@pytest.mark.parametrize(
    'ids_in,ids_out',
    [
        (('a//',), ['a']),
        (('a//', 'a//'), ['a']),
        (('a//', 'b//'), ['a', 'b']),
    ]
)
async def test_parse_ids_workflows(ids_in, ids_out):
    """It should parse workflows & tasks."""
    workflows, _ = await parse_ids_async(*ids_in, constraint='workflows')
    assert list(workflows) == ids_out
    assert list(workflows.values()) == [[] for _ in workflows]


@pytest.mark.parametrize(
    'ids_in,ids_out',
    [
        (('./a',), ['a']),
    ]
)
async def test_parse_ids_workflows_src(ids_in, ids_out, abc_src_dir):
    """It should parse src workflows."""
    workflows, _ = await parse_ids_async(
        *ids_in,
        src=True,
        constraint='workflows',
    )
    assert list(workflows) == ids_out
    assert list(workflows.values()) == [[] for _ in workflows]


@pytest.mark.parametrize(
    'ids_in,ids_out',
    [
        (
            ('a//i',),
            {'a': ['//i']},
        ),
        (
            ('a//i', 'a//j'),
            {'a': ['//i', '//j']},
        ),
        (
            ('a//i', 'b//i'),
            {'a': ['//i'], 'b': ['//i']},
        ),
        (
            ('a//', '//i', 'b//', '//i'),
            {'a': ['//i'], 'b': ['//i']},
        ),
    ]
)
async def test_parse_ids_tasks(ids_in, ids_out):
    """It should parse workflow tasks in two formats."""
    workflows, _ = await parse_ids_async(*ids_in, constraint='tasks')
    assert {
        workflow_id: [detokenise(tokens) for tokens in tokens_list]
        for workflow_id, tokens_list in workflows.items()
    } == ids_out


@pytest.mark.parametrize(
    'ids_in,ids_out',
    [
        (
            ('./a', '//i'),
            {'a': ['//i']}
        ),
        (
            ('./a', '//i', '//j', '//k'),
            {'a': ['//i', '//j', '//k']}
        ),
    ]
)
async def test_parse_ids_tasks_src(ids_in, ids_out, abc_src_dir):
    """It should parse workflow tasks for src workflows."""
    workflows, _ = await parse_ids_async(*ids_in, constraint='tasks', src=True)
    assert {
        workflow_id: [detokenise(tokens) for tokens in tokens_list]
        for workflow_id, tokens_list in workflows.items()
    } == ids_out


@pytest.mark.parametrize(
    'ids_in,ids_out',
    [
        (('a//',), {'a': []}),
        (
            ('a//', 'b//', 'c//'),
            {'a': [], 'b': [], 'c': []}
        ),
        (('a//i',), {'a': ['//i']}),
        (('a//', '//i'), {'a': ['//i']}),
        (
            ('a//', '//i', '//j', '//k'),
            {'a': ['//i', '//j', '//k']},
        ),
        (('a//', '//i', 'b//'), {'a': ['//i'], 'b': []}),
    ]
)
async def test_parse_ids_mixed(ids_in, ids_out):
    """It should parse mixed workflows & tasks."""
    workflows, _ = await parse_ids_async(*ids_in, constraint='mixed')
    assert {
        workflow_id: [detokenise(tokens) for tokens in tokens_list]
        for workflow_id, tokens_list in workflows.items()
    } == ids_out


@pytest.mark.parametrize(
    'ids_in,ids_out',
    [
        (('./a',), {'a': []}),
        (('./a', '//i'), {'a': ['//i']}),
        (('./a', '//i', '//j', '//k'), {'a': ['//i', '//j', '//k']}),
    ]
)
async def test_parse_ids_mixed_src(ids_in, ids_out, abc_src_dir):
    """It should parse mixed workflows & tasks from src workflows."""
    workflows, _ = await parse_ids_async(*ids_in, constraint='mixed', src=True)
    assert {
        workflow_id: [detokenise(tokens) for tokens in tokens_list]
        for workflow_id, tokens_list in workflows.items()
    } == ids_out


@pytest.mark.parametrize(
    'ids_in,errors',
    [
        (('a//',), False),
        (('a//', 'b//'), False),
        (('a//', 'b//', 'c//'), True),
    ]
)
async def test_parse_ids_max_workflows(ids_in, errors):
    """It should validate input against the max_workflows constraint."""
    try:
        await parse_ids_async(*ids_in, constraint='workflows', max_workflows=2)
    except UserInputError:
        if not errors:
            raise
    else:
        if errors:
            raise Exception('Should have raised UserInputError')


@pytest.mark.parametrize(
    'ids_in,errors',
    [
        (('a//', '//i'), False),
        (('a//', '//i', '//j'), False),
        (('a//', '//i', '//j', '//k'), True),
    ]
)
async def test_parse_ids_max_tasks(ids_in, errors):
    """It should validate input against the max_tasks constraint."""
    try:
        await parse_ids_async(*ids_in, constraint='tasks', max_tasks=2)
    except UserInputError:
        if not errors:
            raise
    else:
        if errors:
            raise Exception('Should have raised UserInputError')


async def test_parse_ids_infer_run_name(tmp_run_dir):
    """It should infer the run name for auto-numbered installations."""
    # it doesn't do anything for a named run
    tmp_run_dir('foo/bar', named=True, installed=True)
    workflows, *_ = await parse_ids_async('foo//', constraint='workflows')
    assert list(workflows) == ['foo']

    # it correctly identifies the latest run
    tmp_run_dir('bar/run1')
    workflows, *_ = await parse_ids_async('bar//', constraint='workflows')
    assert list(workflows) == ['bar/run1']
    tmp_run_dir('bar/run2')
    workflows, *_ = await parse_ids_async('bar//', constraint='workflows')
    assert list(workflows) == ['bar/run2']

    # it leaves the ID alone if infer_latest_runs = False
    workflows, *_ = await parse_ids_async(
        'bar//',
        constraint='workflows',
        infer_latest_runs=False,
    )
    assert list(workflows) == ['bar']


@pytest.fixture
def patch_expand_workflow_tokens(monkeypatch):
    """Define the output of scan events."""

    def _patch_expand_workflow_tokens(_ids):

        async def _expand_workflow_tokens_impl(tokens, match_active=True):
            nonlocal _ids
            for id_ in _ids:
                yield tokens.duplicate(workflow=id_)

        monkeypatch.setattr(
            'cylc.flow.id_cli._expand_workflow_tokens_impl',
            _expand_workflow_tokens_impl,
        )

    _patch_expand_workflow_tokens(['xxx'])
    return _patch_expand_workflow_tokens


@pytest.mark.parametrize(
    'ids_in,ids_out,multi_mode',
    [
        # multi mode should be True if multiple workflows are defined
        (['a//'], ['a'], False),
        (['a//', 'b//'], ['a', 'b'], True),
        # or if pattern matching is used, irrespective of the number of matches
        (['*//'], ['xxx'], True),
    ]
)
async def test_parse_ids_multi_mode(
    patch_expand_workflow_tokens,
    ids_in,
    ids_out,
    multi_mode,
):
    """It should glob for workflows.

    Note:
        More advanced tests for this in the integration tests.

    """
    workflows, _multi_mode = await parse_ids_async(
        *ids_in,
        constraint='workflows',
        match_workflows=True,
    )
    assert list(workflows) == ids_out
    assert _multi_mode == multi_mode


@pytest.fixture
def src_dir(tmp_path):
    """A src dir containing a workflow called "a"."""
    cwd_before = Path.cwd()
    src_dir = (tmp_path / 'a')
    src_dir.mkdir()
    src_file = src_dir / 'flow.cylc'
    src_file.touch()
    os.chdir(tmp_path)
    yield src_dir
    os.chdir(cwd_before)


def test_parse_src_path(src_dir, monkeypatch):
    """It should locate src dirs."""
    # valid absolute path
    workflow_id, src_path, src_file_path = _parse_src_path(
        str(src_dir.resolve())
    )
    assert workflow_id == 'a'
    assert src_path == src_dir
    assert src_file_path == src_dir / 'flow.cylc'

    # broken absolute path
    with pytest.raises(UserInputError):
        workflow_id, src_path, src_file_path = _parse_src_path(
            str(src_dir.resolve()) + 'xyz'
        )

    # valid relative path
    workflow_id, src_path, src_file_path = _parse_src_path('./a')
    assert workflow_id == 'a'
    assert src_path == src_dir
    assert src_file_path == src_dir / 'flow.cylc'

    # broken relative path
    with pytest.raises(UserInputError):
        _parse_src_path('./xxx')

    # relative '.' (invalid)
    with pytest.raises(WorkflowFilesError) as exc_ctx:
        workflow_id, src_path, src_file_path = _parse_src_path('.')
    assert 'No flow.cylc or suite.rc in .' in str(exc_ctx.value)

    # move into the src dir
    monkeypatch.chdir(src_dir)

    # relative '.' (valid)
    workflow_id, src_path, src_file_path = _parse_src_path('.')
    assert workflow_id == 'a'
    assert src_path == src_dir
    assert src_file_path == src_dir / 'flow.cylc'

    # relative './<flow-file>'
    workflow_id, src_path, src_file_path = _parse_src_path('./flow.cylc')
    assert workflow_id == 'a'
    assert src_path == src_dir
    assert src_file_path == src_dir / 'flow.cylc'


async def test_parse_ids_src_path(src_dir):
    workflows, src_path = await parse_ids_async(
        './a',
        src=True,
        constraint='workflows',
    )
    assert workflows == {'a': []}


@pytest.mark.parametrize(
    'ids_in,error_msg',
    [
        (
            ['/home/me/whatever'],
            'Invalid ID: /home/me/whatever',
        ),
        (
            ['foo/..'],
            'cannot be a path that points to the cylc-run directory or above',
        ),
        (
            ['~alice/foo'],
            "Operating on others users' workflows is not supported",
        ),
    ]
)
async def test_parse_ids_invalid_ids(ids_in, error_msg):
    """It should error for invalid IDs."""
    with pytest.raises(Exception) as exc_ctx:
        await parse_ids_async(
            *ids_in,
            constraint='workflows',
        )
    assert error_msg in str(exc_ctx.value)


async def test_parse_ids_file(tmp_run_dir):
    """It should reject IDs that are paths to files."""
    tmp_path = tmp_run_dir('x')
    tmp_file = tmp_path / 'tmpfile'
    tmp_file.touch()
    (tmp_path / WorkflowFiles.FLOW_FILE).touch()
    # using a directory should work
    await parse_ids_async(
        str(tmp_path.relative_to(get_cylc_run_dir())),
        constraint='workflows',
    )
    with pytest.raises(Exception) as exc_ctx:
        # using a file should not
        await parse_ids_async(
            str(tmp_file.relative_to(get_cylc_run_dir())),
            constraint='workflows',
        )
    assert 'Workflow ID cannot be a file' in str(exc_ctx.value)


async def test_parse_ids_constraint():
    """It should validate input against the constraint."""
    # constraint: workflows
    await parse_ids_async('a//', constraint='workflows')
    with pytest.raises(UserInputError):
        await parse_ids_async('a//b', constraint='workflows')
    # constraint: tasks
    await parse_ids_async('a//b', constraint='tasks')
    with pytest.raises(UserInputError):
        await parse_ids_async('a//', constraint='tasks')
    # constraint: mixed
    await parse_ids_async('a//', constraint='mixed')
    await parse_ids_async('a//b', constraint='mixed')
    # constraint: invalid
    with pytest.raises(ValueError):
        await parse_ids_async('foo', constraint='bar')


async def test_parse_ids_src_run(abc_src_dir, tmp_run_dir):
    """It should locate the flow file when src=True."""
    # locate flow file for a src workflow
    workflows, flow_file_path = await parse_ids_async(
        './a',
        src=True,
        constraint='workflows',
    )
    assert list(workflows) == ['a']
    assert flow_file_path == abc_src_dir / 'a' / WorkflowFiles.FLOW_FILE

    # locate flow file for a run workflow
    run_dir = tmp_run_dir('b')
    workflows, flow_file_path = await parse_ids_async(
        'b',
        src=True,
        constraint='workflows',
    )
    assert list(workflows) == ['b']
    assert flow_file_path == run_dir / WorkflowFiles.FLOW_FILE


def test_validate_constraint():
    """It should validate tokens against the constraint."""
    # constraint=workflows
    _validate_constraint(Tokens(workflow='a'), constraint='workflows')
    with pytest.raises(UserInputError):
        _validate_constraint(Tokens(cycle='a'), constraint='workflows')
    with pytest.raises(UserInputError):
        _validate_constraint(Tokens(), constraint='workflows')
    # constraint=tasks
    _validate_constraint(Tokens(cycle='a'), constraint='tasks')
    with pytest.raises(UserInputError):
        _validate_constraint(Tokens(workflow='a'), constraint='tasks')
    with pytest.raises(UserInputError):
        _validate_constraint(Tokens(), constraint='tasks')
    # constraint=mixed
    _validate_constraint(Tokens(workflow='a'), constraint='mixed')
    _validate_constraint(Tokens(cycle='a'), constraint='mixed')
    with pytest.raises(UserInputError):
        _validate_constraint(Tokens(), constraint='mixed')


def test_validate_workflow_ids(tmp_run_dir):
    _validate_workflow_ids(Tokens('workflow'), src_path='')
    with pytest.raises(UserInputError):
        _validate_workflow_ids(Tokens('~alice/workflow'), src_path='')
    run_dir = tmp_run_dir('b')
    with pytest.raises(UserInputError):
        _validate_workflow_ids(
            Tokens('workflow'),
            src_path=run_dir / 'flow.cylc',
        )


def test_validate_number():
    _validate_number(Tokens('a'), max_workflows=1)
    with pytest.raises(UserInputError):
        _validate_number(Tokens('a'), Tokens('b'), max_workflows=1)
    t1 = Tokens(cycle='1')
    t2 = Tokens(cycle='2')
    _validate_number(t1, max_tasks=1)
    with pytest.raises(UserInputError):
        _validate_number(t1, t2, max_tasks=1)


@pytest.fixture
def no_scan(monkeypatch):
    """Disable the filesystem part of scan."""

    @pipe
    async def _scan():
        # something that looks like scan but doesn't do anything
        yield

    monkeypatch.setattr('cylc.flow.id_cli.scan', _scan)


async def test_expand_workflow_tokens_impl_selector(no_scan):
    """It should reject filters it can't handle."""
    tokens = tokenise('~user/*')
    await _expand_workflow_tokens([tokens])
    tokens['workflow_sel'] = 'stopped'
    with pytest.raises(UserInputError):
        await _expand_workflow_tokens([tokens])
