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.