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

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.