Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! 
How to use _insert_obj method in pyatom
Best Python code snippet using pyatom_python
test_daos.py
Source:test_daos.py 
...13 )14 def setUp(self):15 super().setUp()16 self.dao = daos.AssignmentConfigDao(self.app.settings)17 def _insert_obj(self):18 return self.dao.insert(AssignmentConfigDaoTest.DEFAULT_OBJECT)19 def test_id_from(self):20 _id = self.dao.id_from("course", "assignment")21 self.assertEqual("course/assignment", _id)22 def test_insert(self):23 result = self._insert_obj()24 self.assertIsNotNone(result.inserted_id)25 def test_find_by_id(self):26 result = self._insert_obj()27 obj = self.dao.find_by_id(result.inserted_id)28 self.assertIsNotNone(obj)29 for var in vars(obj):30 self.assertEqual(31 getattr(obj, var), getattr(AssignmentConfigDaoTest.DEFAULT_OBJECT, var)32 )33 def test_delete(self):34 insert_result = self._insert_obj()35 delete_result = self.dao.delete_by_id(insert_result.inserted_id)36 self.assertGreater(delete_result.deleted_count, 0)37class CourseDaoTest(BaseTest):38 DEFAULT_OBJECT = models.Course(id_="course", tokens=["value"])39 def setUp(self):40 super().setUp()41 self.dao = daos.CourseDao(self.app.settings)42 def _insert_obj(self):43 return self.dao.insert_or_update(CourseDaoTest.DEFAULT_OBJECT)44 def test_insert(self):45 result = self._insert_obj()46 self.assertIsNotNone(result.upserted_id)47 def test_find_by_id(self):48 result = self._insert_obj()49 obj = self.dao.find_by_id(result.upserted_id)50 self.assertIsNotNone(obj)51 for var in vars(obj):52 self.assertEqual(53 getattr(obj, var), getattr(CourseDaoTest.DEFAULT_OBJECT, var)54 )55 def test_update(self):56 insert_result = self._insert_obj()57 obj = self.dao.find_by_id(insert_result.upserted_id)58 obj.tokens = ["new_value"]59 update_result = self.dao.insert_or_update(obj)60 self.assertGreater(update_result.matched_count, 0)61class GradingJobLogDaoTest(BaseTest):62 DEFAULT_OBJECT = models.GradingJobLog(63 job_id="job_id", stdout="output", stderr="errors"64 )65 def setUp(self):66 super().setUp()67 self.dao = daos.GradingJobLogDao(self.app.settings)68 def _insert_obj(self):69 return self.dao.insert(GradingJobLogDaoTest.DEFAULT_OBJECT)70 def test_insert(self):71 result = self._insert_obj()72 self.assertIsNotNone(result.inserted_id)73 def test_find_by_id(self):74 result = self._insert_obj()75 obj = self.dao.find_by_id(result.inserted_id)76 self.assertIsNotNone(obj)77 for var in vars(obj):78 if var == "id":79 self.assertIsNotNone(obj.id)80 else:81 self.assertEqual(82 getattr(obj, var), getattr(GradingJobLogDaoTest.DEFAULT_OBJECT, var)83 )84 def test_find_by_invalid_id(self):85 result = self.dao.find_by_id("$$$")86 self.assertIsNone(result)87class GradingJobDaoTest(BaseTest):88 DEFAULT_OBJECT = models.GradingJob(89 job_type=models.GradingJobType.STUDENT,90 run_id="run123",91 worker_id="worker123",92 queued_at=dt.datetime.utcnow(),93 success=True,94 stages=[{"image": "alpine"}],95 students=[{"STUDENT_ID": "example"}],96 )97 def setUp(self):98 super().setUp()99 self.dao = daos.GradingJobDao(self.app.settings)100 def _insert_obj(self):101 return self.dao.insert(GradingJobDaoTest.DEFAULT_OBJECT)102 def test_insert(self):103 result = self._insert_obj()104 self.assertIsNotNone(result.inserted_id)105 def test_find_by_id(self):106 result = self._insert_obj()107 obj = self.dao.find_by_id(result.inserted_id)108 self.assertIsNotNone(obj)109 for var in vars(obj):110 if var == "id":111 self.assertIsNotNone(obj.id)112 elif var == "queued_at":113 delta = GradingJobDaoTest.DEFAULT_OBJECT.queued_at - obj.queued_at114 self.assertEqual(delta.seconds, 0)115 else:116 self.assertEqual(117 getattr(obj, var), getattr(GradingJobDaoTest.DEFAULT_OBJECT, var)118 )119 def test_find_by_run_id(self):120 result = self._insert_obj()121 objs = self.dao.find_by_run_id(GradingJobDaoTest.DEFAULT_OBJECT.run_id)122 self.assertEqual(len(objs), 1)123 self.assertEqual(objs[0].id, str(result.inserted_id))124 def test_update(self):125 insert_result = self._insert_obj()126 obj = self.dao.find_by_id(insert_result.inserted_id)127 obj.success = False128 update_result = self.dao.update(obj)129 self.assertGreater(update_result.matched_count, 0)130class GradingRunDaoTest(BaseTest):131 DEFAULT_OBJECT = models.GradingRun(132 assignment_id="assignment123",133 state=models.GradingRunState.READY,134 students_env=[{"override": "1"}],135 student_jobs_left=1,136 )137 def setUp(self):138 super().setUp()139 self.dao = daos.GradingRunDao(self.app.settings)140 def _insert_obj(self):141 return self.dao.insert(GradingRunDaoTest.DEFAULT_OBJECT)142 def test_insert(self):143 result = self._insert_obj()144 self.assertIsNotNone(result.inserted_id)145 def test_find_by_id(self):146 result = self._insert_obj()147 obj = self.dao.find_by_id(result.inserted_id)148 self.assertIsNotNone(obj)149 for var in vars(obj):150 if var == "id":151 self.assertIsNotNone(obj.id)152 else:153 self.assertEqual(154 getattr(obj, var), getattr(GradingRunDaoTest.DEFAULT_OBJECT, var)155 )156 def test_update(self):157 insert_result = self._insert_obj()158 obj = self.dao.find_by_id(insert_result.inserted_id)159 obj.student_jobs_left = 0160 update_result = self.dao.update(obj)161 self.assertGreater(update_result.matched_count, 0)162class WorkerNodeDaoTest(BaseTest):163 DEFAULT_OBJECT = models.WorkerNode(id_="holygoply", hostname="example.com")164 def setUp(self):165 super().setUp()166 self.dao = daos.WorkerNodeDao(self.app.settings)167 def _insert_obj(self):168 self.dao.insert(WorkerNodeDaoTest.DEFAULT_OBJECT)169 return WorkerNodeDaoTest.DEFAULT_OBJECT.id170 def test_insert(self):171 worker_id = self._insert_obj()172 self.assertIsNotNone(worker_id)173 def test_find_by_hostname(self):174 self._insert_obj()175 obj = self.dao.find_by_hostname(WorkerNodeDaoTest.DEFAULT_OBJECT.hostname)176 self.assertIsNotNone(obj)177 def test_find_by_liveness(self):178 worker_id = self._insert_obj()179 obj_list = self.dao.find_by_liveness(alive=True)180 no_obj_list = self.dao.find_by_liveness(alive=False)181 self.assertEqual(len(obj_list), 1)182 self.assertEqual(len(no_obj_list), 0)183 self.assertEqual(obj_list[0].id, worker_id)184 def test_update(self):185 worker_id = self._insert_obj()186 obj = self.dao.find_by_id(worker_id)187 obj.is_alive = False188 update_result = self.dao.update(obj)...
ResultData.py
Source:ResultData.py
...20 cur.execute("SELECT * FROM fio_jobs WHERE run_id = ?",21 (d['global']['id'],))22 d['jobs'] = cur.fetchall()23 return d24 def _insert_obj(self, tablename, obj):25 keys = obj.keys()26 values = obj.values()27 cur = self.db.cursor()28 cmd = "INSERT INTO {} ({}) VALUES ({}".format(tablename,29 ",".join(keys),30 '?,' * len(values))31 cmd = cmd[:-1] + ')'32 cur.execute(cmd, tuple(values))33 self.db.commit()34 return cur.lastrowid35 def insert_result(self, result):36 row_id = self._insert_obj('fio_runs', result['global'])37 for job in result['jobs']:38 job['run_id'] = row_id...
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 pyatom 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?
