code: def test_usage_report_e2e(
      monkeypatch,
      ray_start_cluster,
      tmp_path,
      reset_usage_stats,
      gcs_storage_type,
  ):
      """
      Test usage report works e2e with env vars.
      """
      cluster_config_file_path = tmp_path / "ray_bootstrap_config.yaml"
      cluster_config_file_path.write_text(
          """
  cluster_name: minimal
  max_workers: 1
  provider:
      type: aws
      region: us-west-2
      availability_zone: us-west-2a
  """
      )
      with patch.object(
          ray._private.utils,
          "get_current_node_cpu_model_name",
          return_value="TestCPU",
      ), monkeypatch.context() as m:
          m.setenv("HOME", str(tmp_path))
          m.setenv("RAY_USAGE_STATS_ENABLED", "1")
          m.setenv(
              "RAY_USAGE_STATS_REPORT_URL", "http://127.0.0.1:8000"
          )
          m.setenv("RAY_USAGE_STATS_REPORT_INTERVAL_S", "1")
          m.setenv("RAY_USAGE_STATS_EXTRA_TAGS", "extra_k1=extra_v1")
          cluster = ray_start_cluster
          cluster.add_node(num_cpus=3)
          if os.environ.get("RAY_MINIMAL") != "1":
              from ray import train  # noqa: F401
              from ray.rllib.algorithms.ppo import PPO  # noqa: F401
  
          ray_usage_lib.record_extra_usage_tag(
              ray_usage_lib.TagKey._TEST1, "extra_v2"
          )
  
          ray.init(address=cluster.address)
  
          ray_usage_lib.record_extra_usage_tag(
              ray_usage_lib.TagKey._TEST2, "extra_v3"
          )
  
          if os.environ.get("RAY_MINIMAL") != "1":
              from ray import tune  # noqa: F401
  
              def objective(*args):
                  pass
  
              tuner = tune.Tuner(objective)
              tuner.fit()
  
          @ray.remote(num_cpus=0)
          class StatusReporter:
              def __init__(self):
                  self.reported = 0
                  self.payload = None
  
              def report_payload(self, payload):
                  self.payload = payload
  
              def reported(self):
                  self.reported += 1
  
              def get(self):
                  return self.reported
  
              def get_payload(self):
                  return self.payload
  
          reporter = StatusReporter.remote()
  
          class UsageStatsServer(BaseHTTPRequestHandler):
              reporter = None
  
              def do_POST(self):
                  content_length = int(self.headers["Content-Length"])
                  post_data = self.rfile.read(content_length)
                  self.reporter.reported.remote()
                  self.reporter.report_payload.remote(
                      json.loads(post_data)
                  )
                  self.send_response(200)
                  self.send_header("Content-type", "text/html")
                  self.end_headers()
  
          @ray.remote(num_cpus=0)
          def run_usage_stats_server(reporter):
              UsageStatsServer.reporter = reporter
              server = HTTPServer(("127.0.0.1", 8000), UsageStatsServer)
              server.serve_forever()
  
          run_usage_stats_server.remote(reporter)
  
          """
          Verify the usage stats are reported to the server.
          """
          print("Verifying usage stats report.")
          # Since the interval is 1 second, there must have been
          # more than 5 requests sent within 30 seconds.
          try:
              wait_for_condition(
                  lambda: ray.get(reporter.get.remote()) > 5, timeout=30
              )
          except Exception:
              print_dashboard_log()
              raise
          payload = ray.get(reporter.get_payload.remote())
          (
              ray_version,
              python_version,
          ) = ray._private.utils.compute_version_info()
          assert payload["ray_version"] == ray_version
          assert payload["python_version"] == python_version
          assert payload["schema_version"] == "0.1"
          assert payload["os"] == sys.platform
          if sys.platform != "linux":
              payload["libc_version"] is None
          else:
              import platform
  
              assert (
                  payload["libc_version"]
                  == f"{platform.libc_ver()[0]}:{platform.libc_ver()[1]}"
              )
  
          assert payload["source"] == "OSS"
          assert payload["cloud_provider"] == "aws"
          assert payload["min_workers"] is None
          assert payload["max_workers"] == 1
          assert payload["head_node_instance_type"] is None
          assert payload["worker_node_instance_types"] is None
          assert payload["total_num_cpus"] == 3
          assert payload["total_num_gpus"] is None
          assert payload["total_memory_gb"] > 0
          assert payload["total_object_store_memory_gb"] > 0
          assert (
              int(payload["extra_usage_tags"]["actor_num_created"]) >= 0
          )
          assert int(payload["extra_usage_tags"]["pg_num_created"]) >= 0
          assert (
              int(
                  payload["extra_usage_tags"][
                      "num_actor_creation_tasks"
                  ]
              )
              >= 0
          )
          assert (
              int(payload["extra_usage_tags"]["num_actor_tasks"]) >= 0
          )
          assert (
              int(payload["extra_usage_tags"]["num_normal_tasks"]) >= 0
          )
          assert int(payload["extra_usage_tags"]["num_drivers"]) >= 0
          payload["extra_usage_tags"]["actor_num_created"] = "0"
          payload["extra_usage_tags"]["pg_num_created"] = "0"
          payload["extra_usage_tags"]["num_actor_creation_tasks"] = "0"
          payload["extra_usage_tags"]["num_actor_tasks"] = "0"
          payload["extra_usage_tags"]["num_normal_tasks"] = "0"
          payload["extra_usage_tags"]["num_drivers"] = "0"
          expected_payload = {
              "extra_k1": "extra_v1",
              "_test1": "extra_v2",
              "_test2": "extra_v3",
              "dashboard_metrics_grafana_enabled": "False",
              "dashboard_metrics_prometheus_enabled": "False",
              "actor_num_created": "0",
              "pg_num_created": "0",
              "num_actor_creation_tasks": "0",
              "num_actor_tasks": "0",
              "num_normal_tasks": "0",
              "num_drivers": "0",
              "gcs_storage": gcs_storage_type,
              "dashboard_used": "False",
          }
          if os.environ.get("RAY_MINIMAL") != "1":
              expected_payload["tune_scheduler"] = "FIFOScheduler"
              expected_payload[
                  "tune_searcher"
              ] = "BasicVariantGenerator"
              expected_payload["air_entrypoint"] = "Tuner.fit"
              expected_payload["air_storage_configuration"] = "local"
          assert payload["extra_usage_tags"] == expected_payload
          assert payload["total_num_nodes"] == 1
          assert payload["total_num_running_jobs"] == 1
          if os.environ.get("RAY_MINIMAL") == "1":
              assert set(payload["library_usages"]) == set()
          else:
              assert set(payload["library_usages"]) == {
                  "rllib",
                  "train",
                  "tune",
              }
          assert payload["hardware_usages"] == ["TestCPU"]
          validate(instance=payload, schema=schema)
          """
          Verify the usage_stats.json is updated.
          """
          print("Verifying usage stats write.")
          global_node = ray._private.worker._global_node
          temp_dir = pathlib.Path(global_node.get_session_dir_path())
  
          wait_for_condition(lambda: file_exists(temp_dir), timeout=30)
  
          timestamp_old = read_file(temp_dir, "usage_stats")[
              "collect_timestamp_ms"
          ]
          success_old = read_file(temp_dir, "usage_stats")[
              "total_success"
          ]
          # Test if the timestampe has been updated.
          wait_for_condition(
              lambda: timestamp_old
              < read_file(temp_dir, "usage_stats")[
                  "collect_timestamp_ms"
              ]
          )
          wait_for_condition(
              lambda: success_old
              < read_file(temp_dir, "usage_stats")["total_success"]
          )
          assert read_file(temp_dir, "success")
  
imports: import json
  import os
  import pathlib
  import sys
  import time
  from dataclasses import asdict
  from pathlib import Path
  from unittest.mock import patch
  import requests
  import pytest
  from jsonschema import validate
  from http.server import BaseHTTPRequestHandler, HTTPServer
  import ray
  import ray._private.usage.usage_constants
  import ray._private.usage.usage_lib
  from ray._private.test_utils import format_web_url, run_string_as_driver, wait_for_condition, wait_until_server_available
  from ray._private.usage.usage_lib import ClusterConfigToReport, UsageStatsEnabledness
  from ray.autoscaler._private.cli_logger import cli_logger
  from ray.util.placement_group import placement_group
  from ray._private.accelerators import NvidiaGPUAcceleratorManager
label_desc: Use image node_id 27 for calling a global function with example usage: test_usage_report_e2e(monkeypatch, ray_start_cluster, tmp_path, reset_usage_stats, gcs_storage_type) and returns: self, self