test_rosdep.py

import os
import subprocess
import sys

import pytest
from rosdep2.sources_list import download_default_sources_list
from rosdep2.sources_list import load_cached_sources_list
from rosdep2.sources_list import parse_sources_list
from rosdep2.sources_list import update_sources_list


@pytest.fixture(scope='session')
def _ensure_rosdep_sources_list(tmpdir_factory):
    """
    Ensure that a valid sources.list.d is available.

    :returns: Path to a directory containing a temporary sources list, or
      'None' if a valid sources list is already present.
    """
    try:
        sources_list = parse_sources_list()
    except (OSError, InvalidData):
        sources_list = None

    if not sources_list:
        temp_sources_list_dir = str(tmpdir_factory.mktemp('sources.list.d'))
        default_sources_list = download_default_sources_list()
        sources_list = open(os.path.join(temp_sources_list_dir, '10-temp.list'), 'w')
        try:
            sources_list.write(default_sources_list)
        finally:
            sources_list.close()
        return temp_sources_list_dir


def _set_up_rosdep_cache(ros_home, sources_list_dir):
    """Set up a temporary rosdep cache in the given ROS home directory."""
    old_ros_home = os.environ.get('ROS_HOME')
    os.environ['ROS_HOME'] = ros_home
    try:
        assert update_sources_list(
            sources_list_dir=sources_list_dir,
            skip_eol_distros=True,
            quiet=True)
    finally:
        if old_ros_home is not None:
            os.environ['ROS_HOME'] = old_ros_home
        else:
            os.environ.pop('ROS_HOME', None)


@pytest.fixture(scope='session')
def _ensure_rosdep_cache(request, tmpdir_factory):
    """
    Ensure that a valid rosdep cache is available.

    :returns: Path to a ROS home directory containing a temporary cache, or
      'None' if a valid rosdep cache is already present.
    """
    if not load_cached_sources_list():
        temp_ros_home = str(tmpdir_factory.mktemp('ros_home'))
        sources_list_dir = request.getfixturevalue(
            '_ensure_rosdep_sources_list')
        _set_up_rosdep_cache(temp_ros_home, sources_list_dir)
        return temp_ros_home


@pytest.fixture
def ensure_rosdep(request, _ensure_rosdep_cache):
    """
    Ensure that a valid rosdep cache is available for a test to use.
    """
    if _ensure_rosdep_cache is None:
        return

    old_ros_home = os.environ.get('ROS_HOME')
    def _restore_ros_home():
        if old_ros_home is not None:
            os.environ['ROS_HOME'] = old_ros_home
        else:
            os.environ.pop('ROS_HOME', None)
    request.addfinalizer(_restore_ros_home)
    os.environ['ROS_HOME'] = _ensure_rosdep_cache


@pytest.mark.usefixtures('ensure_rosdep')
def test_rosdep():
    subprocess.check_call([sys.executable, '-m', 'rosdep2', 'where-defined', 'python3'])