KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _is_def method in tempest
Best Python code snippet using tempest_python
test_openvswitch.py
Source:test_openvswitch.py 
1import os2import tempfile3from unittest import mock4from . import utils5from hotsos.core import issues6from hotsos.core.issues.utils import IssuesStore, KnownBugsStore7from hotsos.core.config import setup_config8from hotsos.core.plugins.openvswitch import OpenvSwitchBase9from hotsos.core.ycheck.scenarios import YScenarioChecker10from hotsos.plugin_extensions.openvswitch import (11 event_checks,12 summary,13)14LP1917475_LOG = r"""152021-09-22T01:13:33.307Z|00266|ovsdb_idl|WARN|transaction error: {"details":"RBAC rules for client \"compute5\" role \"ovn-controller\" prohibit row insertion into table \"IGMP_Group\".","error":"permission error"}16""" # noqa17DPIF_LOST_PACKETS_LOGS = """182022-03-25T01:55:12.913Z|10324|dpif_netlink(handler10)|WARN|system@ovs-system: lost packet on port channel 0 of handler 0192022-03-25T01:55:12.924Z|05761|dpif_netlink(handler11)|WARN|system@ovs-system: lost packet on port channel 0 of handler 2202022-03-25T01:55:12.972Z|06053|dpif_netlink(handler12)|WARN|system@ovs-system: lost packet on port channel 0 of handler 3212022-03-25T01:55:13.567Z|06054|dpif_netlink(handler12)|WARN|system@ovs-system: lost packet on port channel 0 of handler 3222022-03-25T01:55:13.571Z|06069|dpif_netlink(handler13)|WARN|system@ovs-system: lost packet on port channel 0 of handler 4232022-03-25T01:58:42.750Z|05763|dpif_netlink(handler11)|WARN|system@ovs-system: lost packet on port channel 0 of handler 224""" # noqa25BFD_STATE_CHANGES_TMPLT = """262022-04-21T21:00:{secs}.466Z|01221|bfd(monitor130)|INFO|ovn-abc-ra-f: BFD state change: up->down "Control Detection Time Expired"->"Control Detection Time Expired".27""" # noqa28CR_LRP_CHANGES = """292022-04-21T14:03:{secs}.947Z|1044240|binding|INFO|Claiming lport cr-lrp-31f4fa6a-04cf-462f-aab9-b283fcdb7ce4 for this chassis.302022-04-21T15:03:{secs}.213Z|1044241|binding|INFO|Claiming lport bac8173d-ee39-4139-b699-754a3d17d771 for this chassis.31""" # noqa32DPCTL_SHOW = r"""33 port 6: br-int (internal)34 RX packets:0 errors:0 dropped:1887 overruns:0 frame:035 TX packets:0 errors:0 dropped:0 aborted:0 carrier:036 collisions:037 RX bytes:0 TX bytes:038 port 7: qr-aa623763-fd (internal)39 RX packets:309 errors:0 dropped:1394875 overruns:0 frame:040 TX packets:66312 errors:0 dropped:0 aborted:0 carrier:041 collisions:042 RX bytes:529676 (517.3 KiB) TX bytes:114369124 (109.1 MiB)43 port 8: sg-6df85cb0-c0 (internal)44 RX packets:78 errors:0 dropped:0 overruns:0 frame:045 TX packets:51 errors:0 dropped:0 aborted:0 carrier:046 collisions:047 RX bytes:7878 (7.7 KiB) TX bytes:5026 (4.9 KiB)48""" # noqa49OVS_DB_RECONNECT_ERROR = """502022-07-04 20:03:32.405 2050177 ERROR ovsdbapp.backend.ovs_idl.connection ValueError: non-zero flags not allowed in calls to send() on <class 'eventlet.green.ssl.GreenSSLSocket'>51""" # noqa52BFD_STATE_CHANGES = """532022-07-27T08:49:57.903Z|00007|bfd|INFO|ovn-abc-xa-15: BFD state change: admin_down->down "No Diagnostic"->"No Diagnostic".542022-07-27T08:49:57.903Z|00007|bfd(handler7)|INFO|ovn-abc-xa-15: BFD state change: down->init "No Diagnostic"->"No Diagnostic".552022-07-27T08:49:58.323Z|00066|bfd(handler1)|INFO|ovn-abc-xb-0: BFD state change: down->up "No Diagnostic"->"No Diagnostic".562022-07-27T08:49:58.362Z|00069|bfd(handler1)|INFO|ovn-abc-xa-2: BFD state change: down->up "No Diagnostic"->"No Diagnostic".572022-07-27T08:49:58.844Z|00018|bfd(handler4)|INFO|ovn-abc-xa-15: BFD state change: init->up "No Diagnostic"->"No Diagnostic".58""" # noqa59class TestOpenvswitchBase(utils.BaseTestCase):60 def setUp(self, *args, **kwargs):61 super().setUp(*args, **kwargs)62 setup_config(PLUGIN_NAME='openvswitch')63class TestCoreOpenvSwitch(TestOpenvswitchBase):64 def testBase_offload_disabled(self):65 enabled = OpenvSwitchBase().offload_enabled66 self.assertFalse(enabled)67 def testBase_offload_enabled(self):68 with mock.patch('hotsos.core.plugins.openvswitch.ovs.CLIHelper') as \69 mock_cli:70 mock_cli.return_value = mock.MagicMock()71 f = mock_cli.return_value.ovs_vsctl_get_Open_vSwitch72 f.return_value = '{hw-offload="true", max-idle="30000"}'73 enabled = OpenvSwitchBase().offload_enabled74 self.assertTrue(enabled)75class TestOpenvswitchServiceInfo(TestOpenvswitchBase):76 def test_get_package_checks(self):77 expected = ['libc-bin 2.31-0ubuntu9.2',78 'openssl 1.1.1f-1ubuntu2.10',79 'openvswitch-common 2.13.3-0ubuntu0.20.04.2',80 'openvswitch-switch 2.13.3-0ubuntu0.20.04.2',81 'python3-openssl 19.0.0-1build1',82 'python3-openvswitch 2.13.3-0ubuntu0.20.04.2',83 'python3-ovsdbapp 1.1.0-0ubuntu2']84 inst = summary.OpenvSwitchSummary()85 self.assertEqual(self.part_output_to_actual(inst.output)['dpkg'],86 expected)87 def test_get_resource_checks(self):88 expected = {'systemd': {89 'enabled': [90 'openvswitch-switch'91 ],92 'static': [93 'ovs-vswitchd', 'ovsdb-server'94 ]95 },96 'ps': [97 'ovs-vswitchd (1)',98 'ovsdb-server (1)']}99 inst = summary.OpenvSwitchSummary()100 self.assertEqual(self.part_output_to_actual(inst.output)['services'],101 expected)102 def test_bridge_checks(self):103 expected = {'br-data': [104 {'ens7': {105 'addresses': [],106 'hwaddr': '52:54:00:78:19:c3',107 'state': 'UP',108 'speed': 'Unknown!'}}],109 'br-ex': [],110 'br-int': ['(6 ports)'],111 'br-tun': ['vxlan-0a000072',112 'vxlan-0a000085']}113 inst = summary.OpenvSwitchSummary()114 self.assertEqual(self.part_output_to_actual(inst.output)['bridges'],115 expected)116class TestOpenvswitchEventChecks(TestOpenvswitchBase):117 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',118 new=utils.is_def_filter('ovs-vswitchd.yaml'))119 def test_ovs_vswitchd_checks(self):120 expected = {121 'ovs-vswitchd': {122 'unreasonably-long-poll-interval':123 {'2022-02-10': 3},124 'bridge-no-such-device': {125 '2022-02-10': {'tap6a0486f9-82': 1}}}}126 inst = event_checks.OVSEventChecks()127 self.assertEqual(self.part_output_to_actual(inst.output), expected)128 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',129 new=utils.is_def_filter('errors-and-warnings.yaml'))130 def test_ovs_common_log_checks(self):131 expected = {132 'errors-and-warnings': {133 'ovs-vswitchd': {134 'WARN': {135 '2022-02-04': 56,136 '2022-02-09': 24,137 '2022-02-10': 12}},138 'ovsdb-server': {139 'WARN': {140 '2022-02-04': 6,141 '2022-02-09': 2,142 '2022-02-10': 4}}}}143 inst = event_checks.OVSEventChecks()144 self.assertEqual(self.part_output_to_actual(inst.output), expected)145 @mock.patch('hotsos.core.ycheck.engine.properties.input.CLIHelper')146 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',147 new=utils.is_def_filter('datapath-checks.yaml'))148 def test_ovs_dp_checks(self, mock_helper):149 mock_helper.return_value = mock.MagicMock()150 mock_helper.return_value.ovs_appctl_dpctl_show.return_value = \151 DPCTL_SHOW152 expected = {'datapath-checks-port-stats': {153 'qr-aa623763-fd': {154 'RX': {155 'dropped': 1394875,156 'packets': 309157 }}}}158 inst = event_checks.OVSEventChecks()159 self.assertEqual(self.part_output_to_actual(inst.output), expected)160 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',161 new=utils.is_def_filter('ovn-central.yaml'))162 def test_ovn_central_checks(self):163 expected = {'ovsdb-server-sb': {164 'inactivity-probe': {165 '2022-02-16': {'10.130.11.109': 1,166 '10.130.11.110': 1},167 '2022-02-17': {'10.130.11.109': 1,168 '10.130.11.110': 1}},169 'unreasonably-long-poll-interval': {170 '2022-02-16': 2,171 '2022-02-17': 3}},172 'ovsdb-server-nb': {173 'inactivity-probe': {174 '2022-02-16': {'10.130.11.109': 1},175 '2022-02-17': {'10.130.11.115': 1}},176 'unreasonably-long-poll-interval': {177 '2022-02-16': 2,178 '2022-02-17': 1}179 }}180 inst = event_checks.OVNEventChecks()181 self.assertEqual(self.part_output_to_actual(inst.output), expected)182 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',183 new=utils.is_def_filter('ovn-controller.yaml'))184 def test_ovn_controller_checks(self):185 expected = {'ovn-controller':186 {'unreasonably-long-poll-interval': {187 '2022-02-16': 1,188 '2022-02-17': 1},189 'involuntary-context-switches': {190 '2022-02-16': {'09': 634},191 '2022-02-17': {'04': 136}},192 'bridge-not-found-for-port': {193 '2022-02-16':194 {'provnet-aa3a4fec-a788-42e6-a773-bf3a0cdb52c2':195 1}}}}196 inst = event_checks.OVNEventChecks()197 self.assertEqual(self.part_output_to_actual(inst.output), expected)198 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',199 new=utils.is_def_filter('errors-and-warnings.yaml'))200 def test_ovn_common_log_checks(self):201 expected = {'errors-and-warnings': {202 'ovn-controller': {203 'ERR': {'2022-02-16': 2},204 'WARN': {'2022-02-16': 4, '2022-02-17': 5}},205 'ovn-northd': {206 'ERR': {'2022-02-16': 1, '2022-02-17': 1},207 'WARN': {'2022-02-16': 1, '2022-02-17': 1}},208 'ovsdb-server-nb': {209 'ERR': {'2022-02-16': 1, '2022-02-17': 1},210 'WARN': {'2022-02-16': 12, '2022-02-17': 17}},211 'ovsdb-server-sb': {212 'ERR': {'2022-02-16': 2, '2022-02-17': 2},213 'WARN': {'2022-02-16': 23, '2022-02-17': 23}}}}214 inst = event_checks.OVNEventChecks()215 self.assertEqual(self.part_output_to_actual(inst.output), expected)216 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',217 new=utils.is_def_filter('bfd.yaml'))218 @utils.create_test_files({'var/log/openvswitch/ovs-vswitchd.log':219 BFD_STATE_CHANGES})220 def test_ovs_bfd_state_changes(self):221 expected = {'ovs-vswitchd': {222 'bfd-state-changes': {223 '2022-07-27': {224 'ovn-abc-xa-15': [225 'admin_down->down',226 'down->init',227 'init->up'],228 'ovn-abc-xa-2': [229 'down->up'],230 'ovn-abc-xb-0': [231 'down->up']}}}}232 inst = event_checks.OVSEventChecks()233 self.assertEqual(self.part_output_to_actual(inst.output), expected)234class TestOpenvswitchScenarioChecks(TestOpenvswitchBase):235 @mock.patch('hotsos.core.host_helpers.packaging.CLIHelper')236 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',237 new=utils.is_def_filter('ovn_bugs.yaml'))238 def test_1865127(self, mock_cli):239 mock_cli.return_value = mock.MagicMock()240 mock_cli.return_value.dpkg_l.return_value = \241 ["ii ovn-common 20.12.0 amd64"]242 YScenarioChecker()()243 self.assertEqual(issues.IssuesManager().load_bugs(), {})244 mock_cli.return_value.dpkg_l.return_value = \245 ["ii ovn-common 20.03.2-0ubuntu0.20.04.3 amd64"]246 # we already have this bug in our logs so no need to mock it247 YScenarioChecker()()248 msg = ('The version of ovn on this node is affected by a known bug '249 'where the ovn-controller logs are being spammed with error '250 'messages containing "No bridge for localnet port ..." when '251 'that is in fact not an error. Upgrading to a version >= '252 '20.12.0 will fix the issue.')253 expected = {'bugs-detected':254 [{'id': 'https://bugs.launchpad.net/bugs/1865127',255 'desc': msg,256 'origin': 'openvswitch.01part'}]}257 self.assertEqual(issues.IssuesManager().load_bugs(),258 expected)259 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',260 new=utils.is_def_filter('ovn_bugs.yaml'))261 @utils.create_test_files({'var/log/ovn/ovn-controller.log': LP1917475_LOG})262 def test_1917475(self):263 YScenarioChecker()()264 expected = {'bugs-detected':265 [{'id': 'https://bugs.launchpad.net/bugs/1917475',266 'desc': "known ovn bug identified - db rbac errors",267 'origin': 'openvswitch.01part'}]}268 self.assertEqual(issues.IssuesManager().load_bugs(),269 expected)270 @mock.patch('hotsos.core.host_helpers.packaging.CLIHelper')271 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',272 new=utils.is_def_filter('ovs_bugs.yaml'))273 def test_1839592(self, mock_cli):274 mock_cli.return_value = mock.MagicMock()275 mock_cli.return_value.dpkg_l.return_value = \276 ["ii libc-bin 2.26-3ubuntu1.3 amd64"]277 YScenarioChecker()()278 expected = {'bugs-detected':279 [{'id': 'https://bugs.launchpad.net/bugs/1839592',280 'desc': "Installed package 'libc-bin' with version "281 "2.26-3ubuntu1.3 has a known critical bug which "282 "causes ovs deadlocks. If this environment is "283 "using OVS it should be upgraded asap.",284 'origin': 'openvswitch.01part'}]}285 self.assertEqual(issues.IssuesManager().load_bugs(), expected)286 @mock.patch('hotsos.core.plugins.openvswitch.ovs.CLIHelper')287 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',288 new=utils.is_def_filter('flow_lookup_checks.yaml'))289 def test_flow_lookup_checks_p1(self, mock_cli):290 mock_cli.return_value = mock.MagicMock()291 mock_cli.return_value.ovs_appctl_dpctl_show.return_value = \292 ['lookups: hit:39017272903 missed:137481120 lost:54691089']293 YScenarioChecker()()294 msg = ('OVS datapath is reporting a non-zero amount of "lost" packets '295 '(total=54691089) which implies that packets destined for '296 'userspace (e.g. vm tap) are being dropped. Please check '297 'ovs-appctl dpctl/show to see if the number of lost packets is '298 'still increasing.')299 issues = list(IssuesStore().load().values())[0]300 self.assertEqual([issue['desc'] for issue in issues], [msg])301 @mock.patch('hotsos.core.plugins.openvswitch.ovs.CLIHelper')302 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',303 new=utils.is_def_filter('flow_lookup_checks.yaml'))304 @utils.create_test_files({'var/log/openvswitch/ovs-vswitchd.log':305 DPIF_LOST_PACKETS_LOGS})306 def test_flow_lookup_checks_p2(self, mock_cli):307 mock_cli.return_value = mock.MagicMock()308 mock_cli.return_value.ovs_appctl_dpctl_show.return_value = \309 ['lookups: hit:39017272903 missed:137481120 lost:54691089']310 YScenarioChecker()()311 msg = ('OVS datapath is reporting a non-zero amount of "lost" '312 'packets (total=54691089) which implies that packets '313 'destined for userspace (e.g. vm tap) are being dropped. '314 'ovs-vswitchd is also reporting large numbers of dropped '315 'packets within a 24h period (look for '316 '"system@ovs-system: lost packet on port channel"). '317 'This could be caused by '318 'overloaded system cores blocking ovs threads from '319 'delivering packets in time. Please check ovs-appctl '320 'dpctl/show to see if the number of lost packets is still '321 'increasing.')322 issues = list(IssuesStore().load().values())[0]323 self.assertEqual([issue['desc'] for issue in issues], [msg])324 @mock.patch('hotsos.core.ycheck.engine.properties.search.CLIHelper')325 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',326 new=utils.is_def_filter('bfd_flapping.yaml'))327 def test_bfd_flapping_vswitchd_only(self, mock_cli):328 mock_cli.return_value = mock.MagicMock()329 mock_cli.return_value.date.return_value = '2022-04-21 20:44:21'330 with tempfile.TemporaryDirectory() as dtmp:331 setup_config(DATA_ROOT=dtmp)332 logfile = os.path.join(dtmp,333 'var/log/openvswitch/ovs-vswitchd.log')334 os.makedirs(os.path.dirname(logfile))335 with open(logfile, 'w') as fd:336 for i in range(0, 10):337 fd.write(BFD_STATE_CHANGES_TMPLT.338 format(secs='0{}'.format(i)))339 YScenarioChecker()()340 msg = ('The ovn-controller on this host has experienced 10 BFD '341 'state changes within an hour (and within the last 24 '342 'hours). This is unusual and could be an indication that '343 'something is wrong with the network between this node and '344 'one or more peer chassis nodes.')345 issues = list(IssuesStore().load().values())[0]346 self.assertEqual([issue['desc'] for issue in issues], [msg])347 @mock.patch('hotsos.core.ycheck.engine.properties.search.CLIHelper')348 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',349 new=utils.is_def_filter('bfd_flapping.yaml'))350 def test_bfd_flapping_cr_lrp_changes(self, mock_cli):351 mock_cli.return_value = mock.MagicMock()352 mock_cli.return_value.date.return_value = '2022-04-21 20:44:21'353 with tempfile.TemporaryDirectory() as dtmp:354 setup_config(DATA_ROOT=dtmp)355 logfile = os.path.join(dtmp, 'var/log/ovn/ovn-controller.log')356 os.makedirs(os.path.dirname(logfile))357 with open(logfile, 'w') as fd:358 for i in range(0, 20):359 fd.write(CR_LRP_CHANGES.format(secs='0{}'.format(i)))360 YScenarioChecker()()361 msg = ('The ovn-controller on this host is showing 20 logical '362 'router port (lrp) chassis re-assignments within the last '363 '24 hours that do not appear to have resulted from BFD '364 'state changes. This could indicate that some operator '365 'activity is causing significant load in ovn which may or '366 'may not be expected.')367 issues = list(IssuesStore().load().values())[0]368 self.assertEqual([issue['desc'] for issue in issues], [msg])369 @mock.patch('hotsos.core.host_helpers.packaging.CLIHelper')370 @mock.patch('hotsos.core.ycheck.engine.YDefsLoader._is_def',371 new=utils.is_def_filter('ovsdb_reconnect_errors.yaml'))372 @utils.create_test_files({'var/log/neutron/neutron-server.log':373 OVS_DB_RECONNECT_ERROR})374 def test_ovsdb_reconnect_error(self, mock_cli):375 mock_cli.return_value = mock.MagicMock()376 mock_cli.return_value.dpkg_l.return_value = \377 ["ii python3-openvswitch 2.17.0 amd64"]378 YScenarioChecker()()379 msg = ("Installed package 'python3-openvswitch' with version "380 "2.17.0 has a known bug whereby if connections to the ovn "381 "southbound db are closed, the client fails to reconnect. "382 "This is usually resolved with a service restart and a "383 "fix is available as of openvswitch version 2.17.2.")384 issues = list(KnownBugsStore().load().values())[0]...
render.py
Source:render.py
...44 elif _is_for(key):45 rendered_obj = _render_for(46 key, value, params, defs, rendered_obj47 )48 elif _is_def(key):49 _store_def(key, value, defs)50 elif _is_use(key):51 rendered_obj = _render_use(52 key, value, params, defs, rendered_obj53 )54 else:55 _update_obj(rendered_obj, key, value, params, defs)56 else:57 last_if = None58 _update_obj(rendered_obj, key, value, params, defs)59 if defaults_obj:60 rendered_obj = _deep_merge_dicts(defaults_obj, rendered_obj) # type: ignore61 return rendered_obj62 elif isinstance(obj, list):63 rendered_obj = []64 for elem in obj:65 rendered_elem = render_from_obj(elem, params, defs)66 if _can_extend_list(elem, rendered_elem):67 # Convert rendered_elem to [] if it's {}68 rendered_obj.extend(rendered_elem or []) # type: ignore69 else:70 rendered_obj.append(rendered_elem)71 return rendered_obj72 elif isinstance(obj, str):73 return render_interpolation(obj, params)74 else:75 return obj76def _render_load(77 value: JsonType,78 params: Dict[str, Any],79 defs: Dict[str, Def],80 rendered_obj: JsonType,81) -> JsonType:82 if not isinstance(value, list):83 value = [value]84 for filename in value:85 filename = _parse_filename(filename, params, "load")86 elem = _load_yaml(filename)87 rendered_elem = render_from_obj(elem, params, defs)88 rendered_obj = _shallow_merge(89 f"load: {filename}", rendered_elem, params, defs, rendered_obj90 )91 return rendered_obj92def _parse_filename(filename: JsonType, params: Dict[str, Any], load_type: str) -> str:93 if not isinstance(filename, str):94 raise YATLSyntaxError(95 f"{load_type} must be given a string or list of strings: {filename}"96 )97 filename = render_interpolation(filename, params)98 if not isinstance(filename, str):99 raise YATLSyntaxError(100 f"{load_type} directive must be given a string or list of strings: {filename}"101 )102 return filename103def _load_defaults(104 value: JsonType, params: Dict[str, Any], defs: Dict[str, Def]105) -> dict:106 if not isinstance(value, list):107 value = [value]108 accumulated_defaults: dict = {}109 for filename in value:110 filename = _parse_filename(filename, params, "load_defaults_from")111 defaults = _load_yaml(filename)112 if not isinstance(defaults, dict):113 raise YATLSyntaxError(f"{filename} must be an object at the top-level")114 rendered_defaults = render_from_obj(defaults, params, defs)115 accumulated_defaults = _deep_merge_dicts(116 accumulated_defaults, rendered_defaults117 )118 return accumulated_defaults119def _is_if(key: str) -> bool:120 return bool(re.match(r"\.if\b", key))121def _is_elif(key: str) -> bool:122 return bool(re.match(r"\.elif\b", key))123def _render_if(124 if_key: str,125 if_value: JsonType,126 params: Dict[str, Any],127 defs: Dict[str, Def],128 rendered_obj: JsonType,129) -> Tuple[JsonType, bool]:130 # This regular expression captures both if and elif.131 if_match = re.match(r"\.(?:el)?if\s*\((.*)\)\s*$", if_key)132 if not if_match:133 raise YATLSyntaxError(f"Invalid if statement: {if_key}")134 condition = if_match[1].strip()135 if params[condition]:136 return _shallow_merge(if_key, if_value, params, defs, rendered_obj), True137 return rendered_obj, False138def _shallow_merge(139 key: str,140 value: JsonType,141 params: Dict[str, Any],142 defs: Dict[str, Def],143 rendered_obj: JsonType,144) -> JsonType:145 rendered_value = render_from_obj(value, params, defs)146 if not _can_merge_values(rendered_obj, rendered_value):147 raise YATLSyntaxError(148 f"Cannot merge {_type_name(rendered_value)} with {_type_name(rendered_obj)} in {key}"149 )150 if isinstance(rendered_value, dict):151 # Don't deep-merge, just shallow-update152 return {**rendered_obj, **rendered_value} # type: ignore153 elif isinstance(rendered_value, list):154 rendered_obj = rendered_obj or [] # Convert {} to []155 return rendered_obj + rendered_value # type: ignore156 else:157 return rendered_value158def _can_merge_values(parent_obj: JsonType, if_value: JsonType) -> bool:159 """Handle when there are multiple ifs in an object:160 Valid:161 foo: 1162 .if(y):163 bar: 2164 Invalid:165 .foo: 1166 .if(y):167 - 2168 """169 if not isinstance(parent_obj, dict) and not isinstance(parent_obj, list):170 # The parent object may be a scalar value if there are two ifs:171 # .if(x): 1172 # .if(y): 2173 return False174 if isinstance(parent_obj, dict) and not parent_obj:175 # The parent object is empty, so the if is the first node176 return True177 return type(parent_obj) == type(if_value)178def _type_name(x: Any) -> str:179 return type(x).__name__180def _render_else(181 key: str,182 value: JsonType,183 params: Dict[str, Any],184 defs: Dict[str, Def],185 rendered_obj: JsonType,186) -> JsonType:187 return _shallow_merge(key, value, params, defs, rendered_obj)188def _is_for(key: str) -> bool:189 return bool(re.match(r"\.for\b", key))190def _render_for(191 key: str,192 value: JsonType,193 params: Dict[str, Any],194 defs: Dict[str, Def],195 rendered_obj: JsonType,196) -> JsonType:197 for_match = re.match(198 r"for\s*\(([a-zA-Z_][a-zA-Z0-9_]*)\s+in\s+([a-zA-Z_][a-zA-Z0-9_]*)\)\s*$",199 key[1:],200 )201 if not for_match:202 raise YATLSyntaxError(f"Invalid for statement: {key}")203 var = for_match[1].strip()204 param = for_match[2].strip()205 try:206 iterable = params[param]207 except KeyError:208 raise YATLEnvironmentError(f"Missing parameter {param}")209 rendered_list = []210 for elem in iterable:211 rendered_list.append(render_from_obj(value, {**params, var: elem}, defs))212 return _shallow_merge(key, rendered_list, params, defs, rendered_obj)213def _can_extend_list(elem: JsonType, rendered_elem: JsonType) -> bool:214 if isinstance(elem, dict):215 # All keys must be directives.216 if any(not _is_directive(key) for key in elem):217 return False218 # The rendered object must be a list, or an empty object.219 if isinstance(rendered_elem, list):220 return True221 if isinstance(rendered_elem, dict) and not rendered_elem:222 return True223 return False224def _is_directive(key: str) -> bool:225 return (226 _is_if(key)227 or _is_elif(key)228 or key == ".else"229 or _is_for(key)230 or key == ".load"231 or _is_def(key)232 or _is_use(key)233 )234def _is_def(key: str) -> bool:235 return bool(re.match(r"\.def\b", key))236def _store_def(key: str, value: JsonType, defs: Dict[str, Def]) -> None:237 name, args = _parse_def_parts(key)238 if len(args) != len({*args}):239 raise YATLSyntaxError(f"Duplicate name in def arguments: {key}")240 defs[name] = Def(name, args, value)241def _parse_def_parts(key: str) -> Tuple[str, List[str]]:242 name_match = re.match(243 r"""244 \.def \s+245 ([a-zA-Z_][a-zA-Z0-9_]*) \s* # Capture the name246 """,247 key,248 re.VERBOSE,...
desk_check.py
Source:desk_check.py
...48 return self._build_name_dicts(frame.f_back,49 name_dicts=name_dicts,50 functions=functions)51 return name_dicts, functions52 def _is_def(self, f_name, cur_line):53 return cur_line and 'def ' in cur_line and f_name not in cur_line54 def _trace(self, frame, event, arg):55 f_name = frame.f_code.co_name56 lineno = frame.f_lineno57 # Ignore other modules (change this if additional modules need to be traced)58 module_name = frame.f_globals.get('__name__')59 if module_name != '__main__':60 return self._trace61 try:62 call_line_i = self.stack[-1][0] - 163 except IndexError:64 call_line_i = None65 try:66 cur_line = self.src_lines[lineno - 1]67 except IndexError:68 cur_line = None69 name_dicts, functions = self._build_name_dicts(frame)70 if event == 'line':71 if self.prev_line:72 self.history.append(73 TraceData(self.prev_lineno - 1, self.prev_line, name_dicts,74 call_line_i, None, self.mock_builtins.outputs))75 self.prev_line = cur_line76 self.prev_lineno = lineno77 elif event == 'call' and not self._is_def(f_name, cur_line):78 if self.prev_line:79 self.stack.append((self.prev_lineno, self.prev_line))80 if self.prev_line:81 if self.history:82 last = self.history[-1]83 self.history.append(84 TraceData(self.prev_lineno - 1, self.prev_line,85 *last[2:]))86 self.history.append(87 TraceData(lineno - 1, cur_line, name_dicts,88 self.prev_lineno - 1, None,89 self.mock_builtins.outputs))90 self.prev_line = None91 self.prev_lineno = None...
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?
