text: # vim:ts=4:sw=4:et:
  # Copyright (c) Meta Platforms, Inc. and affiliates.
  #
  # This source code is licensed under the MIT license found in the
  # LICENSE file in the root directory of this source tree.
  
  
  import errno
  import os
  import re
  import subprocess
  
  from watchman.integration.lib import WatchmanInstance
  from watchman.integration.lib import WatchmanTestCase
  
  
  def is_hg_installed() -> bool:
      with open(os.devnull, "wb") as devnull:
          try:
              env = os.environ.copy()
              env["HGPLAIN"] = "1"
              env["WATCHMAN_SOCK"] = (
                  WatchmanInstance.getSharedInstance().getSockPath().legacy_sockpath()
              )
  
              exit_code = subprocess.call(
                  ["hg", "--version"], stdout=devnull, stderr=devnull, env=env
              )
              return exit_code == 0
          except OSError as e:
              if e.errno == errno.ENOENT:
                  return False
              raise
  
  
  @WatchmanTestCase.expand_matrix
  class TestRequestId(WatchmanTestCase.WatchmanTestCase):
      def test_queryRequestId(self) -> None:
          root = self.mkdtemp()
          self.watchmanCommand("watch", root)
          self.watchmanCommand("log-level", "debug")
          self.touchRelative(root, "111")
  
          request_id = "f13bd3bc02c27afe2413932c6fa6c4942b0574b3"
          params = {"since": "c:0:0", "request_id": request_id}
          self.watchmanCommand("query", root, params)
          pat = re.compile(".* \\[client=.*\\] request_id = %s" % request_id)
  
          self.assertWaitFor(
              lambda: any(pat.match(l) for l in self.getServerLogContents()),
              message="request_id logged",
          )
  
      def skipIfNoHgRequestIdSupport(self) -> None:
          root = self.mkdtemp()
          request_id = "bf8a47014bd1b66103a8ab0aece4be7ada871660"
  
          env = os.environ.copy()
          env["HGPLAIN"] = "1"
          env["HGREQUESTID"] = request_id
          env["WATCHMAN_SOCK"] = (
              WatchmanInstance.getSharedInstance().getSockPath().legacy_sockpath()
          )
  
          subprocess.call(["hg", "init"], env=env, cwd=root)
          subprocess.call(["hg", "log"], env=env, cwd=root)
  
          try:
              with open(os.path.join(root, ".hg/blackbox.log")) as f:
                  if request_id in f.read():
                      return
          except IOError:
              pass
  
          self.skipTest("HGREQUESTID is not supported")
  
      def test_scmHgRequestId(self) -> None:
          if not is_hg_installed():
              self.skipTest("Hg not installed")
          self.skipIfNoHgRequestIdSupport()
  
          root = self.mkdtemp()
  
          # In this test, the repo does not necessarily need fsmonitor enabled,
          # since watchman calls HGREQUESTID=... hg status and that would also
          # have request_id logged without fsmonitor.
          env = os.environ.copy()
          env["HGPLAIN"] = "1"
          env["WATCHMAN_SOCK"] = (
              WatchmanInstance.getSharedInstance().getSockPath().legacy_sockpath()
          )
          subprocess.call(["hg", "init"], env=env, cwd=root)
          subprocess.call(
              [
                  "hg",
                  "commit",
                  "-mempty",
                  "-utest",
                  "-d0 0",
                  "--config=ui.allowemptycommit=1",
              ],
              env=env,
              cwd=root,
          )
          commit_hash = subprocess.check_output(
              ["hg", "log", "-r.", "-T{node}"], env=env, cwd=root
          ).decode("utf-8")
  
          # Must watch the directory after it's an HG repo to perform scm-aware
          # queries.
          self.watchmanCommand("watch", root)
          request_id = "4c05a798ea1acc7c97b75e61fec5f640d90f8209"
  
          params = {
              "fields": ["name"],
              "request_id": request_id,
              "since": {"scm": {"mergebase-with": commit_hash}},
          }
          self.watchmanCommand("query", root, params)
  
          blackbox_path = os.path.join(root, ".hg", "blackbox.log")
  
          def try_read_blackbox():
              try:
                  with open(blackbox_path) as f:
                      return f.read()
              except IOError:
                  return ""
  
          self.assertWaitFor(
              lambda: request_id in try_read_blackbox(),
              message="request_id passed to and logged by hg",
          )
  