1  """CLI runner helpers for Auths E2E tests."""
  2  
  3  import json
  4  import subprocess
  5  from dataclasses import dataclass
  6  from pathlib import Path
  7  
  8  
  9  @dataclass
 10  class CLIResult:
 11      """Result of running a CLI command."""
 12  
 13      returncode: int
 14      stdout: str
 15      stderr: str
 16  
 17      @property
 18      def json(self) -> dict:
 19          """Parse stdout as JSON."""
 20          return json.loads(self.stdout)
 21  
 22      def assert_success(self) -> "CLIResult":
 23          """Assert the command exited with code 0."""
 24          assert self.returncode == 0, (
 25              f"Command failed with exit code {self.returncode}\n"
 26              f"stdout: {self.stdout}\n"
 27              f"stderr: {self.stderr}"
 28          )
 29          return self
 30  
 31      def assert_failure(self, exit_code: int | None = None) -> "CLIResult":
 32          """Assert the command failed."""
 33          if exit_code is not None:
 34              assert self.returncode == exit_code, (
 35                  f"Expected exit code {exit_code}, got {self.returncode}\n"
 36                  f"stderr: {self.stderr}"
 37              )
 38          else:
 39              assert self.returncode != 0, (
 40                  f"Expected failure but got exit code 0\n"
 41                  f"stdout: {self.stdout}"
 42              )
 43          return self
 44  
 45  
 46  def run_auths(
 47      binary: Path,
 48      args: list[str],
 49      *,
 50      cwd: Path | None = None,
 51      env: dict[str, str] | None = None,
 52      timeout: int = 60,
 53      stdin_data: str | None = None,
 54  ) -> CLIResult:
 55      """Run an auths CLI command and capture output."""
 56      result = subprocess.run(
 57          [str(binary)] + args,
 58          cwd=cwd,
 59          env=env,
 60          capture_output=True,
 61          text=True,
 62          timeout=timeout,
 63          input=stdin_data,
 64      )
 65      return CLIResult(
 66          returncode=result.returncode,
 67          stdout=result.stdout,
 68          stderr=result.stderr,
 69      )
 70  
 71  
 72  def get_identity_did(binary: Path, env: dict[str, str]) -> str:
 73      """Extract the controller DID from `auths id show --json`."""
 74      result = run_auths(binary, ["id", "show", "--json"], env=env)
 75      result.assert_success()
 76      return result.json["data"]["controller_did"]
 77  
 78  
 79  def get_device_did(binary: Path, env: dict[str, str]) -> str:
 80      """Extract the first device DID from `auths status --json`."""
 81      result = run_auths(binary, ["status", "--json"], env=env)
 82      result.assert_success()
 83      devices = result.json["data"]["devices"]["devices_detail"]
 84      assert len(devices) > 0, "No devices found in status output"
 85      return devices[0]["device_did"]
 86  
 87  
 88  def export_attestation(env: dict[str, str], out_path: Path) -> dict:
 89      """Extract the first attestation from the auths git repo to a file.
 90  
 91      Returns the parsed attestation JSON.
 92      """
 93      auths_home = Path(env["AUTHS_HOME"])
 94  
 95      ls = run_git(
 96          ["ls-tree", "-r", "--name-only", "refs/auths/registry"],
 97          cwd=auths_home,
 98          env=env,
 99      )
100      ls.assert_success()
101  
102      att_path = None
103      for line in ls.stdout.splitlines():
104          if line.endswith("/attestation.json"):
105              att_path = line
106              break
107      assert att_path is not None, "No attestation found in auths repo"
108  
109      show = run_git(
110          ["show", f"refs/auths/registry:{att_path}"],
111          cwd=auths_home,
112          env=env,
113      )
114      show.assert_success()
115  
116      out_path.write_text(show.stdout)
117      return json.loads(show.stdout)
118  
119  
120  def run_git(
121      args: list[str],
122      *,
123      cwd: Path,
124      env: dict[str, str] | None = None,
125      timeout: int = 15,
126  ) -> CLIResult:
127      """Run a git command and capture output."""
128      result = subprocess.run(
129          ["git"] + args,
130          cwd=cwd,
131          env=env,
132          capture_output=True,
133          text=True,
134          timeout=timeout,
135      )
136      return CLIResult(
137          returncode=result.returncode,
138          stdout=result.stdout,
139          stderr=result.stderr,
140      )