KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use update_execution_data method in SeleniumBase
Best Python code snippet using SeleniumBase
test_server.py
Source:test_server.py 
...482 {'worker_id': '', 'token': token},483 {'exit_status': 'success', 'token': token},484 {'time_terminated': 'Tue, 02 Apr 2013 10:29:13 GMT', 'token': token}485 ]486 def update_execution_data(update):487 return patch({'update_execution_data': update}, self.entry,488 self.cred.worker_key, 'tasks', task_ids[2])489 for update in good_updates:490 resp = update_execution_data(update)491 self.assertEqual(resp.status_code, requests.codes.ok)492 for update in bad_updates:493 resp = update_execution_data(update)494 self.assertEqual(resp.status_code, requests.codes.unprocessable_entity)495class TestCancellation(unittest.TestCase):496 """497 Verifies that the behavior of task cancellation is as expected.498 """499 def __init__(self, entry, db, cred, *args, **kwargs):500 super().__init__(*args, **kwargs)501 self.entry = entry502 self.db = db503 self.cred = cred504 def _add_continuations(self, child_ids, parent_id):505 return post(child_ids, self.entry, self.cred.provider_key, 'tasks', parent_id,506 'add_continuations')507 def test_cancellation_depth_1(self):...
validation.py
Source:validation.py
1# -*- coding: utf-8 -*-2"""3banyan.server.validation4------------------------5Defines custom validators for resource schema.6"""7from flask import abort, g, current_app as app8from eve.methods.common import serialize9import eve.io.mongo10from banyan.server.state import legal_provider_transitions, legal_worker_transitions11from banyan.server.constants import *12from banyan.server.mongo_common import find_by_id13"""14One thing I dislike about Eve is that the configuration accepts a single global validator. This has15the following disadvantages:161. The single custom validator may become complex and hard to maintain.172. We cannot easily "overload" custom validation rules by giving them different meanings when used18 for different resources.19"""20class ValidatorBase(eve.io.mongo.Validator):21 """22 Subclass of ``eve.io.mongo.Validator`` with support for the custom validation rules used by23 Banyan. This validator should not be used directly, because it does not support virtual24 resources. Use ``Validator`` instead.25 Support for virtual resources can't be added here, because ``BulkUpdateValidator`` derives26 from this class.27 """28 def __init__(self, schema, resource=None, allow_unknown=False,29 transparent_schema_rules=False):30 """31 The last two arguments are retained despite the fact that they are unused, in order32 to maintain compatibility with Eve's validator interface.33 """34 self.db = app.data.driver.db35 """36 ``transparent_schema_rules`` is set to ``True``, so that the additional rules for37 virtual resources will not cause any issues.38 """39 super().__init__(40 schema=schema,41 resource=resource,42 allow_unknown=False,43 transparent_schema_rules=True44 )45 """46 This is a hack that is necessary in order to get the base class to pass the47 ``resource`` parameter to the constructors of the validators for nested schema and48 virtual resources.49 """50 self._Validator__config['resource'] = self.resource51 def validate(self, document, schema=None, update=False, context=None):52 """53 Called after each POST request, e.g. when a new task is created. Also called each54 time a subfield needs to be validated. In this case, ``context`` will be set to the55 parent field's value, and ``update`` may be `True`.56 """57 """58 As described above, ``context`` is non-null iff we are validating a subfield. Since59 we don't have any special constraints to enforce for subfields, we just perform the60 default validation.61 """62 if context:63 return super().validate(document, schema, update, context)64 # See https://github.com/nicolaiarocci/eve/issues/796 for more information about65 # this.66 assert not update, "Expected 'validate_update' to be called for updates " \67 "instead of 'validate' with 'update=True'."68 if 'state' in document and document['state'] not in ['inactive', 'available']:69 self._error('state', "Tasks can only be created in the 'inactive' and "70 "'available' states.")71 return False72 # We call the parent's `validate` function now, so that we can ensure that all73 # ObjectIDs for continuations are valid.74 if not super().validate(document, schema, update, context):75 return False76 for func in [self.validate_continuations, self.validate_worker_id]:77 if not func(document):78 return False79 return True80 def validate_continuation(self, child):81 state = child['state']82 if state != 'inactive':83 self._error('continuations', "Continuation '{}' should be 'inactive', but "84 "is in the '{}' state.".format(child['_id'], state))85 return False86 return True87 def validate_continuations(self, document):88 if 'continuations' not in document:89 return True90 children = [self.db['tasks'].find_one({'_id': _id}) for _id in91 document['continuations']]92 # It's not practical to check for cyclic dependencies, but we do perform a basic93 # sanity check.94 if '_id' in document:95 if document['_id'] in children:96 self._error('continuations', "Task cannot have itself as a "97 "continuation.")98 return False99 for child in children:100 if not self.validate_continuation(child):101 return False102 return True103 def validate_worker_id(self, document):104 if 'worker_id' not in document:105 return True106 worker = find_by_id('users', document['worker_id'], self.db, {'role': True})107 if worker['role'] != 'worker':108 self._error('worker_id', "User with given id is a {}, not a {}.".format(109 worker['role'], 'worker'))110 return False111 return True112 def validate_execution_data_token(self, document, original_document):113 ex_data = document['update_execution_data']114 if 'token' not in ex_data:115 abort(401, description="Workers must provide the execution data token in "116 "order to update the task status.")117 ex_data_id = original_document['execution_data_id']118 orig_ex_data = find_by_id('execution_info', ex_data_id, self.db, {'token': True})119 if ex_data['token'] != orig_ex_data['token']:120 abort(403, description="Provided execution data token does not match.")121 def ensure_worker_has_permission(self, perm):122 assert g.user_info is not None123 if perm not in g.user_info['permissions']:124 abort(403, description="Worker lacks '{}' permission.".format(perm))125 def validate_state_change(self, final_state, update, required_fields):126 def fail():127 self._error('state', "State cannot be changed to '{}' without setting "128 "fields '{}' using 'update_execution_data'.".format(final_state,129 required_fields))130 return False131 if 'update_execution_data' not in update:132 return fail()133 updated_fields = set(update['update_execution_data'].keys())134 if len(updated_fields & required_fields) != len(required_fields):135 return fail()136 return True137 def validate_update(self, document, _id, original_document=None):138 """139 Called after each PATCH request, e.g. when a task is updated in some way.140 """141 # The authorization token and corresponding user are set by the authenticator.142 assert g.user is not None143 role = g.user['role']144 ex_data_update = 'update_execution_data' in document145 """146 Ensure that if a worker is updating a task that is has already claimed, then the147 execution data token that is has provided is correct.148 """149 if ex_data_update and ('state' not in document or document['state'] != 'running'):150 self.validate_execution_data_token(document, original_document)151 if 'state' in document:152 si = original_document['state']153 sf = document['state']154 legal_trans = legal_provider_transitions if role == 'provider' else \155 legal_worker_transitions156 if sf not in legal_trans[si]:157 self._error('state', "User with role '{}' is not authorized to make "158 "state transition from '{}' to '{}'.".format(role, si, sf))159 return False160 if sf == 'running':161 self.ensure_worker_has_permission('claim')162 if not self.validate_state_change(sf, document,163 required_fields={'worker_id'}):164 return False165 elif sf == 'terminated' or (sf == 'cancelled' and role == 'worker'):166 self.ensure_worker_has_permission('report')167 if not self.validate_state_change(sf, document,168 required_fields={'exit_status', 'time_terminated'}):169 return False170 elif ex_data_update:171 self.ensure_worker_has_permission('report')172 keys = set(document['update_execution_data'].keys())173 special_keys = {'worker_id', 'exit_status', 'time_terminated'}174 common_keys = keys & special_keys175 if len(common_keys) != 0:176 self._error('update_execution_data', "The fields {} cannot be "177 "changed independently of the task state.".178 format(common_keys))179 return False180 """181 We call the parent's `validate_update` function now, so that we can ensure that all182 ObjectIds for continuations are valid.183 """184 if not super().validate_update(document, _id, original_document):185 return False186 return self.validate_continuations(document)187 def _validate_empty(self, empty, field, value):188 """189 Extends the 'empty' rule so that it can also be applied to lists. Taken from Nicola190 Iarocci's answer here_.191 .. _here: http://stackoverflow.com/a/23708110192 """193 super()._validate_empty(empty, field, value)194 if isinstance(field, list) and not empty and len(value) == 0:195 self._error(field, "list cannot be empty")196 def _validate_allows_duplicates(self, allows_duplicates, field, value):197 if not isinstance(value, list):198 self._error(field, "Value must be a list.")199 return False200 if not allows_duplicates and len(set(value)) != len(value):201 self._error(field, "Field contains duplicates.")202 return False203 return True204 def _validate_createonly(self, createonly, field, value):205 """206 A field that is marked ``createonly`` may only be initialized if no value is already207 associated with it for the given document.208 """209 if not createonly or not self._original_document:210 return True211 if field in self._original_document:212 self._error(field, "Cannot modify field '{}' once it has been set.".213 format(field))214 return False215 return True216 def _validate_creatable_iff_inactive(self, createonly, field, value):217 """218 Like ``createonly``, but with the additional restriction that the task must also be219 in the ``inactive`` state in order for the field to be set.220 """221 if not createonly or not self._original_document:222 return True223 if self._original_document['state'] != 'inactive':224 self._error(field, "Cannot set field '{}' when task is no longer in "225 "'inactive' state.".format(field))226 return False227 return self._validate_createonly(createonly, field, value)228 def _validate_mutable_iff_inactive(self, mutable, field, value):229 """230 Like ``creatable_iff_inactive``, but allows mutation instead of only initialization.231 """232 if not mutable or not self._original_document or \233 self._original_document['state'] == 'inactive':234 return True235 if field in self._original_document and value != self._original_document[field]:236 self._error(field, "Cannot change value of field '{}' when task is no "237 "longer in 'inactive' state.".format(field))238 return False239 return True240class BulkUpdateValidator(ValidatorBase):241 """242 A validator specialized for validating virtual resources.243 """244 def __init__(self, schema, resource=None, allow_unknown=False,245 transparent_schema_rules=False):246 """247 The last two arguments are retained despite the fact that they are unused, in order248 to maintain compatibility with Eve's validator interface.249 """250 super().__init__(schema, resource)251 def validate_update_format(self, updates):252 """253 Ensures that the JSON array of updates is in the correct format, without validating254 the content of any update. This function is factored out of ``validate_update`` so255 that derived classes can perform intermedidate validation before calling256 ``validate_update_content``.257 """258 if not isinstance(updates, list):259 self._error('updates', "Updates string must be a JSON array.")260 return False261 if len(updates) > max_update_list_length:262 self._error('updates', "Updates list can have at most {} elements.".263 format(max_update_list_length))264 return False265 virtual_resource_keys = ['targets', 'values']266 for i, update in enumerate(updates):267 if not isinstance(update, dict):268 self._error('update {}'.format(i), "Array element is not a dict.")269 continue270 cur_issues = []271 for key in virtual_resource_keys:272 if key not in update:273 cur_issues.append("Missing field '{}'.".format(key))274 invalid_keys = update.keys() - virtual_resource_keys275 if len(invalid_keys) != 0:276 cur_issues.append("Invalid keys '{}'.".format(invalid_keys))277 targets, values = update['targets'], update['values']278 if not isinstance(targets, list):279 cur_issues.append("'targets' field must be a list.")280 if not isinstance(values, list) and not isinstance(values, dict):281 cur_issues.append("'values' field must be a list or a dict.")282 if len(cur_issues) > 0:283 self._error('update {}'.format(i), str(cur_issues))284 if len(self._errors) != 0:285 return False286 for i, update in enumerate(updates):287 # TODO: fill in default fields for the schema.288 # TODO: fill in default values for list elements, if specified.289 """290 This deserializes some values that require special handlers, such as291 ObjectIds and datetime objects. If this becomes a bottleneck, then write a292 custom function that only modifies the fields that are needed.293 """294 updates[i] = serialize(update, schema=self.schema)295 return True296 def validate_update_content(self, updates, original_ids=None, original_documents=None):297 assert (original_ids and original_documents) or \298 (not original_ids) and (not original_documents)299 for i, update in enumerate(updates):300 if original_ids:301 """302 To process rules like ``createonly``, we need the previous values of303 the fields.304 """305 super().validate_update(updates[i], original_ids[i],306 original_documents[i])307 else:308 super().validate(updates[i])309 return len(self._errors) == 0310 def validate_update(self, updates, original_ids=None, original_documents=None):311 """312 All requests made to virtual resources are considered to be updates, since according313 to our design, virtual resources can only be used to modify fields, not update them.314 The fields ``original_ids`` and ``original_documents`` only need to be provided if315 the documents being altered by the updates have validation rules that require316 knowledge of the previous values of fields (e.g. ``createonly``).317 Args:318 updates: The parsed ``dict`` of updates.319 original_ids: An array of the ids corresponding to the documents in320 ``original_documents`` (optional).321 original_documents: An array of the original documents corresponding to the322 IDs in ``updates['targets']`` (optional).323 """324 if not self.validate_update_format(updates):325 return False326 return self.validate_update_content(updates, original_ids, original_documents)327class Validator(ValidatorBase):328 """329 Adds support for virtual resources to ``ValidatorBase``.330 """331 def __init__(self, schema, resource, allow_unknown=False, transparent_schema_rules=False):332 """333 The last two arguments are retained despite the fact that they are unused, in order334 to maintain compatibility with Eve's validator interface.335 """336 self.db = app.data.driver.db337 self.virtual_validators = {}338 # We use a local import here to avoid cyclic dependencies.339 from schema import virtual_resources340 for parent_res, virtuals in virtual_resources.items():341 if parent_res != resource:342 continue343 for virtual_res, v_schema in virtuals.items():344 if 'validator' not in v_schema:345 validator = BulkUpdateValidator346 else:347 validator = v_schema['validator']348 self.virtual_validators[virtual_res] = validator(349 schema=v_schema['schema'], resource=resource)350 super().__init__(schema, resource)351 def _validate_virtual_resource(self, virtual_resource, field, value):352 if not virtual_resource:353 return True354 assert self._id is not None355 assert isinstance(value, list) or isinstance(value, dict)356 dummy = [{'targets': [self._id], 'values': value}]357 validator = self.virtual_validators[field]358 if not validator.validate_update(dummy):359 for k, v in validator.errors.items():360 self._error(k, v)361 return False...
event_hooks.py
Source:event_hooks.py
...155 and we release all of the continuations.156 """157 for x in cont_list:158 continuations.release(x, db)159def update_execution_data(updates, original):160 """161 If a task is set to 'running' *for the first time*, then the following changes are made:162 - The attempt count is incremented.163 - A new instance of execution data is inserted into `execution_info`,164 along with the fields specified in `update_execution_data`.165 If a task is set to 'terminated' and its attempt count is less than its maximum attempt166 count, then the following changes are made:167 - The attempt count is incremented.168 - The state of the task is set to `available`.169 - A new instance of execution data is inserted into `execution_info`.170 Otherwise, the changes given in `update_execution_data` are applied independently.171 """172 db = app.data.driver.db173 id_ = original[config.ID_FIELD]...
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 SeleniumBase 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?
