KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _allocate method in stestr
Best Python code snippet using stestr_python
test_jobs.py
Source:test_jobs.py 
...200 alloc = [1]201 j = jobs.Job("1", "w", 1, prof, 0)202 j._submit(0)203 with pytest.raises(ValueError) as excinfo:204 j._allocate(alloc)205 assert "mapping" in str(excinfo.value)206 def test_allocate_staging_job_with_missing_dest_mapping_must_raise(self):207 prof = jobs.DataStagingJobProfile("p", 1, "a", "b")208 alloc = [1]209 j = jobs.Job("1", "w", 1, prof, 0)210 j._submit(0)211 with pytest.raises(ValueError) as excinfo:212 j._allocate(alloc, {"a": 1})213 assert "b" in str(excinfo.value)214 def test_allocate_staging_job_with_missing_src_mapping_must_raise(self):215 prof = jobs.DataStagingJobProfile("p", 1, "a", "b")216 alloc = [1]217 j = jobs.Job("1", "w", 1, prof, 0)218 j._submit(0)219 with pytest.raises(ValueError) as excinfo:220 j._allocate(alloc, {"b": 1})221 assert "a" in str(excinfo.value)222 def test_allocate_pfs_job_with_missing_storage_must_raise(self):223 prof = jobs.ParallelHomogeneousPFSJobProfile("p", 10, 20, "a")224 alloc = [1]225 j = jobs.Job("1", "w", 1, prof, 0)226 j._submit(0)227 with pytest.raises(ValueError) as excinfo:228 j._allocate(alloc, {"b": 1})229 assert "a" in str(excinfo.value)230 def test_allocate_pfs_job_without_mapping_must_raise(self):231 prof = jobs.ParallelHomogeneousPFSJobProfile("p", 10, 20, "a")232 alloc = [1]233 j = jobs.Job("1", "w", 1, prof, 0)234 j._submit(0)235 with pytest.raises(ValueError) as excinfo:236 j._allocate(alloc)237 assert "mapping" in str(excinfo.value)238 def test_allocate_not_storage_job_with_mapping_must_raise(self):239 prof = jobs.DelayJobProfile("p", 100)240 alloc = [1]241 j = jobs.Job("1", "w", 1, prof, 0)242 j._submit(0)243 with pytest.raises(ValueError) as excinfo:244 j._allocate(alloc, {"b": 1})245 assert "mapping" in str(excinfo.value)246 def test_allocate_valid(self):247 alloc = [1]248 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)249 j._submit(0)250 j._allocate(alloc)251 assert j.is_runnable and j.state == jobs.JobState.ALLOCATED252 assert j.allocation == alloc253 def test_allocate_cannot_be_changed(self):254 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)255 j._submit(0)256 j._allocate([1])257 with pytest.raises(RuntimeError):258 j._allocate([3])259 def test_allocate_not_submitted_job_must_raise(self):260 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)261 with pytest.raises(RuntimeError) as excinfo:262 j._allocate([3])263 assert "queue" in str(excinfo.value)264 def test_allocate_size_bigger_than_requested_must_raise(self):265 j = jobs.Job("1", "w", 3, jobs.DelayJobProfile("p", 100), 0)266 j._submit(0)267 with pytest.raises(ValueError) as excinfo:268 j._allocate([3, 2, 5, 10])269 assert "hosts" in str(excinfo.value)270 def test_allocate_size_less_than_requested_must_raise(self):271 j = jobs.Job("1", "w", 3, jobs.DelayJobProfile("p", 100), 0)272 j._submit(0)273 with pytest.raises(ValueError) as excinfo:274 j._allocate([3, 2])275 assert "hosts" in str(excinfo.value)276 def test_reject_job_not_submitted_must_raise(self):277 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)278 with pytest.raises(RuntimeError) as excinfo:279 j._reject()280 assert 'queue' in str(excinfo.value)281 def test_reject_job_submitted_valid(self):282 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)283 j._submit(0)284 j._reject()285 assert j.state == jobs.JobState.REJECTED and j.is_rejected286 def test_reject_allocated_job_must_raise(self):287 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)288 j._submit(0)289 j._allocate([1])290 with pytest.raises(RuntimeError) as excinfo:291 j._reject()292 assert 'queue' in str(excinfo.value)293 def test_reject_finished_job_must_raise(self):294 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)295 j._submit(0)296 j._allocate([1])297 j._start(0)298 j._terminate(10, jobs.JobState.COMPLETED_SUCCESSFULLY)299 with pytest.raises(RuntimeError) as excinfo:300 j._reject()301 assert 'queue' in str(excinfo.value)302 def test_submit_valid(self):303 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)304 j._submit(0)305 assert j.subtime == 0306 assert j.state == jobs.JobState.SUBMITTED and j.is_submitted307 def test_submit_submitted_job_must_raise(self):308 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)309 j._submit(0)310 with pytest.raises(RuntimeError) as excinfo:311 j._submit(0)312 assert 'submitted' in str(excinfo.value)313 def test_submit_invalid_time_must_raise(self):314 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)315 with pytest.raises(ValueError) as excinfo:316 j._submit(-1)317 assert 'subtime' in str(excinfo.value)318 def test_start_valid(self):319 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)320 j._submit(0)321 j._allocate([1])322 j._start(2)323 assert j.is_running324 assert j.state == jobs.JobState.RUNNING325 assert j.start_time == 2326 def test_start_not_allocated_job_must_raise(self):327 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)328 j._submit(0)329 with pytest.raises(RuntimeError) as excinfo:330 j._start(0)331 assert 'runnable' in str(excinfo.value)332 def test_start_finished_job_must_raise(self):333 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)334 j._submit(0)335 j._allocate([1])336 j._start(1)337 j._terminate(2, jobs.JobState.COMPLETED_FAILED)338 with pytest.raises(RuntimeError) as excinfo:339 j._start(2)340 assert 'runnable' in str(excinfo.value)341 def test_start_running_job_must_raise(self):342 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)343 j._submit(0)344 j._allocate([1])345 j._start(1)346 with pytest.raises(RuntimeError) as excinfo:347 j._start(2)348 assert 'runnable' in str(excinfo.value)349 def test_start_rejected_job_must_raise(self):350 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)351 j._submit(0)352 j._reject()353 with pytest.raises(RuntimeError) as excinfo:354 j._start(2)355 assert 'runnable' in str(excinfo.value)356 def test_start_invalid_time_must_raise(self):357 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)358 j._submit(2)359 j._allocate([1])360 with pytest.raises(ValueError) as excinfo:361 j._start(1)362 assert 'current_time' in str(excinfo.value)363 def test_terminate_with_invalid_final_state_must_raise(self):364 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)365 j._submit(0)366 j._allocate([1])367 j._start(0)368 with pytest.raises(ValueError) as excinfo:369 j._terminate(10, jobs.JobState.ALLOCATED)370 assert 'final_state' in str(excinfo.value)371 with pytest.raises(ValueError) as excinfo:372 j._terminate(10, jobs.JobState.REJECTED)373 assert 'final_state' in str(excinfo.value)374 def test_terminate_successfully_valid(self):375 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)376 j._submit(0)377 j._allocate([1])378 j._start(0)379 j._terminate(10, jobs.JobState.COMPLETED_SUCCESSFULLY)380 assert j.is_finished381 assert j.state == jobs.JobState.COMPLETED_SUCCESSFULLY382 assert j.stop_time == 10383 def test_terminate_failed_valid(self):384 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)385 j._submit(0)386 j._allocate([1])387 j._start(0)388 j._terminate(10, jobs.JobState.COMPLETED_FAILED)389 assert j.is_finished390 assert j.state == jobs.JobState.COMPLETED_FAILED391 assert j.stop_time == 10392 def test_terminate_killed_valid(self):393 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)394 j._submit(0)395 j._allocate([1])396 j._start(0)397 j._terminate(10, jobs.JobState.COMPLETED_KILLED)398 assert j.is_finished399 assert j.state == jobs.JobState.COMPLETED_KILLED400 assert j.stop_time == 10401 def test_terminate_walltime_reached_valid(self):402 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)403 j._submit(0)404 j._allocate([1])405 j._start(0)406 j._terminate(10, jobs.JobState.COMPLETED_WALLTIME_REACHED)407 assert j.is_finished408 assert j.state == jobs.JobState.COMPLETED_WALLTIME_REACHED409 assert j.stop_time == 10410 def test_terminate_finished_job_must_raise(self):411 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)412 j._submit(0)413 j._allocate([1])414 j._start(0)415 j._terminate(10, jobs.JobState.COMPLETED_SUCCESSFULLY)416 with pytest.raises(RuntimeError) as excinfo:417 j._terminate(10, jobs.JobState.COMPLETED_SUCCESSFULLY)418 assert 'running' in str(excinfo.value)419 def test_terminate_not_running_job_must_raise(self):420 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)421 j._submit(0)422 j._allocate([1])423 with pytest.raises(RuntimeError) as excinfo:424 j._terminate(10, jobs.JobState.COMPLETED_SUCCESSFULLY)425 assert 'running' in str(excinfo.value)426 def test_terminate_invalid_time_must_raise(self):427 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)428 j._submit(0)429 j._allocate([1])430 j._start(2)431 with pytest.raises(ValueError) as excinfo:432 j._terminate(1, jobs.JobState.COMPLETED_SUCCESSFULLY)433 assert 'current_time' in str(excinfo.value)434 def test_waiting_time(self):435 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)436 j._submit(0)437 j._allocate([1])438 j._start(10)439 assert j.waiting_time == 10440 def test_waiting_time_when_not_started_must_not_raise(self):441 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)442 j._submit(0)443 j._allocate([1])444 assert j.waiting_time is None445 def test_runtime(self):446 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)447 j._submit(0)448 j._allocate([1])449 j._start(0)450 j._terminate(10, jobs.JobState.COMPLETED_SUCCESSFULLY)451 assert j.runtime == 10452 def test_runtime_when_not_finished_must_not_raise(self):453 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)454 j._submit(0)455 j._allocate([1])456 j._start(0)457 assert j.runtime is None458 def test_stretch_with_walltime(self):459 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0, 100)460 j._submit(0)461 j._allocate([1])462 j._start(10)463 j._terminate(20, jobs.JobState.COMPLETED_SUCCESSFULLY)464 assert j.stretch == j.waiting_time / j.walltime465 def test_stretch_with_runtime(self):466 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)467 j._submit(0)468 j._allocate([1])469 j._start(10)470 j._terminate(30, jobs.JobState.COMPLETED_SUCCESSFULLY)471 assert j.stretch == j.waiting_time / j.runtime472 def test_turnaround_time(self):473 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)474 j._submit(0)475 j._allocate([1])476 j._start(10)477 j._terminate(30, jobs.JobState.COMPLETED_SUCCESSFULLY)478 assert j.turnaround_time == j.waiting_time + j.runtime479 def test_turnaround_time_when_not_finished_must_not_raise(self):480 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)481 j._submit(0)482 j._allocate([1])483 j._start(10)484 assert j.turnaround_time is None485 def test_per_processor_slowdown(self):486 j = jobs.Job("1", "w", 2, jobs.DelayJobProfile("p", 100), 0)487 j._submit(0)488 j._allocate([1, 2])489 j._start(10)490 j._terminate(30, jobs.JobState.COMPLETED_SUCCESSFULLY)491 assert j.per_processor_slowdown == max(492 1, j.turnaround_time / (j.res * j.runtime))493 def test_per_processor_slowdown_when_not_finished_must_not_raise(self):494 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)495 j._submit(0)496 j._allocate([1])497 j._start(10)498 assert j.per_processor_slowdown is None499 def test_slowdown(self):500 j = jobs.Job("1", "w", 2, jobs.DelayJobProfile("p", 100), 0)501 j._submit(0)502 j._allocate([1, 2])503 j._start(10)504 j._terminate(30, jobs.JobState.COMPLETED_SUCCESSFULLY)505 assert j.slowdown == max(1, j.turnaround_time / j.runtime)506 def test_slowdown_when_not_finished_must_not_raise(self):507 j = jobs.Job("1", "w", 1, jobs.DelayJobProfile("p", 100), 0)508 j._submit(0)509 j._allocate([1])510 j._start(10)...
SQLiteOperate.py
Source:SQLiteOperate.py
1# -*- coding: utf-8 -*-2import sqlite33"""4setTable5 Type: INTEGER, FLOAT, REAL, NUMERIC, BOOLEAN, TIME, DATE, TIMESTAMP, VARCHAR, NVARCHAR, TEXT, BLOB6 Option: PRIMARY KEY, AUTOINCREMENT, NOT NULL, UNIQUE7"""8class SQLiteOperate:9 def __init__(self):10 self.conn = ''11 def connectOrCreatDB(self, _name):12 self.conn = sqlite3.connect(_name + '.sqlite')13 def newTable(self, _name, _setTable):14 _SQL = ''15 for _setTableValue in _setTable:16 if _setTableValue['Default']:17 _setTableValue['Default'] = 'DEFAULT ' + _setTableValue['Default']18 _SQL += _setTableValue['name'] + ' ' + _setTableValue['Type'] + ' ' + _setTableValue['Default'] + ' ' + _setTableValue['Option']19 _SQL += ','20 _SQL = _SQL.rstrip(',')21 self.conn.execute('''CREATE TABLE ''' + _name + '''(''' + _SQL + ''');''')22 def addData(self, _tableName, _fields='', _value=''):23 if _fields:24 _SQL = "INSERT INTO " + _tableName + " (" + _fields + ") VALUES (" + _value + ")"25 else:26 _SQL = "INSERT INTO " + _tableName + " VALUES (" + _value + ")"27 # print('addData SQL = ' + _SQL)28 self.conn.execute(_SQL)29 self.conn.commit()30 def getData(self, _tableName, _fields, _allocate=''):31 # print('_allocate= ' + _allocate)32 _SQL = "SELECT " + _fields + " from " + _tableName33 if _allocate:34 if _allocate.find('LIKE') < 0 and _allocate.find('ORDER BY') >= 0:35 _SQL += _allocate36 else:37 _SQL += " where " + _allocate38 # print('getData= ' + _SQL)39 return self.conn.execute(_SQL)40 def updateData(self, _tableName, _updateValue, _allocate=''):41 _SQL = "UPDATE " + _tableName + " set " + _updateValue42 if _allocate:43 _SQL += " where " + _allocate44 # print(_SQL)45 self.conn.execute(_SQL)46 self.conn.commit()47 def deleteData(self, _tableName, _allocate):48 _SQL = "DELETE from " + _tableName + " where " + _allocate49 self.conn.execute(_SQL)50 self.conn.commit()51 def getFieldName(self, _tableName):52 _SQL = "PRAGMA table_info(" + _tableName + ")"53 _cursor = self.conn.execute(_SQL)54 # _fields = []55 _fields = ''56 for _row in _cursor:57 # _fields.append(_row[1])58 _fields += _row[1] + ','59 _fields = _fields.rstrip(',')60 return _fields61 def close(self):62 self.conn.close()63if __name__ == '__main__':64 SQ3 = SQLiteOperate()65 SQ3.connectOrCreatDB('.\static\\temp\\videoDB')66 setTable = [{'name': 'ID', 'Type': 'INTEGER', 'Default': '', 'Option': 'PRIMARY KEY AUTOINCREMENT'},67 {'name': 'VideoPath', 'Type': 'TEXT', 'Default': '', 'Option': 'NOT NULL'},68 {'name': 'VideoName', 'Type': 'TEXT', 'Default': '', 'Option': 'NOT NULL'},69 {'name': 'Score', 'Type': 'INTEGER', 'Default': '0', 'Option': 'NOT NULL'},70 {'name': 'Views', 'Type': 'INTEGER', 'Default': '0', 'Option': 'NOT NULL'}]71 # SQ3.newTable('firstTry7', setTable)72 # SQ3.addData('firstTry7', 'VideoPath, VideoName, Score, Views', _str)73 # SQ3.addData('firstTry7', _value='4, "C:/", "testVideo.mp4", 95, 10')74 # cursor = SQ3.getData('firstTry7', 'ID, VideoPath, VideoName, Score, Views')75 # SQ3.updateData('firstTry7', 'VideoPath="F:/", Score=50', 'ID=2')76 # cursor = SQ3.getData('firstTry7', 'ID, VideoPath, VideoName, Score, Views', 'ID=2')77 cursor = SQ3.getData('videoList', '*', 'dbVideoName LIKE "%AOA%"')78 # cursor = SQ3.getFieldName('videoList')79 print(cursor)80 for row in cursor:81 print("ID = ", row[0])82 print("VideoPath = ", row[1])83 print("VideoName = ", row[2])84 print("Score = ", row[3])85 print("Views = ", row[4])86 # SQ3.deleteData('firstTry7', 'ID=2')...
nametoken_test.py
Source:nametoken_test.py
1from datagenius.names.nametoken import Nametoken2def test_allocate():3 n = Nametoken(["George", "G.", "Carlin"])4 n._allocate()5 assert n.fname == "George"6 assert n.mname == "G."7 assert n.lname == "Carlin"8 n = Nametoken(["Jonathan", "Strange"])9 n._allocate()10 assert n.fname == "Jonathan"11 assert n.mname is None12 assert n.lname == "Strange"13 n = Nametoken(["Heather and Rob", None, "Vandemar"])14 n._allocate()15 assert n.fname == "Heather"16 assert n.lname == "Vandemar"17 assert n.fname2 == "Rob"18 assert n.lname2 == "Vandemar"19def assign_ampersand_split():20 n = Nametoken(["Joe & Roger", None, "Jones"])21 assert n.name_list == ["Joe", None, "Jones"]22 assert n.fname2 == "Roger"23 n = Nametoken(["Penny & Alice", None, "Orville & Stabs"])24 assert n.name_list == ["Penny", "Orville"]25 assert n.fname2 == "Alice"26 assert n.lname2 == "Stabs"27 n._allocate()28 assert n.fname == "Penny"29 assert n.lname == "Orville"30 assert n.fname2 == "Alice"31 assert n.lname2 == "Stabs"32def test_assign_trailing_middle_initial():33 n = Nametoken(["George G.", "Carlin"])34 assert n.name_list == ["George", "Carlin"]35 assert n.mname == "G."36 n = Nametoken(["Mary Jo R.", "Williamson"])37 assert n.name_list == ["Mary Jo", "Williamson"]38 assert n.mname == "R."39 # Make sure mname stays in place after allocation:40 n._allocate()41 assert n.fname == "Mary Jo"42 assert n.mname == "R."43 assert n.lname == "Williamson"44 # Test single character fnames:45 n = Nametoken(["S", "Ramachandran"])46 n._allocate()47 assert n.fname == "S"...
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?
