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

How to use run_with_test_data method in Contexts

Best Python code snippet using Contexts

executor.py

Source:executor.py

...328        """æ¿æ¢æè¨ä¸çåæ°"""329        for a in asserts:330            self.replace_cls(params, a, "expected", "actually")331    @staticmethod332    async def run_with_test_data(env, data, report_id, case_id, params_pool: dict = None,333                                 request_param: dict = None, path='ä¸»case', name: str = ""):334        start_at = datetime.now()335        executor = Executor()336        result, err = await executor.run(env, case_id, params_pool, request_param, path)337        finished_at = datetime.now()338        cost = "{}s".format((finished_at - start_at).seconds)339        if err is not None:340            status = 2341        else:342            if result.get("status"):343                status = 0344            else:345                status = 1346        asserts = result.get("asserts")347        url = result.get("url")348        case_logs = result.get("logs")349        body = result.get("request_data")350        status_code = result.get("status_code")351        request_method = result.get("request_method")352        request_headers = result.get("request_headers")353        response = result.get("response")354        case_name = result.get("case_name")355        response_headers = result.get("response_headers")356        cookies = result.get("cookies")357        req = json.dumps(request_param, ensure_ascii=False)358        data[case_id].append(status)359        await TestResultDao.insert(report_id, case_id, case_name, status,360                                   case_logs, start_at, finished_at,361                                   url, body, request_method, request_headers, cost,362                                   asserts, response_headers, response,363                                   status_code, cookies, 0, req, name)364    @staticmethod365    async def run_single(env: int, data, report_id, case_id, params_pool: dict = None, path="ä¸»case"):366        test_data = await PityTestcaseDataDao.list_testcase_data_by_env(env, case_id)367        if not test_data:368            await Executor.run_with_test_data(env, data, report_id, case_id, params_pool, dict(), path,369                                              "é»è®¤æ°æ®")370            return371        await asyncio.gather(372            *(Executor.run_with_test_data(env, data, report_id, case_id, params_pool, Executor.get_dict(x.json_data),373                                          path,374                                          x.name)375              for x in test_data))376    @case_log377    def replace_body(self, req_params, body, body_type=1):378        """æ ¹æ®ä¼ å¥çæé åæ°è¿è¡åæ°æ¿æ¢"""379        if body_type != Config.BodyType.json:380            self.append("å½åè¯·æ±æ°æ®ä¸ä¸ºjson, è·³è¿æ¿æ¢")381            return body382        try:383            if body:384                data = json.loads(body)385                for k, v in req_params.items():386                    if data.get(k) is not None:...

mmt_client.py

Source:mmt_client.py

1host = "localhost"2port = 909034import sys56import socket7# your gen-py dir8sys.path.append('gen-py')910# GamaThrift files11from GamaThrift import MmtService12from GamaThrift import PlannerService13from GamaThrift import *14from GamaThrift.ttypes import *1516# Thrift files17from thrift import Thrift18from thrift.transport import TSocket19from thrift.transport import TTransport20from thrift.protocol import TBinaryProtocol21from thrift.server import TServer2223from thrift_server_class import Handler2425import time26import os2728from threading import Lock, Thread29import xmlParser30import json31import copy3233path = os.getcwd() + "\ip_config.txt"3435test_data = "@b@@r@Client@b@@r@server_group@b@@r@<ummisco.gama.network.common.CompositeGamaMessage>@n@  <unread>true</unread>@n@  <sender class='string'>Client</sender>@n@  <receivers class='string'>server_group</receivers>@n@  <contents class='string'>&lt;string&gt;1;76.3275146484;38.3087158203;0,;0.0$2;97.2351074219;87.4938964844;3,1,;0.0$3;76.3275146484;38.3087158203;0,1,;0.0$4;76.3275146484;38.3087158203;3,;0.0$5;71.1517333984;8.80432128906;1,3,;0.0$6;71.1517333984;8.80432128906;1,0,;0.0$7;71.1517333984;8.80432128906;1,0,;0.0$8;71.1517333984;8.80432128906;3,;0.0$9;97.2351074219;87.4938964844;3,;0.0$10;97.2351074219;87.4938964844;1,0,3,;0.0$+29;15.0146484375;96.1303710938;3,;0.0;19;21;4$43;29.931640625;88.8519287109;3,;0.0;22;24;4$44;77.4658203125;23.9532470703;1,;0.0;1;3;5$45;3.88488769531;2.18811035156;3,;0.0;18;20;8$1;29.1931152344;32.5958251953;0,;0.0;9;11;7$27;32.7239990234;17.2760009766;3,;0.0;4;6;4$23;64.9108886719;52.2979736328;1,;0.0;6;8;5$20;39.1021728516;32.5500488281;0,;0.0;6;8;7$19;6.80541992188;5.908203125;3,;0.0;16;18;8$36;11.0870361328;85.3179931641;3,;0.0;15;17;5$2;42.5659179688;72.3480224609;1,;0.0;22;24;2$48;11.7248535156;70.6024169922;1,;0.0;12;14;6$3;80.4016113281;66.8487548828;1,;0.0;5;7;10$14;49.2218017578;52.0141601562;3,;0.0;11;13;9$42;37.5366210938;4.58984375;0,;0.0;7;9;3$12;89.1143798828;1.57470703125;1,;0.0;13;15;10$10;41.7510986328;32.8308105469;1,;0.0;3;5;7$0;25.4058837891;36.4990234375;3,;0.0;8;10;4$9;82.6446533203;79.6661376953;1,;0.0;7;9;2$32;25.3723144531;72.8637695312;0,;0.0;19;21;6$13;2.68249511719;98.7396240234;1,;0.0;20;22;5$8;0.302124023438;11.2091064453;0,;0.0;17;19;3$33;97.6806640625;89.2700195312;3,;0.0;0;2;2$26;16.7022705078;12.1490478516;0,;0.0;13;15;3$18;33.8043212891;2.40478515625;1,;0.0;9;11;3$6;7.83996582031;64.8956298828;0,;0.0;15;17;6$16;18.2769775391;63.1225585938;1,;0.0;9;11;6$21;32.0556640625;23.0255126953;3,;0.0;16;18;9$31;53.4912109375;11.7156982422;1,;0.0;3;5;3$46;71.6247558594;81.8084716797;3,;0.0;2;4;9$4;30.4138183594;65.6829833984;0,;0.0;21;23;1$41;85.7788085938;70.6878662109;1,;0.0;10;12;2$35;9.72595214844;28.7078857422;0,;0.0;21;23;3$24;95.2789306641;69.4091796875;1,;0.0;4;6;2$30;77.7862548828;64.4287109375;3,;0.0;6;8;9$5;21.4935302734;14.4622802734;3,;0.0;20;22;9$47;17.7093505859;41.8304443359;1,;0.0;12;14;7$22;60.1928710938;80.5633544922;3,;0.0;18;20;2$28;85.3820800781;68.5943603516;0,;0.0;13;15;1$40;80.7495117188;81.9122314453;0,;0.0;1;3;10$25;61.4959716797;49.1882324219;0,;0.0;8;10;1$15;91.9097900391;36.9689941406;0,;0.0;1;3;1$11;67.9870605469;61.5661621094;1,;0.0;14;16;2$17;29.6356201172;68.1976318359;3,;0.0;14;16;4$7;34.5581054688;40.5151367188;1,;0.0;4;6;6$49;1.49536132812;91.3208007812;1,;0.0;18;20;5$39;85.6964111328;56.0699462891;3,;0.0;4;6;8$37;85.0738525391;42.4987792969;0,;0.0;4;6;1$38;17.1691894531;83.1298828125;1,;0.0;22;24;6$34;66.5374755859;58.7677001953;1,;0.0;19;21;7$&lt;/string&gt;</contents>@n@  <emissionTimeStamp>8</emissionTimeStamp>@n@</ummisco.gama.network.common.CompositeGamaMessage>\n"36run_with_test_data = False37run = True38lock = Lock()394041def waitOnPlanner_Thread(no_agents):42    try: 43        processor = MmtService.Processor(handler)44        port = 999945        with open(path, 'r') as f:46            lines = f.readlines()47        transport = TSocket.TServerSocket(lines[1].strip('\n'), port)48        tfactory = TTransport.TBufferedTransportFactory()49        pfactory = TBinaryProtocol.TBinaryProtocolFactory()50        51        server = TServer.TSimpleServer(processor, transport, tfactory, pfactory)52        53        print "Starting thrift-based server in environment on port %d" % port54        server.serve()55        print "done!"56    except Thrift.TException, tx:57        print "%s" % (tx.message)5859noag=1060handler = Handler()61waitonplanner_thread = Thread(target=waitOnPlanner_Thread, args=(noag,))62waitonplanner_thread.daemon = True63waitonplanner_thread.start()646566def ping_planner(msg):67    try:6869        with open(path, 'r') as f:70            lines = f.readlines()71        # Make socket72        transport = TSocket.TSocket(lines[3].strip('\n'), 9097)73        74        # Buffering is critical. Raw sockets are very slow75        transport = TTransport.TBufferedTransport(transport)76        77        # Wrap in a protocol78        protocol = TBinaryProtocol.TBinaryProtocol(transport)79        80        # Create a client to use the protocol encoder81        client = PlannerService.Client(protocol)82        83        # Connect!84        transport.open()85        print "trying to ping"86        client.ping()87        print msg88        89        transport.close()90        91    except Thrift.TException, tx:92        print "%s" % (tx.message)9394def receiveFromGama():95    global handler96    # Create a TCP/IP socket97    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)9899    # Bind the socket to the port100    server_address = ('130.243.73.174', 10000)101    print 'starting up on port %s' % str(server_address)102    sock.bind(server_address)103104    # Listen for incoming connections105    sock.listen(1)106    # Wait for a connection107    print 'waiting for a connection'  108    connection, client_address = sock.accept()109    110    print 'connection from %s' % str(client_address)111    while run:112        # Receive the data in small chunks113        data = connection.recv(2600).decode()114        #print(type(data))115        #print 'received "%s"' % data116        if data:117            print "Checking data from Gama"118            #print data119            msg_content, msg_timestamp = xmlParser.toXML(data)120            #print msg_content121            print msg_timestamp122            print msg_content123            if msg_content == "Mission Complete":124                # Make socket125                transport = TSocket.TSocket('localhost', 9097)126                # Buffering is critical. Raw sockets are very slow127                transport = TTransport.TBufferedTransport(transport)128                # Wrap in a protocol129                protocol = TBinaryProtocol.TBinaryProtocol(transport)130                # Create a client to use the protocol encoder131                client = PlannerService.Client(protocol)132                # Connect!133                transport.open()134                #send to planner mission complete135                if handler.currentPlan.vehicles:136                    handler.currentPlan.vehicles[0].onboardPlanner = True137                    client.send_computePlan(handler.currentPlan)138                transport.close()139                140            elif "replan" in msg_content:141                # Make socket142                transport = TSocket.TSocket('localhost', 9097)143                # Buffering is critical. Raw sockets are very slow144                transport = TTransport.TBufferedTransport(transport)145                # Wrap in a protocol146                protocol = TBinaryProtocol.TBinaryProtocol(transport)147                # Create a client to use the protocol encoder148                client = PlannerService.Client(protocol)149                # Connect!150                transport.open()151                #send to planner mission complete152                json_content = json.loads(msg_content)153                agents_alive = json_content['agents']154                completed_tasks = json_content['tasks']155                #TODO add for failed equipment156                #keep only alive agents in the request157158                ag = [x.id for x in handler.currentPlan.vehicles]159                act = [x.actionId for x in handler.currentPlan.actions]160161                dead =  [x for x in ag if not x in agents_alive]162                for d in dead:163                    for x in handler.currentPlan.vehicles:164                        if int(x.id) == int(d):165                            print handler.currentPlan.vehicles.index(x)166                            handler.currentPlan.vehicles.pop(handler.currentPlan.vehicles.index(x))167                            break168                #keep only uncompleted tasks in the request169                complete = [x for x in act if x in completed_tasks]170                for c in complete:171                    for x in handler.currentPlan.actions:172                        if int(x.actionId) == int(c):173                            handler.currentPlan.actions.pop(handler.currentPlan.actions.index(x))174                            break175176                ag = [x.id for x in handler.currentPlan.vehicles]177                print ag178                act = [x.actionId for x in handler.currentPlan.actions]179                print act 180181                print "Remaining agents: " + str(len(handler.currentPlan.vehicles))182                print "Remaining tasks: " + str(len(handler.currentPlan.actions))183                print "Ask for a new plan"184                client.send_computePlan(handler.currentPlan)185                186                transport.close()187            elif msg_content == "Ping":188                ping_planner("Hello blue monday")189190        else:191            #print 'no more data from %s' % str(client_address)192            pass193194    connection.close()195    with lock:196        mission_done = True197    print("Shutting down gama connection")198    199try:200    print("Running with test_data: ", run_with_test_data)201202    if run_with_test_data:203        thread = Thread(target = receiveFromGama)204        thread.start()205        time.sleep(15)206        #send the saved string as an initial plan to the agents207        hostGama = 'localhost'208        portGama = 10002 #The same port as used by the server209        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)210        s.connect((hostGama, portGama))211        s.sendall(test_data.encode())212        print("Just sent dummy data to GAMA agents and exit")213        run = False214        s.close()215    else:216        thread = Thread(target = receiveFromGama)217        thread.start()218        #ping_planner("Hello blue monday")219        #time.sleep(5)220        print("Ready for Gama")221        #ping_planner("Hello blue monday")222        hostGama = '130.243.73.174'223        portGama = 10002 #The same port as used by the server224        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)225        s.connect((hostGama, portGama))226        while True:227            if handler.isPlanNew:228                print "New plan - Send to Gama agents"229                230                string2send = str(handler.currentPlan)231                agents = ''232                for x in handler.currentPlan.vehicles:233                    eq = ''234                    for e in x.equipments:235                        eq += str(e.type) + ","236                    #print eq237                    agents += str(x.id) + ';' + str(x.location.latitude) + ';' + str(x.location.longitude) + ';' + eq + ';' + str(x.currentSpeed) + '$'238                tasks = ''239                actions = 0240                for x in handler.currentPlan.actions:241                    actions = actions + 1242                    eqt = ''243                    for e in x.relatedTask.requiredTypes:244                        eqt += str(e) + ','245                    tasks += str(x.actionId) + ';' + str(x.area.area[0].latitude) + ';' + str(x.area.area[0].longitude) + ';' + str(eqt) + ';' +  str(x.speed) +  ';' + str(x.startTime) + ';' + str(x.endTime) + ';' + str(x.assignedVehicleId) + '$'246                #print agents247                #print tasks248                string2send = agents + "+" + tasks249                data = "@b@@r@Client@b@@r@server_group@b@@r@<ummisco.gama.network.common.CompositeGamaMessage>@n@  <unread>true</unread>@n@  <sender class='string'>Client</sender>@n@  <receivers class='string'>server_group</receivers>@n@  <contents class='string'>&lt;string&gt;"+string2send+"&lt;/string&gt;</contents>@n@  <emissionTimeStamp>8</emissionTimeStamp>@n@</ummisco.gama.network.common.CompositeGamaMessage>\n"250                print "Below is example data"251                print "Number of actions to send %s " % str(actions)252                s.sendall(data.encode())253                handler.setPlanNewValue(False)254255                ag = [x.id for x in handler.currentPlan.vehicles]256                print ag257                act = [x.actionId for x in handler.currentPlan.actions]258                print act259except KeyboardInterrupt:260    run = False
...

core.py

Source:core.py

...162            finally:163                self.run_teardown()164    def run_setup(self):165        for setup in self.setups:166            run_with_test_data(setup, self.example)167    def run_action(self):168        run_with_test_data(self.action, self.example)169    def run_assertions(self):170        for assertion in self.assertions:171            assertion.run(self.example)172    def run_teardown(self):173        for teardown in self.teardowns:174            run_with_test_data(teardown, self.example)175def bind_methods(funcs, instance):176    return [types.MethodType(func, instance) for func in funcs]177class Assertion(object):178    def __init__(self, func, plugin_composite):179        self.func = func180        self.name = func.__name__181        self.plugin_composite = plugin_composite182        self.exception_handler = ExceptionHandler(self.plugin_composite)183    def run(self, test_data):184        with self.exception_handler.run_assertion(self):185            run_with_test_data(self.func, test_data)186def run_with_test_data(func, test_data):187    sig = inspect.signature(func)188    if test_data is not NO_EXAMPLE and sig.parameters:189        if isinstance(test_data, tuple) and len(sig.parameters) == len(test_data):190            func(*test_data)191        else:192            func(test_data)193    else:194        func()195class ExceptionHandler(object):196    def __init__(self, plugin_composite):197        self.plugin_composite = plugin_composite198    @contextmanager199    def run_test_run(self, test_run):200        self.plugin_composite.test_run_started()...

resource_monitor_unittest.py

Source:resource_monitor_unittest.py

...181    def setUp(self):182        self._res_dir = os.path.join(183                            os.path.dirname(os.path.realpath(__file__)),184                            'res_resource_monitor')185    def run_with_test_data(self, testdata_file, testans_file):186        """Parses testdata_file with the parses, and checks that results187            are the same as those in testans_file.188        @param testdata_file string filename containing top output to test.189        @param testans_file string filename containing answers to the test.190        """191        parsed_results = resource_monitor.ResourceMonitorParsedResult(192                testdata_file)193        with open(testans_file, "rb") as testans:194            csvreader = csv.reader(testans)195            columns = csvreader.next()196            self.assertEqual(list(columns),197                    resource_monitor.ResourceMonitorParsedResult._columns)198            utils_over_time = []199            for util_val in imap(200                    resource_monitor.201                            ResourceMonitorParsedResult.UtilValues._make,202                    csvreader):203                utils_over_time.append(util_val)204            self.assertEqual(utils_over_time, parsed_results._utils_over_time)205    def test_full_data(self):206        """General test with many possible changes to input."""207        self.run_with_test_data(208                os.path.join(self._res_dir, 'top_test_data.txt'),209                os.path.join(self._res_dir, 'top_test_data_ans.csv'))210    def test_whitespace_ridden(self):211        """Tests resilience to arbitrary whitespace characters between fields"""212        self.run_with_test_data(213                os.path.join(self._res_dir, 'top_whitespace_ridden.txt'),214                os.path.join(self._res_dir, 'top_whitespace_ridden_ans.csv'))215    def test_field_order_changed(self):216        """Tests resilience to changes in the order of fields217            (for e.g, if the Mem free/used fields change orders in the input).218        """219        self.run_with_test_data(220                os.path.join(self._res_dir, 'top_field_order_changed.txt'),221                os.path.join(self._res_dir, 'top_field_order_changed_ans.csv'))222if __name__ == '__main__':...

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.