Best Python code snippet using autotest_python
rpc_interface.py
Source:rpc_interface.py
...221 """222 rpc_utils.check_modify_host(kwargs)223 host = models.Host.smart_get(id)224 try:225 rpc_utils.check_modify_host_locking(host, kwargs)226 except model_logic.ValidationError as e:227 if not kwargs.get('force_modify_locking', False):228 raise229 logging.exception('The following exception will be ignored and lock '230 'modification will be enforced. %s', e)231 # This is required to make `lock_time` for a host be exactly same232 # between the master and a shard.233 if kwargs.get('locked', None) and 'lock_time' not in kwargs:234 kwargs['lock_time'] = datetime.datetime.now()235 host.update_object(kwargs)236 # force_modifying_locking is not an internal field in database, remove.237 kwargs.pop('force_modify_locking', None)238 rpc_utils.fanout_rpc([host], 'modify_host_local',239 include_hostnames=False, id=id, **kwargs)240def modify_host_local(id, **kwargs):241 """Modify host attributes in local DB.242 @param id: Host id.243 @param kwargs: key=value pairs of values to set on the host.244 """245 models.Host.smart_get(id).update_object(kwargs)246@rpc_utils.route_rpc_to_master247def modify_hosts(host_filter_data, update_data):248 """Modify local attributes of multiple hosts.249 If this is called on the master, but one of the hosts in that match the250 filters is assigned to a shard, this will call `modify_hosts_local` RPC251 to the responsible shard.252 When this is called on a shard, the shard just routes the RPC to the master253 and does nothing.254 The filters are always applied on the master, not on the shards. This means255 if the states of a host differ on the master and a shard, the state on the256 master will be used. I.e. this means:257 A host was synced to Shard 1. On Shard 1 the status of the host was set to258 'Repair Failed'.259 - A call to modify_hosts with host_filter_data={'status': 'Ready'} will260 update the host (both on the shard and on the master), because the state261 of the host as the master knows it is still 'Ready'.262 - A call to modify_hosts with host_filter_data={'status': 'Repair failed'263 will not update the host, because the filter doesn't apply on the master.264 @param host_filter_data: Filters out which hosts to modify.265 @param update_data: A dictionary with the changes to make to the hosts.266 """267 update_data = update_data.copy()268 rpc_utils.check_modify_host(update_data)269 hosts = models.Host.query_objects(host_filter_data)270 affected_shard_hostnames = set()271 affected_host_ids = []272 # Check all hosts before changing data for exception safety.273 for host in hosts:274 try:275 rpc_utils.check_modify_host_locking(host, update_data)276 except model_logic.ValidationError as e:277 if not update_data.get('force_modify_locking', False):278 raise279 logging.exception('The following exception will be ignored and '280 'lock modification will be enforced. %s', e)281 if host.shard:282 affected_shard_hostnames.add(host.shard.rpc_hostname())283 affected_host_ids.append(host.id)284 # This is required to make `lock_time` for a host be exactly same285 # between the master and a shard.286 if update_data.get('locked', None) and 'lock_time' not in update_data:287 update_data['lock_time'] = datetime.datetime.now()288 for host in hosts:289 host.update_object(update_data)...
rpc_utils.py
Source:rpc_utils.py
...291 # beneath tasks being run on a host.292 if 'status' in update_data:293 raise model_logic.ValidationError({294 'status': 'Host status can not be modified by the frontend.'})295def check_modify_host_locking(host, update_data):296 """297 Checks when locking/unlocking has been requested if the host is already298 locked/unlocked.299 :param host: models.Host object to be modified300 :param update_data: A dictionary with the changes to make to the host.301 """302 locked = update_data.get('locked', None)303 if locked is not None:304 if locked and host.locked:305 raise model_logic.ValidationError({306 'locked': 'Host already locked by %s on %s.' %307 (host.locked_by, host.lock_time)})308 if not locked and not host.locked:309 raise model_logic.ValidationError({...
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!!