docstr: Check that we validate _on the receiving_ side, not just sending side
docstr_tok: ["Check","that","we","validate","_on","the","receiving_","side",",","not","just","sending","side"]
code: def test_command_validation(self, command, raise_exception):
      """Check that we validate _on the receiving_ side, not just sending side"""
      expected_context = contextlib.nullcontext()
      if raise_exception:
          expected_context = pytest.raises(
              ValueError,
              match=r'The command must start with \["airflow", "tasks", "run"\]\.',
          )
  
      with mock.patch(
          "airflow.providers.celery.executors.celery_executor_utils._execute_in_subprocess"
      ) as mock_subproc, mock.patch(
          "airflow.providers.celery.executors.celery_executor_utils._execute_in_fork"
      ) as mock_fork, mock.patch(
          "celery.app.task.Task.request"
      ) as mock_task:
          mock_task.id = "abcdef-124215-abcdef"
          with expected_context:
              celery_executor_utils.execute_command(command)
          if raise_exception:
              mock_subproc.assert_not_called()
              mock_fork.assert_not_called()
          else:
              # One of these should be called.
              assert mock_subproc.call_args == (
                  (command, "abcdef-124215-abcdef"),
              ) or mock_fork.call_args == (
                  (command, "abcdef-124215-abcdef"),
              )
  
imports: from __future__ import annotations
  import contextlib
  import logging
  import os
  import signal
  import sys
  from datetime import datetime, timedelta
  from unittest import mock
  import celery.contrib.testing.tasks
  import pytest
  import time_machine
  from celery import Celery
  from celery.result import AsyncResult
  from kombu.asynchronous import set_event_loop
  from airflow.configuration import conf
  from airflow.models.baseoperator import BaseOperator
  from airflow.models.dag import DAG
  from airflow.models.taskinstance import TaskInstance, TaskInstanceKey
  from airflow.providers.celery.executors import celery_executor, celery_executor_utils, default_celery
  from airflow.providers.celery.executors.celery_executor import CeleryExecutor
  from airflow.utils import timezone
  from airflow.utils.state import State
  from tests.test_utils import db
  from tests.test_utils.config import conf_vars
label_desc: Use image node_id 8 for calling the TestCeleryExecutor obj's underlying member method code with example usage: obj.test_command_validation(command, raise_exception) without return types