test_egon-data.test_airflow() - Code Metrics - Inspection of "Fix tests and GitHub actions" - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#475)

by Stephan

created 2021-11-26 12:55 UTC

test_egon-data.test_airflow() A

↳ Parent: test_egon-data

Complexity

Conditions

Size

Total Lines	12
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	8
nop	0
dl	0
loc	12
rs	10
c	0
b	0
f	0

from importlib import import_module

from click.testing import CliRunner
import pytest

from egon.data import __version__
from egon.data.cli import egon_data


def test_main():
    runner = CliRunner()
    with runner.isolated_filesystem():
        result = runner.invoke(egon_data, ["--version"])

    assert result.output == "{name}, version {version}\n".format(
        name=egon_data.name, version=__version__
    )
    assert result.exit_code == 0


@pytest.mark.skip(
    reason=(
        "Needs `docker` and/or PostgreSQL and we're currently not making sure"
        "\nthese are present on the continuous integration service(s) we use."
    )
)
def test_airflow():
    """ Test that `egon-data airflow` correctly forwards to airflow. """
    runner = CliRunner()
    with runner.isolated_filesystem():
        result = runner.invoke(egon_data, ["airflow", "--help"])
    assert result.output == ""


def test_pipeline_importability():
    error = None
    for m in ["egon.data.airflow.dags.pipeline"]:
        try:
            import_module(m)
        except Exception as e:
            error = e
        assert error is None, (
            "\nDid not expect an error when importing:\n\n  `{}`\n\nGot: {}"
        ).format(m, error)


1			from importlib import import_module
2
3			from click.testing import CliRunner
4			import pytest
5
6			from egon.data import __version__
7			from egon.data.cli import egon_data
8
9
10			def test_main():
11			runner = CliRunner()
12			with runner.isolated_filesystem():
13			result = runner.invoke(egon_data, ["--version"])
14
15			assert result.output == "{name}, version {version}\n".format(
16			name=egon_data.name, version=__version__
17			)
18			assert result.exit_code == 0
19
20
21			@pytest.mark.skip(
22			reason=(
23			"Needs `docker` and/or PostgreSQL and we're currently not making sure"
24			"\nthese are present on the continuous integration service(s) we use."
25			)
26			)
27			def test_airflow():
28			""" Test that `egon-data airflow` correctly forwards to airflow. """
29			runner = CliRunner()
30			with runner.isolated_filesystem():
31			result = runner.invoke(egon_data, ["airflow", "--help"])
32			assert result.output == ""
33
34
35			def test_pipeline_importability():
36			error = None
37			for m in ["egon.data.airflow.dags.pipeline"]:
38			try:
39			import_module(m)
40			except Exception as e:
41			error = e
42			assert error is None, (
43			"\nDid not expect an error when importing:\n\n `{}`\n\nGot: {}"
44			).format(m, error)
45

openego / eGon-data

Pull Request — dev (#475)

test_egon-data.test_airflow() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like