KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use state_file method in molecule
Best Python code snippet using molecule_python
cmd.py
Source:cmd.py 
1# -*- coding: utf-8 -*-2'''3Tests for the file state4'''5# Import python libs6import os7import textwrap8import tempfile9# Import Salt Testing libs10from salttesting.helpers import ensure_in_syspath11ensure_in_syspath('../../')12# Import salt libs13import integration14import salt.utils15STATE_DIR = os.path.join(integration.FILES, 'file', 'base')16class CMDTest(integration.ModuleCase,17 integration.SaltReturnAssertsMixIn):18 '''19 Validate the cmd state20 '''21 def test_run_simple(self):22 '''23 cmd.run24 '''25 ret = self.run_state('cmd.run', name='ls', cwd=tempfile.gettempdir())26 self.assertSaltTrueReturn(ret)27 def test_test_run_simple(self):28 '''29 cmd.run test interface30 '''31 ret = self.run_state('cmd.run', name='ls',32 cwd=tempfile.gettempdir(), test=True)33 self.assertSaltNoneReturn(ret)34 def test_run_redirect(self):35 '''36 test cmd.run with shell redirect37 '''38 state_name = 'run_redirect'39 state_filename = state_name + '.sls'40 state_file = os.path.join(STATE_DIR, state_filename)41 date_file = tempfile.mkstemp()[1]42 state_key = 'cmd_|-date > {0}_|-date > {0}_|-run'.format(date_file)43 try:44 salt.utils.fopen(state_file, 'w').write(textwrap.dedent('''\45 date > {0}:46 cmd.run47 '''.format(date_file)))48 ret = self.run_function('state.sls', [state_name])49 self.assertTrue(ret[state_key]['result'])50 finally:51 os.remove(state_file)52 os.remove(date_file)53 def test_run_unless(self):54 '''55 test cmd.run unless56 '''57 state_name = 'run_redirect'58 state_filename = state_name + '.sls'59 state_file = os.path.join(STATE_DIR, state_filename)60 unless_file = tempfile.mkstemp()[1]61 state_key = 'cmd_|-/var/log/messages_|-/var/log/messages_|-run'62 try:63 salt.utils.fopen(state_file, 'w').write(textwrap.dedent('''\64 /var/log/messages:65 cmd.run:66 - unless: echo cheese > {0}67 '''.format(unless_file)))68 ret = self.run_function('state.sls', [state_name])69 self.assertTrue(ret[state_key]['result'])70 finally:71 os.remove(state_file)72 os.remove(unless_file)73 def test_run_creates_exists(self):74 '''75 test cmd.run creates already there76 '''77 state_name = 'run_redirect'78 state_filename = state_name + '.sls'79 state_file = os.path.join(STATE_DIR, state_filename)80 creates_file = tempfile.mkstemp()[1]81 state_key = 'cmd_|-touch {0}_|-touch {0}_|-run'.format(creates_file)82 try:83 salt.utils.fopen(state_file, 'w').write(textwrap.dedent('''\84 touch {0}:85 cmd.run:86 - creates: {0}87 '''.format(creates_file)))88 ret = self.run_function('state.sls', [state_name])89 self.assertTrue(ret[state_key]['result'])90 self.assertEqual(len(ret[state_key]['changes']), 0)91 finally:92 os.remove(state_file)93 os.remove(creates_file)94 def test_run_creates_new(self):95 '''96 test cmd.run creates not there97 '''98 state_name = 'run_redirect'99 state_filename = state_name + '.sls'100 state_file = os.path.join(STATE_DIR, state_filename)101 creates_file = tempfile.mkstemp()[1]102 os.remove(creates_file)103 state_key = 'cmd_|-touch {0}_|-touch {0}_|-run'.format(creates_file)104 try:105 salt.utils.fopen(state_file, 'w').write(textwrap.dedent('''\106 touch {0}:107 cmd.run:108 - creates: {0}109 '''.format(creates_file)))110 ret = self.run_function('state.sls', [state_name])111 self.assertTrue(ret[state_key]['result'])112 self.assertEqual(len(ret[state_key]['changes']), 4)113 finally:114 os.remove(state_file)115 os.remove(creates_file)116 def test_run_watch(self):117 '''118 test cmd.run watch119 '''120 state_name = 'run_watch'121 state_filename = state_name + '.sls'122 state_file = os.path.join(STATE_DIR, state_filename)123 saltines_key = 'cmd_|-saltines_|-/bin/true_|-run'124 biscuits_key = 'cmd_|-biscuits_|-echo hello_|-wait'125 try:126 salt.utils.fopen(state_file, 'w').write(textwrap.dedent('''\127 saltines:128 cmd.run:129 - name: /bin/true130 - cwd: /131 - stateful: True132 biscuits:133 cmd.wait:134 - name: echo hello135 - cwd: /136 - watch:137 - cmd: saltines138 '''))139 ret = self.run_function('state.sls', [state_name])140 self.assertTrue(ret[saltines_key]['result'])141 self.assertTrue(ret[biscuits_key]['result'])142 finally:143 os.remove(state_file)144if __name__ == '__main__':145 from integration import run_tests...
water_environment.py
Source:water_environment.py
1import gym2from gym import spaces3import numpy as np4from reward_machines.rm_environment import RewardMachineEnv5from envs.water.water_world import WaterWorld, WaterWorldParams, play6class WaterEnv(gym.Env):7 def __init__(self, state_file):8 params = WaterWorldParams(state_file, b_radius=15, max_x=400, max_y=400, b_num_per_color=2, use_velocities=True, ball_disappear=False)9 self.params = params10 self.action_space = spaces.Discrete(5) # noop, up, right, down, left11 self.observation_space = spaces.Box(low=-2, high=2, shape=(52,), dtype=np.float)12 self.env = WaterWorld(params)13 def get_events(self):14 return self.env.get_true_propositions()15 def step(self, action):16 self.env.execute_action(action)17 obs = self.env.get_features()18 reward = 0 # all the reward comes from the RM19 done = False20 info = {}21 return obs, reward, done, info22 def reset(self):23 self.env.reset()24 return self.env.get_features()25# MULTITASK --------------------------------------------------------------------------------------------------26class WaterRMEnv(RewardMachineEnv):27 def __init__(self, state_file):28 env = WaterEnv(state_file)29 rm_files = ["./envs/water/reward_machines/t%d.txt"%i for i in range(1,11)]30 super().__init__(env, rm_files)31 def render(self, mode='human'):32 if mode == 'human':33 play(self)34 else:35 raise NotImplementedError36class WaterRMEnvM0(WaterRMEnv):37 def __init__(self):38 state_file = "./envs/water/maps/world_0.pkl"39 super().__init__(state_file)40class WaterRMEnvM1(WaterRMEnv):41 def __init__(self):42 state_file = "./envs/water/maps/world_1.pkl"43 super().__init__(state_file)44class WaterRMEnvM2(WaterRMEnv):45 def __init__(self):46 state_file = "./envs/water/maps/world_2.pkl"47 super().__init__(state_file)48class WaterRMEnvM3(WaterRMEnv):49 def __init__(self):50 state_file = "./envs/water/maps/world_3.pkl"51 super().__init__(state_file)52class WaterRMEnvM4(WaterRMEnv):53 def __init__(self):54 state_file = "./envs/water/maps/world_4.pkl"55 super().__init__(state_file)56class WaterRMEnvM5(WaterRMEnv):57 def __init__(self):58 state_file = "./envs/water/maps/world_5.pkl"59 super().__init__(state_file)60class WaterRMEnvM6(WaterRMEnv):61 def __init__(self):62 state_file = "./envs/water/maps/world_6.pkl"63 super().__init__(state_file)64class WaterRMEnvM7(WaterRMEnv):65 def __init__(self):66 state_file = "./envs/water/maps/world_7.pkl"67 super().__init__(state_file)68class WaterRMEnvM8(WaterRMEnv):69 def __init__(self):70 state_file = "./envs/water/maps/world_8.pkl"71 super().__init__(state_file)72class WaterRMEnvM9(WaterRMEnv):73 def __init__(self):74 state_file = "./envs/water/maps/world_9.pkl"75 super().__init__(state_file)76class WaterRMEnvM10(WaterRMEnv):77 def __init__(self):78 state_file = "./envs/water/maps/world_10.pkl"79 super().__init__(state_file)80# SINGLE TASK --------------------------------------------------------------------------------------------------81class WaterRM10Env(RewardMachineEnv):82 def __init__(self, state_file):83 env = WaterEnv(state_file)84 rm_files = ["./envs/water/reward_machines/t10.txt"]85 super().__init__(env, rm_files)86 def render(self, mode='human'):87 if mode == 'human':88 play(self)89 else:90 raise NotImplementedError91class WaterRM10EnvM0(WaterRM10Env):92 def __init__(self):93 state_file = "./envs/water/maps/world_0.pkl"94 super().__init__(state_file)95class WaterRM10EnvM1(WaterRM10Env):96 def __init__(self):97 state_file = "./envs/water/maps/world_1.pkl"98 super().__init__(state_file)99class WaterRM10EnvM2(WaterRM10Env):100 def __init__(self):101 state_file = "./envs/water/maps/world_2.pkl"102 super().__init__(state_file)103class WaterRM10EnvM3(WaterRM10Env):104 def __init__(self):105 state_file = "./envs/water/maps/world_3.pkl"106 super().__init__(state_file)107class WaterRM10EnvM4(WaterRM10Env):108 def __init__(self):109 state_file = "./envs/water/maps/world_4.pkl"110 super().__init__(state_file)111class WaterRM10EnvM5(WaterRM10Env):112 def __init__(self):113 state_file = "./envs/water/maps/world_5.pkl"114 super().__init__(state_file)115class WaterRM10EnvM6(WaterRM10Env):116 def __init__(self):117 state_file = "./envs/water/maps/world_6.pkl"118 super().__init__(state_file)119class WaterRM10EnvM7(WaterRM10Env):120 def __init__(self):121 state_file = "./envs/water/maps/world_7.pkl"122 super().__init__(state_file)123class WaterRM10EnvM8(WaterRM10Env):124 def __init__(self):125 state_file = "./envs/water/maps/world_8.pkl"126 super().__init__(state_file)127class WaterRM10EnvM9(WaterRM10Env):128 def __init__(self):129 state_file = "./envs/water/maps/world_9.pkl"130 super().__init__(state_file)131class WaterRM10EnvM10(WaterRM10Env):132 def __init__(self):133 state_file = "./envs/water/maps/world_10.pkl"...
fill.py
Source:fill.py
1"""2Python script for generating .sql files from cleaned .csv files3"""4import csv5from database import mysqlconnect6def func_1():7 rows = []8 state_file = csv.reader(open("../data/states.csv", "r"))9 fields = next(state_file)10 for row in state_file:11 rows.append(row)12 query = "insert into States (code, name) values "13 values = ""14 for row in rows:15 values += f'("{row[0]}", "{row[1]}"),'16 query += values17 query = query[:len(query) - 1] + ";"18 with open("../sql/states_fill.sql", "w") as ofile:19 ofile.write("use mapm;")20 ofile.write(query)21def func_2():22 rows = []23 state_file = csv.reader(open("../data/counties.csv", "r"))24 fields = next(state_file)25 for row in state_file:26 rows.append(row)27 query = "insert into Counties (id, name, state, lat, lng) values "28 values = ""29 for row in rows:30 values += f'("{row[0]}", "{row[1]}", "{row[2]}", "{row[3]}", "{row[4]}"),'31 query += values32 query = query[:len(query) - 1] + ";"33 with open("../sql/counties_fill.sql", "w") as ofile:34 ofile.write("use mapm;")35 ofile.write(query)36def func_3():37 rows = []38 state_file = csv.reader(open("../data/real_estate/county_prices.csv", "r"))39 fields = next(state_file)40 for row in state_file:41 rows.append(row)42 query = "insert into County_real_estate (cnty, price) values "43 values = ""44 for row in rows:45 values += f'("{row[0]}", "{row[2]}"),'46 query += values47 query = query[:len(query) - 1] + ";"48 with open("../sql/counties_realty_fill.sql", "w") as ofile:49 ofile.write("use mapm;")50 ofile.write(query)51def func_4():52 rows = []53 state_file = csv.reader(open("../data/real_estate/state_prices.csv", "r"))54 fields = next(state_file)55 for row in state_file:56 rows.append(row)57 query = "insert into State_real_estate (state, price) values "58 values = ""59 for row in rows:60 values += f'("{row[0]}", "{row[1]}"),'61 query += values62 query = query[:len(query) - 1] + ";"63 with open("../sql/state_realty_fill.sql", "w") as ofile:64 ofile.write("use mapm;")65 ofile.write(query)66def func_5():67 rows = []68 state_file = csv.reader(69 open("../data/life_expectancy/county_life_expectancy.csv", "r"))70 fields = next(state_file)71 for row in state_file:72 rows.append(row)73 query = "insert into County_life_expec (cnty, age) values "74 values = ""75 for row in rows:76 values += f'("{row[0]}", "{row[3]}"),'77 query += values78 query = query[:len(query) - 1] + ";"79 with open("../sql/county_life_expec.sql", "w") as ofile:80 ofile.write("use mapm;")81 ofile.write(query)82def func_6():83 rows = []84 state_file = csv.reader(85 open("../data/life_expectancy/states_life_expectancy.csv", "r"))86 fields = next(state_file)87 for row in state_file:88 rows.append(row)89 query = "insert into State_life_expec (state, age) values "90 values = ""91 for row in rows:92 values += f'("{row[0]}", "{row[1]}"),'93 query += values94 query = query[:len(query) - 1] + ";"95 with open("../sql/state_life_expec.sql", "w") as ofile:96 ofile.write("use mapm;")97 ofile.write(query)98def func_7():99 rows = []100 state_file = csv.reader(open("../data/covid/vaccinations.csv", "r"))101 fields = next(state_file)102 for row in state_file:103 rows.append(row)104 query = "insert into State_vaccinations \105 (state, total_vaccinations, total_distributed, people_vaccinated, \106 people_fully_vaccinated_per_hundred, total_vaccinations_per_hundred, \107 people_fully_vaccinated, people_vaccinated_per_hundred, \108 distributed_per_hundred, daily_vaccinations_raw, \109 daily_vaccinations, daily_vaccinations_per_million, \110 share_doses_used) values "111 values = ""112 for row in rows:113 values += \114 f'("{row[0]}", "{row[3]}", "{row[4]}", \115 "{row[5]}", "{row[6]}", "{row[7]}", \116 "{row[8]}", "{row[9]}", "{row[10]}", \117 "{row[11]}", "{row[12]}", "{row[13]}", \118 "{row[14]}"),'119 query += values120 query = query[:len(query) - 1] + ";"121 with open("../sql/state_covid_vacc.sql", "w") as ofile:122 ofile.write("use mapm;")123 ofile.write(query)124if __name__ == "__main__":125 func_1()126 func_2()127 func_3()128 func_4()129 func_5()130 func_6()...
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 molecule 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?
