Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use test_output_table method in stestr
Best Python code snippet using stestr_python
test_execution_handler.py
Source:test_execution_handler.py 
1import os2import sys3import pytest4from typing import List5from datetime import datetime, timezone6sys.path.append(os.getcwd())7from unittest import mock8from bin.env_utils import ExecutionEnvironment, get_env9from seamster.dbt.model import DBTModel10from seamster.metadata.reader import StarscreamDataSourceType11from seamster.metadata.trino.execution_handler import TrinoExecutionHandler, ExecutionStatus, ExecutionHandlerResult, ModelPublisherResult, ModelPublisherStatus12from seamster.metadata.metadata_writer import MetadataWriterInput13from seamster.metadata.engine_table import EngineTableInput, EngineTableOutput, EngineInput14from seamster.metadata.utils import SpanResult15from seamster.dbt.profile import DBTProfile16from seamster.dbt.profile_config import DBTProfileConfig17from seamster.metadata.airflow_xcom_writer import AirflowXcomWriter18TEST_SOURCES_DIR = 'seamster/tests/fixtures/sources'19TEST_MODELS_DIR = 'seamster/tests/fixtures/models'20@mock.patch('seamster.dbt.model.MODELS_DIR', TEST_MODELS_DIR)21@mock.patch('seamster.dbt.model_attrs.MODELS_DIR', TEST_MODELS_DIR)22@mock.patch('seamster.dbt.source.SOURCES_DIR', TEST_SOURCES_DIR) 23@mock.patch.object(DBTProfileConfig, 'ENVIRONMENT_PROFILE_PATH_MAP', 24 {ExecutionEnvironment.development: 'profiles/development/profiles.yml' if get_env() == ExecutionEnvironment.development else '/app/profiles/development/profiles.yml',25 ExecutionEnvironment.production: 'profiles/production/profiles.yml' if get_env() == ExecutionEnvironment.development else '/app/profiles/production/profiles.yml',26 ExecutionEnvironment.ci: 'profiles/ci/profiles.yml' if get_env() == ExecutionEnvironment.development else '/app/profiles/ci/profiles.yml'27 })28@mock.patch('seamster.metadata.trino.execution_handler.get_user_id', lambda: 'test_user_id')29@mock.patch('seamster.metadata.trino.execution_handler.uuid4', lambda: 'test-uuid')30@mock.patch('seamster.metadata.trino.execution_handler.get_sha', lambda: 'abc123')31class TestTrinoExecutionHandler(object):32 @pytest.fixture33 def test_engine_table(self) -> EngineTableInput:34 return EngineTableInput(35 catalog_name='scratch',36 dataset_name='test_dataset',37 table_name='from_starscream__test_trino_table',38 exclude_from_span=False,39 engine_name=DBTProfile.Trino,40 environment=ExecutionEnvironment.production41 )42 @pytest.mark.parametrize('model_run_arg, environment, output_table, previous_output_table, expected_table_inputs',43 [44 ('test_trino_table',45 ExecutionEnvironment.development, 46 EngineTableOutput('seamster_adhoc', 'seamster_dev_test_user_id', 'test_trino_table_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.development),47 EngineTableOutput('seamster_adhoc', 'seamster_dev_test_user_id', 'test_trino_table', DBTProfile.Trino, ExecutionEnvironment.development),48 [EngineTableInput('scratch','test_dataset','from_starscream__test_trino_table',DBTProfile.Trino,ExecutionEnvironment.production,False)]),49 ('test_trino_table',50 ExecutionEnvironment.production,51 EngineTableOutput('seamster', 'seamster_base_test_project', 'test_trino_table_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.production),52 EngineTableOutput('seamster', 'seamster_base_test_project', 'test_trino_table', DBTProfile.Trino, ExecutionEnvironment.production),53 [EngineTableInput('scratch','test_dataset','from_starscream__test_trino_table',DBTProfile.Trino,ExecutionEnvironment.production,False)]),54 ('test_model_ref_trino_model', 55 ExecutionEnvironment.production,56 EngineTableOutput('seamster', 'seamster_backroom_test_project', 'test_model_ref_trino_model_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.production),57 EngineTableOutput('seamster', 'seamster_backroom_test_project', 'test_model_ref_trino_model', DBTProfile.Trino, ExecutionEnvironment.production),58 [EngineTableInput('seamster', 'seamster_backroom_test_project', 'test_trino_model', DBTProfile.Trino, ExecutionEnvironment.production,False)]),59 ('test_trino_table_sensitive',60 ExecutionEnvironment.production,61 EngineTableOutput('seamster_sensitive', 'seamster_base_test_project', 'test_trino_table_sensitive_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.production),62 EngineTableOutput('seamster_sensitive', 'seamster_base_test_project', 'test_trino_table_sensitive', DBTProfile.Trino, ExecutionEnvironment.production),63 [EngineTableInput('scratch','test_dataset','from_starscream__test_trino_table', DBTProfile.Trino, ExecutionEnvironment.production,False)]),64 ('test_trino_table_sensitive',65 ExecutionEnvironment.development,66 EngineTableOutput('seamster_adhoc', 'seamster_dev_test_user_id', 'test_trino_table_sensitive_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.development),67 EngineTableOutput('seamster_adhoc', 'seamster_dev_test_user_id', 'test_trino_table_sensitive', DBTProfile.Trino, ExecutionEnvironment.development),68 [EngineTableInput('scratch','test_dataset','from_starscream__test_trino_table', DBTProfile.Trino, ExecutionEnvironment.production,False)])])69 @mock.patch.object(EngineInput, 'from_name')70 def test_init_inputs_and_outputs(self, mock_from_name: mock.MagicMock, model_run_arg: str, environment: ExecutionEnvironment, output_table: EngineTableOutput, 71 previous_output_table: EngineTableOutput, test_engine_table: EngineTableInput, expected_table_inputs: List[EngineTableInput]) -> None:72 mock_from_name.return_value = test_engine_table73 actual = TrinoExecutionHandler(model_run_arg=model_run_arg, 74 run_vars={}, 75 environment=environment)76 assert actual.output == output_table77 assert actual.previous_output == previous_output_table78 assert actual.inputs == expected_table_inputs79 @mock.patch('seamster.metadata.trino.execution_handler.get_utc_now_with_tzinfo', lambda: datetime(2021, 12, 12, 12, 12, 12, 12, timezone.utc))80 @mock.patch('seamster.metadata.trino.execution_handler.get_unix_timestamp', lambda: 1639311132)81 @mock.patch('seamster.metadata.trino.execution_handler.OverseerClient')82 @mock.patch('seamster.metadata.trino.execution_handler.DBT.run')83 @mock.patch('seamster.metadata.trino.execution_handler.MetadataHandler')84 @mock.patch('seamster.metadata.trino.execution_handler.TrinoModelPublisher')85 @mock.patch.object(EngineInput, 'from_name')86 @mock.patch.object(AirflowXcomWriter, 'write_skip_run_results')87 def test_run_development(self, mock_xcom_writer: mock.MagicMock, mock_from_name: mock.MagicMock, mock_model_publisher: mock.MagicMock, mock_metadata_handler: mock.MagicMock, 88 mock_dbt_run_result: mock.MagicMock, mock_overseer_client: mock.MagicMock, test_input_thresholds_result: SpanResult) -> None:89 mock_metadata_handler.return_value.calculate_span.return_value = test_input_thresholds_result90 mock_dbt_run_result.return_value.return_code = 091 mock_metadata_handler.return_value.verify_and_write_metadata.return_value = True92 mock_model_publisher.return_value.publish.return_value = ModelPublisherResult(93 status=ModelPublisherStatus.SUCCESS,94 published_path='seamster_dev_test_user_id/test_trino_table_1639311132/20221118195959',95 published_table_fqn='seamster_adhoc.seamster_dev_test_user_id.test_trino_table'96 )97 test_output_table = EngineTableOutput('seamster_adhoc', 'seamster_dev_test_user_id', 'test_trino_table_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.development)98 execution_handler = TrinoExecutionHandler(99 model_run_arg='test_trino_table',100 run_vars={'scratch': True, 'target-node': 'test_trino_table'},101 environment=ExecutionEnvironment.development102 )103 execution_result = execution_handler.run()104 assert mock_metadata_handler.return_value.calculate_span.called105 mock_dbt_run_result.assert_called_once_with(106 model='test_trino_table',107 vars={108 'trino-model-external-location': 'temporary/temporary_test_uuid/seamster_dev_test_user_id/test_trino_table_1639311132/20221118195959',109 'custom-suffix': '_temporary_test_uuid', 110 'target-node': 'test_trino_table',111 'scratch': True,112 'custom-schema': 'test_user_id'113 },114 flags=[],115 capture_stdout=True,116 suppress_exception=False,117 profile=DBTProfile.Trino.value,118 tags=[DBTProfile.Trino.value]119 )120 mock_metadata_handler.return_value.verify_and_write_metadata.assert_called_once_with(121 MetadataWriterInput(122 output_table=test_output_table,123 span_result=test_input_thresholds_result,124 path='temporary/temporary_test_uuid/seamster_dev_test_user_id/test_trino_table_1639311132/20221118195959',125 source='seamster',126 processed_at=datetime(2021, 12, 12, 12, 12, 12, 12, timezone.utc),127 process_ended_at=datetime(2021, 12, 12, 12, 12, 12, 12, timezone.utc),128 git_sha='abc123',129 format='parquet',130 job_name='seamster_adhoc.seamster_dev_test_user_id.test_trino_table_temporary_test_uuid',131 datasource_type=StarscreamDataSourceType.DropDataSource,132 overseer_uuid='',133 is_sensitive=False134 )135 )136 mock_model_publisher.return_value.publish.assert_called_once_with(staging_table=test_output_table, output_path='temporary/temporary_test_uuid/seamster_dev_test_user_id/test_trino_table_1639311132/20221118195959')137 mock_overseer_client.return_value.assert_not_called138 assert execution_result == ExecutionHandlerResult(139 execution_status=ExecutionStatus.SUCCESS,140 published_table_fqn='seamster_adhoc.seamster_dev_test_user_id.test_trino_table',141 published_output_path='seamster_dev_test_user_id/test_trino_table_1639311132/20221118195959'142 )143 @mock.patch('seamster.metadata.trino.execution_handler.get_utc_now_with_tzinfo', lambda: datetime(2021, 12, 12, 12, 12, 12, 12, timezone.utc))144 @mock.patch('seamster.metadata.trino.execution_handler.get_unix_timestamp', lambda: 1639311132)145 @mock.patch('seamster.metadata.trino.execution_handler.OverseerClient')146 @mock.patch('seamster.metadata.trino.execution_handler.DBT.run')147 @mock.patch('seamster.metadata.trino.execution_handler.MetadataHandler')148 @mock.patch('seamster.metadata.trino.execution_handler.TrinoModelPublisher')149 @mock.patch.object(EngineInput, 'from_name')150 @mock.patch.object(AirflowXcomWriter, 'write_skip_run_results')151 def test_run_production(self, mock_xcom_writer: mock.MagicMock, mock_from_name: mock.MagicMock, mock_model_publisher: mock.MagicMock, mock_metadata_handler: mock.MagicMock, 152 mock_dbt_run_result: mock.MagicMock, mock_overseer_client: mock.MagicMock, test_input_thresholds_result: SpanResult) -> None:153 mock_overseer_client.return_value.notify_job_start.return_value='123'154 mock_metadata_handler.return_value.calculate_span.return_value = test_input_thresholds_result155 mock_dbt_run_result.return_value.return_code = 0156 mock_metadata_handler.return_value.verify_and_write_metadata.return_value = True157 mock_model_publisher.return_value.publish.return_value = ModelPublisherResult(158 status=ModelPublisherStatus.SUCCESS,159 published_path='seamster_base_test_project/test_trino_table/20221118195959',160 published_table_fqn='seamster.seamster_base_test_project.test_trino_table'161 )162 test_output_table = EngineTableOutput('seamster', 'seamster_base_test_project', 'test_trino_table_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.production)163 execution_handler = TrinoExecutionHandler(164 model_run_arg='test_trino_table',165 run_vars={},166 environment=ExecutionEnvironment.production167 )168 execution_result = execution_handler.run()169 assert mock_metadata_handler.return_value.calculate_span.called170 mock_dbt_run_result.assert_called_once_with(171 model='test_trino_table',172 vars={173 'trino-model-external-location': 'temporary/temporary_test_uuid/seamster_base_test_project/test_trino_table/20221118195959',174 'custom-suffix': '_temporary_test_uuid', 175 'target-node': 'test_trino_table'176 },177 flags=[],178 capture_stdout=True,179 suppress_exception=False,180 profile=DBTProfile.Trino.value,181 tags=[DBTProfile.Trino.value]182 )183 mock_metadata_handler.return_value.verify_and_write_metadata.assert_called_once_with(184 MetadataWriterInput(185 output_table=test_output_table,186 span_result=test_input_thresholds_result,187 path='temporary/temporary_test_uuid/seamster_base_test_project/test_trino_table/20221118195959',188 source='seamster',189 processed_at=datetime(2021, 12, 12, 12, 12, 12, 12, timezone.utc),190 process_ended_at=datetime(2021, 12, 12, 12, 12, 12, 12, timezone.utc),191 git_sha='abc123',192 format='parquet',193 job_name='seamster.seamster_base_test_project.test_trino_table_temporary_test_uuid',194 datasource_type=StarscreamDataSourceType.DropDataSource,195 overseer_uuid='123',196 is_sensitive=False197 )198 )199 mock_model_publisher.return_value.publish.assert_called_once_with(staging_table=test_output_table, output_path='temporary/temporary_test_uuid/seamster_base_test_project/test_trino_table/20221118195959')200 test_model = DBTModel('test_trino_table')201 mock_overseer_client.return_value.notify_job_start.assert_called_once_with(test_model)202 mock_overseer_client.return_value.notify_job_end.assert_called_once_with(execution_id='123',dbt_run_exit_code=0)203 mock_overseer_client.return_value.notify_freshness.assert_called_once_with(test_model, datetime(2022, 11, 18, 19, 59, 59, 999999))204 assert execution_result == ExecutionHandlerResult(205 execution_status=ExecutionStatus.SUCCESS,206 published_table_fqn='seamster.seamster_base_test_project.test_trino_table',207 published_output_path='seamster_base_test_project/test_trino_table/20221118195959'208 )209 @mock.patch('seamster.metadata.trino.execution_handler.get_utc_now_with_tzinfo', lambda: datetime(2021, 12, 12, 12, 12, 12, 12, timezone.utc))210 @mock.patch('seamster.metadata.trino.execution_handler.get_unix_timestamp', lambda: 1639311132)211 @mock.patch('seamster.metadata.trino.execution_handler.OverseerClient')212 @mock.patch('seamster.metadata.trino.execution_handler.DBT.run')213 @mock.patch('seamster.metadata.trino.execution_handler.MetadataHandler')214 @mock.patch('seamster.metadata.trino.execution_handler.TrinoModelPublisher')215 @mock.patch.object(EngineInput, 'from_name')216 @mock.patch.object(AirflowXcomWriter, 'write_skip_run_results')217 def test_run_production_sensitive(self, mock_xcom_writer: mock.MagicMock, mock_from_name: mock.MagicMock, mock_model_publisher: mock.MagicMock, mock_metadata_handler: mock.MagicMock, 218 mock_dbt_run_result: mock.MagicMock, mock_overseer_client: mock.MagicMock, test_input_thresholds_result: SpanResult) -> None:219 mock_overseer_client.return_value.notify_job_start.return_value='123'220 mock_metadata_handler.return_value.calculate_span.return_value = test_input_thresholds_result221 mock_dbt_run_result.return_value.return_code = 0222 mock_metadata_handler.return_value.verify_and_write_metadata.return_value = True223 mock_model_publisher.return_value.publish.return_value = ModelPublisherResult(224 status=ModelPublisherStatus.SUCCESS,225 published_path='seamster_base_test_project/test_trino_table_sensitive/20221118195959',226 published_table_fqn='seamster_sensitive.seamster_base_test_project.test_trino_table_sensitive'227 )228 test_output_table = EngineTableOutput('seamster_sensitive', 'seamster_base_test_project', 'test_trino_table_sensitive_temporary_test_uuid', DBTProfile.Trino, ExecutionEnvironment.production)229 execution_handler = TrinoExecutionHandler(230 model_run_arg='test_trino_table_sensitive',231 run_vars={},232 environment=ExecutionEnvironment.production233 )234 execution_result = execution_handler.run()235 assert mock_metadata_handler.return_value.calculate_span.called236 mock_dbt_run_result.assert_called_once_with(237 model='test_trino_table_sensitive',238 vars={239 'trino-model-external-location': 'temporary/temporary_test_uuid/seamster_base_test_project/test_trino_table_sensitive/20221118195959',240 'custom-suffix': '_temporary_test_uuid', 241 'target-node': 'test_trino_table_sensitive',242 'trino-production-external-gcs-bucket': 'gs://seamster-sensitive/',243 'custom_database_name': 'seamster_sensitive'244 },245 flags=[],246 capture_stdout=True,247 suppress_exception=False,248 profile=DBTProfile.Trino.value,249 tags=[DBTProfile.Trino.value]250 )251 assert mock_metadata_handler.return_value.verify_and_write_metadata.mock_calls[0].args[0].is_sensitive252 mock_model_publisher.return_value.publish.assert_called_once_with(staging_table=test_output_table, output_path='temporary/temporary_test_uuid/seamster_base_test_project/test_trino_table_sensitive/20221118195959')253 test_model = DBTModel('test_trino_table_sensitive')254 mock_overseer_client.return_value.notify_job_start.assert_called_once_with(test_model)255 mock_overseer_client.return_value.notify_job_end.assert_called_once_with(execution_id='123',dbt_run_exit_code=0)256 mock_overseer_client.return_value.notify_freshness.assert_called_once_with(test_model, datetime(2022, 11, 18, 19, 59, 59, 999999))257 assert execution_result == ExecutionHandlerResult(258 execution_status=ExecutionStatus.SUCCESS,259 published_table_fqn='seamster_sensitive.seamster_base_test_project.test_trino_table_sensitive',260 published_output_path='seamster_base_test_project/test_trino_table_sensitive/20221118195959'...
test_robot.py
Source:test_robot.py
1import unittest2from util import redirect_stdout3from battleground import Battleground4from robot import Robot5from searches.follow_right import FollowRight6from searches.follow_left import FollowLeft7from searches.dfs import DFS8class TestRobot(unittest.TestCase):9 small_terrain = [10 [0,0],11 [0,0]12 ]13 small_terrain_repr = """14= _ 15_ _ 16"""17 empty_terrain = [18 [0,0,0,0,0,0,0],19 [0,0,0,0,0,0,0],20 [0,0,0,0,0,0,0],21 [0,0,0,0,0,0,0],22 [0,0,0,0,0,0,0],23 [0,0,0,0,0,0,0],24 [0,0,0,0,0,0,0]25 ]26 empty_terrain_repr = """27= _ _ _ _ _ _ 28_ _ _ _ _ _ _ 29_ _ _ _ _ _ _ 30_ _ _ _ _ _ _ 31_ _ _ _ _ _ _ 32_ _ _ _ _ _ _ 33_ _ _ _ _ _ _ 34"""35 # Currently failing to follow_side36 blocked_terrain = [37 [0,0,0,0,0,0,0],38 [0,0,0,0,0,0,0],39 [0,0,0,0,0,0,0],40 [3,3,3,3,3,3,3],41 [0,0,0,0,0,0,0],42 [0,0,0,0,0,0,0],43 [0,0,0,0,0,0,0],44 ]45 blocked_terrain_repr = """46= _ _ _ _ _ _ 47_ _ _ _ _ _ _ 48_ _ _ _ _ _ _ 49^ ^ ^ ^ ^ ^ ^ 50_ _ _ _ _ _ _ 51_ _ _ _ _ _ _ 52_ _ _ _ _ _ _ 53"""54 narrow_terrain = [55 [0,0,0,0,0,0,0],56 [0,3,3,3,3,3,3],57 [0,3,0,0,0,0,0],58 [0,3,0,3,3,3,0],59 [0,3,0,3,0,0,0],60 [0,3,0,3,0,3,3],61 [0,0,0,3,0,0,0]62 ]63 narrow_terrain_repr = """64= _ _ _ _ _ _ 65_ ^ ^ ^ ^ ^ ^ 66_ ^ _ _ _ _ _ 67_ ^ _ ^ ^ ^ _ 68_ ^ _ ^ _ _ _ 69_ ^ _ ^ _ ^ ^ 70_ _ _ ^ _ _ _ 71"""72 test_behaviour_table = [73 {74 "terrain": small_terrain,75 "goal": None,76 "solvable": True,77 "moves_left": 2,78 "moves_right": 279 },80 {81 "terrain": small_terrain,82 "goal": (1,0),83 "solvable": True,84 "moves_left": 3,85 "moves_right": 1,86 },87 {88 "terrain": empty_terrain,89 "goal": None,90 "solvable": True,91 "moves_left": 12,92 "moves_right": 1293 },94 {95 "terrain": empty_terrain,96 "goal": (6,0),97 "solvable": True,98 "moves_left": 18,99 "moves_right": 6,100 },101 {102 "terrain": blocked_terrain,103 "goal": None,104 "solvable": False,105 "moves_left": 16,106 "moves_right": 16,107 },108 {109 "terrain": narrow_terrain,110 "goal": None,111 "solvable": True,112 "moves_left": 36,113 "moves_right": 24,114 }115 ]116 test_output_table = [117 {118 "terrain": small_terrain,119 "repr": small_terrain_repr120 },121 {122 "terrain": empty_terrain,123 "repr": empty_terrain_repr124 },125 {126 "terrain": blocked_terrain,127 "repr": blocked_terrain_repr128 }129 ]130 def test_robot_behaviour(self):131 for entry in TestRobot.test_behaviour_table:132 if entry["goal"] is None and entry["solvable"]:133 goal = (len(entry["terrain"]) - 1, len(entry["terrain"]) - 1)134 elif entry["goal"] is None:135 goal = (0,0)136 else:137 goal = entry["goal"]138 battleground = Battleground(terrain=entry["terrain"])139 robot = Robot(battleground, search=Robot.SEARCH_DFS, goal=entry["goal"])140 self.assertEqual(robot.run(), entry["solvable"])141 self.assertEqual(goal, (robot.x, robot.y))142 battleground = Battleground(terrain=entry["terrain"])143 robot = Robot(battleground, search=Robot.SEARCH_FOLLOW_LEFT, goal=entry["goal"])144 self.assertEqual(robot.run(), entry["solvable"])145 self.assertEqual(robot.moves(), entry["moves_left"])146 self.assertEqual(goal, (robot.x, robot.y))147 battleground = Battleground(terrain=entry["terrain"])148 robot = Robot(battleground, search=Robot.SEARCH_FOLLOW_RIGHT, goal=entry["goal"])149 self.assertEqual(robot.run(), entry["solvable"])150 self.assertEqual(robot.moves(), entry["moves_right"])151 self.assertEqual(goal, (robot.x, robot.y))152 def test_robot_output(self):153 for entry in TestRobot.test_output_table:154 with redirect_stdout() as file:155 battleground = Battleground(terrain=entry["terrain"])156 robot = Robot(battleground)157 print(robot)158 with open(file, 'r') as f:159 self.assertEqual(f.read().strip(), entry["repr"].strip())160 with redirect_stdout() as file:161 battleground = Battleground(terrain=entry["terrain"])162 robot = Robot(battleground, display=Robot.DISPLAY_COLOR)163 robot.render()164 with open(file, 'r') as f:...
register_execute_test.py
Source:register_execute_test.py
1import gpudb2import os3from util import test_environment as te4import argparse5TEST_INPUT_TABLE = te.LOAN_TEST_DATA_TABLE_NAME6TEST_OUTPUT_TABLE = te.LOAN_INFERENCE_TABLE_NAME7def main(db_handle):8 proc_name = 'distributed_test'9 file_paths = ["dt_test.py", "../../kinetica_proc.py", "../../util/test_environment.py"]10 # Read proc code in as bytes and add to a file data array11 files = {}12 for script_path in file_paths:13 script_name = os.path.basename(script_path)14 with open(script_path, 'rb') as f:15 files[script_name] = f.read()16 # Remove proc if it exists from a prior registration17 if db_handle.has_proc(proc_name)['proc_exists']:18 db_handle.delete_proc(proc_name)19 print("Registering proc...")20 response = db_handle.create_proc(proc_name, 'distributed', files, 'python', [file_paths[0]], {})21 print(response)22 print("Executing proc...")23 response = db_handle.execute_proc(proc_name, {}, {}, [TEST_INPUT_TABLE], {}, [TEST_OUTPUT_TABLE], {})24 print(response)25if __name__ == "__main__":26 # Set up args27 parser = argparse.ArgumentParser(description='Register and execute the Python Distributed Model Testing UDF example.')28 parser.add_argument('--host', default='127.0.0.1', help='Kinetica host to run example against')29 parser.add_argument('--username', default='', help='Username of user to run example with')30 parser.add_argument('--password', default='', help='Password of user')31 args = parser.parse_args()...
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 stestr 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?
