How to use fork_start method in autotest

Best Python code snippet using autotest_python

subcommand_unittest.py

Source:subcommand_unittest.py Github

copy

Full Screen

...59 self.god.stub_function(subcommand.os, 'fork')60 subcommand.os.fork.expect_call().and_return(1000)61 func = self.god.create_mock_function('func')62 cmd = _create_subcommand(func, [])63 cmd.fork_start()64 return cmd65 def test_fork_start_parent(self):66 cmd = self._setup_fork_start_parent()67 self.assertEquals(cmd.pid, 1000)68 self.god.check_playback()69 def _setup_fork_start_child(self):70 self.god.stub_function(subcommand.os, 'pipe')71 self.god.stub_function(subcommand.os, 'fork')72 self.god.stub_function(subcommand.os, 'close')73 self.god.stub_function(subcommand.os, 'write')74 self.god.stub_function(subcommand.cPickle, 'dumps')75 self.god.stub_function(subcommand.os, '_exit')76 def test_fork_start_child(self):77 self._setup_fork_start_child()78 func = self.god.create_mock_function('func')79 fork_hook = self.god.create_mock_function('fork_hook')80 join_hook = self.god.create_mock_function('join_hook')81 subcommand.subcommand.register_fork_hook(fork_hook)82 subcommand.subcommand.register_join_hook(join_hook)83 cmd = _create_subcommand(func, (1, 2))84 subcommand.os.pipe.expect_call().and_return((10, 20))85 subcommand.os.fork.expect_call().and_return(0)86 subcommand.os.close.expect_call(10)87 fork_hook.expect_call(cmd)88 func.expect_call(1, 2).and_return(True)89 subcommand.cPickle.dumps.expect_call(True,90 subcommand.cPickle.HIGHEST_PROTOCOL).and_return('True')91 subcommand.os.write.expect_call(20, 'True')92 subcommand.os.close.expect_call(20)93 join_hook.expect_call(cmd)94 subcommand.os._exit.expect_call(0)95 cmd.fork_start()96 self.god.check_playback()97 def test_fork_start_child_error(self):98 self._setup_fork_start_child()99 self.god.stub_function(subcommand.logging, 'exception')100 func = self.god.create_mock_function('func')101 cmd = _create_subcommand(func, (1, 2))102 error = Exception('some error')103 subcommand.os.pipe.expect_call().and_return((10, 20))104 subcommand.os.fork.expect_call().and_return(0)105 subcommand.os.close.expect_call(10)106 func.expect_call(1, 2).and_raises(error)107 subcommand.logging.exception.expect_call('function failed')108 subcommand.cPickle.dumps.expect_call(error,109 subcommand.cPickle.HIGHEST_PROTOCOL).and_return('error')110 subcommand.os.write.expect_call(20, 'error')111 subcommand.os.close.expect_call(20)112 subcommand.os._exit.expect_call(1)113 cmd.fork_start()114 self.god.check_playback()115 def _setup_poll(self):116 cmd = self._setup_fork_start_parent()117 self.god.stub_function(subcommand.os, 'waitpid')118 return cmd119 def test_poll_running(self):120 cmd = self._setup_poll()121 (subcommand.os.waitpid.expect_call(1000, subcommand.os.WNOHANG)122 .and_raises(subcommand.os.error('waitpid')))123 self.assertEquals(cmd.poll(), None)124 self.god.check_playback()125 def test_poll_finished_success(self):126 cmd = self._setup_poll()127 (subcommand.os.waitpid.expect_call(1000, subcommand.os.WNOHANG)128 .and_return((1000, 0)))129 self.assertEquals(cmd.poll(), 0)130 self.god.check_playback()131 def test_poll_finished_failure(self):132 cmd = self._setup_poll()133 self.god.stub_function(cmd, '_handle_exitstatus')134 (subcommand.os.waitpid.expect_call(1000, subcommand.os.WNOHANG)135 .and_return((1000, 10)))136 cmd._handle_exitstatus.expect_call(10).and_raises(Exception('fail'))137 self.assertRaises(Exception, cmd.poll)138 self.god.check_playback()139 def test_wait_success(self):140 cmd = self._setup_poll()141 (subcommand.os.waitpid.expect_call(1000, 0)142 .and_return((1000, 0)))143 self.assertEquals(cmd.wait(), 0)144 self.god.check_playback()145 def test_wait_failure(self):146 cmd = self._setup_poll()147 self.god.stub_function(cmd, '_handle_exitstatus')148 (subcommand.os.waitpid.expect_call(1000, 0)149 .and_return((1000, 10)))150 cmd._handle_exitstatus.expect_call(10).and_raises(Exception('fail'))151 self.assertRaises(Exception, cmd.wait)152 self.god.check_playback()153class real_subcommand_test(unittest.TestCase):154 """Test actually running subcommands (without mocking)."""155 def _setup_subcommand(self, func, *args):156 cmd = subcommand.subcommand(func, args)157 cmd.fork_start()158 return cmd159 def test_fork_waitfor_no_timeout(self):160 """Test fork_waitfor success with no timeout."""161 cmd = self._setup_subcommand(lambda: None)162 self.assertEquals(cmd.fork_waitfor(), 0)163 def test_fork_waitfor_timeout(self):164 """Test fork_waitfor success with a timeout."""165 cmd = self._setup_subcommand(lambda: None)166 self.assertEquals(cmd.fork_waitfor(timeout=60), 0)167 def test_fork_waitfor_exception(self):168 """Test fork_waitfor failure with an exception."""169 cmd = self._setup_subcommand(lambda: None, 'foo')170 with self.assertRaises(error.AutoservSubcommandError):171 cmd.fork_waitfor(timeout=60)...

Full Screen

Full Screen

app.py

Source:app.py Github

copy

Full Screen

1from flask import Flask, jsonify2from flask_cors import CORS3from dotenv import load_dotenv4from MySQLdb import _mysql5import os6load_dotenv()7db = _mysql.connect(host=os.getenv("DB_HOST"), user=os.getenv("DB_USER"), passwd=os.getenv("DB_PASS"), db=os.getenv("DB_NAME"), charset='utf8')8app = Flask(__name__)9CORS(app)10# Регулируемые параметры маршрутов11app_conf = { 12 # Количество записей на странице списка13 'PAGE_CAPACITY': 1014}15@app.route('/list/<int:page>', methods=['GET'])16def lst(page):17 start = (page - 1) * app_conf['PAGE_CAPACITY']18 db.query("SELECT COUNT(*) from vacancy")19 count = list(db.store_result().fetch_row(maxrows=0, how=1))[0]['COUNT(*)']20 db.query("SELECT * from vacancy, teacher WHERE vacancy.teacher_teacher_id=teacher.teacher_id LIMIT {}, {}".format(start, app_conf['PAGE_CAPACITY']))21 vacancies = db.store_result().fetch_row(maxrows=0, how=1)22 vacancies = map(lambda row: {23 'id': int(row['vacancy_id']),24 'pay': { 'start': int(row['fork_start']), 'end': None if row['fork_end'] is None else int(row['fork_end']) },25 'title': row['title'].decode('utf8'),26 'age': 1,27 'seniority': 1,28 'short_name': row['short_name'].decode('utf8'),29 'phone_number': row['phone_number'].decode('utf8'),30 'description': row['description'].decode('utf8')31 }, vacancies)32 vacancies = list(vacancies)33 return jsonify({ 'count': int(count), 'list': vacancies })34@app.route('/read/<int:id>', methods=['GET'])35def read(id):36 db.query("SELECT * from vacancy, teacher WHERE vacancy_id={} AND vacancy.teacher_teacher_id=teacher.teacher_id".format(id))37 vacancy = list(db.store_result().fetch_row(maxrows=0, how=1))[0]38 db.query("SELECT s.speciality_id, s.name from speciality AS s, teacher_has_speciality AS ths WHERE ths.teacher_teacher_id={} AND s.speciality_id = ths.speciality_speciality_id".format(int(vacancy['teacher_id']))39 specialities = db.store_result().fetch_row(maxrows=0, how=1)40 specialities = map(lambda row: [int(row['speciality_id']), row['name'].decode("utf-8")], specialities)41 specialities = list(specialities)42 db.query("SELECT s.speciality_id, s.name from speciality AS s")43 all_specialities = db.store_result().fetch_row(maxrows=0, how=1)44 all_specialities = map(lambda row: [int(row['speciality_id']), row['name'].decode("utf-8")], all_specialities)45 all_specialities = list(all_specialities)46 return jsonify({47 'title': vacancy['title'].decode('utf-8'),48 'pay': { 'start': vacancy['fork_start'].decode('utf-8'), 'end': vacancy['fork_end'].decode('utf-8') },49 'seniority': Date.now().year() - int(vacancy['date_of_birth']),50 'career_start': int(vacancy['career_start']),51 'description': vacancy['description'].decode('utf-8'),52 'phone_number': vacancy['phone_number'].decode('utf-8'),53 'email': vacancy['email'].decode('utf-8'),54 'telegram': vacancy['telegram'].decode('utf-8'),55 'full_name': vacancy['full_name'].decode('utf-8'),56 'short_name': vacancy['short_name'].decode('utf-8'),57 'age': Date.now().year() - vacancy['date_of_birth'].year(),58 'date_of_birth': vacancy['date_of_birth'].format("%Y-%m-%d"),59 'about': vacancy['about'],60 'specialities': specialities,61 'all_specialities': all_specialities,62 'teacher_id': int(vacancy['teacher_id'])63 })64@app.route('/specialities', methods=['GET'])65def specialities():66 db.query("SELECT s.speciality_id, s.name from speciality AS s")67 all_specialities = db.store_result().fetch_row(maxrows=0, how=1)68 all_specialities = list(map(lambda v: [int(v['speciality_id']), v['name'].decode("utf-8")], all_specialities))69 return jsonify({ 'all_specialities': all_specialities })70@app.route('/delete/<int:id>', methods=['DELETE'])71def delete(id):72 db.query("DELETE FROM vacancy WHERE vacancy_id={}".format(id))73 return 'Success', 20074@app.route('/create', methods=['POST'])75def create():76 payload = request.get_json()77 data = payload78 speciality_id = data['speciality_id']79 pay_start = data['pay']['start']80 pay_start = 'NULL' if len(pay_start) == 0 else pay_start 81 pay_end = data['pay']['end']82 pay_start = 'NULL' if len(pay_end) == 0 else pay_end 83 db.query("INSERT INTO teacher VALUES (DEFAULT, '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}')".format(data['full_name'], data['short_name'], data['date_of_birth'], data['about'], data['career_start'], data['email'], data['telegram']))84 teacher_id = db.store_result().last_id()85 db.query("INSERT INTO vacancy VALUES (DEFAULT, {}, {}, '{}', '{}', {}, {})".format(pay_start, pay_end, data['title'], data['description'], speciality_id, teacher_id))86 vacancy_id = db.store_result().last_id()87 db.query("INSERT INTO teacher_has_speciality VALUES ({}, {})".format(teacher_id, speciality_id))88 return jsonify({ 'id': vacancy_id })89@app.route('/update/<int:id>', methods=['PUT'])90def update(id):91 payload = request.get_json()92 data = payload93 vacancy_id = data['vacancy_id']94 pay_start = data['pay']['start']95 pay_start = 'NULL' if len(pay_start) == 0 else pay_start 96 pay_end = data['pay']['end']97 pay_start = 'NULL' if len(pay_end) == 0 else pay_end 98 db.query("UPDATE teacher SET full_name='{}', short_name='{}', date_of_birth='{}', about='{}', career_start='{}', phone_number='{}', email='{}', telegram='{}' WHERE teacher_id={}".format(data['full_name'], data['short_name'], data['date_of_birth'], data['about'], data['career_start'], data['email'], data['telegram']))99 db.query("UPDATE vacancy SET fork_start={}, fork_end={}, title='{}', description='{}', speciality_speciality_id={} WHERE vacancy_id={}".format(pay_start, pay_end, data['title'], data['description'], speciality_id, teacher_id))100 db.query("UPDATE teacher_has_speciality SET speciality_speciality_id={} WHERE teacher_teacher_id={}".format(teacher_id, speciality_id))101 return jsonify({ 'id': vacancy_id })102if __name__ == '__main__':...

Full Screen

Full Screen

call.py

Source:call.py Github

copy

Full Screen

1from telnyx.api_resources.abstract import CreateableAPIResource2from telnyx.api_resources.abstract.meta import Action, ResourceMeta3from telnyx.six import with_metaclass4class BaseCall:5 OBJECT_NAME = "call"6 reject = Action("reject")7 answer = Action("answer")8 hangup = Action("hangup")9 bridge = Action("bridge")10 speak = Action("speak")11 fork_start = Action("fork_start")12 fork_stop = Action("fork_stop")13 gather_using_audio = Action("gather_using_audio")14 gather_using_speak = Action("gather_using_speak")15 playback_start = Action("playback_start")16 playback_stop = Action("playback_stop")17 record_start = Action("record_start")18 record_stop = Action("record_stop")19 send_dtmf = Action("send_dtmf")20 transfer = Action("transfer")21class Call(with_metaclass(ResourceMeta, CreateableAPIResource, BaseCall)):...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful