Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use generate_task_id method in lisa

Best Python code snippet using lisa_python

test_group_queue.py

Source:test_group_queue.py

...5    def test_generate_job_id(self):6        task_group = TaskGroup(number_of_shards=3)7        group_id = task_group.group_id8        all_task_ids = (9            task_group.generate_task_id(),10            task_group.generate_task_id(),11            task_group.generate_task_id(),12            task_group.generate_task_id(),13        )14        # checking basic signature of the task_id and it's shard key component15        # note roll back into '1' because of number_of_shards=316        shard_ids_should_be = ('1', '2', '0', '1')17        task_keys_should_be = ('1', '2', '3', '4')18        parts = zip(all_task_ids, shard_ids_should_be, task_keys_should_be)19        _loop_id = 020        for task_id, shard_id_should_be, task_key_should_be in parts:21            _loop_id += 122            print('Loop ', _loop_id)23            assert type(task_id) is tuple24            shard_key, job_key = task_id25            assert shard_key == group_id + '-' + shard_id_should_be26            assert job_key == task_key_should_be27    def test_tasks_count(self):28        task_group = TaskGroup(number_of_shards=3)29        assert task_group.get_remaining_tasks_count() == 030        task1_id = task_group.generate_task_id()31        task2_id = task_group.generate_task_id()32        task_group.report_task_active(task1_id)33        task_group.report_task_active(task2_id)34        assert task_group.get_remaining_tasks_count() == 235        task_group.report_task_done(task2_id)36        assert task_group.get_remaining_tasks_count() == 137        task_group.report_task_done(task1_id)38        assert task_group.get_remaining_tasks_count() == 039    def test_join(self):40        task_group = TaskGroup(number_of_shards=3)41        task1_id = task_group.generate_task_id()42        task_group.report_task_active(task1_id)43        join_timeout = 0.144        with self.assertRaises(TimeoutError) as ctx:45            task_group.join(join_timeout)46        assert 'has 1 remaining tasks after timeout' in str(ctx.exception)47        task_group.report_task_done(task1_id)48        task_group.join(join_timeout)49    def test_task_context(self):50        task_group = TaskGroup(number_of_shards=3)51        task_id = task_group.generate_task_id()52        join_timeout = 0.153        assert task_group.get_remaining_tasks_count() == 054        with TaskGroup.task_context(task_id) as task_context:55            assert task_group.get_remaining_tasks_count() == 156            last_marker_1 = task_group.get_task_data(task_id)57            assert last_marker_1 is not None58            # also make sure manual kicking of task's "still working on it" marker works59            task_context.report_active()60            assert task_group.get_remaining_tasks_count() == 161            last_marker_2 = task_group.get_task_data(task_id)62            # Not normative test really, just sanity test for us to ensure63            # that `task_context.report_active()` did something64            # Now we are just abusing knowledge of .report_active() just bumping65            # the value by new Unix timestamp, so all newer markers will be bigger...

test_task.py

Source:test_task.py

...24    # Test get existing task25    assert db.task(new_task["id"]) == new_task26    # Test create task with specific ID27    task_name = generate_task_name()28    task_id = generate_task_id()29    new_task = db.create_task(30        name=task_name,31        command=test_command,32        params={"foo": 1, "bar": 2},33        offset=1,34        period=10,35        task_id=task_id,36    )37    assert new_task["name"] == task_name38    assert new_task["id"] == task_id39    assert "print(params)" in new_task["command"]40    assert new_task["type"] == "periodic"41    assert new_task["database"] == db.name42    assert isinstance(new_task["created"], float)43    assert db.task(new_task["id"]) == new_task44    # Test create duplicate task45    with assert_raises(TaskCreateError) as err:46        db.create_task(47            name=task_name,48            command=test_command,49            params={"foo": 1, "bar": 2},50            task_id=task_id,51        )52    assert err.value.error_code == 185153    # Test list tasks54    for task in sys_db.tasks():55        assert task["type"] in {"periodic", "timed"}56        assert isinstance(task["id"], str)57        assert isinstance(task["name"], str)58        assert isinstance(task["created"], float)59        assert isinstance(task["command"], str)60    # Test list tasks with bad database61    with assert_raises(TaskListError) as err:62        bad_db.tasks()63    assert err.value.error_code in {11, 1228}64    # Test get missing task65    with assert_raises(TaskGetError) as err:66        db.task(generate_task_id())67    assert err.value.error_code == 185268    # Test delete existing task69    assert task_id in extract("id", db.tasks())70    assert db.delete_task(task_id) is True71    assert task_id not in extract("id", db.tasks())72    with assert_raises(TaskGetError) as err:73        db.task(task_id)74    assert err.value.error_code == 185275    # Test delete missing task76    with assert_raises(TaskDeleteError) as err:77        db.delete_task(generate_task_id(), ignore_missing=False)78    assert err.value.error_code == 1852...

tasks.py

Source:tasks.py

...7        self.model = model8        # generate a task ID if not specified9        id = self.config.id10        if not id:11            id = str(self.generate_task_id())12        self.id = id13    @staticmethod14    def create_dataset(dataset_config):15        """ Builds the input dataset using the provided configuration. """16        classname, config = dataset_config.classname, dataset_config.config17        cls = get_class(classname)18        if not issubclass(cls, BaseDataset):19            raise ValueError("Data input class does not inherit from BaseInput.")20        dataset = cls(config)21        return dataset22    def generate_task_id(self):23        """Generates a task ID"""24        return current_ts()25    def run(self):26        """Runs the task."""...

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.