1# Copyright (c) 2012, Willow Garage, Inc.
2# All rights reserved.
3#
4# Redistribution and use in source and binary forms, with or without
5# modification, are permitted provided that the following conditions are met:
6#
7#     * Redistributions of source code must retain the above copyright
8#       notice, this list of conditions and the following disclaimer.
9#     * Redistributions in binary form must reproduce the above copyright
10#       notice, this list of conditions and the following disclaimer in the
11#       documentation and/or other materials provided with the distribution.
12#     * Neither the name of the Willow Garage, Inc. nor the names of its
13#       contributors may be used to endorse or promote products derived from
14#       this software without specific prior written permission.
15#
16# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
17# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
20# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
21# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
22# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
23# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
24# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
25# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
26# POSSIBILITY OF SUCH DAMAGE.
27
28import os
29import tempfile
30import yaml
31try:
32    from urllib.request import urlopen
33    from urllib.error import URLError
34except ImportError:
35    from urllib2 import urlopen
36    from urllib2 import URLError
37
38import rospkg.distro
39import rosdep2.sources_list
40
41GITHUB_BASE_URL = 'https://raw.githubusercontent.com/ros/rosdistro/master/rosdep/base.yaml'
42
43
44def get_test_dir():
45    return os.path.abspath(os.path.join(os.path.dirname(__file__), 'sources.list.d'))
46
47
48def test_get_sources_list_dir():
49    assert rosdep2.sources_list.get_sources_list_dir()
50
51
52def test_get_sources_cache_dir():
53    assert rosdep2.sources_list.get_sources_cache_dir()
54
55
56def test_url_constants():
57    from rosdep2.sources_list import DEFAULT_SOURCES_LIST_URL
58    for url_name, url in [('DEFAULT_SOURCES_LIST_URL', DEFAULT_SOURCES_LIST_URL)]:
59        try:
60            f = urlopen(url)
61            f.read()
62            f.close()
63        except Exception:
64            assert False, 'URL [%s][%s] failed to download' % (url_name, url)
65
66
67def test_download_default_sources_list():
68    from rosdep2.sources_list import download_default_sources_list
69    data = download_default_sources_list()
70    assert 'http' in data, data  # sanity check, all sources files have urls
71    try:
72        download_default_sources_list(url='http://bad.ros.org/foo.yaml')
73        assert False, 'should not have succeeded/valdiated'
74    except URLError:
75        pass
76
77
78def test_CachedDataSource():
79    from rosdep2.sources_list import CachedDataSource, DataSource, TYPE_GBPDISTRO, TYPE_YAML
80    type_ = TYPE_GBPDISTRO
81    url = 'http://fake.willowgarage.com/foo'
82    tags = ['tag1']
83    rosdep_data = {'key': {}}
84    origin = '/tmp/bar'
85    cds = CachedDataSource(type_, url, tags, rosdep_data, origin=origin)
86    assert cds == CachedDataSource(type_, url, tags, rosdep_data, origin=origin)
87    assert cds != CachedDataSource(type_, url, tags, rosdep_data, origin=None)
88    assert cds != CachedDataSource(type_, url, tags, {}, origin=origin)
89    assert cds != CachedDataSource(TYPE_YAML, url, tags, rosdep_data, origin=origin)
90    assert cds != CachedDataSource(type_, 'http://ros.org/foo.yaml', tags, rosdep_data, origin=origin)
91    assert cds != DataSource(type_, url, tags, origin=origin)
92    assert DataSource(type_, url, tags, origin=origin) != cds
93    assert cds.type == type_
94    assert cds.url == url
95    assert cds.origin == origin
96    assert cds.rosdep_data == rosdep_data
97    assert type_ in str(cds)
98    assert type_ in repr(cds)
99    assert url in str(cds)
100    assert url in repr(cds)
101    assert tags[0] in str(cds)
102    assert tags[0] in repr(cds)
103    assert 'key' in str(cds)
104    assert 'key' in repr(cds)
105
106
107def test_DataSource():
108    from rosdep2.sources_list import DataSource
109    data_source = DataSource('yaml', 'http://fake/url', ['tag1', 'tag2'])
110    assert data_source == rosdep2.sources_list.DataSource('yaml', 'http://fake/url', ['tag1', 'tag2'])
111    assert 'yaml' == data_source.type
112    assert 'http://fake/url' == data_source.url
113    assert ['tag1', 'tag2'] == data_source.tags
114    assert 'yaml http://fake/url tag1 tag2' == str(data_source)
115
116    data_source_foo = DataSource('yaml', 'http://fake/url', ['tag1', 'tag2'], origin='foo')
117    assert data_source_foo != data_source
118    assert data_source_foo.origin == 'foo'
119    assert '[foo]:\nyaml http://fake/url tag1 tag2' == str(data_source_foo), str(data_source_foo)
120
121    assert repr(data_source)
122
123    try:
124        rosdep2.sources_list.DataSource('yaml', 'http://fake/url', 'tag1', origin='foo')
125        assert False, 'should have raised'
126    except ValueError:
127        pass
128    try:
129        rosdep2.sources_list.DataSource('yaml', 'non url', ['tag1'], origin='foo')
130        assert False, 'should have raised'
131    except ValueError:
132        pass
133    try:
134        rosdep2.sources_list.DataSource('bad', 'http://fake/url', ['tag1'], origin='foo')
135        assert False, 'should have raised'
136    except ValueError:
137        pass
138    try:
139        rosdep2.sources_list.DataSource('yaml', 'http://host.no.path/', ['tag1'], origin='foo')
140        assert False, 'should have raised'
141    except ValueError:
142        pass
143
144
145def test_parse_sources_file():
146    from rosdep2.sources_list import parse_sources_file
147    from rosdep2 import InvalidData
148    for f in ['20-default.list', '30-nonexistent.list']:
149        path = os.path.join(get_test_dir(), f)
150        sources = parse_sources_file(path)
151        assert sources[0].type == 'yaml'
152        assert sources[0].origin == path, sources[0].origin
153    try:
154        sources = parse_sources_file('bad')
155    except InvalidData:
156        pass
157
158
159def test_parse_sources_list():
160    from rosdep2.sources_list import parse_sources_list
161    from rosdep2 import InvalidData
162    # test with non-existent dir, should return with empty list as
163    # directory is not required to exist.
164    assert [] == parse_sources_list(sources_list_dir='/not/a/real/path')
165
166    # test with real dir
167    path = get_test_dir()
168    sources_list = parse_sources_list(sources_list_dir=get_test_dir())
169    # at time test was written, at least two sources files
170    assert len(sources_list) > 1
171    # make sure files got loaded in intended order
172    assert sources_list[0].origin.endswith('20-default.list')
173    assert sources_list[1].origin.endswith('20-default.list')
174    assert sources_list[2].origin.endswith('30-nonexistent.list')
175
176    # tripwire -- we don't know what the actual return value is, but
177    # should not error on a correctly configured test system.
178    parse_sources_list()
179
180
181def test_write_cache_file():
182    from rosdep2.sources_list import write_cache_file, compute_filename_hash, PICKLE_CACHE_EXT
183    try:
184        import cPickle as pickle
185    except ImportError:
186        import pickle
187    tempdir = tempfile.mkdtemp()
188
189    filepath = write_cache_file(tempdir, 'foo', {'data': 1}) + PICKLE_CACHE_EXT
190    computed_path = os.path.join(tempdir, compute_filename_hash('foo')) + PICKLE_CACHE_EXT
191    assert os.path.samefile(filepath, computed_path)
192    with open(filepath, 'rb') as f:
193        assert {'data': 1} == pickle.loads(f.read())
194
195
196def test_update_sources_list():
197    from rosdep2.sources_list import update_sources_list, InvalidData, compute_filename_hash, PICKLE_CACHE_EXT
198    try:
199        import cPickle as pickle
200    except ImportError:
201        import pickle
202    try:
203        from urllib.request import pathname2url
204    except ImportError:
205        from urllib import pathname2url
206    sources_list_dir = get_test_dir()
207    index_path = os.path.abspath(os.path.join(os.path.dirname(__file__), 'rosdistro', 'index.yaml'))
208    index_url = 'file://' + pathname2url(index_path)
209    os.environ['ROSDISTRO_INDEX_URL'] = index_url
210    tempdir = tempfile.mkdtemp()
211    # use a subdirectory of test dir to make sure rosdep creates the necessary substructure
212    tempdir = os.path.join(tempdir, 'newdir')
213
214    errors = []
215
216    def error_handler(loc, e):
217        errors.append((loc, e))
218    retval = update_sources_list(sources_list_dir=sources_list_dir,
219                                 sources_cache_dir=tempdir, error_handler=error_handler)
220    assert retval
221    assert len(retval) == 2, retval
222    # one of our sources is intentionally bad, this should be a softfail
223    assert len(errors) == 1, errors
224    assert errors[0][0].url == 'https://badhostname.willowgarage.com/rosdep.yaml'
225
226    source0, path0 = retval[0]
227    assert source0.origin.endswith('20-default.list'), source0
228    hash1 = compute_filename_hash(GITHUB_URL)
229    hash2 = compute_filename_hash(BADHOSTNAME_URL)
230    filepath = os.path.join(tempdir, hash1)
231    assert filepath == path0, '%s vs %s' % (filepath, path0)
232    with open(filepath + PICKLE_CACHE_EXT, 'rb') as f:
233        data = pickle.loads(f.read())
234        assert 'cmake' in data
235
236    # verify that cache index exists. contract specifies that even
237    # failed downloads are specified in the index, just in case old
238    # download data is present.
239    with open(os.path.join(tempdir, 'index'), 'r') as f:
240        index = f.read().strip()
241    expected = "#autogenerated by rosdep, do not edit. use 'rosdep update' instead\n"\
242               'yaml %s \n'\
243               'yaml %s python\n'\
244               'yaml %s ubuntu' % (GITHUB_URL, GITHUB_PYTHON_URL, BADHOSTNAME_URL)
245    assert expected == index, '\n[%s]\nvs\n[%s]' % (expected, index)
246
247
248def test_load_cached_sources_list():
249    from rosdep2.sources_list import load_cached_sources_list, update_sources_list
250    tempdir = tempfile.mkdtemp()
251
252    # test behavior on empty cache
253    assert [] == load_cached_sources_list(sources_cache_dir=tempdir)
254
255    # pull in cache data
256    sources_list_dir = get_test_dir()
257    retval = update_sources_list(sources_list_dir=sources_list_dir,
258                                 sources_cache_dir=tempdir, error_handler=None)
259    assert retval
260
261    # now test with cached data
262    retval = load_cached_sources_list(sources_cache_dir=tempdir)
263    assert len(retval) == 3, 'len(%s) != 3' % retval
264    source0 = retval[0]
265    source1 = retval[1]
266    source2 = retval[2]
267
268    # this should be the 'default' source
269    assert 'python' in source1.rosdep_data
270    assert not source0.tags
271
272    # this should be the 'non-existent' source
273    assert source2.rosdep_data == {}
274    assert source2.tags == ['ubuntu']
275
276
277def test_DataSourceMatcher():
278    empty_data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', [])
279    assert empty_data_source == rosdep2.sources_list.DataSource('yaml', 'http://fake/url', [])
280
281    # matcher must match 'all' tags
282    data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', ['tag1', 'tag2'])
283    partial_data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', ['tag1'])
284
285    # same tags as test data source
286    matcher = rosdep2.sources_list.DataSourceMatcher(['tag1', 'tag2'])
287    assert matcher.matches(data_source)
288    assert matcher.matches(partial_data_source)
289    assert matcher.matches(empty_data_source)
290
291    # alter one tag
292    matcher = rosdep2.sources_list.DataSourceMatcher(['tag1', 'tag3'])
293    assert not matcher.matches(data_source)
294    assert matcher.matches(empty_data_source)
295    matcher = rosdep2.sources_list.DataSourceMatcher(['tag1'])
296    assert not matcher.matches(data_source)
297
298
299def test_download_rosdep_data():
300    from rosdep2.sources_list import download_rosdep_data
301    from rosdep2 import DownloadFailure
302    url = GITHUB_BASE_URL
303    data = download_rosdep_data(url)
304    assert 'boost' in data  # sanity check
305
306    # try with a bad URL
307    try:
308        data = download_rosdep_data('http://badhost.willowgarage.com/rosdep.yaml')
309        assert False, 'should have raised'
310    except DownloadFailure as e:
311        pass
312    # try to trigger both non-dict clause and YAMLError clause
313    for url in [
314        'https://code.ros.org/svn/release/trunk/distros/',
315        'https://code.ros.org/svn/release/trunk/distros/manifest.xml',
316    ]:
317        try:
318            data = download_rosdep_data(url)
319            assert False, 'should have raised'
320        except DownloadFailure as e:
321            pass
322
323
324BADHOSTNAME_URL = 'https://badhostname.willowgarage.com/rosdep.yaml'
325GITHUB_URL = 'https://github.com/ros/rosdistro/raw/master/rosdep/base.yaml'
326GITHUB_PYTHON_URL = 'https://github.com/ros/rosdistro/raw/master/rosdep/python.yaml'
327GITHUB_FUERTE_URL = 'https://raw.githubusercontent.com/ros-infrastructure/rosdep_rules/master/rosdep_fuerte.yaml'
328EXAMPLE_SOURCES_DATA_BAD_TYPE = 'YAML %s' % (GITHUB_URL)
329EXAMPLE_SOURCES_DATA_BAD_URL = 'yaml not-a-url tag1 tag2'
330EXAMPLE_SOURCES_DATA_BAD_LEN = 'yaml'
331EXAMPLE_SOURCES_DATA_NO_TAGS = 'yaml %s' % (GITHUB_URL)
332EXAMPLE_SOURCES_DATA = 'yaml %s fuerte ubuntu' % (GITHUB_URL)
333EXAMPLE_SOURCES_DATA_MULTILINE = """
334# this is a comment, above and below are empty lines
335
336yaml %s
337yaml %s fuerte ubuntu
338""" % (GITHUB_URL, GITHUB_FUERTE_URL)
339
340
341def test_parse_sources_data():
342    from rosdep2.sources_list import parse_sources_data, TYPE_YAML, InvalidData
343
344    retval = parse_sources_data(EXAMPLE_SOURCES_DATA, origin='foo')
345    assert len(retval) == 1
346    sd = retval[0]
347    assert sd.type == TYPE_YAML, sd.type
348    assert sd.url == GITHUB_URL
349    assert sd.tags == ['fuerte', 'ubuntu']
350    assert sd.origin == 'foo'
351
352    retval = parse_sources_data(EXAMPLE_SOURCES_DATA_NO_TAGS)
353    assert len(retval) == 1
354    sd = retval[0]
355    assert sd.type == TYPE_YAML
356    assert sd.url == GITHUB_URL
357    assert sd.tags == []
358    assert sd.origin == '<string>'
359
360    retval = parse_sources_data(EXAMPLE_SOURCES_DATA_MULTILINE)
361    assert len(retval) == 2
362    sd = retval[0]
363    assert sd.type == TYPE_YAML
364    assert sd.url == GITHUB_URL
365    assert sd.tags == []
366
367    sd = retval[1]
368    assert sd.type == TYPE_YAML
369    assert sd.url == GITHUB_FUERTE_URL
370    assert sd.tags == ['fuerte', 'ubuntu']
371
372    for bad in [EXAMPLE_SOURCES_DATA_BAD_URL,
373                EXAMPLE_SOURCES_DATA_BAD_TYPE,
374                EXAMPLE_SOURCES_DATA_BAD_LEN]:
375        try:
376            parse_sources_data(bad)
377            assert False, 'should have raised: %s' % (bad)
378        except InvalidData as e:
379            pass
380
381
382def test_DataSourceMatcher_create_default():
383    distro_name = rospkg.distro.current_distro_codename()
384    os_detect = rospkg.os_detect.OsDetect()
385    os_name, os_version, os_codename = os_detect.detect_os()
386
387    matcher = rosdep2.sources_list.DataSourceMatcher.create_default()
388
389    # matches full
390    os_data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', [distro_name, os_name, os_codename])
391    assert matcher.matches(os_data_source)
392
393    # matches against current os
394    os_data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', [os_name, os_codename])
395    assert matcher.matches(os_data_source)
396
397    # matches against current distro
398    distro_data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', [distro_name])
399    assert matcher.matches(distro_data_source)
400
401    # test matcher with os override
402    matcher = rosdep2.sources_list.DataSourceMatcher.create_default(os_override=('fubuntu', 'flucid'))
403    assert not matcher.matches(os_data_source)
404    data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', ['fubuntu'])
405    assert matcher.matches(data_source)
406    data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', ['flucid'])
407    assert matcher.matches(data_source)
408    data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', ['flucid', 'fubuntu'])
409    assert matcher.matches(data_source)
410    data_source = rosdep2.sources_list.DataSource('yaml', 'http://fake/url', ['kubuntu', 'lucid'])
411    assert not matcher.matches(data_source)
412
413
414def test_SourcesListLoader_create_default():
415    from rosdep2.sources_list import update_sources_list, SourcesListLoader, DataSourceMatcher
416    # create temp dir for holding sources cache
417    tempdir = tempfile.mkdtemp()
418
419    # pull in cache data
420    sources_list_dir = get_test_dir()
421    retval = update_sources_list(sources_list_dir=sources_list_dir,
422                                 sources_cache_dir=tempdir, error_handler=None)
423    assert retval
424
425    # now test with cached data
426    matcher = rosdep2.sources_list.DataSourceMatcher(['ubuntu', 'lucid'])
427    loader = SourcesListLoader.create_default(matcher, sources_cache_dir=tempdir)
428    assert loader.sources
429    sources0 = loader.sources
430    assert not any([s for s in loader.sources if not matcher.matches(s)])
431
432    loader = SourcesListLoader.create_default(matcher, sources_cache_dir=tempdir)
433    assert sources0 == loader.sources
434
435    # now test with different matcher
436    matcher2 = rosdep2.sources_list.DataSourceMatcher(['python'])
437    loader2 = SourcesListLoader.create_default(matcher2, sources_cache_dir=tempdir)
438    assert loader2.sources
439    # - should have filtered down to python-only
440    assert sources0 != loader2.sources
441    assert not any([s for s in loader2.sources if not matcher2.matches(s)])
442
443    # test API
444
445    # very simple, always raises RNF
446    try:
447        loader.get_rosdeps('foo')
448    except rospkg.ResourceNotFound:
449        pass
450    try:
451        loader.get_view_key('foo')
452    except rospkg.ResourceNotFound:
453        pass
454
455    assert [] == loader.get_loadable_resources()
456    all_sources = [x.url for x in loader.sources]
457    assert all_sources == loader.get_loadable_views()
458
459    # test get_source early to make sure model matches expected
460    try:
461        loader.get_source('foo')
462        assert False, 'should have raised'
463    except rospkg.ResourceNotFound:
464        pass
465    s = loader.get_source(GITHUB_URL)
466    assert s.url == GITHUB_URL
467
468    # get_view_dependencies
469    # - loader doesn't new view name, so assume everything
470    assert all_sources == loader.get_view_dependencies('foo')
471    # - actual views don't depend on anything
472    assert [] == loader.get_view_dependencies(GITHUB_URL)
473
474    # load_view
475    from rosdep2.model import RosdepDatabase
476    for verbose in [True, False]:
477        rosdep_db = RosdepDatabase()
478        loader.load_view(GITHUB_URL, rosdep_db, verbose=verbose)
479        assert rosdep_db.is_loaded(GITHUB_URL)
480        assert [] == rosdep_db.get_view_dependencies(GITHUB_URL)
481        entry = rosdep_db.get_view_data(GITHUB_URL)
482        assert 'cmake' in entry.rosdep_data
483        assert GITHUB_URL == entry.origin
484
485    #  - coverage, repeat loader, should noop
486    loader.load_view(GITHUB_URL, rosdep_db)
487
488
489def test_unpickle_same_results():
490    try:
491        import cPickle as pickle
492    except ImportError:
493        import pickle
494    with open(os.path.join('test', 'fixtures', 'python2cache.pickle'), 'rb') as py2_cache:
495        py2_result = pickle.loads(py2_cache.read())
496    with open(os.path.join('test', 'fixtures', 'python3cache.pickle'), 'rb') as py3_cache:
497        py3_result = pickle.loads(py3_cache.read())
498    assert py2_result == py3_result
499