KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use execute_node method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_vm.py

Source:test_vm.py

...28    return PeekParser()29def assert_called_with(vm, *args, **kwargs):30    vm.context['debug'].assert_called_with(vm.app, *args, **kwargs)31def test_peek_vm_data_types(peek_vm, parser):32    peek_vm.execute_node(parser.parse('debug "42" 42 true 4.2 [4, 2] {4:2} @a42 foo="bar"')[0])33    assert_called_with(peek_vm, '42', 42, True, 4.2, [4, 2], {4: 2}, **{'@': ['a42'], 'foo': 'bar'})34def test_peek_vm_expr(peek_vm, parser):35    peek_vm.execute_node(parser.parse('debug 1 + 2 * 3 + (5-1) {"a": 42}.@a [4, 2, 1].1 foo="a" + "b"')[0])36    assert_called_with(peek_vm, 11, 42, 2, **{'foo': 'ab'})37def test_str_and_number(peek_vm, parser):38    peek_vm.execute_node(parser.parse('debug "hello" + 42')[0])39    assert_called_with(peek_vm, 'hello42')40def test_peek_vm_func(peek_vm, parser):41    peek_vm.execute_node(parser.parse('debug echo(0 1 2)')[0])42    assert_called_with(peek_vm, '0 1 2')43    peek_vm.execute_node(parser.parse('debug [echo].0(0 1 2)')[0])44    assert_called_with(peek_vm, '0 1 2')45    peek_vm.execute_node(parser.parse('debug echo(42) + "hello"')[0])46    assert_called_with(peek_vm, '42hello')47def test_peek_vm_es_api_call(peek_vm, parser):48    peek_vm.execute_node(parser.parse('GET /')[0])49    peek_vm.app.display.info.assert_called_with(50        '{"foo": [1, 2, 3, 4], "bar": {"hello": [42, "world"]}}', header_text='')51    assert peek_vm.get_value('_') == {'foo': [1, 2, 3, 4], 'bar': {'hello': [42, 'world']}}52    peek_vm.execute_node(parser.parse('debug _."bar".@hello.1')[0])53    assert_called_with(peek_vm, 'world')54    peek_vm.execute_node(parser.parse('GET ("foo" + "/" + 42)')[0])55    peek_vm.app.es_client_manager.current.perform_request.assert_called_with('GET', '/foo/42', None, headers=None)56    peek_vm.execute_node(parser.parse('PUT /<my-index-{now/d}>')[0])57    peek_vm.app.es_client_manager.current.perform_request.assert_called_with('PUT', '/%3Cmy-index-%7Bnow%2Fd%7D%3E',58                                                                             None, headers=None)59def test_es_api_call_quiet(peek_vm, parser):60    peek_vm.execute_node(parser.parse('GET / quiet=true')[0])61    peek_vm.app.display.info.assert_not_called()62    assert peek_vm.get_value('_') == {'foo': [1, 2, 3, 4], 'bar': {'hello': [42, 'world']}}63def test_peek_vm_let(peek_vm, parser):64    peek_vm.execute_node(parser.parse('let foo={"a": [3, 4, 5]} bar=["hello", 42] c="world"')[0])65    peek_vm.execute_node(parser.parse('debug foo bar c')[0])66    assert_called_with(peek_vm, {"a": [3, 4, 5]}, ["hello", 42], "world")67    peek_vm.execute_node(parser.parse('let foo.@a.1 = 42')[0])68    peek_vm.execute_node(parser.parse('debug foo')[0])69    assert_called_with(peek_vm, {"a": [3, 42, 5]})70def test_invalid_let(peek_vm, parser):71    peek_vm.execute_node(parser.parse('let foo=[[1, 2], echo]')[0])72    with pytest.raises(PeekError) as e:73        peek_vm.execute_node(parser.parse('let foo.0."x" = 42')[0])74    assert "Invalid lhs for assignment: ['foo', 0, 'x']" in str(e.value)75    with pytest.raises(PeekError) as e:76        peek_vm.execute_node(parser.parse('let foo.1."x" = 42')[0])77    assert "Invalid lhs for assignment: ['foo', 1, 'x']" in str(e.value)78def test_for_in(peek_vm, parser):79    peek_vm.execute_node(parser.parse('''for x in [1, 2, 3] {80    let y = x81    debug x82}''')[0])83    assert peek_vm.get_value('y') == 384    peek_vm.context['debug'].assert_has_calls(85        [call(peek_vm.app, 1), call(peek_vm.app, 2), call(peek_vm.app, 3)])86def test_payload_file(peek_vm, parser):87    peek_vm.execute_node(parser.parse('''let data = {88    "category": "click",89    "tags": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],90}''')[0])91    payload_file = os.path.join(os.path.dirname(__file__), 'payload.json')92    peek_vm.execute_node(parser.parse('''PUT _bulk93@{}'''.format(payload_file))[0])94    peek_vm.app.es_client_manager.current.perform_request.assert_called_with(95        'PUT',96        '/_bulk',97        '{"index": {"_index": "index", "_id": "1"}}\n'98        '{"category": "click", "tag": 1}\n'99        '{"index": {"_index": "index", "_id": "2"}}\n'100        '{"category": "click", "tag": 2}\n'101        '{"index": {"_index": "index", "_id": "3"}}\n'102        '{"category": "click", "tag": 3}\n'103        '{"index": {"_index": "index", "_id": "4"}}\n'104        '{"category": "click", "tag": 4}\n'105        '{"index": {"_index": "index", "_id": "5"}}\n'106        '{"category": "click", "tag": 5}\n'107        '{"index": {"_index": "index", "_id": "6"}}\n'108        '{"category": "click", "tag": 6}\n'109        '{"index": {"_index": "index", "_id": "7"}}\n'110        '{"category": "click", "tag": 7}\n'111        '{"index": {"_index": "index", "_id": "8"}}\n'112        '{"category": "click", "tag": 8}\n'113        '{"index": {"_index": "index", "_id": "9"}}\n'114        '{"category": "click", "tag": 9}\n'115        '{"index": {"_index": "index", "_id": "10"}}\n'116        '{"category": "click", "tag": 10}\n',117        headers=None,118    )119def test_warning_header(peek_vm, parser):120    import elasticsearch121    if elasticsearch.__version__ < (7, 7, 0):122        return123    import warnings124    from elasticsearch.exceptions import ElasticsearchDeprecationWarning125    peek_vm.app.display.warn = MagicMock(return_value=None)126    es_client = peek_vm.app.es_client_manager.get_client()127    message = 'This is a warning message'128    def perform_request(*args, **kwargs):129        warnings.warn(message, ElasticsearchDeprecationWarning)130    es_client.perform_request = MagicMock(side_effect=perform_request)131    peek_vm.execute_node(parser.parse('GET /')[0])132    peek_vm.app.display.warn.assert_called_with(message)133def test_maybe_encode_date_math():134    assert _maybe_encode_date_math('/<my-index-{now/d}>') == '/%3Cmy-index-%7Bnow%2Fd%7D%3E'135    assert (_maybe_encode_date_math(136        '/<logstash-{now/d-2d}>,<logstash-{now/d-1d}>,<logstash-{now/d}>/_search') ==...

ast_parse.py

Source:ast_parse.py

...52            class_hierarchy.extend([i.__name__ for i in subclasses])53            for subclass in subclasses:54                self.get_class_hierarchy(subclass, class_hierarchy=class_hierarchy)55        return class_hierarchy56    def execute_node(self, node, namespace):        57        try:58            if not node in self.index:59                self.index[node]= {'executed':False}60            if not self.index[node]['executed']:61                exec(node.dumps(), namespace)62                self.index[node]['executed'] = True63                self.src_executed += node.dumps() + '\n'64        except NameError as err:65            var_name = re.search(r"(?<=').*(?=')", err.args[0]).group()66            if var_name in self.all_assigns_dict:67                for dependency in self.all_assigns_dict[var_name]:68                    if dependency.absolute_bounding_box.top_left.line < node.absolute_bounding_box.top_left.line:69                        self.execute_node(dependency, namespace)70                    else:71                        assert True72            else:73                assignment_deep_search = self.all_assigns.find("name", var_name)74                if assignment_deep_search:75                    self.execute_node(assignment_deep_search.parent, namespace)76            self.execute_node(node, namespace)77    def skip_node(self, node):78        pass79    def check_list_node_for_types(self, list_node, acceptable_types):80        types = set(acceptable_types)81        for i in list_node.value:82            if not i.type in types:83                return False84        return True85    def execute_ast(self, namespace):86        for i in self.fst:87            if i.find(['import','from_import','dotted_as_name','name_as_name']):88                self.execute_node(i, namespace)89            elif i.find('def'):90                if i.find('name', self.psyneulink_calls):91                    self.psyneulink_calls.append(i.name)92                self.execute_node(i, namespace)93            elif i.find('assign') or i.find('call'):94                acceptable_types = ['int', 'float', 'binary',95                                    'string', 'raw_string',96                                    'binary_string','string_chain']97                if hasattr(i.value, "type") and i.value.type in acceptable_types or \98                        hasattr(i.value, "type") and i.value.type == 'list' and \99                        self.check_list_node_for_types(i,acceptable_types) or \100                        i.find('name', self.psyneulink_calls):101                    self.execute_node(i, namespace)102            elif i.find('call'):103                if i.find('name',self.psyneulink_calls):104                    self.execute_node(i, namespace)105        gdict = self.fst.find('assign',lambda x: x.find('name','pnlv_graphics_spec'))106        if gdict:107            self.execute_node(gdict, namespace)108        else:...

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.