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!!