Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.