code: def test_memory_color():
      def config(**kwargs):
          return dask.config.set(
              {
                  f"distributed.worker.memory.{k}": v
                  for k, v in kwargs.items()
              }
          )
  
      with config(target=0.6, spill=0.7, pause=0.8, terminate=0.95):
          c = MemoryColor()
          assert c._memory_color(50, 100, Status.running) == "blue"
          assert c._memory_color(60, 100, Status.running) == "orange"
          assert c._memory_color(75, 100, Status.running) == "orange"
          # Pause is not impacted by the paused threshold, but by the worker status
          assert c._memory_color(85, 100, Status.running) == "orange"
          assert c._memory_color(0, 100, Status.paused) == "red"
          assert (
              c._memory_color(0, 100, Status.closing_gracefully)
              == "red"
          )
          # Passing the terminate threshold will turn the bar red, regardless of pause
          assert c._memory_color(95, 100, Status.running) == "red"
          # Disabling memory limit disables all threshold-related color changes
          assert c._memory_color(100, 0, Status.running) == "blue"
          assert c._memory_color(100, 0, Status.paused) == "red"
  
      # target disabled
      with config(target=False, spill=0.7):
          c = MemoryColor()
          assert c._memory_color(60, 100, Status.running) == "blue"
          assert c._memory_color(75, 100, Status.running) == "orange"
  
      # spilling disabled
      with config(target=False, spill=False, pause=0.8, terminate=0.95):
          c = MemoryColor()
          assert c._memory_color(94, 100, Status.running) == "blue"
          assert (
              c._memory_color(0, 100, Status.closing_gracefully)
              == "red"
          )
          assert c._memory_color(95, 100, Status.running) == "red"
  
      # terminate disabled; fall back to 100%
      with config(target=False, spill=False, terminate=False):
          c = MemoryColor()
          assert c._memory_color(99, 100, Status.running) == "blue"
          assert c._memory_color(100, 100, Status.running) == "red"
          assert c._memory_color(110, 100, Status.running) == "red"
  
imports: from __future__ import annotations
  import asyncio
  import json
  import re
  import ssl
  import sys
  import pytest
  from urllib.parse import quote_plus
  from bokeh.server.server import BokehTornado
  from tlz import first
  from tornado.httpclient import AsyncHTTPClient, HTTPRequest
  import dask
  from dask.core import flatten
  from distributed import Event
  from distributed.client import wait
  from distributed.core import Status
  from distributed.dashboard import scheduler
  from distributed.dashboard.components.scheduler import AggregateAction, ClusterMemory, ComputePerKey, Contention, CurrentLoad, Events, FinePerformanceMetrics, Hardware, MemoryByKey, MemoryColor, Occupancy, ProcessingHistogram, ProfileServer, Shuffling, StealingEvents, StealingTimeSeries, SystemMonitor, SystemTimeseries, TaskGraph, TaskGroupGraph, TaskProgress, TaskStream, WorkerNetworkBandwidth, WorkersMemory, WorkersMemoryHistogram, WorkerTable
  from distributed.dashboard.components.worker import Counters
  from distributed.dashboard.scheduler import applications
  from distributed.diagnostics.task_stream import TaskStreamPlugin
  from distributed.metrics import context_meter, time
  from distributed.spans import span
  from distributed.utils import format_dashboard_link
  from distributed.utils_test import block_on_event, dec, div, gen_cluster, get_cert, inc, slowinc
  from distributed.worker import Worker
label_desc: Use image node_id 1 for calling a global function with example usage: test_memory_color() and returns: dask