code: def test_03_02_pipeline_preferences(self):
      #
      # Walk the worker up through pipelines and preferences.
      #
      self.awthread = self.AWThread(self.work_addr, self.zmq_context)
      self.awthread.start()
      self.set_work_socket()
      self.awthread.ex(
          self.awthread.aw.do_job,
          anareply.Work(
              image_set_numbers=[1],
              worker_runs_post_group=False,
              wants_dictionary=True,
          ),
      )
      #
      # The worker should ask for the pipeline and preferences next.
      #
      req = self.awthread.recv(self.work_socket)
      self.assertIsInstance(req, anarequest.PipelinePreferences)
      self.assertEqual(req.analysis_id, self.analysis_id)
  
      tests.core.modules.maybe_download_example_image(
          ["ExampleSBSImages"], "Channel1-01-A-01.tif"
      )
      tests.core.modules.maybe_download_example_image(
          ["ExampleHT29"], "AS_09125_050116030001_D03f00d0.tif"
      )
      input_dir = os.path.normcase(
          os.path.join(
              tests.core.modules.example_images_directory(),
              "ExampleSBSImages",
          )
      )
      output_dir = os.path.normcase(
          os.path.join(
              tests.core.modules.example_images_directory(),
              "ExampleHT29",
          )
      )
      cellprofiler_core.preferences.set_default_image_directory(
          input_dir
      )
      input_dir = (
          cellprofiler_core.preferences.get_default_image_directory()
      )
      cellprofiler_core.preferences.set_default_output_directory(
          output_dir
      )
      output_dir = (
          cellprofiler_core.preferences.get_default_output_directory()
      )
      preferences = {
          cellprofiler_core.preferences.DEFAULT_IMAGE_DIRECTORY: cellprofiler_core.preferences.config_read(
              cellprofiler_core.preferences.DEFAULT_IMAGE_DIRECTORY
          ),
          cellprofiler_core.preferences.DEFAULT_OUTPUT_DIRECTORY: cellprofiler_core.preferences.config_read(
              cellprofiler_core.preferences.DEFAULT_OUTPUT_DIRECTORY
          ),
      }
      cellprofiler_core.preferences.set_default_image_directory(
          tests.core.modules.example_images_directory()
      )
      cellprofiler_core.preferences.set_default_output_directory(
          tests.core.modules.example_images_directory()
      )
      rep = cellprofiler_core.utilities.zmq.Reply(
          pipeline_blob=numpy.array(GOOD_PIPELINE),
          preferences=preferences,
      )
      req.reply(rep)
      #
      # Get the next request so that we know the worker has
      # processed the preferences.
      #
      req = self.awthread.recv(self.work_socket)
      self.assertEqual(
          cellprofiler_core.preferences.get_default_image_directory(),
          input_dir,
      )
      self.assertEqual(
          cellprofiler_core.preferences.get_default_output_directory(),
          output_dir,
      )
      pipe = self.awthread.aw.pipeline
      prefs = self.awthread.aw.preferences
      assert pipe is not None
      assert prefs is not None
      self.assertEqual(len(pipe.modules()), NUM_MODULES)
      #
      # Cancel and check for exit
      #
      req.reply(anareply.ServerExited())
      self.assertRaises(
          cellprofiler_core.pipeline.event.CancelledException,
          self.awthread.ecute,
      )
  
imports: import io
  import queue
  import time
  import os
  import pytest
  import pickle
  import tempfile
  import threading
  import traceback
  import unittest
  import uuid
  import zlib
  import numpy
  import zmq
  import cellprofiler_core.analysis
  import cellprofiler_core.analysis.reply
  import cellprofiler_core.constants.measurement
  import cellprofiler_core.constants.worker
  import cellprofiler_core.measurement
  import cellprofiler_core.module._identify
  import cellprofiler_core.modules.namesandtypes
  import cellprofiler_core.pipeline
  import cellprofiler_core.preferences
  import cellprofiler_core.utilities.image
  import cellprofiler_core.utilities.measurement
  import cellprofiler_core.utilities.pathname
  import cellprofiler_core.utilities.zmq
  import cellprofiler_core.worker
  import cellprofiler_core.analysis.request
  from cellprofiler_core.constants.modules.namesandtypes import M_IMAGE_SET_ZIP_DICTIONARY
  from cellprofiler_core.pipeline import ImageFile
  from cellprofiler_core.pipeline import ImagePlane
  import tests.core
  import tests.core.modules
label_desc: Use image node_id 13 for calling the TestAnalysisWorker obj's underlying member method code with example usage: obj.test_03_02_pipeline_preferences() without return types