code: def test_enforces_concurrency_limit(tmp_path: Path, is_async: bool):
      concurrency_limit = 3
  
      q = queue.Queue()
  
      api = UploadBlockingMockApi()
  
      def save_fn_sync(path: Path, *args, **kwargs):
          api.upload_file_retry(f"http://dst/{path}", path.open("rb"))
  
      async def save_fn_async(path: Path, *args, **kwargs):
          await api.upload_file_retry_async(
              f"http://dst/{path}", path.open("rb")
          )
  
      def add_job():
          path = make_tmp_file(tmp_path)
          q.put(
              make_request_upload(
                  path,
                  save_fn=(
                      Mock(
                          side_effect=Exception(
                              "save_fn should not be called in async test"
                          )
                      )
                      if is_async
                      else functools.partial(save_fn_sync, path)
                  ),
                  save_fn_async=(
                      functools.partial(save_fn_async, path)
                      if is_async
                      else Mock(
                          side_effect=Exception(
                              "save_fn_async should not be called in sync test"
                          )
                      )
                  ),
              )
          )
  
      step_upload = make_step_upload(
          api=api,
          event_queue=q,
          max_threads=(concurrency_limit + 10)
          if is_async
          else concurrency_limit,
          settings=make_async_settings(
              concurrency_limit=concurrency_limit if is_async else None
          ),
      )
      step_upload.start()
  
      waiters = []
  
      # first few jobs should start without blocking
      for _ in range(concurrency_limit):
          add_job()
          waiters.append(api.wait_for_upload(0.1))
  
      # next job should block...
      add_job()
      assert not api.wait_for_upload(0.1)
  
      # ...until we release one of the first jobs
      waiters.pop()()
      waiters.append(api.wait_for_upload(0.1))
  
      # let all jobs finish, to release the threads
      for w in waiters:
          w()
  
      finish_and_wait(q)
  
imports: import asyncio
  import concurrent.futures
  import functools
  import queue
  import random
  import threading
  import time
  from pathlib import Path
  from typing import Any, Callable, Iterable, MutableSequence, Optional
  from unittest.mock import DEFAULT, Mock, patch
  import pytest
  from wandb.filesync import stats
  from wandb.filesync.step_upload import Event, RequestCommitArtifact, RequestFinish, RequestUpload, StepUpload
  from wandb.sdk.internal import file_stream, internal_api
  from wandb.sdk.internal.settings_static import SettingsStatic
  from wandb.sdk.wandb_settings import Settings
label_desc: Use image node_id 13 for calling a global function with example usage: test_enforces_concurrency_limit(tmp_path, is_async) without return types