|
4 | 4 | import json |
5 | 5 | import io |
6 | 6 | import os |
| 7 | +import runpy |
7 | 8 | import select |
8 | 9 | import socket |
9 | 10 | import subprocess |
10 | 11 | import sys |
11 | 12 | import threading |
12 | 13 | import time |
13 | 14 | from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| 15 | +from pathlib import Path |
14 | 16 |
|
15 | 17 | import pytest |
16 | 18 |
|
@@ -176,6 +178,49 @@ def test_real_agent_cli_keeps_json_clean_and_sends_only_after_disclosure(isolate |
176 | 178 | assert usage_ping.control('status')['consent'] == 'disabled' |
177 | 179 |
|
178 | 180 |
|
| 181 | +@pytest.mark.parametrize('parent_opt_out', [None, '0', '1']) |
| 182 | +def test_claude_smoke_environment_disables_real_cli_and_sender( |
| 183 | + isolated, collector, monkeypatch, parent_opt_out, |
| 184 | +): |
| 185 | + endpoint, received, accepted, release = collector |
| 186 | + release.set() |
| 187 | + monkeypatch.setenv('LOOPX_USAGE_PING_ENDPOINT', endpoint) |
| 188 | + # An already acknowledged installation must still obey synthetic isolation. |
| 189 | + usage_ping.control('enable') |
| 190 | + path = usage_ping.state_path() |
| 191 | + before = path.read_bytes() |
| 192 | + generation = json.loads(before)['generation'] |
| 193 | + if parent_opt_out is None: |
| 194 | + monkeypatch.delenv('LOOPX_USAGE_PING', raising=False) |
| 195 | + else: |
| 196 | + monkeypatch.setenv('LOOPX_USAGE_PING', parent_opt_out) |
| 197 | + smoke = runpy.run_path(str(Path(__file__).resolve().parents[1] / 'examples/claude-install-optin-smoke.py')) |
| 198 | + env = smoke['_isolated_host_env'](isolated / 'home') |
| 199 | + # Even a regression may only reach this disposable collector. |
| 200 | + env['LOOPX_USAGE_PING_ENDPOINT'] = endpoint |
| 201 | + setup = ( |
| 202 | + 'import json,sys; from pathlib import Path; from loopx import usage_ping; ' |
| 203 | + 'usage_ping.DEFAULT_RUNTIME_ROOT=Path(sys.argv[1]); ' |
| 204 | + ) |
| 205 | + cli = subprocess.run( |
| 206 | + [sys.executable, '-c', setup + 'from loopx.cli_runtime import main; ' |
| 207 | + 'raise SystemExit(main(["version", "--format", "json"]))', str(isolated)], |
| 208 | + env=env, capture_output=True, text=True, timeout=30, |
| 209 | + ) |
| 210 | + assert cli.returncode == 0 and json.loads(cli.stdout)['ok'], cli.stderr |
| 211 | + assert cli.stderr == '' |
| 212 | + sender = subprocess.run( |
| 213 | + [sys.executable, '-c', setup + |
| 214 | + 'print(json.dumps(usage_ping.control("start", generation=sys.argv[2])))', |
| 215 | + str(isolated), generation], |
| 216 | + env=env, capture_output=True, text=True, timeout=30, |
| 217 | + ) |
| 218 | + assert sender.returncode == 0, sender.stderr |
| 219 | + assert json.loads(sender.stdout) == {'sent': False, 'reason': 'blocked'} |
| 220 | + assert not accepted.wait(0.3) and received == [] |
| 221 | + assert path.read_bytes() == before |
| 222 | + |
| 223 | + |
179 | 224 | def test_real_cli_returns_while_http_response_is_held_and_disable_survives(isolated, collector, monkeypatch): |
180 | 225 | endpoint, received, accepted, release = collector |
181 | 226 | monkeypatch.setenv('LOOPX_USAGE_PING_ENDPOINT', endpoint) |
|
0 commit comments