KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use enqueue_job method in autotest
Best Python code snippet using autotest_python
test_action.py
Source:test_action.py 
...26from ..helpers import (assert_anonymous_access, assert_authorized, assert_equal,27 assert_no_anonymous_access, assert_no_metadata,28 assert_not_authorized, get_metadata, fake_process,29 assert_validation_fails)30def enqueue_job(*args, **kwargs):31 return mock.Mock(id=str(uuid.uuid4()))32class TestExtractorList(FunctionalTestBase):33 def test_extractor_list_empty(self):34 """35 extractor_list when no metadata exist.36 """37 assert_equal(call_action('extractor_list'), [])38 def test_extractor_list_inprogress(self):39 """40 extractor_list does not list metadata that is in progress.41 """42 factories.Resource(format='pdf')43 assert_equal(call_action('extractor_list'), [])44 def test_extractor_list_some(self):...
test_plugin.py
Source:test_plugin.py
1import pytest2import mock3from ckan.tests.helpers import call_action, reset_db4from ckan.tests import factories5from ckanext.validation.model import create_tables, tables_exist6from ckanext.validation.jobs import run_validation_job7@pytest.mark.usefixtures("clean_db", "validation_setup", "with_plugins")8class TestResourceControllerHooksUpdate(object):9 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)10 @mock.patch("ckanext.validation.logic.enqueue_job")11 def test_validation_does_not_run_on_other_fields(self, mock_enqueue):12 resource = {"format": "CSV"}13 dataset = factories.Dataset(resources=[resource])14 dataset["resources"][0]["description"] = "Some resource"15 call_action("resource_update", {}, **dataset["resources"][0])16 mock_enqueue.assert_not_called()17 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)18 @mock.patch("ckanext.validation.logic.enqueue_job")19 def test_validation_does_not_run_on_other_formats(self, mock_enqueue):20 resource = {"format": "PDF"}21 dataset = factories.Dataset(resources=[resource])22 call_action("resource_update", {}, **dataset["resources"][0])23 mock_enqueue.assert_not_called()24 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)25 @mock.patch("ckanext.validation.logic.enqueue_job")26 def test_validation_run_on_upload(self, mock_enqueue):27 resource = {"format": "CSV", "upload": "mock_upload", "url_type": "upload"}28 dataset = factories.Dataset(resources=[resource])29 call_action("resource_update", {}, **dataset["resources"][0])30 assert mock_enqueue.call_count == 131 assert mock_enqueue.call_args[0][0] == run_validation_job32 assert mock_enqueue.call_args[0][1][0]["id"] == dataset["resources"][0]["id"]33 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)34 @mock.patch("ckanext.validation.logic.enqueue_job")35 def test_validation_run_on_url_change(self, mock_enqueue):36 resource = {"format": "CSV", "url": "https://some.url"}37 dataset = factories.Dataset(resources=[resource])38 dataset["resources"][0]["url"] = "https://some.new.url"39 call_action("resource_update", {}, **dataset["resources"][0])40 assert mock_enqueue.call_count == 141 assert mock_enqueue.call_args[0][0] == run_validation_job42 assert mock_enqueue.call_args[0][1][0]["id"] == dataset["resources"][0]["id"]43 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)44 @mock.patch("ckanext.validation.logic.enqueue_job")45 def test_validation_run_on_schema_change(self, mock_enqueue):46 resource = {47 "url": "http://some.url",48 "format": "CSV",49 "schema": {"fields": [{"name": "code"}]},50 }51 dataset = factories.Dataset(resources=[resource])52 dataset["resources"][0]["schema"] = {53 "fields": [{"name": "code"}, {"name": "date"}]54 }55 call_action("resource_update", {}, **dataset["resources"][0])56 assert mock_enqueue.call_count == 157 assert mock_enqueue.call_args[0][0] == run_validation_job58 assert mock_enqueue.call_args[0][1][0]["id"] == dataset["resources"][0]["id"]59 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)60 @mock.patch("ckanext.validation.logic.enqueue_job")61 def test_validation_run_on_format_change(self, mock_enqueue):62 resource = factories.Resource()63 resource["format"] = "CSV"64 call_action("resource_update", {}, **resource)65 assert mock_enqueue.call_count == 166 assert mock_enqueue.call_args[0][0] == run_validation_job67 assert mock_enqueue.call_args[0][1][0]["id"] == resource["id"]68 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)69 @pytest.mark.ckan_config("ckanext.validation.run_on_update_async", False)70 @mock.patch("ckanext.validation.logic.enqueue_job")71 def test_validation_does_not_run_when_config_false(self, mock_enqueue):72 resource = factories.Resource(format="CSV")73 resource["url"] = "http://some.new.url"74 call_action("resource_update", {}, **resource)75 mock_enqueue.assert_not_called()76@pytest.mark.usefixtures("clean_db", "validation_setup", "with_plugins")77class TestResourceControllerHooksCreate(object):78 @mock.patch("ckanext.validation.logic.enqueue_job")79 def test_validation_does_not_run_on_other_formats(self, mock_enqueue):80 factories.Resource(format="PDF")81 mock_enqueue.assert_not_called()82 @mock.patch("ckanext.validation.logic.enqueue_job")83 @pytest.mark.ckan_config("ckanext.validation.run_on_update_async", False)84 def test_validation_run_with_upload(self, mock_enqueue):85 resource = factories.Resource(format="CSV", url_type="upload")86 assert mock_enqueue.call_count == 187 assert mock_enqueue.call_args[0][0] == run_validation_job88 assert mock_enqueue.call_args[0][1][0]["id"] == resource["id"]89 @mock.patch("ckanext.validation.logic.enqueue_job")90 @pytest.mark.ckan_config("ckanext.validation.run_on_update_async", False)91 def test_validation_run_with_url(self, mock_enqueue):92 resource = factories.Resource(format="CSV", url="http://some.data")93 assert mock_enqueue.call_count == 194 assert mock_enqueue.call_args[0][0] == run_validation_job95 assert mock_enqueue.call_args[0][1][0]["id"] == resource["id"]96 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)97 @pytest.mark.ckan_config("ckanext.validation.run_on_update_async", False)98 @mock.patch("ckanext.validation.logic.enqueue_job")99 def test_validation_does_not_run_when_config_false(self, mock_enqueue):100 dataset = factories.Dataset()101 resource = {102 "format": "CSV",103 "url": "http://some.data",104 "package_id": dataset["id"],105 }106 call_action("resource_create", {}, **resource)107 mock_enqueue.assert_not_called()108@pytest.mark.usefixtures("clean_db", "validation_setup", "with_plugins")109class TestPackageControllerHooksCreate(object):110 @mock.patch("ckanext.validation.logic.enqueue_job")111 def test_validation_does_not_run_on_other_formats(self, mock_enqueue):112 factories.Dataset(resources=[{"format": "PDF"}])113 mock_enqueue.assert_not_called()114 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)115 @mock.patch("ckanext.validation.logic.enqueue_job")116 def test_validation_does_not_run_when_config_false(self, mock_enqueue):117 factories.Dataset(resources=[{"format": "CSV", "url": "http://some.data"}])118 mock_enqueue.assert_not_called()119 @mock.patch("ckanext.validation.logic.enqueue_job")120 def test_validation_run_with_upload(self, mock_enqueue):121 resource = {"id": "test-resource-id", "format": "CSV", "url_type": "upload"}122 factories.Dataset(resources=[resource])123 assert mock_enqueue.call_count == 1124 assert mock_enqueue.call_args[0][0] == run_validation_job125 assert mock_enqueue.call_args[0][1][0]["id"] == resource["id"]126 @mock.patch("ckanext.validation.logic.enqueue_job")127 def test_validation_run_with_url(self, mock_enqueue):128 resource = {129 "id": "test-resource-id",130 "format": "CSV",131 "url": "http://some.data",132 }133 factories.Dataset(resources=[resource])134 assert mock_enqueue.call_count == 1135 assert mock_enqueue.call_args[0][0] == run_validation_job136 assert mock_enqueue.call_args[0][1][0]["id"] == resource["id"]137 @mock.patch("ckanext.validation.logic.enqueue_job")138 def test_validation_run_only_supported_formats(self, mock_enqueue):139 resource1 = {140 "id": "test-resource-id-1",141 "format": "CSV",142 "url": "http://some.data",143 }144 resource2 = {145 "id": "test-resource-id-2",146 "format": "PDF",147 "url": "http://some.doc",148 }149 factories.Dataset(resources=[resource1, resource2])150 assert mock_enqueue.call_count == 1151 assert mock_enqueue.call_args[0][0] == run_validation_job152 assert mock_enqueue.call_args[0][1][0]["id"] == resource1["id"]153@pytest.mark.usefixtures("clean_db", "validation_setup", "with_plugins")154class TestPackageControllerHooksUpdate(object):155 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)156 @mock.patch("ckanext.validation.logic.enqueue_job")157 def test_validation_runs_with_url(self, mock_enqueue):158 resource = {159 "id": "test-resource-id",160 "format": "CSV",161 "url": "http://some.data",162 }163 dataset = factories.Dataset(resources=[resource], id="myid")164 mock_enqueue.assert_not_called()165 dataset["resources"][0]["url"] = "http://some.other.data"166 call_action("package_update", {}, **dataset)167 assert mock_enqueue.call_count == 1168 assert mock_enqueue.call_args[0][0] == run_validation_job169 assert mock_enqueue.call_args[0][1][0]["id"] == resource["id"]170 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)171 @mock.patch("ckanext.validation.logic.enqueue_job")172 def test_validation_runs_with_upload(self, mock_enqueue):173 resource = {"id": "test-resource-id", "format": "CSV", "url_type": "upload"}174 dataset = factories.Dataset(resources=[resource])175 mock_enqueue.assert_not_called()176 dataset["resources"][0]["url"] = "http://some.other.data"177 call_action("package_update", {}, **dataset)178 assert mock_enqueue.call_count == 1179 assert mock_enqueue.call_args[0][0] == run_validation_job180 assert mock_enqueue.call_args[0][1][0]["id"] == resource["id"]181 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)182 @mock.patch("ckanext.validation.logic.enqueue_job")183 def test_validation_does_not_run_on_other_formats(self, mock_enqueue):184 resource = {"id": "test-resource-id", "format": "PDF", "url": "http://some.doc"}185 dataset = factories.Dataset(resources=[resource])186 mock_enqueue.assert_not_called()187 dataset["resources"][0]["url"] = "http://some.other.doc"188 call_action("package_update", {}, **dataset)189 mock_enqueue.assert_not_called()190 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)191 @mock.patch("ckanext.validation.logic.enqueue_job")192 def test_validation_run_only_supported_formats(self, mock_enqueue):193 resource1 = {194 "id": "test-resource-id-1",195 "format": "CSV",196 "url": "http://some.data",197 }198 resource2 = {199 "id": "test-resource-id-2",200 "format": "PDF",201 "url": "http://some.doc",202 }203 dataset = factories.Dataset(resources=[resource1, resource2])204 mock_enqueue.assert_not_called()205 dataset["resources"][0]["url"] = "http://some.other.data"206 call_action("package_update", {}, **dataset)207 assert mock_enqueue.call_count == 1208 assert mock_enqueue.call_args[0][0] == run_validation_job209 assert mock_enqueue.call_args[0][1][0]["id"] == resource1["id"]210 @pytest.mark.ckan_config("ckanext.validation.run_on_create_async", False)211 @pytest.mark.ckan_config("ckanext.validation.run_on_update_async", False)212 @mock.patch("ckanext.validation.logic.enqueue_job")213 def test_validation_does_not_run_when_config_false(self, mock_enqueue):214 resource = {215 "id": "test-resource-id",216 "format": "CSV",217 "url": "http://some.data",218 }219 dataset = factories.Dataset(resources=[resource])220 call_action("package_update", {}, **dataset)...
testJobHandler.py
Source:testJobHandler.py
...63 self._handler.set_job_enqueue_observer(on_enqueue_mock)64 self._handler._on_job_enqueue(self.valid_job)65 on_enqueue_mock.assert_called_once_with(self.valid_job)66 # enqueue_event67 def test_enqueue_job(self):68 self._handler.enqueue_job(self.valid_job)69 self.assertEqual(self._handler._job_deque, deque([self.valid_job]))70 def test_enqueue_multiple_jobs(self):71 self._handler.enqueue_job(self.valid_job)72 self._handler.enqueue_job(self.other_valid_job)73 self.assertEqual(self._handler._job_deque,74 deque([self.valid_job, self.other_valid_job]))75 def test_enqueue_job_runs_start_sending_events(self):76 self._handler._start_sending_events = Mock()77 self._handler.enqueue_job(self.valid_job)78 # Simulate state change that is not run due to mock function79 self._handler._state = JobHandler.State.PENDING80 self._handler.enqueue_job(self.other_valid_job)81 self._handler._start_sending_events.assert_called_once_with()82 def test_enqueue_job_raises_on_non_job(self):83 self.assertRaises(TypeError, self._handler.enqueue_job)84 self.assertRaises(TypeError, self._handler.enqueue_job, self.image_event)85 # dequeue86 def test_dequeue_job(self):87 self._handler.enqueue_job(self.valid_job)88 self.assertEqual(self._handler._job_deque,89 deque([self.valid_job]))90 self._handler._dequeue_job()91 self.assertEqual(self._handler._job_deque, deque())92 def test_dequeue_event_exception(self):93 self.assertRaises(IndexError, self._handler._dequeue_job)94 # peek95 def test_peek_current_job_when_empty(self):96 self.assertIsNone(self._handler._peek_current_event())97 def test_peek_current_job(self):98 self._handler.enqueue_job(self.valid_job)99 self.assertEqual(self._handler._peek_current_event(),100 self.valid_job.current_event)101 def test_peek_current_job_multiple(self):102 self._handler.enqueue_job(self.valid_job)103 self._handler.enqueue_job(self.other_valid_job)104 self.assertEqual(self._handler._peek_current_event(),105 self.valid_job.current_event)106 self.assertNotEqual(self._handler._peek_current_event(),107 self.other_valid_job.current_event)108 # peek_last109 def test_peek_last_added_job_when_empty(self):110 self.assertIsNone(self._handler._peek_last_added_job())111 def test_peek_last_added_job_not_empty(self):112 self._handler.enqueue_job(self.valid_job)113 self.assertEqual(self._handler._peek_last_added_job(),114 self.valid_job)115 def test_peek_last_added_event_multiple(self):116 self._handler.enqueue_job(self.valid_job)117 self._handler.enqueue_job(self.other_valid_job)118 self.assertEqual(self._handler._peek_last_added_job(),119 self.other_valid_job)120 # any_jobs121 def test_any_jobs_not_empty(self):122 self._handler._job_deque.append(self.valid_job)123 self.assertTrue(self._handler.has_jobs)124 self._handler._job_deque.append(self.other_valid_job)125 self.assertTrue(self._handler.has_jobs)126 def test_any_events_empty(self):127 self.assertFalse(self._handler.has_jobs)128 # is_valid_element129 def test_positive_is_valid_element(self):130 self.assertTrue(JobHandler.is_valid_element(self.valid_job))131 def test_negative_is_valid_element(self):132 self.assertFalse(JobHandler.is_valid_element(1))133 self.assertFalse(JobHandler.is_valid_element('my_string'))134 self.assertFalse(JobHandler.is_valid_element(None))135 self.assertFalse(JobHandler.is_valid_element(self.image_event))136 # properties137 def test_property_state(self):138 self.assertEqual(self._handler.state, self._handler._state)139 def test_has_jobs(self):140 self.assertFalse(self._handler.has_jobs)141 self._handler._job_deque.append(self.valid_job)142 self.assertTrue(self._handler.has_jobs)143 # state changes144 def test_state_init(self):145 self.assertEqual(self._handler._state, JobHandler.State.STARTED)146 def test_state_force_stop(self):147 self._handler.force_stop()148 self.assertEqual(self._handler._state, JobHandler.State.STOPPED)149 def test_state_enqueue(self):150 self._handler.enqueue_job(self.valid_job)151 self.assertEqual(self._handler._state, JobHandler.State.PENDING)152 def test_state_stop(self):153 self._handler.enqueue_job(self.valid_job)154 self._handler.stop()155 self.assertEqual(self._handler._state, JobHandler.State.PENDING_STOP_ON_RESOLVE)156 def test_state_restart_empty(self):157 self._handler.enqueue_job(self.valid_job)158 self._handler.force_stop()159 self._handler._job_deque.popleft()160 self._handler.restart()161 self.assertEqual(self._handler._state, JobHandler.State.STARTED)162 def test_state_restart_has_jobs(self):163 self._handler.enqueue_job(self.valid_job)164 self._handler.force_stop()165 self._handler.restart()166 self.assertEqual(self._handler._state, JobHandler.State.PENDING)167 # can_state_resolve168 def test_can_state_resolve(self):169 self._handler._state = JobHandler.State.PENDING_STOP_ON_RESOLVE170 self.assertTrue(self._handler._can_state_resolve())171 self._handler._state = JobHandler.State.PENDING172 self.assertTrue(self._handler._can_state_resolve())173 self._handler._state = JobHandler.State.STARTED174 self.assertRaises(JobHandler.NotAllowedInStateException, self._handler._can_state_resolve)175 self._handler._state = JobHandler.State.STOPPED176 self.assertRaises(JobHandler.NotAllowedInStateException, self._handler._can_state_resolve)177if __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.
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 autotest 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?
