code: def test_duplicate_within_pool_and_db():
      """Test that an algo suggesting multiple trials can have a few registered even
      if one of them is a duplicate with db.
      """
      with create_producer() as (producer, storage):
          trials_in_db_before = len(storage._fetch_trials({}))
          new_trials_in_db_before = len(
              storage._fetch_trials({"status": "new"})
          )
  
          # Avoid limiting number of samples from the within the algorithm.
          producer.experiment.algorithm.unwrapped.pool_size = 1000
  
          producer.experiment.algorithm.unwrapped.possible_values = [
              (v,) for v in [0, 1, 2]
          ]
  
          assert producer.produce(3) == 1
  
          # Algorithm was required to suggest some trials
          num_new_trials = producer.experiment.algorithm.unwrapped._num
          assert num_new_trials == 3  # pool size
  
          # `num_new_trials` new trials were registered at database
          assert (
              len(storage._fetch_trials({})) == trials_in_db_before + 1
          )
          assert (
              len(storage._fetch_trials({"status": "new"}))
              == new_trials_in_db_before + 1
          )
          new_trials = list(storage._fetch_trials({"status": "new"}))
          assert new_trials[0].experiment == producer.experiment.id
          assert new_trials[0].start_time is None
          assert new_trials[0].end_time is None
          assert new_trials[0].results == []
          assert new_trials[0].params == {"x": 1}
  
file_path: orion/tests/unittests/core/worker/test_producer.py
imports: import contextlib
  import copy
  import threading
  import time
  import pytest
  from orion.core.io.experiment_builder import build
  from orion.core.worker.producer import Producer
  from orion.testing import OrionState, generate_trials
label_desc: Use image node_id 8 for calling a global function with example usage: test_duplicate_within_pool_and_db() without return types