Best Python code snippet using pytest-django_python
test_genmsg_msg_loader.py
Source:test_genmsg_msg_loader.py  
1# Software License Agreement (BSD License)2#3# Copyright (c) 2009, Willow Garage, Inc.4# All rights reserved.5#6# Redistribution and use in source and binary forms, with or without7# modification, are permitted provided that the following conditions8# are met:9#10#  * Redistributions of source code must retain the above copyright11#    notice, this list of conditions and the following disclaimer.12#  * Redistributions in binary form must reproduce the above13#    copyright notice, this list of conditions and the following14#    disclaimer in the documentation and/or other materials provided15#    with the distribution.16#  * Neither the name of Willow Garage, Inc. nor the names of its17#    contributors may be used to endorse or promote products derived18#    from this software without specific prior written permission.19#20# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS21# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT22# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS23# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE24# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,25# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,26# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;27# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER28# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT29# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN30# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE31# POSSIBILITY OF SUCH DAMAGE.32import os33import sys34import random35def get_test_dir():36    return os.path.abspath(os.path.join(os.path.dirname(__file__), 'files'))37def test_exceptions():38    from genmsg import MsgNotFound39    try:40        raise MsgNotFound('hello')41    except MsgNotFound:42        pass43def test__convert_constant_value():44    from genmsg.msg_loader import convert_constant_value45    from genmsg import InvalidMsgSpec46    assert 0. == convert_constant_value('float32', '0.0')47    assert 0. == convert_constant_value('float64', '0.0')48    49    assert 'fo o' == convert_constant_value('string', '   fo o ')50    assert 1 == convert_constant_value('byte', '1')51    assert 1 == convert_constant_value('char', '1')52    assert 1 == convert_constant_value('int8', '1')53    assert 12 == convert_constant_value('int16', '12')54    assert -13 == convert_constant_value('int32', '-13')55    assert 14 == convert_constant_value('int64', '14')56    assert 0 == convert_constant_value('uint8', '0')57    assert 18 == convert_constant_value('uint16', '18')58    assert 19 == convert_constant_value('uint32', '19')59    assert 20 == convert_constant_value('uint64', '20')60    assert True == convert_constant_value('bool', '1')61    assert False == convert_constant_value('bool', '0')    62    width_fail = [('int8', '129'), ('uint8', '256'),63                  ('int16', '35536'), ('uint16', '-1'),('uint16', '65536'),64                  ('int32', '3000000000'),('int32', '-2700000000'),65                  ('uint32', '-1'),('uint32', '41000000000'),66                  ('uint64', '-1')]67    for t, v in width_fail:68        try:69            convert_constant_value(t, v)70            assert False, "should have failed width check: %s, %s"%(t, v)71        except InvalidMsgSpec:72            pass73    type_fail = [('int32', 'f'), ('float32', 'baz')]74    for t, v in type_fail:75        try:76            convert_constant_value(t, v)77            assert False, "should have failed type check: %s, %s"%(t, v)78        except ValueError:79            pass80    try:81        convert_constant_value('foo', '1')82        assert False, "should have failed invalid type"83    except InvalidMsgSpec:84        pass85def test__load_constant_line():86    from genmsg.msgs import Constant, InvalidMsgSpec87    from genmsg.msg_loader import _load_constant_line88    try:89        _load_constant_line("int8 field=alpha")90        assert False, "should have raised"91    except InvalidMsgSpec:92        pass93    try:94        _load_constant_line("int8 field=")95        assert False, "should have raised"96    except InvalidMsgSpec:97        pass98    try:99        _load_constant_line("faketype field=1")100        assert False, "should have raised"101    except InvalidMsgSpec:102        pass103    104    c = _load_constant_line("int8 field=1")105    assert c == Constant('int8', 'field', 1, '1')106    c = _load_constant_line("string val=hello #world")107    assert c == Constant('string', 'val', 'hello #world', 'hello #world')108    109def test__load_field_line():110    from genmsg.msgs import InvalidMsgSpec, Field111    from genmsg.msg_loader import _load_field_line, InvalidMsgSpec, Field, is_valid_msg_field_name112    try:113       _load_field_line("string", 'foo')114       assert False, "should have raised"115    except InvalidMsgSpec:116        pass117    assert not is_valid_msg_field_name('string[')118    try:119       _load_field_line("string data!", 'foo')120       assert False, "should have raised"121    except InvalidMsgSpec:122        pass123    try:124       _load_field_line("string[ data", 'foo')125       assert False, "should have raised"126    except InvalidMsgSpec:127        pass128    129    f =_load_field_line("string str", 'foo')130    assert f == ('string', 'str')131    132    f =_load_field_line("string str #nonsense", 'foo')133    assert f == ('string', 'str')134    f =_load_field_line("String str #nonsense", '')135    assert f == ('String', 'str')136    f =_load_field_line("String str #nonsense", 'foo')137    assert f == ('foo/String', 'str')138    # make sure Header is mapped139    f =_load_field_line("Header header #nonsense", 'somewhere')140    assert f == ('std_msgs/Header', 'header'), f141    f =_load_field_line("Header header #nonsense", '')142    assert f == ('std_msgs/Header', 'header'), f143def test_load_msg_from_string():144    # make sure Header -> std_msgs/Header conversion works145    from genmsg.msgs import Constant146    from genmsg.msg_loader import load_msg_from_string, MsgContext147    context = MsgContext.create_default()148    msgspec = load_msg_from_string(context, "Header header", 'test_pkg/HeaderTest')149    print(msgspec)150    assert msgspec.has_header()151    assert msgspec.types == ['std_msgs/Header']152    assert msgspec.names == ['header']153    assert msgspec.constants == []154    assert msgspec.short_name == 'HeaderTest'155    assert msgspec.package == 'test_pkg'156    assert msgspec.full_name == 'test_pkg/HeaderTest'157    158    msgspec = load_msg_from_string(context, "int8 c=1\nHeader header\nint64 data", 'test_pkg/HeaderValsTest')159    assert msgspec.has_header()160    assert msgspec.types == ['std_msgs/Header', 'int64']161    assert msgspec.names == ['header', 'data']162    assert msgspec.constants == [Constant('int8', 'c', 1, '1')]163    assert msgspec.short_name == 'HeaderValsTest'164    assert msgspec.package == 'test_pkg'165    assert msgspec.full_name == 'test_pkg/HeaderValsTest'166    167    msgspec = load_msg_from_string(context, "string data\nint64 data2", 'test_pkg/ValsTest')168    assert not msgspec.has_header()169    assert msgspec.types == ['string', 'int64']170    assert msgspec.names == ['data', 'data2']171    assert msgspec.constants == []172    assert msgspec.short_name == 'ValsTest'173    assert msgspec.full_name == 'test_pkg/ValsTest'174def _validate_TestString(msgspec):175    assert ['caller_id', 'orig_caller_id', 'data'] == msgspec.names, msgspec.names176    assert ['string', 'string', 'string'] == msgspec.types, msgspec.types177def test_load_msg_from_file():178    from genmsg.msgs import InvalidMsgSpec179    from genmsg.msg_loader import load_msg_from_file, MsgContext180    test_d = get_test_dir()181    test_ros_dir = os.path.join(test_d, 'test_ros', 'msg')182    test_string_path = os.path.join(test_ros_dir, 'TestString.msg')183    msg_context = MsgContext.create_default()184    spec = load_msg_from_file(msg_context, test_string_path, 'test_ros/TestString')185    assert spec.full_name == 'test_ros/TestString'186    assert spec.package == 'test_ros'187    assert spec.short_name == 'TestString'188    _validate_TestString(spec)189    # test repeat190    spec_2 = load_msg_from_file(msg_context, test_string_path, 'test_ros/TestString')191    assert spec == spec_2192    assert spec.package == spec_2.package193    assert spec.short_name == spec_2.short_name194    195    # test w/ bad file196    test_bad_path = os.path.join(test_ros_dir, 'Bad.msg')197    try:198        load_msg_from_file(msg_context, test_bad_path, 'test_ros/Bad')199        assert False, "should have raised"200    except InvalidMsgSpec:201        pass202    203    # supposed to register204    assert msg_context.is_registered('test_ros/TestString'), msg_context205    206def test_load_msg_from_string_TestString():207    from genmsg.msg_loader import load_msg_from_string, MsgContext208    test_d = get_test_dir()209    test_ros_dir = os.path.join(test_d, 'test_ros', 'msg')210    test_string_path = os.path.join(test_ros_dir, 'TestString.msg')211    with open(test_string_path) as f:212        text = f.read()213    msg_context = MsgContext.create_default()214    _validate_TestString(load_msg_from_string(msg_context, text, 'test_ros/TestString'))215    # supposed to register216    assert msg_context.is_registered('test_ros/TestString'), msg_context217def test_load_msg_by_type():218    from genmsg.msg_loader import load_msg_by_type, MsgContext, MsgNotFound219    220    test_d = get_test_dir()221    geometry_d = os.path.join(test_d, 'geometry_msgs', 'msg')222    test_ros_dir = os.path.join(test_d, 'test_ros', 'msg')223    test_string_path = os.path.join(test_ros_dir, 'TestString.msg')224    search_path = {225        'test_ros': [ test_ros_dir ],226        'geometry_msgs': [ geometry_d ],227        }228    msg_context = MsgContext.create_default()229    msgspec = load_msg_by_type(msg_context, 'test_ros/TestString', search_path)230    _validate_TestString(msgspec)231    # supposed to register232    assert msg_context.is_registered('test_ros/TestString'), msg_context233    # test invalid search path234    try:235        load_msg_by_type(msg_context, 'test_ros/TestString', [test_string_path])236        assert False, "should have raised"237    except ValueError:238        pass239    # test not found240    try:241        load_msg_by_type(msg_context, 'test_ros/Fake', search_path)242        assert False, "should have raised"243    except MsgNotFound:244        pass245    # test all the known geometry msgs246    test_d = get_test_dir()247    for f in os.listdir(geometry_d):248        if f.endswith('.msg'):249            short = f[:-4]250            msg_type = 'geometry_msgs/%s'%short251            spec = load_msg_by_type(msg_context, msg_type, search_path)252            assert spec is not None253            assert spec.package == 'geometry_msgs'254            assert spec.full_name == msg_type255            assert spec.short_name == short256            with open(os.path.join(geometry_d, f)) as file_h:257                assert spec.text == file_h.read()258            # all types with 'Stamped' in name have headers259            if 'Stamped' in f:260                assert spec.has_header(), msg_type261    262def test_get_msg_file():263    from genmsg import MsgNotFound264    from genmsg.msg_loader import get_msg_file265    test_d = get_test_dir()266    test_ros_dir = os.path.join(test_d, 'test_ros', 'msg')267    test_string_path = os.path.join(test_ros_dir, 'TestString.msg')268    search_path = {269        'test_ros': [ test_ros_dir ],270        }271    assert test_string_path == get_msg_file('test_ros', 'TestString', search_path)272    try:273        get_msg_file('test_ros', 'DNE', search_path)274        assert False, "should have raised"275    except MsgNotFound:276        pass277    try:278        get_msg_file('bad_pkg', 'TestString', search_path)279        assert False, "should have raised"280    except MsgNotFound:281        pass282    # test with invalid search path283    try:284        get_msg_file('test_ros', 'TestString', [test_string_path])285        assert False, "should have raised"286    except ValueError:287        pass288def test_get_srv_file():289    from genmsg import MsgNotFound290    from genmsg.msg_loader import get_srv_file291    test_d = get_test_dir()292    test_ros_dir = os.path.join(test_d, 'test_ros', 'srv')293    std_srvs_dir = os.path.join(test_d, 'std_srvs', 'srv')294    empty_path = os.path.join(std_srvs_dir, 'Empty.srv')295    search_path = {296        'test_ros': [ test_ros_dir ],297        'std_srvs': [ std_srvs_dir ],298        }299    assert empty_path == get_srv_file('std_srvs', 'Empty', search_path)300    try:301        get_srv_file('test_ros', 'DNE', search_path)302        assert False, "should have raised"303    except MsgNotFound:304        pass305    try:306        get_srv_file('bad_pkg', 'TestString', search_path)307        assert False, "should have raised"308    except MsgNotFound:309        pass310    # test with invalid search path311    try:312        get_srv_file('std_srvs', 'Empty', [std_srvs_dir])313        assert False, "should have raised"314    except ValueError:315        pass316def test_MsgContext():317    from genmsg.msg_loader import MsgContext, load_msg_from_file318    msg_context = MsgContext()319    assert not msg_context.is_registered('time')320    assert not msg_context.is_registered('duration')321    322    msg_context = MsgContext.create_default()323    # tripwires324    repr(msg_context)325    str(msg_context)326    assert msg_context.is_registered('time'), msg_context._registered_packages327    assert msg_context.is_registered('duration')328    assert not msg_context.is_registered('test_ros/TestString')329    assert not msg_context.is_registered('Header')330    331    # start loading stuff into context332    test_d = get_test_dir()333    test_ros_dir = os.path.join(test_d, 'test_ros', 'msg')334    test_string_path = os.path.join(test_ros_dir, 'TestString.msg')335    spec = load_msg_from_file(msg_context, test_string_path, 'test_ros/TestString')336    msg_context.register('test_ros/TestString', spec)337    assert msg_context.get_registered('test_ros/TestString') == spec338    try:339        msg_context.get_registered('bad/TestString')340        assert False, 'should have raised'341    except KeyError:342        pass343    344    assert msg_context.is_registered('test_ros/TestString')345    # test Header346    assert not msg_context.is_registered('Header')347    assert not msg_context.is_registered('std_msgs/Header')348    349    msg_context.register('std_msgs/Header', spec)350    assert msg_context.is_registered('std_msgs/Header')351def test_load_srv_from_file():352    from genmsg.msg_loader import MsgContext, load_srv_from_file353        354    msg_context = MsgContext.create_default()355    356    d = get_test_dir()357    filename = os.path.join(d, 'test_ros', 'srv', 'AddTwoInts.srv')358    with open(filename, 'r') as f:359        text = f.read()360        361    full_name = 'test_ros/AddTwoInts'362    spec = load_srv_from_file(msg_context, filename, full_name)363    assert spec == load_srv_from_file(msg_context, filename, full_name)364    assert ['int64', 'int64'] == spec.request.types, spec.request.types365    assert ['a', 'b'] == spec.request.names366    assert text == spec.text367    assert full_name == spec.full_name368def test_load_msg_depends():369    #TODO: should there just be a 'load_msg, implicit=True?'370    from genmsg.msg_loader import MsgContext, load_msg_by_type, load_msg_depends, MsgNotFound371    test_d = get_test_dir()372    search_path = {373        'test_ros': [ os.path.join(test_d, 'test_ros', 'msg') ],374        'std_msgs': [ os.path.join(test_d, 'std_msgs', 'msg') ],375        'geometry_msgs': [ os.path.join(test_d, 'geometry_msgs', 'msg') ],376        'sensor_msgs': [ os.path.join(test_d, 'sensor_msgs', 'msg') ],377        'invalid': [ os.path.join(test_d, 'invalid', 'msg') ],378        }379    380    # Test not found381    msg_context = MsgContext.create_default()382    root_spec = load_msg_by_type(msg_context, 'invalid/BadDepend', search_path)383    try:384        load_msg_depends(msg_context, root_spec, search_path)385        assert False, "should have raised MsgNotFound"386    except MsgNotFound:387        pass388    root_spec = load_msg_by_type(msg_context, 'invalid/BadLocalDepend', search_path)389    try:390        load_msg_depends(msg_context, root_spec, search_path)391        assert False, "should have raised MsgNotFound"392    except MsgNotFound:393        pass394    msg_context = MsgContext.create_default()395    root_spec = load_msg_by_type(msg_context, 'std_msgs/Int32', search_path)396    load_msg_depends(msg_context, root_spec, search_path)397    file_p = os.path.join(test_d, 'std_msgs', 'msg', 'Int32.msg')398    assert file_p == msg_context.get_file('std_msgs/Int32')399    assert [] == msg_context.get_depends('std_msgs/Int32')400    msg_context = MsgContext.create_default()401    root_spec = load_msg_by_type(msg_context, 'std_msgs/Header', search_path)402    load_msg_depends(msg_context, root_spec, search_path)403    file_p = os.path.join(test_d, 'std_msgs', 'msg', 'Header.msg')404    assert file_p == msg_context.get_file('std_msgs/Header')405    assert [] == msg_context.get_depends('std_msgs/Header')406    msg_context = MsgContext.create_default()407    root_spec = load_msg_by_type(msg_context, 'Header', search_path)408    load_msg_depends(msg_context, root_spec, search_path)409    file_p = os.path.join(test_d, 'std_msgs', 'msg', 'Header.msg')410    assert file_p == msg_context.get_file('std_msgs/Header')411    assert [] == msg_context.get_depends('std_msgs/Header')412    msg_context = MsgContext.create_default()413    root_spec = load_msg_by_type(msg_context, 'std_msgs/Int32MultiArray', search_path)414    load_msg_depends(msg_context, root_spec, search_path)415    file_p = os.path.join(test_d, 'std_msgs', 'msg', 'Int32MultiArray.msg')416    assert file_p == msg_context.get_file('std_msgs/Int32MultiArray')417    val = msg_context.get_all_depends('std_msgs/Int32MultiArray')418    assert set(['std_msgs/MultiArrayLayout', 'std_msgs/MultiArrayDimension']) == set(val), val419    assert 2 == len(val), val420    val = msg_context.get_depends('std_msgs/Int32MultiArray')421    assert set(['std_msgs/MultiArrayLayout']) == set(val), val422    for s in ['MultiArrayLayout', 'MultiArrayDimension']:423        file_p = os.path.join(test_d, 'std_msgs', 'msg', '%s.msg'%s)424        assert file_p == msg_context.get_file('std_msgs/%s'%s)425def test_load_msg_depends_stamped():426    #TODO: should there just be a 'load_msg, implicit=True?'427    from genmsg.msg_loader import MsgContext, load_msg_by_type, load_msg_depends428    test_d = get_test_dir()429    geometry_d = os.path.join(test_d, 'geometry_msgs', 'msg')430    search_path = {431        'test_ros': [ os.path.join(test_d, 'test_ros', 'msg') ],432        'std_msgs': [ os.path.join(test_d, 'std_msgs', 'msg') ],433        'geometry_msgs': [ geometry_d ],434        'sensor_msgs': [ os.path.join(test_d, 'sensor_msgs', 'msg') ],435        }436    # Test with Stamped and deeper hierarchies, Header437    msg_context = MsgContext.create_default()438    root_spec = load_msg_by_type(msg_context, 'geometry_msgs/PoseStamped', search_path)439    load_msg_depends(msg_context, root_spec, search_path)440    file_p = os.path.join(test_d, 'geometry_msgs', 'msg', 'PoseStamped.msg')441    assert file_p == msg_context.get_file('geometry_msgs/PoseStamped')442    val = msg_context.get_all_depends('geometry_msgs/PoseStamped')443    assert set(['std_msgs/Header', 'geometry_msgs/Pose', 'geometry_msgs/Point', 'geometry_msgs/Quaternion']) == set(val), val444    val = msg_context.get_depends('geometry_msgs/PoseStamped')445    assert set(['std_msgs/Header', 'geometry_msgs/Pose']) == set(val), val446    for s in ['Header']:447        file_p = os.path.join(test_d, 'std_msgs', 'msg', '%s.msg'%s)448        assert file_p == msg_context.get_file('std_msgs/%s'%s)449    for s in ['Pose', 'Point', 'Quaternion']:450        file_p = os.path.join(geometry_d, '%s.msg'%s)451        assert file_p == msg_context.get_file('geometry_msgs/%s'%s)452    msg_context = MsgContext.create_default()453    root_spec = load_msg_by_type(msg_context, 'geometry_msgs/TwistWithCovarianceStamped', search_path)454    load_msg_depends(msg_context, root_spec, search_path)455    file_p = os.path.join(test_d, 'geometry_msgs', 'msg', 'TwistWithCovarianceStamped.msg')456    assert file_p == msg_context.get_file('geometry_msgs/TwistWithCovarianceStamped')457    val = msg_context.get_all_depends('geometry_msgs/TwistWithCovarianceStamped')458    assert set(['std_msgs/Header', 'geometry_msgs/TwistWithCovariance', 'geometry_msgs/Twist', 'geometry_msgs/Vector3']) == set(val), val459    val = msg_context.get_depends('geometry_msgs/TwistWithCovarianceStamped')460    assert set(['std_msgs/Header', 'geometry_msgs/TwistWithCovariance']) == set(val), val461    for s in ['Header']:462        file_p = os.path.join(test_d, 'std_msgs', 'msg', '%s.msg'%s)463        assert file_p == msg_context.get_file('std_msgs/%s'%s)464    for s in ['TwistWithCovariance', 'Twist', 'Vector3']:465        file_p = os.path.join(geometry_d, '%s.msg'%s)466        assert file_p == msg_context.get_file('geometry_msgs/%s'%s)467    msg_context = MsgContext.create_default()468    root_spec = load_msg_by_type(msg_context, 'sensor_msgs/Imu', search_path)469    load_msg_depends(msg_context, root_spec, search_path)470    file_p = os.path.join(test_d, 'sensor_msgs', 'msg', 'Imu.msg')471    assert file_p == msg_context.get_file('sensor_msgs/Imu')472    val = msg_context.get_all_depends('sensor_msgs/Imu')473    assert set(['std_msgs/Header', 'geometry_msgs/Quaternion', 'geometry_msgs/Vector3']) == set(val), val474    val = msg_context.get_depends('sensor_msgs/Imu')475    assert set(['std_msgs/Header', 'geometry_msgs/Quaternion', 'geometry_msgs/Vector3']) == set(val), val476    for s in ['Header']:477        file_p = os.path.join(test_d, 'std_msgs', 'msg', '%s.msg'%s)478        assert file_p == msg_context.get_file('std_msgs/%s'%s)479    for s in ['Quaternion', 'Vector3']:480        file_p = os.path.join(geometry_d, '%s.msg'%s)481        assert file_p == msg_context.get_file('geometry_msgs/%s'%s)482def test_load_depends_msg():483    from genmsg.msg_loader import MsgContext, load_msg_by_type, load_depends, MsgNotFound, load_srv_by_type484    test_d = get_test_dir()485    geometry_d = os.path.join(test_d, 'geometry_msgs', 'msg')486    msg_search_path = {487        'test_ros': [ os.path.join(test_d, 'test_ros', 'msg') ],488        'std_msgs': [ os.path.join(test_d, 'std_msgs', 'msg') ],489        'geometry_msgs': [ geometry_d ],490        'sensor_msgs': [ os.path.join(test_d, 'sensor_msgs', 'msg') ],491        'invalid': [ os.path.join(test_d, 'invalid', 'msg') ],492        }493    # Test not found494    msg_context = MsgContext.create_default()495    root_spec = load_msg_by_type(msg_context, 'invalid/BadDepend', msg_search_path)496    try:497        load_depends(msg_context, root_spec, msg_search_path)498        assert False, "should have raised MsgNotFound"499    except MsgNotFound:500        pass501    root_spec = load_msg_by_type(msg_context, 'invalid/BadLocalDepend', msg_search_path)502    try:503        load_depends(msg_context, root_spec, msg_search_path)504        assert False, "should have raised MsgNotFound"505    except MsgNotFound:506        pass507    508    # Test with msgs509    msg_context = MsgContext.create_default()510    root_spec = load_msg_by_type(msg_context, 'geometry_msgs/PoseStamped', msg_search_path)511    load_depends(msg_context, root_spec, msg_search_path)512    file_p = os.path.join(test_d, 'geometry_msgs', 'msg', 'PoseStamped.msg')513    assert file_p == msg_context.get_file('geometry_msgs/PoseStamped')514    val = msg_context.get_all_depends('geometry_msgs/PoseStamped')515    assert set(['std_msgs/Header', 'geometry_msgs/Pose', 'geometry_msgs/Point', 'geometry_msgs/Quaternion']) == set(val), val516    val = msg_context.get_depends('geometry_msgs/PoseStamped')517    assert set(['std_msgs/Header', 'geometry_msgs/Pose']) == set(val), val518    for s in ['Header']:519        file_p = os.path.join(test_d, 'std_msgs', 'msg', '%s.msg'%s)520        assert file_p == msg_context.get_file('std_msgs/%s'%s)521    for s in ['Pose', 'Point', 'Quaternion']:522        file_p = os.path.join(geometry_d, '%s.msg'%s)523        assert file_p == msg_context.get_file('geometry_msgs/%s'%s)524    msg_context = MsgContext.create_default()525    root_spec = load_msg_by_type(msg_context, 'sensor_msgs/Imu', msg_search_path)526    load_depends(msg_context, root_spec, msg_search_path)527    file_p = os.path.join(test_d, 'sensor_msgs', 'msg', 'Imu.msg')528    assert file_p == msg_context.get_file('sensor_msgs/Imu')529    val = msg_context.get_depends('sensor_msgs/Imu')530    assert set(['std_msgs/Header', 'geometry_msgs/Quaternion', 'geometry_msgs/Vector3']) == set(val), val531    for s in ['Header']:532        file_p = os.path.join(test_d, 'std_msgs', 'msg', '%s.msg'%s)533        assert file_p == msg_context.get_file('std_msgs/%s'%s)534    for s in ['Quaternion', 'Vector3']:535        file_p = os.path.join(geometry_d, '%s.msg'%s)536        assert file_p == msg_context.get_file('geometry_msgs/%s'%s)537def test_load_depends_srv():538    from genmsg.msg_loader import MsgContext, load_msg_by_type, load_depends, MsgNotFound, load_srv_by_type539    test_d = get_test_dir()540    geometry_d = os.path.join(test_d, 'geometry_msgs', 'msg')541    msg_search_path = {542        'test_ros': [ os.path.join(test_d, 'test_ros', 'msg') ],543        'std_msgs': [ os.path.join(test_d, 'std_msgs', 'msg') ],544        'geometry_msgs': [ geometry_d ],545        'sensor_msgs': [ os.path.join(test_d, 'sensor_msgs', 'msg') ],546        'invalid': [ os.path.join(test_d, 'invalid', 'msg') ],547        }548    # Test with srvs549    srv_search_path = {550        'test_ros': [ os.path.join(test_d, 'test_ros', 'srv') ],551        'std_srvs': [ os.path.join(test_d, 'std_srvs', 'srv') ],552        }553    msg_context = MsgContext.create_default()554    root_spec = load_srv_by_type(msg_context, 'test_ros/AddTwoInts', srv_search_path)555    load_depends(msg_context, root_spec, msg_search_path)556    val = msg_context.get_depends('test_ros/AddTwoIntsRequest')557    assert val == [], val558    val = msg_context.get_depends('test_ros/AddTwoIntsResponse')    559    assert val == [], val560    # test with srv that has depends561    msg_context = MsgContext.create_default()562    response_deps = ['std_msgs/Header', 'geometry_msgs/Pose', 'geometry_msgs/PoseStamped', 'geometry_msgs/Point', 'geometry_msgs/Quaternion']563    root_spec = load_srv_by_type(msg_context, 'test_ros/GetPoseStamped', srv_search_path)564    load_depends(msg_context, root_spec, msg_search_path)565    for d in response_deps:566        assert msg_context.is_registered(d)567    val = msg_context.get_depends('test_ros/GetPoseStampedRequest')568    assert val == [], val569    val = msg_context.get_depends('test_ros/GetPoseStampedResponse')    570    assert val == ['geometry_msgs/PoseStamped']571    572    # Test with nonsense573    class Foo(object): pass574    try:575        load_depends(msg_context, Foo(), msg_search_path)576        assert False, "should have raised"577    except ValueError:578        pass579    580def test_load_srv_by_type():581    from genmsg.msg_loader import load_srv_by_type, MsgContext, MsgNotFound582    583    test_d = get_test_dir()584    test_ros_dir = os.path.join(test_d, 'test_ros', 'srv')585    std_srvs_dir = os.path.join(test_d, 'std_srvs', 'srv')586    empty_path = os.path.join(std_srvs_dir, 'Empty.srv')587    a2i_path = os.path.join(std_srvs_dir, 'AddTwoInts.srv')588    search_path = {589        'test_ros': [ test_ros_dir ],590        'std_srvs': [ std_srvs_dir ],591        }592    msg_context = MsgContext.create_default()593    spec = load_srv_by_type(msg_context, 'std_srvs/Empty', search_path)594    assert msg_context.is_registered('std_srvs/EmptyRequest')595    assert msg_context.is_registered('std_srvs/EmptyResponse')596    assert msg_context.get_registered('std_srvs/EmptyRequest') == spec.request597    assert msg_context.get_registered('std_srvs/EmptyResponse') == spec.response598    assert msg_context.get_file('std_srvs/EmptyRequest') == empty_path, msg_context.get_file('std_srvs/EmptyRequest')599    assert msg_context.get_file('std_srvs/EmptyResponse') == empty_path,msg_context.get_file('std_srvs/EmptyResponse')600    assert spec.request.full_name == 'std_srvs/EmptyRequest'601    assert spec.response.full_name == 'std_srvs/EmptyResponse'602    assert spec.request.short_name == 'EmptyRequest'603    assert spec.response.short_name == 'EmptyResponse'604    assert spec.request.package == 'std_srvs'605    assert spec.response.package == 'std_srvs'606    for f in [spec.request.names, spec.request.types, spec.response.names, spec.response.types]:607        assert [] == f608    spec = load_srv_by_type(msg_context, 'test_ros/AddTwoInts', search_path)609    assert msg_context.is_registered('test_ros/AddTwoIntsRequest')610    assert msg_context.is_registered('test_ros/AddTwoIntsResponse')611    assert msg_context.get_registered('test_ros/AddTwoIntsRequest') == spec.request612    assert msg_context.get_registered('test_ros/AddTwoIntsResponse') == spec.response613    assert spec.request.types == ['int64', 'int64'], spec.request.types614    assert spec.request.names == ['a', 'b'], spec.request.names615    assert spec.response.types == ['int64'], spec.response.types616    assert spec.response.names == ['sum'], spec.response.names617    # test invalid search path618    try:619        load_srv_by_type(msg_context, 'test_ros/AddTwoInts', [std_srvs_dir])620        assert False, "should have raised"621    except ValueError:622        pass623    # test not found624    try:625        load_srv_by_type(msg_context, 'test_ros/Fake', search_path)626        assert False, "should have raised"627    except MsgNotFound:...preprocess_test_outlier.py
Source:preprocess_test_outlier.py  
1import pandas as pd2import numpy as np3import datetime4import matplotlib.pyplot as plt5from argparse import ArgumentParser6parser = ArgumentParser()7args = parser.parse_args()8test_d = pd.read_csv("./Data/test.csv")9def strTodatetime(datestr, format):10    return datetime.datetime.strptime(datestr, format)11hotel_label = []12arrival_date = []13#country_label = []14#revenue = []15countries = ['PRT', 'GBR', 'FRA', 'ESP', 'DEU']16'''17#drop data which has no country18test_d = test_d.drop(['agent','company'],axis=1)19test_d = test_d.dropna(axis = 0)20'''21#Deal with Outliers22test_d.loc[test_d.lead_time > 500,'lead_time'] = 50023test_d.loc[test_d.days_in_waiting_list > 0,'days_in_waiting_list'] = 124test_d.loc[test_d.stays_in_weekend_nights >=  5,'stays_in_weekend_nights'] = 525test_d.loc[test_d.adults > 4,'adults'] = 426test_d.loc[test_d.previous_bookings_not_canceled > 0,'previous_bookings_not_canceled'] = 127test_d.loc[test_d.previous_cancellations > 0,'previous_cancellations'] = 128test_d.loc[test_d.stays_in_week_nights > 10,'stays_in_week_nights'] = 1029test_d.loc[test_d.booking_changes > 5,'booking_changes']=  530test_d.loc[test_d.babies > 8,'babies'] = 031test_d.loc[test_d.required_car_parking_spaces > 5,'required_car_parking_spaces'] = 032test_d.loc[test_d.children > 8,'children']  = 033#combine children and babies together as kids34test_d['kids'] = test_d.children + test_d.babies35#combine total mumbers by adding kids and adults36test_d['total_members'] = test_d.kids + test_d.adults37for i in range(test_d.shape[0]):38    """39    æ´çæ¢æData40    """41    #hotel42    if test_d["hotel"][i] == 'Resort Hotel':43        hotel_label.append(0)44    else:45        hotel_label.append(1)46    47    #date48    arr_date = str(test_d['arrival_date_year'][i]) + str(test_d['arrival_date_month'][i])+ str(test_d['arrival_date_day_of_month'][i])49    arrival_date.append(strTodatetime(arr_date,"%Y%B%d"))50    ''' 51    #Country, å
æ'PRT'è¨çº1, å
¶é¤çº052    if test_d["country"][i] == 'PRT':53        country_label.append(1)54    else:55        country_label.append(0)56    '''57    if test_d["country"][i] not in countries:58        test_d['country'][i] = 'Others'59    if i % 10000 == 0:60        print('{} finished'.format(round(i/test_d.shape[0], 2)))61test_d = test_d.join(pd.get_dummies(test_d.meal, prefix='meal')) #meal for one hot62test_d = test_d.join(pd.get_dummies(test_d.market_segment, prefix='segment')) #market_segment for one hot63test_d = test_d.join(pd.get_dummies(test_d.assigned_room_type, prefix = 'assigned_room'))64test_d = test_d.join(pd.get_dummies(test_d.deposit_type, prefix = 'deposit_type'))65test_d = test_d.join(pd.get_dummies(test_d.customer_type, prefix = 'customer_type'))66test_d = test_d.join(pd.get_dummies(test_d.distribution_channel, prefix = 'distribution_channel'))67test_d = test_d.join(pd.get_dummies(test_d.reserved_room_type, prefix = 'reserved_room_type'))68test_d = test_d.join(pd.get_dummies(test_d.country, prefix = 'country'))69test_d['hotel_label'] = hotel_label70test_d['arrival_date'] = arrival_date71#test_d['country_label'] = country_label72#Drop column73drop_list = ['meal','market_segment','assigned_room_type','deposit_type','customer_type','distribution_channel','reserved_room_type',74             'hotel','country','arrival_date_year','arrival_date_month','arrival_date_day_of_month',75             'ID','company','agent']76test_d = test_d.drop(drop_list, axis=1)77#è£naå¼78for i in range(test_d.shape[0]):79    if np.isnan(test_d['children'][i]):80        test_d['children'][i] = 081#test_d缺å°å¹¾åtrain_dæçcolumn82#åå¥çºsegment_Undefined/ assigned_room_L/ distribution_channel_Undefined83segment_Undefined = np.zeros(test_d.shape[0])84assigned_room_L = np.zeros(test_d.shape[0])85distribution_channel_Undefined = np.zeros(test_d.shape[0])86reserved_room_type_L = np.zeros(test_d.shape[0])87test_d['segment_Undefined'] = segment_Undefined88test_d['assigned_room_L'] = assigned_room_L89test_d['distribution_channel_Undefined'] = distribution_channel_Undefined90test_d['reserved_room_type_L'] = reserved_room_type_L...preprocess_test_adr_roomtype.py
Source:preprocess_test_adr_roomtype.py  
1import pandas as pd2import numpy as np3import datetime4import matplotlib.pyplot as plt5from argparse import ArgumentParser6parser = ArgumentParser()7args = parser.parse_args()8test_d = pd.read_csv("./Data/test.csv")9def strTodatetime(datestr, format):10    return datetime.datetime.strptime(datestr, format)11hotel_label = []12arrival_date = []13country_label = []14#revenue = []15for i in range(test_d.shape[0]):16    """17    æ´çæ¢æData18    """19    #hotel20    if test_d["hotel"][i] == 'Resort Hotel':21        hotel_label.append(0)22    else:23        hotel_label.append(1)24    25    #date26    arr_date = str(test_d['arrival_date_year'][i]) + str(test_d['arrival_date_month'][i])+ str(test_d['arrival_date_day_of_month'][i])27    arrival_date.append(strTodatetime(arr_date,"%Y%B%d"))28        29    #Country, å
æ'PRT'è¨çº1, å
¶é¤çº030    if test_d["country"][i] == 'PRT':31        country_label.append(1)32    else:33        country_label.append(0)34            35    if i % 10000 == 0:36        print('{} finished'.format(round(i/test_d.shape[0], 2)))37test_d = test_d.join(pd.get_dummies(test_d.meal, prefix='meal')) #meal for one hot38test_d = test_d.join(pd.get_dummies(test_d.market_segment, prefix='segment')) #market_segment for one hot39test_d = test_d.join(pd.get_dummies(test_d.assigned_room_type, prefix = 'assigned_room'))40test_d = test_d.join(pd.get_dummies(test_d.deposit_type, prefix = 'deposit_type'))41test_d = test_d.join(pd.get_dummies(test_d.customer_type, prefix = 'customer_type'))42test_d = test_d.join(pd.get_dummies(test_d.distribution_channel, prefix = 'distribution_channel'))43test_d = test_d.join(pd.get_dummies(test_d.reserved_room_type, prefix = 'reserved_room_type'))44test_d['hotel_label'] = hotel_label45test_d['arrival_date'] = arrival_date46test_d['country_label'] = country_label47#Drop column48drop_list = ['meal','market_segment','assigned_room_type','deposit_type','customer_type','distribution_channel','reserved_room_type',49             'hotel','country','arrival_date_year','arrival_date_month','arrival_date_day_of_month',50             'ID','company','agent']51test_d = test_d.drop(drop_list, axis=1)52#è£naå¼53for i in range(test_d.shape[0]):54    if np.isnan(test_d['children'][i]):55        test_d['children'][i] = 056#test_d缺å°å¹¾åtrain_dæçcolumn57#åå¥çºsegment_Undefined/ assigned_room_L/ distribution_channel_Undefined58segment_Undefined = np.zeros(test_d.shape[0])59assigned_room_L = np.zeros(test_d.shape[0])60distribution_channel_Undefined = np.zeros(test_d.shape[0])61reserved_room_type_L = np.zeros(test_d.shape[0])62test_d['segment_Undefined'] = segment_Undefined63test_d['assigned_room_L'] = assigned_room_L64test_d['distribution_channel_Undefined'] = distribution_channel_Undefined65test_d['reserved_room_type_L'] = reserved_room_type_L...preprocess_test.py
Source:preprocess_test.py  
1import pandas as pd2import numpy as np3import datetime4import matplotlib.pyplot as plt5from argparse import ArgumentParser6parser = ArgumentParser()7args = parser.parse_args()8test_d = pd.read_csv("./Data/test.csv")9def strTodatetime(datestr, format):10    return datetime.datetime.strptime(datestr, format)11hotel_label = []12arrival_date = []13country_label = []14#revenue = []15for i in range(test_d.shape[0]):16    """17    æ´çæ¢æData18    """19    #hotel20    if test_d["hotel"][i] == 'Resort Hotel':21        hotel_label.append(0)22    else:23        hotel_label.append(1)24    25    #date26    arr_date = str(test_d['arrival_date_year'][i]) + str(test_d['arrival_date_month'][i])+ str(test_d['arrival_date_day_of_month'][i])27    arrival_date.append(strTodatetime(arr_date,"%Y%B%d"))28        29    #Country, å
æ'PRT'è¨çº1, å
¶é¤çº030    if test_d["country"][i] == 'PRT':31        country_label.append(1)32    else:33        country_label.append(0)34            35    if i % 10000 == 0:36        print('{} finished'.format(round(i/test_d.shape[0], 2)))37test_d = test_d.join(pd.get_dummies(test_d.meal, prefix='meal')) #meal for one hot38test_d = test_d.join(pd.get_dummies(test_d.market_segment, prefix='segment')) #market_segment for one hot39test_d = test_d.join(pd.get_dummies(test_d.assigned_room_type, prefix = 'assigned_room'))40test_d = test_d.join(pd.get_dummies(test_d.deposit_type, prefix = 'deposit_type'))41test_d = test_d.join(pd.get_dummies(test_d.customer_type, prefix = 'customer_type'))42test_d = test_d.join(pd.get_dummies(test_d.distribution_channel, prefix = 'distribution_channel'))43test_d['hotel_label'] = hotel_label44test_d['arrival_date'] = arrival_date45test_d['country_label'] = country_label46#Drop column47drop_list = ['meal','market_segment','assigned_room_type','deposit_type','customer_type',48             'hotel','country','arrival_date_year','arrival_date_month','arrival_date_day_of_month',49             'ID', 'distribution_channel','company','agent','reserved_room_type']50test_d = test_d.drop(drop_list, axis=1)51#è£naå¼52for i in range(test_d.shape[0]):53    if np.isnan(test_d['children'][i]):54        test_d['children'][i] = 055#test_d缺å°å¹¾åtrain_dæçcolumn56#åå¥çºsegment_Undefined/ assigned_room_L/ distribution_channel_Undefined57segment_Undefined = np.zeros(test_d.shape[0])58assigned_room_L = np.zeros(test_d.shape[0])59distribution_channel_Undefined = np.zeros(test_d.shape[0])60test_d['segment_Undefined'] = segment_Undefined61test_d['assigned_room_L'] = assigned_room_L62test_d['distribution_channel_Undefined'] = distribution_channel_Undefined...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
