KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use update_spec method in autotest
Best Python code snippet using autotest_python
test_doc_manager_base.py
Source:test_doc_manager_base.py 
1# Copyright 2017 MongoDB, Inc.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import sys15sys.path[0:0] = [""] # noqa16from mongo_connector.connector import (17 get_mininum_mongodb_version,18 update_mininum_mongodb_version,19)20from mongo_connector.doc_managers.doc_manager_base import (21 DocManagerBase,22 UpdateDoesNotApply,23)24from mongo_connector.test_utils import TESTARGS25from mongo_connector.version import Version26from tests import unittest27UPDATE_SUCCESS_TEST_CASES = [28 {29 "description": "Update whole document.",30 "doc": {"a": 1},31 "update_spec": {"a": 2, "b": 3},32 "result": {"a": 2, "b": 3},33 },34 {35 "description": "Update by un-setting an existing field.",36 "doc": {"a": 1, "b": 2},37 "update_spec": {"$unset": {"a": True}},38 "result": {"b": 2},39 },40 {41 "description": "Update by un-setting an existing nested field.",42 "doc": {"a": {"b": 2, "c": 3}},43 "update_spec": {"$unset": {"a.b": True}},44 "result": {"a": {"c": 3}},45 },46 {47 "description": "Update by un-setting an array element.",48 "doc": {"a": 1, "b": [0, 1, 2, 3]},49 "update_spec": {"$unset": {"b.1": True}},50 "result": {"a": 1, "b": [0, None, 2, 3]},51 },52 {53 "description": "Update by adding a field.",54 "doc": {"a": 1},55 "update_spec": {"$set": {"b": [{"c": 1}, {"d": 2}]}},56 "result": {"a": 1, "b": [{"c": 1}, {"d": 2}]},57 },58 {59 "description": "Update by adding a nested field.",60 "doc": {"a": 1},61 "update_spec": {"$set": {"b.c.d": 2}},62 "result": {"a": 1, "b": {"c": {"d": 2}}},63 },64 {65 "description": "Update by adding and removing a fields.",66 "doc": {"a": 1, "c": 3},67 "update_spec": {"$unset": {"a": True, "c": True}, "$set": {"b": 2, "d": 4}},68 "result": {"b": 2, "d": 4},69 },70 {71 "description": "Update by setting an element far beyond the end of" "an array.",72 "doc": {"a": 1, "b": [{"c": 1}]},73 "update_spec": {"$set": {"b.4": {"c": 2}}},74 "result": {"a": 1, "b": [{"c": 1}, None, None, None, {"c": 2}]},75 },76 {77 "description": "Update by setting an element right beyond the end "78 "of an array.",79 "doc": {"a": 1, "b": [{"c": 1}]},80 "update_spec": {"$set": {"b.1": {"c": 2}}},81 "result": {"a": 1, "b": [{"c": 1}, {"c": 2}]},82 },83 {84 "description": "Update by setting an attribute of a sub-document far "85 "beyond the end of an array.",86 "doc": {"a": 1, "b": [{"c": 1}]},87 "update_spec": {"$set": {"b.4.c": 2}},88 "result": {"a": 1, "b": [{"c": 1}, None, None, None, {"c": 2}]},89 },90 {91 "description": "Update by setting an attribute of a sub-document "92 "right beyond the end of an array.",93 "doc": {"a": 1, "b": [{"c": 1}]},94 "update_spec": {"$set": {"b.1.c": 2}},95 "result": {"a": 1, "b": [{"c": 1}, {"c": 2}]},96 },97 {98 "description": "Update by changing a field within an array element.",99 "doc": {"a": 1, "b": [{"c": 1}]},100 "update_spec": {"$set": {"b.0.c": 2}},101 "result": {"a": 1, "b": [{"c": 2}]},102 },103 {104 "description": "Update by adding a field within an array element.",105 "doc": {"a": 1, "b": [{"c": 1}, {"d": 2}]},106 "update_spec": {"$set": {"b.1.c": 3}},107 "result": {"a": 1, "b": [{"c": 1}, {"c": 3, "d": 2}]},108 },109 {110 "description": "Update by replacing an array element.",111 "doc": {"a": 1, "b": [0, 1, 2, 3]},112 "update_spec": {"$set": {"b.2": {"new": 2}}},113 "result": {"a": 1, "b": [0, 1, {"new": 2}, 3]},114 },115 {116 "description": "Update by replacing an array with a document with "117 "int field.",118 "doc": {"a": 1, "b": [{"c": 1}, {"d": 2}]},119 "update_spec": {"$set": {"b": {"0": {"e": 100}}}},120 "result": {"a": 1, "b": {"0": {"e": 100}}},121 },122]123UNSET_FAILURE_TEST_CASES = [124 {125 "description": "Update by un-setting a non-existent field.",126 "doc": {"a": 1, "b": 2},127 "update_spec": {"$unset": {"not-present": True}},128 "result": {"a": 1, "b": 2},129 },130 {131 "description": "Update by un-setting a non-existent nested field.",132 "doc": {"a": 1, "b": {"c": {"d": 1}}},133 "update_spec": {"$unset": {"b.not-present.foo": True}},134 "result": {"a": 1, "b": {"c": {"d": 1}}},135 },136 {137 "description": "Update by un-setting invalid array index.",138 "doc": {"a": 1, "b": [0, 1, 2, 3]},139 "update_spec": {"$unset": {"b.not-an-index": True}},140 "result": {"a": 1, "b": [0, 1, 2, 3]},141 },142 {143 "description": "Update by un-setting invalid nested array index.",144 "doc": {"a": 1, "b": [0, 1, 2, 3]},145 "update_spec": {"$unset": {"b.not-an-index.not-present": True}},146 "result": {"a": 1, "b": [0, 1, 2, 3]},147 },148 {149 "description": "Update by un-setting a non-existent array element.",150 "doc": {"a": 1, "b": [0, 1, 2]},151 "update_spec": {"$unset": {"b.4": True}},152 "result": {"a": 1, "b": [0, 1, 2]},153 },154 {155 "description": "Update by un-setting a non-existent field in an array"156 "element.",157 "doc": {"a": 1, "b": [0, {"c": 1}, 2]},158 "update_spec": {"$unset": {"b.1.not-present": True}},159 "result": {"a": 1, "b": [0, {"c": 1}, 2]},160 },161 {162 "description": "Update by adding and removing a non-existent field.",163 "doc": {"a": 1},164 "update_spec": {"$unset": {"a": True, "not-present": True}, "$set": {"b": 2}},165 "result": {"b": 2},166 },167]168UPDATE_FAILURE_TEST_CASES = [169 {170 "description": "Using array notation on non-array field.",171 "doc": {"a": 1},172 "update_spec": {"$set": {"a.0": 2}},173 },174 {175 "description": "Using array notation on non-array field.",176 "doc": {"a": 1},177 "update_spec": {"$set": {"a.0.1": 2}},178 },179 {180 "description": "Using nested field notation on non-object.",181 "doc": {"a": 1},182 "update_spec": {"$set": {"a.b": 2}},183 },184 {185 "description": "Using deeply nested field notation on non-object.",186 "doc": {"a": 1},187 "update_spec": {"$set": {"a.b.c.b": 2}},188 },189 {190 "description": "Setting a field on an array field.",191 "doc": {"a": [{"c": 1}, {"c": 2}]},192 "update_spec": {"$set": {"a.c": 2}},193 },194 {195 "description": "Setting a field on a null array element.",196 "doc": {"a": [None, None]},197 "update_spec": {"$set": {"a.0.c": 2}},198 },199]200class TestDocManagerBase(unittest.TestCase):201 """Unit tests for DocManagerBase"""202 def setUp(self):203 self.base = DocManagerBase()204 def assertUpdateTestSucceeds(self, test):205 self.assertEqual(206 self.base.apply_update(test["doc"], test["update_spec"]),207 test["result"],208 msg=test["description"],209 )210 def assertUpdateTestFails(self, test):211 try:212 doc = self.base.apply_update(test["doc"], test["update_spec"])213 self.fail(214 "UpdateDoesNotApply on MongoDB verison %s not raised for "215 "test: %s, applied %r to %r and got %r"216 % (217 get_mininum_mongodb_version(),218 test["description"],219 test["update_spec"],220 test["doc"],221 doc,222 )223 )224 except UpdateDoesNotApply:225 pass226 def test_apply_update(self):227 for test in UPDATE_SUCCESS_TEST_CASES:228 self.assertUpdateTestSucceeds(test)229 def test_apply_update_fails(self):230 for test in UPDATE_FAILURE_TEST_CASES:231 self.assertUpdateTestFails(test)232 def test_apply_update_unset_failures(self):233 # Reset the minimum MongoDB version at the start and end.234 update_mininum_mongodb_version(None)235 for mock_mongodb_version in [(3, 4), (3, 2), (3, 0), (2, 6), (2, 4), None]:236 if mock_mongodb_version is None:237 update_mininum_mongodb_version(None)238 else:239 update_mininum_mongodb_version(Version(*mock_mongodb_version))240 for test in UNSET_FAILURE_TEST_CASES:241 if mock_mongodb_version == (2, 4):242 self.assertUpdateTestSucceeds(test)243 else:244 self.assertUpdateTestFails(test)245 def test_bulk_upsert(self):246 with self.assertRaises(NotImplementedError):247 self.base.bulk_upsert([{}], *TESTARGS)248 def test_update(self):249 with self.assertRaises(NotImplementedError):250 self.base.update({}, {}, *TESTARGS)251 def test_upsert(self):252 with self.assertRaises(NotImplementedError):253 self.base.upsert({}, *TESTARGS)254 def test_remove(self):255 with self.assertRaises(NotImplementedError):256 self.base.remove(1, *TESTARGS)257 def test_insert_file(self):258 with self.assertRaises(NotImplementedError):259 self.base.insert_file(None, *TESTARGS)260 def test_handle_command(self):261 with self.assertRaises(NotImplementedError):262 self.base.handle_command({}, *TESTARGS)263 def test_search(self):264 with self.assertRaises(NotImplementedError):265 self.base.search(0, 1)266 def test_commit(self):267 with self.assertRaises(NotImplementedError):268 self.base.commit()269 def test_get_last_doc(self):270 with self.assertRaises(NotImplementedError):271 self.base.get_last_doc()272 def test_stop(self):273 with self.assertRaises(NotImplementedError):274 self.base.stop()275if __name__ == "__main__":...
nodes_and_relationships_updater.py
Source:nodes_and_relationships_updater.py
1"""2Class that handles Nodes and Relationships3updates from a Mongo document.4"""5import re6import logging7from mongo_connector.doc_managers.nodes_and_relationships_builder import NodesAndRelationshipsBuilder8LOG = logging.getLogger(__name__)9class NodesAndRelationshipsUpdater(object):10 def __init__(self):11 self.statements_with_params = []12 def run_update(self, update_spec, doc_id, doc_type):13 params_dict = {"doc_id": doc_id}14 set_dict = {}15 for spec in update_spec.keys():16 if spec=='$set':17 update_value_list = update_spec['$set']18 for update_value in update_value_list.keys():19 if (self.is_relationship_update(update_value_list[update_value])):20 self.update_relationship(update_value_list[update_value], doc_type, update_value, doc_id)21 else:22 set_dict.update({update_value: update_value_list[update_value]})23 params_dict.update({"set_parameter": set_dict})24 statement = "MATCH (d:Document:`{doc_type}`) WHERE d._id={{doc_id}} SET d+={{set_parameter}}".format(doc_type=doc_type)25 self.statements_with_params.append({statement: params_dict})26 elif spec=='$unset':27 update_value_list = update_spec['$unset']28 for update_value in update_value_list.keys():29 statement = "MATCH (d:Document:{doc_type}), (c:Document:{update_value}) WHERE d._id={{doc_id}} OPTIONAL MATCH (d)-[r]-(c) DELETE r WITH d, c OPTIONAL MATCH (c)-[s]-() WITH d,c,s, CASE WHEN s IS NULL THEN c ELSE NULL END AS n DELETE n".format(doc_type=doc_type, update_value=update_value)30 self.statements_with_params.append({statement: params_dict})31 statement = "MATCH (d:Document:`{doc_type}` {{ _id: {{doc_id}} }} ) REMOVE d.`{remove_parameter}` ".format(doc_type=doc_type, remove_parameter=update_value)32 self.statements_with_params.append({statement: params_dict})33 else:34 self.handle_replacement(update_spec, doc_id, doc_type)35 break36 def handle_replacement(self, update_spec, doc_id, doc_type):37 params_dict = {"doc_id": doc_id}38 self.remove_legacy_nodes(doc_id, doc_type)39 for spec in update_spec.keys():40 if self.drop_id_spec(spec):41 if (self.is_relationship_update(update_spec[spec])):42 self.clear_node(doc_type, doc_id)43 self.update_relationship(update_spec[spec], doc_type, spec, doc_id)44 else:45 params_dict.update({"set_parameter": {spec: update_spec[spec]}})46 statement = "MATCH (d:Document:`{doc_type}`) WHERE d._id={{doc_id}} SET d={{set_parameter}}".format(doc_type=doc_type) 47 self.statements_with_params.append({statement: params_dict})48 def is_relationship_update(self, update_param):49 return (type(update_param) is dict)50 def update_relationship(self, document, root_type, doc_type, doc_id):51 builder = NodesAndRelationshipsBuilder(document, doc_type, doc_id, [root_type])52 builder.build_relationships_query(root_type, doc_type, doc_id, doc_id)53 self.statements_with_params.append(builder.query_nodes)54 self.statements_with_params.append(builder.relationships_query)55 def remove_legacy_nodes(self, doc_id, doc_type):56 params_dict = {"doc_id": doc_id}57 statement = "MATCH (d:Document:`{doc_type}`) WHERE d._id={{doc_id}} OPTIONAL MATCH (d)-[r]-(c) DELETE r WITH d, c OPTIONAL MATCH (c)-[s]-() WITH d,c,s, CASE WHEN s IS NULL THEN c ELSE NULL END AS n DELETE n".format(doc_type=doc_type)58 self.statements_with_params.append({statement: params_dict})59 def clear_node(self, doc_type, doc_id):60 params_dict = {"doc_id": doc_id}61 params_dict.update({"parameters": {"_id": doc_id}})62 statement = "MATCH (d:Document:`{doc_type}`) WHERE d._id={{doc_id}} SET d={{}} SET d={{parameters}}".format(doc_type=doc_type) 63 self.statements_with_params.append({statement: params_dict})64 def drop_id_spec(self, spec):65 if spec=='_id':66 return False67 return True...
util.py
Source:util.py
1import pathlib2from typing import Union3import yaml4def update_dict_from_spec(data: Union[dict, list], update_spec: dict):5 """6 Update a dict (or a list) according to a given update specification.7 """8 for key in update_spec:9 if isinstance(update_spec[key], dict):10 update_dict_from_spec(data[key], update_spec[key])11 else:12 data[key] = update_spec[key]13def build_yaml(template_path: Union[str, pathlib.Path], output_path: Union[str, pathlib.Path], update_spec: dict):14 """15 Load a YAML template file, update some properties, and write the result16 to a new YAML file.17 This function takes an update specification to inject the modified values18 at the right location in the YAML structure. This specification takes the19 form of a Python dictionary that follows the structure of the property to20 update.21 This is easier to understand with an example. Let us say we have the22 following YAML template file:23 ```yaml24 foo:25 bar:26 baz: 027 qux: 128 ```29 Then, suppose we apply the following update specification:30 ```python31 spec = {32 "foo": {33 "bar": {34 "qux": 123435 }36 }37 }38 ```39 The resulting YAML file will be:40 ```yaml41 foo:42 bar:43 baz: 044 qux: 123445 ```46 Note that the other property `bar` has not been modified.47 """48 with pathlib.Path(template_path).open("r") as template_file:49 data = yaml.safe_load(template_file)50 update_dict_from_spec(data, update_spec)51 with pathlib.Path(output_path).open("w") as output_file:...
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 autotest 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?
