File: test_cylc.py

package info (click to toggle)
cylc-flow 8.6.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 14,368 kB
  • sloc: python: 87,751; sh: 17,109; sql: 233; xml: 171; javascript: 78; lisp: 55; makefile: 11
file content (142 lines) | stat: -rw-r--r-- 4,789 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
#!/usr/bin/env python3
# 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
import sys
from types import SimpleNamespace
from typing import Callable
from unittest.mock import Mock

import pytest

from cylc.flow.scripts.cylc import iter_commands, pythonpath_manip


@pytest.fixture
def mock_entry_points(monkeypatch: pytest.MonkeyPatch):
    """Mock a range of entry points."""
    def _load_fail(*args, **kwargs):
        raise ModuleNotFoundError('foo')

    def _resolve_ok(*args, **kwargs):
        return Mock()

    def _require_ok(*args, **kwargs):
        return

    def _mocked_entry_points(include_bad: bool = False):
        commands = {
            # an entry point with all dependencies installed:
            'good': SimpleNamespace(
                name='good',
                module='os.path',
                load=_resolve_ok,
                extras=[],
                dist=SimpleNamespace(name='a'),
            ),
            # an entry point with optional dependencies missing:
            'missing': SimpleNamespace(
                name='missing',
                module='not.a.python.module',  # force an import error
                load=_load_fail,
                extras=[],
                dist=SimpleNamespace(name='foo'),
            ),
        }
        if include_bad:
            # an entry point with non-optional dependencies unexpectedly
            # missing:
            commands['bad'] = SimpleNamespace(
                name='bad',
                module='not.a.python.module',
                load=_load_fail,
                require=_require_ok,
                extras=[],
                dist=SimpleNamespace(name='d'),
            )
        monkeypatch.setattr('cylc.flow.scripts.cylc.COMMANDS', commands)

    return _mocked_entry_points


def test_iter_commands(mock_entry_points):
    """Test listing commands works ok.

    It should exclude commands with missing optional dependencies.
    """
    mock_entry_points()
    commands = list(iter_commands())
    assert [i[0] for i in commands] == ['good']


def test_iter_commands_bad(mock_entry_points):
    """Test listing commands doesn't fail on import error."""
    mock_entry_points(include_bad=True)
    list(iter_commands())


def test_execute_cmd(
    mock_entry_points,
    capsys: pytest.CaptureFixture,
):
    """It should fail with a warning for commands with missing dependencies."""
    # (stop IDEs reporting code as unreachable in this test)
    execute_cmd: Callable
    from cylc.flow.scripts.cylc import execute_cmd

    mock_entry_points(include_bad=True)

    # the "good" entry point should exit 0 (exit with no args)
    assert execute_cmd('good') == 0
    assert capsys.readouterr().err == ''

    # the "missing" entry point should exit 1 with a warning to stderr
    assert execute_cmd('missing') == 1
    assert capsys.readouterr().err.strip() == (
        '"cylc missing" requires "foo"\n\nModuleNotFoundError: foo'
    )

    # the "bad" entry point should log an error
    assert execute_cmd('bad') == 1

    stderr = capsys.readouterr().err.strip()
    assert '"cylc bad" requires "d"' in stderr
    assert 'ModuleNotFoundError: foo' in stderr


def test_pythonpath_manip(monkeypatch):
    """pythonpath_manip removes items in PYTHONPATH from sys.path

    and adds items from CYLC_PYTHONPATH
    """

    # Local CYLC_PYTHONPATH can mess with this test.
    monkeypatch.delenv('CYLC_PYTHONPATH', raising=False)

    monkeypatch.setenv('PYTHONPATH', '/remove1:/remove2')
    monkeypatch.setattr('sys.path', ['/leave-alone', '/remove1', '/remove2'])
    pythonpath_manip()
    # ... we don't change PYTHONPATH
    assert os.environ['PYTHONPATH'] == '/remove1:/remove2'
    # ... but we do remove PYTHONPATH items from sys.path, and don't remove
    # items there not in PYTHONPATH
    assert sys.path == ['/leave-alone']
    # If CYLC_PYTHONPATH is set we retrieve its contents and
    # add them to the sys.path:
    monkeypatch.setenv('CYLC_PYTHONPATH', '/add1:/add2')
    pythonpath_manip()
    assert sys.path == ['/add1', '/add2', '/leave-alone']