Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _create_object method in tempest
Best Python code snippet using tempest_python
test_object_manager.py
Source:test_object_manager.py 
...104 )105 )106 # Run through (de)serializer to fill in any missing vars107 return self.deserializer.deserialize(self.serializer.serialize(msg))108 def _create_object(self, local_id=None, full_id=None, parent_id=None, pos=None, rot=None,109 pcode=None, namevalue=None, region_handle=None) -> Object:110 msg = self._create_object_update(111 local_id=local_id, full_id=full_id, parent_id=parent_id, pos=pos, rot=rot,112 pcode=pcode, namevalue=namevalue, region_handle=region_handle)113 self.message_handler.handle(msg)114 actual_handle = msg["RegionData"]["RegionHandle"]115 region = self.session.region_by_handle(actual_handle)116 return region.objects.lookup_fullid(msg["ObjectData"]["FullID"])117 def _create_kill_object(self, local_id) -> Message:118 return Message(119 "KillObject",120 Block(121 "ObjectData",122 ID=local_id,123 )124 )125 def _kill_object(self, local_id: int):126 self.message_handler.handle(self._create_kill_object(local_id))127 def _create_object_update_cached(self, local_id: int, region_handle: int = 123,128 crc: int = 22, flags: int = 4321):129 return Message(130 'ObjectUpdateCached',131 Block("RegionData", TimeDilation=102, RegionHandle=region_handle),132 Block(133 "ObjectData",134 ID=local_id,135 CRC=crc,136 UpdateFlags=flags,137 )138 )139 def _get_avatar_positions(self) -> Dict[UUID, Vector3]:140 return {av.FullID: av.RegionPosition for av in self.region_object_manager.all_avatars}141class RegionObjectManagerTests(ObjectManagerTestMixin, unittest.IsolatedAsyncioTestCase):142 def test_basic_tracking(self):143 """Does creating an object result in it being tracked?"""144 msg = self._create_object_update()145 self.message_handler.handle(msg)146 obj = self.region_object_manager.lookup_fullid(msg["ObjectData"]["FullID"])147 self.assertIsNotNone(obj)148 def test_terse_object_update(self):149 msg = self._create_object_update(pos=Vector3(1, 2, 3))150 self.message_handler.handle(msg)151 local_id = msg["ObjectData"]["ID"]152 msg = Message(153 'ImprovedTerseObjectUpdate',154 Block('RegionData', RegionHandle=123, TimeDilation=65345),155 Block(156 'ObjectData',157 Data_={158 'ID': local_id,159 'State': 0,160 'FootCollisionPlane': None,161 'Position': Vector3(-2, -3, -4),162 'Velocity': Vector3(-0.0, -0.0, -0.0),163 'Acceleration': Vector3(-0.0, -0.0, -0.0),164 'Rotation': Quaternion(0, 0, 0, 1),165 'AngularVelocity': Vector3(-0.0, -0.0, -0.0)166 },167 TextureEntry_=None,168 ),169 )170 self.message_handler.handle(msg)171 obj = self.region_object_manager.lookup_localid(local_id)172 self.assertEqual(obj.Position, Vector3(-2, -3, -4))173 def test_parent_tracking(self):174 """Are basic parenting scenarios handled?"""175 parent = self._create_object()176 child = self._create_object(parent_id=parent.LocalID)177 self.assertSequenceEqual([child.LocalID], parent.ChildIDs)178 def test_orphan_parent_tracking(self):179 child = self._create_object(local_id=2, parent_id=1)180 self.assertEqual({1}, self.region_object_manager.missing_locals)181 parent = self._create_object(local_id=1)182 self.assertEqual(set(), self.region_object_manager.missing_locals)183 self.assertSequenceEqual([child.LocalID], parent.ChildIDs)184 def test_killing_parent_kills_children(self):185 _child = self._create_object(local_id=2, parent_id=1)186 parent = self._create_object(local_id=1)187 # This should orphan the child again188 self._kill_object(parent.LocalID)189 parent = self._create_object(local_id=1)190 # We should not have picked up any children191 self.assertSequenceEqual([], parent.ChildIDs)192 def test_hierarchy_killed(self):193 _child = self._create_object(local_id=3, parent_id=2)194 _other_child = self._create_object(local_id=4, parent_id=2)195 _parent = self._create_object(local_id=2, parent_id=1)196 grandparent = self._create_object(local_id=1)197 # KillObject implicitly kills all known descendents at that point198 self._kill_object(grandparent.LocalID)199 self.assertEqual(0, len(self.region_object_manager))200 def test_hierarchy_avatar_not_killed(self):201 _child = self._create_object(local_id=3, parent_id=2)202 _parent = self._create_object(local_id=2, parent_id=1, pcode=PCode.AVATAR)203 grandparent = self._create_object(local_id=1)204 # KillObject should only "unsit" child avatars (does this require an ObjectUpdate205 # or is ParentID=0 implied?)206 self._kill_object(grandparent.LocalID)207 self.assertEqual(2, len(self.region_object_manager))208 self.assertIsNotNone(self.region_object_manager.lookup_localid(2))209 def test_attachment_orphan_parent_tracking(self):210 """211 Test that multi-level parenting trees handle orphaning correctly.212 Technically there can be at least 4 levels of parenting if sitting.213 object -> seated agent -> attachment root -> attachment child214 """215 child = self._create_object(local_id=3, parent_id=2)216 parent = self._create_object(local_id=2, parent_id=1)217 self.assertSequenceEqual([child.LocalID], parent.ChildIDs)218 def test_unparenting_succeeds(self):219 child = self._create_object(local_id=3, parent_id=2)220 parent = self._create_object(local_id=2)221 msg = self._create_object_update(local_id=child.LocalID, full_id=child.FullID, parent_id=0)222 self.message_handler.handle(msg)223 self.assertEqual(0, child.ParentID)224 self.assertSequenceEqual([], parent.ChildIDs)225 def test_reparenting_succeeds(self):226 child = self._create_object(local_id=3, parent_id=2)227 parent = self._create_object(local_id=2)228 second_parent = self._create_object(local_id=1)229 msg = self._create_object_update(local_id=child.LocalID,230 full_id=child.FullID, parent_id=second_parent.LocalID)231 self.message_handler.handle(msg)232 self.assertEqual(second_parent.LocalID, child.ParentID)233 self.assertSequenceEqual([], parent.ChildIDs)234 self.assertSequenceEqual([child.LocalID], second_parent.ChildIDs)235 def test_reparenting_without_known_parent_succeeds(self):236 child = self._create_object(local_id=3, parent_id=2)237 second_parent = self._create_object(local_id=1)238 msg = self._create_object_update(local_id=child.LocalID,239 full_id=child.FullID, parent_id=second_parent.LocalID)240 self.message_handler.handle(msg)241 # Create the original parent after its former child has been reparented242 parent = self._create_object(local_id=2)243 self.assertEqual(second_parent.LocalID, child.ParentID)244 self.assertSequenceEqual([], parent.ChildIDs)245 self.assertSequenceEqual([child.LocalID], second_parent.ChildIDs)246 def test_reparenting_with_neither_parent_known_succeeds(self):247 child = self._create_object(local_id=3, parent_id=2)248 msg = self._create_object_update(local_id=child.LocalID,249 full_id=child.FullID, parent_id=1)250 self.message_handler.handle(msg)251 second_parent = self._create_object(local_id=1)252 self.assertEqual(second_parent.LocalID, child.ParentID)253 self.assertSequenceEqual([child.LocalID], second_parent.ChildIDs)254 def test_property_changes_reported_correctly(self):255 obj = self._create_object(local_id=1)256 msg = self._create_object_update(local_id=obj.LocalID, full_id=obj.FullID, pos=(2.0, 2.0, 2.0))257 self.message_handler.handle(msg)258 events = self.object_addon.events259 self.assertEqual(2, len(events))260 self.assertEqual({"Position", "TextureEntry"}, events[1][2])261 def test_region_position(self):262 parent = self._create_object(pos=(0.0, 1.0, 0.0))263 child = self._create_object(parent_id=parent.LocalID, pos=(0.0, 1, 0.0))264 self.assertEqual(parent.RegionPosition, (0.0, 1.0, 0.0))265 self.assertEqual(child.RegionPosition, (0.0, 2.0, 0.0))266 def test_orphan_region_position(self):267 child = self._create_object(local_id=2, parent_id=1, pos=(0.0, 1, 0.0))268 with self.assertRaises(ValueError):269 getattr(child, "RegionPosition")270 def test_rotated_region_position(self):271 parent = self._create_object(pos=(0.0, 1.0, 0.0), rot=Quaternion.from_euler(0, 0, 180, True))272 child = self._create_object(parent_id=parent.LocalID, pos=(0.0, 1.0, 0.0))273 self.assertEqual(parent.RegionPosition, (0.0, 1.0, 0.0))274 self.assertEqual(child.RegionPosition, (0.0, 0.0, 0.0))275 def test_rotated_region_position_multi_level(self):276 rot = Quaternion.from_euler(0, 0, 180, True)277 grandparent = self._create_object(pos=(0.0, 1.0, 0.0), rot=rot)278 parent = self._create_object(parent_id=grandparent.LocalID, pos=(0.0, 1.0, 0.0), rot=rot)279 child = self._create_object(parent_id=parent.LocalID, pos=(1.0, 2.0, 0.0))280 self.assertEqual(grandparent.RegionPosition, (0.0, 1.0, 0.0))281 self.assertEqual(parent.RegionPosition, (0.0, 0.0, 0.0))282 self.assertEqual(child.RegionPosition, (1.0, 2.0, 0.0))283 def test_global_position(self):284 obj = self._create_object(pos=(0.0, 0.0, 0.0))285 self.assertEqual(obj.GlobalPosition, (0.0, 123.0, 0.0))286 def test_avatar_locations(self):287 agent1_id = UUID.random()288 agent2_id = UUID.random()289 self.message_handler.handle(Message(290 "CoarseLocationUpdate",291 Block("AgentData", AgentID=agent1_id),292 Block("AgentData", AgentID=agent2_id),293 Block("Location", X=1, Y=2, Z=3),294 Block("Location", X=2, Y=3, Z=4),295 ))296 self.assertDictEqual(self._get_avatar_positions(), {297 # CoarseLocation's Z axis is multiplied by 4298 agent1_id: Vector3(1, 2, 12),299 agent2_id: Vector3(2, 3, 16),300 })301 # Simulate an avatar sitting on an object302 seat_object = self._create_object(pos=(0, 0, 3))303 # If we have a real object pos it should override coarse pos304 avatar_obj = self._create_object(full_id=agent1_id, pcode=PCode.AVATAR,305 parent_id=seat_object.LocalID, pos=Vector3(0, 0, 2))306 self.assertDictEqual(self._get_avatar_positions(), {307 # Agent is seated, make sure this is region and not local pos308 agent1_id: Vector3(0, 0, 5),309 agent2_id: Vector3(2, 3, 16),310 })311 # Simulate missing parent for agent312 self._kill_object(seat_object.LocalID)313 self.assertDictEqual(self._get_avatar_positions(), {314 # Agent is seated, but we don't know its parent. We have315 # to use the coarse location.316 agent1_id: Vector3(1, 2, 12),317 agent2_id: Vector3(2, 3, 16),318 })319 # If the object is killed and no coarse pos, it shouldn't be in the dict320 # CoarseLocationUpdates are expected to be complete, so any agents missing321 # are no longer in the sim.322 self._kill_object(avatar_obj.LocalID)323 self.message_handler.handle(Message(324 "CoarseLocationUpdate",325 Block("AgentData", AgentID=agent2_id),326 Block("Location", X=2, Y=3, Z=4),327 ))328 self.assertDictEqual(self._get_avatar_positions(), {329 agent2_id: Vector3(2, 3, 16),330 })331 # 255 on Z axis means we can't guess the real Z332 self.message_handler.handle(Message(333 "CoarseLocationUpdate",334 Block("AgentData", AgentID=agent2_id),335 Block("Location", X=2, Y=3, Z=math.inf),336 ))337 self.assertDictEqual(self._get_avatar_positions(), {338 agent2_id: Vector3(2, 3, math.inf),339 })340 agent2_avatar = self.region_object_manager.lookup_avatar(agent2_id)341 self.assertEqual(agent2_avatar.GlobalPosition, Vector3(2, 126, math.inf))342 def test_name_cache(self):343 # Receiving an update with a NameValue for an avatar should update NameCache344 obj = self._create_object(345 pcode=PCode.AVATAR,346 namevalue='DisplayName STRING RW DS ð²ð«ð¦ð ð¬ð¡ð¢ð«ððªð¢\n'347 'FirstName STRING RW DS firstname\n'348 'LastName STRING RW DS Resident\n'349 'Title STRING RW DS foo'.encode("utf8"),350 )351 self.assertEqual(self.session_manager.name_cache.lookup(obj.FullID).first_name, "firstname")352 av = self.region_object_manager.lookup_avatar(obj.FullID)353 self.assertEqual(av.Name, "ð²ð«ð¦ð ð¬ð¡ð¢ð«ððªð¢ (firstname Resident)")354 self.assertEqual(av.PreferredName, "ð²ð«ð¦ð ð¬ð¡ð¢ð«ððªð¢")355 def test_normalize_cache_data(self):356 normalized = normalize_object_update_compressed_data(OBJECT_UPDATE_COMPRESSED_DATA)357 expected = {358 'PSBlock': None,359 'ParentID': 0,360 'LocalID': 1234,361 'FullID': UUID('121210bf-1658-427e-8fb4-fb001acd9be5'),362 'PCode': PCode.PRIMITIVE,363 'State': 0,364 'CRC': 18381,365 'Material': 3,366 'ClickAction': 0,367 'Scale': Vector3(39.0, 39.0, 0.10000000149011612),368 'Position': Vector3(43.07024002075195, 79.34690856933594, 20.049999237060547),369 'Rotation': Quaternion(0.0, 0.0, -0.48543819785118103, 0.8742709854884798),370 'OwnerID': UUID('6b9bc4fe-330a-4f61-bbe2-e4b243ac37bd'),371 'AngularVelocity': Vector3(0.0, 0.0, 0.0791015625),372 'TreeSpecies': None,373 'ScratchPad': None,374 'Text': b'',375 'TextColor': b'',376 'MediaURL': b'',377 'Sound': UUID(),378 'SoundGain': 0.0,379 'SoundFlags': 0,380 'SoundRadius': 0.0,381 'NameValue': [],382 'PathCurve': 32,383 'ProfileCurve': 0,384 'PathBegin': 0,385 'PathEnd': 25600,386 'PathScaleX': 150,387 'PathScaleY': 0,388 'PathShearX': 0,389 'PathShearY': 0,390 'PathTwist': 0,391 'PathTwistBegin': 0,392 'PathRadiusOffset': 0,393 'PathTaperX': 0,394 'PathTaperY': 0,395 'PathRevolutions': 0,396 'PathSkew': 0,397 'ProfileBegin': 0,398 'ProfileEnd': 0,399 'ProfileHollow': 0,400 }401 filtered_normalized = {k: v for k, v in normalized.items() if k in expected}402 self.assertDictEqual(filtered_normalized, expected)403 sculpt_texture = normalized["ExtraParams"][ExtraParamType.SCULPT]["Texture"]404 self.assertEqual(sculpt_texture, UUID('89556747-24cb-43ed-920b-47caed15465f'))405 self.assertIsNotNone(normalized['TextureAnim'])406 self.assertIsNotNone(normalized['TextureEntry'])407 def test_object_cache(self):408 self.mock_get_region_object_cache_chain.return_value = RegionViewerObjectCacheChain([409 RegionViewerObjectCache(self.region.cache_id, [410 ViewerObjectCacheEntry(411 local_id=1234,412 crc=22,413 data=OBJECT_UPDATE_COMPRESSED_DATA,414 )415 ])416 ])417 cache_msg = self._create_object_update_cached(1234, flags=4321)418 obj = self.region_object_manager.lookup_localid(1234)419 self.assertIsNone(obj)420 self.region_object_manager.load_cache()421 self.message_handler.handle(cache_msg)422 obj = self.region_object_manager.lookup_localid(1234)423 self.assertEqual(obj.FullID, UUID('121210bf-1658-427e-8fb4-fb001acd9be5'))424 # Flags from the ObjectUpdateCached should have been merged in425 self.assertEqual(obj.UpdateFlags, 4321)426 async def test_request_objects(self):427 # request five objects, three of which won't receive an ObjectUpdate428 futures = self.region_object_manager.request_objects((1234, 1235, 1236, 1237))429 self._create_object(1234)430 self._create_object(1235)431 done, pending = await asyncio.wait(futures, timeout=0.0001)432 objects = await asyncio.gather(*done)433 # wait() returns unordered results, so use a set.434 self.assertEqual(set(o.LocalID for o in objects), {1234, 1235})435 pending = list(pending)436 self.assertEqual(2, len(pending))437 pending_1, pending_2 = pending438 # Timing out should cancel439 with self.assertRaises(asyncio.TimeoutError):440 await asyncio.wait_for(pending_1, 0.00001)441 self.assertTrue(pending_1.cancelled())442 fut = self.region_object_manager.request_objects(1238)[0]443 self._kill_object(1238)444 self.assertTrue(fut.cancelled())445 # Object manager being cleared due to region death should cancel446 self.assertFalse(pending_2.cancelled())447 self.region_object_manager.clear()448 self.assertTrue(pending_2.cancelled())449 # The clear should have triggered the objects to be removed from the world view as well450 # as the region view451 self.assertEqual(0, len(self.session.objects))452 self.assertEqual(0, len(self.region_object_manager))453class SessionObjectManagerTests(ObjectManagerTestMixin, unittest.IsolatedAsyncioTestCase):454 async def asyncSetUp(self) -> None:455 await super().asyncSetUp()456 self.second_region = self.session.register_region(457 ("127.0.0.1", 9), "https://localhost:5", 124458 )459 self.session.objects.track_region_objects(124)460 self._setup_region_circuit(self.second_region)461 def test_get_fullid(self):462 obj = self._create_object()463 self.assertIs(self.session.objects.lookup_fullid(obj.FullID), obj)464 self._kill_object(obj.LocalID)465 self.assertIsNone(self.session.objects.lookup_fullid(obj.FullID))466 def test_region_handle_change(self):467 obj = self._create_object(region_handle=123)468 self.assertEqual(obj.RegionHandle, 123)469 self.assertIs(self.region.objects.lookup_fullid(obj.FullID), obj)470 self.assertIs(self.region.objects.lookup_localid(obj.LocalID), obj)471 # Send an update moving the object to the new region472 self._create_object(local_id=~obj.LocalID & 0xFFffFFff, full_id=obj.FullID, region_handle=124)473 self.assertEqual(obj.RegionHandle, 124)474 self.assertIsNone(self.region.objects.lookup_fullid(obj.FullID))475 self.assertIsNone(self.region.objects.lookup_localid(obj.LocalID))476 self.assertIs(self.second_region.objects.lookup_fullid(obj.FullID), obj)477 self.assertIs(self.second_region.objects.lookup_localid(obj.LocalID), obj)478 self.assertEqual(1, len(self.session.objects))479 self.assertEqual(0, len(self.region.objects))480 self.assertEqual(1, len(self.second_region.objects))481 def test_object_moved_to_bad_region(self):482 obj = self._create_object(region_handle=123)483 msg = self._create_object_update(484 local_id=~obj.LocalID & 0xFFffFFff, full_id=obj.FullID, region_handle=999)485 self.message_handler.handle(msg)486 # Should not be in this region anymore487 self.assertEqual(0, len(self.region_object_manager))488 # Should still be tracked by the session489 self.assertEqual(1, len(self.session.objects))490 self.assertIsNotNone(self.session.objects.lookup_fullid(obj.FullID))491 def test_linkset_region_handle_change(self):492 parent = self._create_object(region_handle=123)493 child = self._create_object(region_handle=123, parent_id=parent.LocalID)494 self._create_object(local_id=~parent.LocalID & 0xFFffFFff, full_id=parent.FullID, region_handle=124)495 # Children reference their parents, not the other way around. Moving this to a new region496 # should have cleared the list because it now has no children in the same region.497 self.assertEqual([], parent.ChildIDs)498 # Move the child to the same region499 self._create_object(500 local_id=child.LocalID, full_id=child.FullID, region_handle=124, parent_id=parent.LocalID)501 # Child should be back in the children list502 self.assertEqual([child.LocalID], parent.ChildIDs)503 self.assertEqual(parent.LocalID, child.ParentID)504 self.assertEqual(0, len(self.region.objects))505 self.assertEqual(2, len(self.second_region.objects))506 self.assertEqual(0, len(self.region.objects.missing_locals))507 self.assertEqual(0, len(self.second_region.objects.missing_locals))508 def test_all_objects(self):509 obj = self._create_object()510 self.assertEqual([obj], list(self.session.objects.all_objects))511 def test_all_avatars(self):512 obj = self._create_object(pcode=PCode.AVATAR)513 av_list = list(self.session.objects.all_avatars)514 self.assertEqual(1, len(av_list))515 self.assertEqual(obj, av_list[0].Object)516 def test_avatars_preference(self):517 # If we have a coarselocation for an avatar in one region and518 # an actual object in another, we should always prefer the519 # one with the actual object.520 av_1 = self._create_object(pcode=PCode.AVATAR, region_handle=123)521 av_2 = self._create_object(pcode=PCode.AVATAR, region_handle=124)522 # Coarse location shouldn't be used for either of these523 self.session.region_by_handle(123).message_handler.handle(Message(524 "CoarseLocationUpdate",525 Block("AgentData", AgentID=av_2.FullID),526 Block("Location", X=2, Y=3, Z=4),527 ))528 self.session.region_by_handle(124).message_handler.handle(Message(529 "CoarseLocationUpdate",530 Block("AgentData", AgentID=av_1.FullID),531 Block("Location", X=2, Y=3, Z=4),532 ))533 av_list = list(self.session.objects.all_avatars)534 self.assertEqual(2, len(av_list))535 self.assertTrue(all(a.Object for a in av_list))536 def test_lookup_avatar(self):537 av_1 = self._create_object(pcode=PCode.AVATAR)538 av_obj = self.session.objects.lookup_avatar(av_1.FullID)539 self.assertEqual(av_1.FullID, av_obj.FullID)540 async def test_requesting_properties(self):541 obj = self._create_object()542 futs = self.session.objects.request_object_properties(obj)543 self.message_handler.handle(Message(544 "ObjectProperties",545 Block("ObjectData", ObjectID=obj.FullID, Name="Foobar", TextureID=b""),546 ))547 await asyncio.wait_for(futs[0], timeout=0.0001)548 self.assertEqual(obj.Name, "Foobar")549 async def test_load_ancestors(self):550 child = self._create_object(region_handle=123, parent_id=1)551 parentless = self._create_object(region_handle=123)552 orphaned = self._create_object(region_handle=123, parent_id=9)553 async def _create_after():554 await asyncio.sleep(0.001)555 self._create_object(region_handle=123, local_id=child.ParentID)556 asyncio.create_task(_create_after())557 await self.session.objects.load_ancestors(child)558 await self.session.objects.load_ancestors(parentless)559 with self.assertRaises(asyncio.TimeoutError):560 await self.session.objects.load_ancestors(orphaned, wait_time=0.005)561 async def test_auto_request_objects(self):562 self.session_manager.settings.AUTOMATICALLY_REQUEST_MISSING_OBJECTS = True563 self.message_handler.handle(self._create_object_update_cached(1234))564 self.message_handler.handle(self._create_object_update_cached(1235))565 self.assertEqual({1234, 1235}, self.region_object_manager.queued_cache_misses)566 # Pretend viewer sent out its own RequestMultipleObjects567 self.message_handler.handle(Message(568 'RequestMultipleObjects',569 Block("RegionData", SessionID=self.session.id, AgentID=self.session.agent_id),570 Block(571 "ObjectData",572 ID=1234,573 )574 ))575 # Proxy should have killed its pending request for 1234576 self.assertEqual({1235}, self.region_object_manager.queued_cache_misses)577 async def test_auto_request_avatar_seats(self):578 # Avatars' parent links should always be requested regardless of579 # object auto-request setting's value.580 seat_id = 999581 av = self._create_object(pcode=PCode.AVATAR, parent_id=seat_id)582 self.assertEqual({seat_id}, self.region_object_manager.queued_cache_misses)583 # Need to wait for it to decide it's worth requesting584 await asyncio.sleep(0.22)585 self.assertEqual(set(), self.region_object_manager.queued_cache_misses)586 # Make sure we sent a request after the timeout587 req_msg = self.deserializer.deserialize(self.transport.packets[-1][0])588 self.assertEqual("RequestMultipleObjects", req_msg.name)589 self.assertEqual(590 [{'CacheMissType': 0, 'ID': seat_id}],591 req_msg.to_dict()['body']['ObjectData'],592 )593 # Parent should not be requested again if an unrelated property like pos changes594 self._create_object(local_id=av.LocalID, full_id=av.FullID,595 pcode=PCode.AVATAR, parent_id=seat_id, pos=(1, 2, 9))596 self.assertEqual(set(), self.region_object_manager.queued_cache_misses)597 async def test_handle_object_update_event(self):598 with self.session.objects.events.subscribe_async(599 message_names=(ObjectUpdateType.OBJECT_UPDATE,),600 predicate=lambda e: e.object.UpdateFlags & JUST_CREATED_FLAGS and "LocalID" in e.updated,601 ) as get_events:602 self._create_object(local_id=999)603 evt = await asyncio.wait_for(get_events(), 1.0)604 self.assertEqual(999, evt.object.LocalID)605 async def test_handle_object_update_predicate(self):606 with self.session.objects.events.subscribe_async(607 message_names=(ObjectUpdateType.OBJECT_UPDATE,),608 ) as get_events:609 self._create_object(local_id=999)610 evt = await asyncio.wait_for(get_events(), 1.0)611 self.assertEqual(999, evt.object.LocalID)612 async def test_handle_object_update_events_two_subscribers(self):613 with self.session.objects.events.subscribe_async(614 message_names=(ObjectUpdateType.OBJECT_UPDATE,),615 ) as get_events:616 with self.session.objects.events.subscribe_async(617 message_names=(ObjectUpdateType.OBJECT_UPDATE,),618 ) as get_events2:619 self._create_object(local_id=999)620 evt = await asyncio.wait_for(get_events(), 1.0)621 evt2 = await asyncio.wait_for(get_events2(), 1.0)622 self.assertEqual(999, evt.object.LocalID)623 self.assertEqual(evt, evt2)624 async def test_handle_object_update_events_two_subscribers_timeout(self):625 with self.session.objects.events.subscribe_async(626 message_names=(ObjectUpdateType.OBJECT_UPDATE,),627 ) as get_events:628 with self.session.objects.events.subscribe_async(629 message_names=(ObjectUpdateType.OBJECT_UPDATE,),630 ) as get_events2:631 self._create_object(local_id=999)632 evt = asyncio.wait_for(get_events(), 0.01)633 evt2 = asyncio.wait_for(get_events2(), 0.01)...
base.py
Source:base.py
...34 return self.client.login(username=user.username,35 password=password)36 def create_project(self, title, description='', **kwargs):37 kwargs['theme'] = kwargs.get('theme', self.default_theme)38 return self._create_object(39 Project, ('title', title), ('description', description),40 **kwargs41 )42 def create_user(self, username, **kwargs):43 kwargs.update({'username': username})44 kwargs['password'] = kwargs.get('password', username)45 user = User.objects.create_user(**kwargs)46 user.save()47 return user48 def _create_object(self, cls, *args, **kwargs):49 for name, val in args:50 kwargs[name] = val51 obj = cls.objects.create(**kwargs)52 obj.save()53 return obj54 def create_access(self, user, project, can_edit=False):55 return self._create_object(56 ProjectAccess,57 ('user', user), ('project', project), ('can_edit', can_edit),58 )59 def create_page(self, title, **kwargs):60 return self._create_object(Page, ('title', title), **kwargs)61 def create_post(self, title, content, **kwargs):62 timestamp = timezone.now()63 kwargs['date_created'] = kwargs.get('date_created', timestamp)64 kwargs['date_updated'] = kwargs.get('date_updated', timestamp)65 return self._create_object(66 Post,67 ('title', title), ('content', content),68 **kwargs69 )70 def create_category(self, title, **kwargs):71 return self._create_object(Category, ('title', title), **kwargs)72 def create_tag(self, title, **kwargs):73 return self._create_object(Tag, ('title', title), **kwargs)74 def create_theme(self, title, markup, **kwargs):75 kwargs['creator'] = kwargs.get('creator', self.admin_user)76 kwargs['filepath'] = kwargs.get('filepath', '')77 return self._create_object(78 Theme,79 ('title', title), ('body_markup', markup),80 **kwargs81 )82 def create_project_plugin(self, title, **kwargs):83 kwargs['markup'] = kwargs.get('markup', 'markup')84 return self._create_object(ProjectPlugin, ('title', title), **kwargs)85 def create_page_plugin(self, title, **kwargs):86 kwargs['head_markup'] = kwargs.get('head_markup', 'head_markup')87 kwargs['body_markup'] = kwargs.get('body_markup', 'body_markup')88 return self._create_object(PagePlugin, ('title', title), **kwargs)89class FuglViewTestCase(FuglTestCase):90 def setUp(self):91 super().setUpTheme()92 self.client = Client()93 def tearDown(self):...
factories.py
Source:factories.py
1from test_app.models import *2def _create_object(model, **kw):3 obj = model(**kw)4 obj.save()5 return obj6def create_entity(**kw):7 if 'fk_entity' not in kw:8 kw['fk_entity'] = _create_object(FKEntity, value='FK Value')9 if 'o2o_entity' not in kw:10 kw['o2o_entity'] = _create_object(OneFromEntity, value='Value')11 return _create_object(Entity, **kw)12def create_related_entity(**kw):13 return _create_object(RelatedEntity, **kw)14def create_deep_related_entity(**kw):15 return _create_object(DeepRelatedEntity, **kw)16def create_sort_entity(**kw):17 return _create_object(SortEntity, **kw)18def create_sort_related_entity_a(**kw):19 return _create_object(SortRelatedEntityA, **kw)20def create_sort_related_entity_b(**kw):21 return _create_object(SortRelatedEntityB, **kw)22def create_sort_related_entity_c(**kw):23 return _create_object(SortRelatedEntityC, **kw)24def create_O2O_to_entity(**kw):...
Automation Testing Tutorials
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tempest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
