linux observer
feat(cli): add non-interactive flags to setup
+225
-4
+134
-3
src/solstone_linux/cli.py
+134
-3
src/solstone_linux/cli.py
···
24
24
import sys
25
25
from pathlib import Path
26
26
27
-
from . import doctor
27
+
from . import doctor, streams
28
28
from .config import load_config, save_config
29
29
from .streams import stream_name
30
30
···
69
69
70
70
def cmd_setup(args: argparse.Namespace) -> int:
71
71
"""Interactive setup — configure server URL and register."""
72
+
cli_token = args.token if getattr(args, "token", None) else None
73
+
env_token = os.environ.get("SOLSTONE_TOKEN")
74
+
token = cli_token or env_token
75
+
non_interactive = getattr(args, "non_interactive", False)
76
+
77
+
if (
78
+
cli_token is None
79
+
and env_token is None
80
+
and getattr(args, "server_url", None) is None
81
+
and getattr(args, "stream_name", None) is None
82
+
and not non_interactive
83
+
):
84
+
return _cmd_setup_interactive()
85
+
86
+
if cli_token:
87
+
print(
88
+
"warning: --token on the command line may be visible in shell history and /proc on shared machines",
89
+
file=sys.stderr,
90
+
)
91
+
92
+
from .upload import UploadClient
93
+
94
+
config = load_config()
95
+
96
+
server_url = getattr(args, "server_url", None) or config.server_url
97
+
if not server_url:
98
+
if non_interactive:
99
+
print(
100
+
"error: --server-url required with --non-interactive", file=sys.stderr
101
+
)
102
+
return 2
103
+
default_url = config.server_url or ""
104
+
url = input(f"Solstone server URL [{default_url}]: ").strip()
105
+
if url:
106
+
server_url = url
107
+
elif not config.server_url:
108
+
print("Error: server URL is required", file=sys.stderr)
109
+
return 1
110
+
config.server_url = server_url
111
+
112
+
stream_override = getattr(args, "stream_name", None)
113
+
if stream_override:
114
+
config.stream = stream_override
115
+
elif not config.stream:
116
+
try:
117
+
config.stream = streams.stream_name(host=socket.gethostname())
118
+
except ValueError as e:
119
+
print(f"Error deriving stream name: {e}", file=sys.stderr)
120
+
return 1
121
+
122
+
config.ensure_dirs()
123
+
124
+
if token:
125
+
config.key = token
126
+
save_config(config)
127
+
print(f"Server: {config.server_url}")
128
+
print(f"Stream: {config.stream}")
129
+
print("Using provided token; skipping registration.")
130
+
print(f"\nConfig saved to {config.config_path}")
131
+
print(f"Captures will go to {config.captures_dir}")
132
+
print(
133
+
"\nRun 'solstone-linux run' to start, or 'solstone-linux install-service' for systemd."
134
+
)
135
+
return 0
136
+
137
+
print(f"Stream: {config.stream}")
138
+
save_config(config)
139
+
140
+
if not config.key:
141
+
sol = shutil.which("sol")
142
+
if sol:
143
+
print("Registering via sol CLI...")
144
+
try:
145
+
result = subprocess.run(
146
+
[sol, "observer", "--json", "create", config.stream],
147
+
capture_output=True,
148
+
text=True,
149
+
timeout=10,
150
+
)
151
+
if result.returncode == 0:
152
+
data = json.loads(result.stdout)
153
+
config.key = data["key"]
154
+
save_config(config)
155
+
print(f"Registered (key: {config.key[:8]}...)")
156
+
else:
157
+
print("CLI registration failed, trying HTTP...")
158
+
except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, OSError):
159
+
print("CLI registration failed, trying HTTP...")
160
+
161
+
if not config.key:
162
+
print("Registering with server...")
163
+
client = UploadClient(config)
164
+
if client.ensure_registered(config):
165
+
config = load_config()
166
+
print(f"Registered (key: {config.key[:8]}...)")
167
+
else:
168
+
print(
169
+
"Warning: registration failed. Run setup again when server is available."
170
+
)
171
+
if non_interactive:
172
+
return 1
173
+
else:
174
+
print(f"Already registered (key: {config.key[:8]}...)")
175
+
176
+
print(f"\nConfig saved to {config.config_path}")
177
+
print(f"Captures will go to {config.captures_dir}")
178
+
print(
179
+
"\nRun 'solstone-linux run' to start, or 'solstone-linux install-service' for systemd."
180
+
)
181
+
return 0
182
+
183
+
184
+
def _cmd_setup_interactive() -> int:
185
+
# Keep the legacy no-flags setup path separate so its prompt/output stays byte-identical.
72
186
from .upload import UploadClient
73
187
74
188
config = load_config()
···
325
439
)
326
440
327
441
# setup
328
-
subparsers.add_parser("setup", help="Interactive configuration")
442
+
setup_parser = subparsers.add_parser("setup", help="Interactive configuration")
443
+
setup_parser.add_argument("--server-url", help="Server URL (skips prompt)")
444
+
setup_parser.add_argument(
445
+
"--token",
446
+
help="Pre-issued registration key; skips server registration",
447
+
)
448
+
setup_parser.add_argument(
449
+
"--stream-name",
450
+
help="Stream name (defaults to hostname-derived)",
451
+
)
452
+
setup_parser.add_argument(
453
+
"--non-interactive",
454
+
action="store_true",
455
+
help="Fail instead of prompting for missing values",
456
+
)
329
457
330
458
# doctor
331
-
subparsers.add_parser("doctor", help="Verify install prerequisites")
459
+
subparsers.add_parser(
460
+
"doctor",
461
+
help="Verify install prerequisites",
462
+
)
332
463
333
464
# install-service
334
465
subparsers.add_parser("install-service", help="Install systemd user service")
+91
-1
tests/test_cli.py
+91
-1
tests/test_cli.py
···
7
7
from unittest.mock import patch
8
8
9
9
from solstone_linux import cli as cli_module
10
-
from solstone_linux.cli import cmd_install_service
10
+
from solstone_linux.cli import cmd_install_service, cmd_setup
11
+
from solstone_linux.config import Config
11
12
12
13
13
14
def _args() -> argparse.Namespace:
···
116
117
captured = capsys.readouterr()
117
118
assert "nothing to do" not in captured.out.lower()
118
119
assert run_mock.call_count == 8
120
+
121
+
122
+
def test_cmd_setup_non_interactive_happy_path(tmp_path: Path):
123
+
args = argparse.Namespace(
124
+
server_url="https://x",
125
+
token="t",
126
+
stream_name=None,
127
+
non_interactive=True,
128
+
)
129
+
config = Config(base_dir=tmp_path)
130
+
131
+
with patch("solstone_linux.cli.load_config", return_value=config):
132
+
with patch("solstone_linux.cli.save_config") as save_mock:
133
+
with patch("solstone_linux.cli.streams.stream_name", return_value="host-a"):
134
+
with patch("solstone_linux.upload.UploadClient.ensure_registered"):
135
+
assert cmd_setup(args) == 0
136
+
137
+
saved_config = save_mock.call_args.args[0]
138
+
assert saved_config.server_url == "https://x"
139
+
assert saved_config.key == "t"
140
+
assert saved_config.stream == "host-a"
141
+
142
+
143
+
def test_cmd_setup_non_interactive_missing_server_url_fails(tmp_path: Path, capsys):
144
+
args = argparse.Namespace(
145
+
server_url=None,
146
+
token=None,
147
+
stream_name=None,
148
+
non_interactive=True,
149
+
)
150
+
config = Config(base_dir=tmp_path)
151
+
152
+
with patch.dict(os.environ, {}, clear=True):
153
+
with patch("solstone_linux.cli.load_config", return_value=config):
154
+
with patch("solstone_linux.upload.UploadClient.ensure_registered"):
155
+
assert cmd_setup(args) == 2
156
+
157
+
captured = capsys.readouterr()
158
+
assert "--server-url" in captured.err
159
+
160
+
161
+
def test_cmd_setup_env_token_fallback(tmp_path: Path, capsys):
162
+
args = argparse.Namespace(
163
+
server_url="https://x",
164
+
token=None,
165
+
stream_name=None,
166
+
non_interactive=True,
167
+
)
168
+
config = Config(base_dir=tmp_path)
169
+
170
+
with patch.dict(os.environ, {"SOLSTONE_TOKEN": "envtok"}, clear=True):
171
+
with patch("solstone_linux.cli.load_config", return_value=config):
172
+
with patch("solstone_linux.cli.save_config") as save_mock:
173
+
with patch(
174
+
"solstone_linux.cli.streams.stream_name",
175
+
return_value="host-a",
176
+
):
177
+
with patch("solstone_linux.upload.UploadClient.ensure_registered"):
178
+
assert cmd_setup(args) == 0
179
+
180
+
saved_config = save_mock.call_args.args[0]
181
+
captured = capsys.readouterr()
182
+
assert saved_config.key == "envtok"
183
+
assert "shared machines" not in captured.err
184
+
185
+
186
+
def test_cmd_setup_cli_token_beats_env(tmp_path: Path, capsys):
187
+
args = argparse.Namespace(
188
+
server_url="https://x",
189
+
token="clitok",
190
+
stream_name=None,
191
+
non_interactive=True,
192
+
)
193
+
config = Config(base_dir=tmp_path)
194
+
195
+
with patch.dict(os.environ, {"SOLSTONE_TOKEN": "envtok"}, clear=True):
196
+
with patch("solstone_linux.cli.load_config", return_value=config):
197
+
with patch("solstone_linux.cli.save_config") as save_mock:
198
+
with patch(
199
+
"solstone_linux.cli.streams.stream_name",
200
+
return_value="host-a",
201
+
):
202
+
with patch("solstone_linux.upload.UploadClient.ensure_registered"):
203
+
assert cmd_setup(args) == 0
204
+
205
+
saved_config = save_mock.call_args.args[0]
206
+
captured = capsys.readouterr()
207
+
assert saved_config.key == "clitok"
208
+
assert "shared machines" in captured.err