|
1
|
"""Thin wrapper around the fossil binary for write operations.""" |
|
2
|
|
|
3
|
import logging |
|
4
|
import os |
|
5
|
import subprocess |
|
6
|
from pathlib import Path |
|
7
|
|
|
8
|
logger = logging.getLogger(__name__) |
|
9
|
|
|
10
|
|
|
11
|
class FossilCLI: |
|
12
|
"""Wrapper around the fossil binary for write operations.""" |
|
13
|
|
|
14
|
def __init__(self, binary: str | None = None): |
|
15
|
if binary is None: |
|
16
|
from constance import config |
|
17
|
|
|
18
|
binary = config.FOSSIL_BINARY_PATH |
|
19
|
self.binary = binary |
|
20
|
|
|
21
|
@property |
|
22
|
def _env(self): |
|
23
|
import os |
|
24
|
|
|
25
|
return {**os.environ, "USER": "fossilrepo"} |
|
26
|
|
|
27
|
def _run(self, *args: str, timeout: int = 30) -> subprocess.CompletedProcess: |
|
28
|
cmd = [self.binary, *args] |
|
29
|
return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=True, env=self._env) |
|
30
|
|
|
31
|
def ensure_default_user(self, repo_path: Path, username: str = "fossilrepo") -> None: |
|
32
|
"""Ensure a default user exists in the repo. Creates if needed.""" |
|
33
|
try: |
|
34
|
# Check if user exists |
|
35
|
result = subprocess.run( |
|
36
|
[self.binary, "user", "list", "-R", str(repo_path)], |
|
37
|
capture_output=True, |
|
38
|
text=True, |
|
39
|
timeout=10, |
|
40
|
env=self._env, |
|
41
|
) |
|
42
|
if username not in result.stdout: |
|
43
|
subprocess.run( |
|
44
|
[self.binary, "user", "new", username, "", username, "-R", str(repo_path)], |
|
45
|
capture_output=True, |
|
46
|
text=True, |
|
47
|
timeout=10, |
|
48
|
env=self._env, |
|
49
|
) |
|
50
|
subprocess.run( |
|
51
|
[self.binary, "user", "default", username, "-R", str(repo_path)], |
|
52
|
capture_output=True, |
|
53
|
text=True, |
|
54
|
timeout=10, |
|
55
|
env=self._env, |
|
56
|
) |
|
57
|
except Exception: |
|
58
|
pass |
|
59
|
|
|
60
|
def init(self, path: Path) -> Path: |
|
61
|
"""Create a new .fossil repository.""" |
|
62
|
path.parent.mkdir(parents=True, exist_ok=True) |
|
63
|
self._run("init", str(path)) |
|
64
|
return path |
|
65
|
|
|
66
|
def version(self) -> str: |
|
67
|
result = self._run("version") |
|
68
|
return result.stdout.strip() |
|
69
|
|
|
70
|
def is_available(self) -> bool: |
|
71
|
try: |
|
72
|
self._run("version") |
|
73
|
return True |
|
74
|
except (FileNotFoundError, subprocess.CalledProcessError): |
|
75
|
return False |
|
76
|
|
|
77
|
def render_pikchr(self, source: str) -> str: |
|
78
|
"""Render Pikchr markup to SVG. Returns SVG string or empty on failure.""" |
|
79
|
try: |
|
80
|
result = subprocess.run( |
|
81
|
[self.binary, "pikchr", "-"], |
|
82
|
input=source, |
|
83
|
capture_output=True, |
|
84
|
text=True, |
|
85
|
timeout=10, |
|
86
|
) |
|
87
|
if result.returncode == 0: |
|
88
|
return result.stdout |
|
89
|
except (FileNotFoundError, subprocess.TimeoutExpired): |
|
90
|
pass |
|
91
|
return "" |
|
92
|
|
|
93
|
def diff(self, repo_path: Path, from_version: str, to_version: str) -> str: |
|
94
|
"""Run fossil diff between two versions. Returns raw unified diff output.""" |
|
95
|
try: |
|
96
|
result = subprocess.run( |
|
97
|
[self.binary, "diff", "--from", from_version, "--to", to_version, "-R", str(repo_path)], |
|
98
|
capture_output=True, |
|
99
|
text=True, |
|
100
|
timeout=60, |
|
101
|
env=self._env, |
|
102
|
) |
|
103
|
return result.stdout |
|
104
|
except (FileNotFoundError, subprocess.TimeoutExpired): |
|
105
|
return "" |
|
106
|
|
|
107
|
def tarball(self, repo_path: Path, checkin: str) -> bytes: |
|
108
|
"""Generate a tar.gz archive of a checkin. Returns raw bytes.""" |
|
109
|
result = subprocess.run( |
|
110
|
[self.binary, "tarball", checkin, "-R", str(repo_path), "/dev/stdout"], |
|
111
|
capture_output=True, |
|
112
|
timeout=120, |
|
113
|
env=self._env, |
|
114
|
) |
|
115
|
if result.returncode != 0: |
|
116
|
return b"" |
|
117
|
return result.stdout |
|
118
|
|
|
119
|
def zip_archive(self, repo_path: Path, checkin: str) -> bytes: |
|
120
|
"""Generate a zip archive of a checkin. Returns raw bytes.""" |
|
121
|
import tempfile |
|
122
|
|
|
123
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=True) as tmp: |
|
124
|
result = subprocess.run( |
|
125
|
[self.binary, "zip", checkin, tmp.name, "-R", str(repo_path)], |
|
126
|
capture_output=True, |
|
127
|
text=True, |
|
128
|
timeout=120, |
|
129
|
env=self._env, |
|
130
|
) |
|
131
|
if result.returncode != 0: |
|
132
|
return b"" |
|
133
|
return Path(tmp.name).read_bytes() |
|
134
|
|
|
135
|
def blame(self, repo_path: Path, filename: str) -> list[dict]: |
|
136
|
"""Run fossil blame on a file. Returns [{user, uuid, line_num, text}]. |
|
137
|
|
|
138
|
Requires creating a temp checkout since blame needs an open checkout. |
|
139
|
""" |
|
140
|
import tempfile |
|
141
|
|
|
142
|
lines = [] |
|
143
|
tmpdir = tempfile.mkdtemp(prefix="fossilrepo-blame-") |
|
144
|
try: |
|
145
|
# Open a checkout in the temp dir |
|
146
|
subprocess.run( |
|
147
|
[self.binary, "open", str(repo_path), "--workdir", tmpdir], |
|
148
|
capture_output=True, |
|
149
|
text=True, |
|
150
|
timeout=30, |
|
151
|
cwd=tmpdir, |
|
152
|
) |
|
153
|
# Run blame |
|
154
|
result = subprocess.run( |
|
155
|
[self.binary, "blame", filename], |
|
156
|
capture_output=True, |
|
157
|
text=True, |
|
158
|
timeout=30, |
|
159
|
cwd=tmpdir, |
|
160
|
) |
|
161
|
if result.returncode == 0: |
|
162
|
import re |
|
163
|
|
|
164
|
for line in result.stdout.splitlines(): |
|
165
|
# Format: "hash date user: code" |
|
166
|
m = re.match(r"([0-9a-f]+)\s+(\S+)\s+([^:]+):\s?(.*)", line) |
|
167
|
if m: |
|
168
|
lines.append( |
|
169
|
{ |
|
170
|
"uuid": m.group(1), |
|
171
|
"date": m.group(2), |
|
172
|
"user": m.group(3).strip(), |
|
173
|
"text": m.group(4), |
|
174
|
} |
|
175
|
) |
|
176
|
# Close checkout |
|
177
|
subprocess.run([self.binary, "close", "--force"], capture_output=True, cwd=tmpdir, timeout=10, env=self._env) |
|
178
|
except Exception: |
|
179
|
pass |
|
180
|
finally: |
|
181
|
import shutil |
|
182
|
|
|
183
|
shutil.rmtree(tmpdir, ignore_errors=True) |
|
184
|
return lines |
|
185
|
|
|
186
|
def push(self, repo_path: Path, remote_url: str = "") -> dict: |
|
187
|
"""Push to the remote. Returns {success, artifacts_sent, message}.""" |
|
188
|
import re |
|
189
|
|
|
190
|
cmd = [self.binary, "push", "-R", str(repo_path)] |
|
191
|
if remote_url: |
|
192
|
cmd.append(remote_url) |
|
193
|
try: |
|
194
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, env=self._env) |
|
195
|
artifacts = 0 |
|
196
|
m = re.search(r"sent:\s*(\d+)", result.stdout) |
|
197
|
if m: |
|
198
|
artifacts = int(m.group(1)) |
|
199
|
return {"success": result.returncode == 0, "artifacts_sent": artifacts, "message": result.stdout.strip()} |
|
200
|
except (FileNotFoundError, subprocess.TimeoutExpired) as e: |
|
201
|
return {"success": False, "artifacts_sent": 0, "message": str(e)} |
|
202
|
|
|
203
|
def sync(self, repo_path: Path, remote_url: str = "") -> dict: |
|
204
|
"""Bidirectional sync with remote. Returns {success, message}.""" |
|
205
|
cmd = [self.binary, "sync", "-R", str(repo_path)] |
|
206
|
if remote_url: |
|
207
|
cmd.append(remote_url) |
|
208
|
try: |
|
209
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, env=self._env) |
|
210
|
return {"success": result.returncode == 0, "message": result.stdout.strip()} |
|
211
|
except (FileNotFoundError, subprocess.TimeoutExpired) as e: |
|
212
|
return {"success": False, "message": str(e)} |
|
213
|
|
|
214
|
def pull(self, repo_path: Path) -> dict: |
|
215
|
"""Pull updates from the remote. Returns {success, artifacts_received, message}.""" |
|
216
|
try: |
|
217
|
result = subprocess.run( |
|
218
|
[self.binary, "pull", "-R", str(repo_path)], |
|
219
|
capture_output=True, |
|
220
|
text=True, |
|
221
|
timeout=60, |
|
222
|
) |
|
223
|
import re |
|
224
|
|
|
225
|
artifacts = 0 |
|
226
|
m = re.search(r"received:\s*(\d+)", result.stdout) |
|
227
|
if m: |
|
228
|
artifacts = int(m.group(1)) |
|
229
|
return {"success": result.returncode == 0, "artifacts_received": artifacts, "message": result.stdout.strip()} |
|
230
|
except (FileNotFoundError, subprocess.TimeoutExpired) as e: |
|
231
|
return {"success": False, "artifacts_received": 0, "message": str(e)} |
|
232
|
|
|
233
|
def get_remote_url(self, repo_path: Path) -> str: |
|
234
|
"""Get the configured remote URL for a repo.""" |
|
235
|
try: |
|
236
|
result = subprocess.run( |
|
237
|
[self.binary, "remote", "-R", str(repo_path)], |
|
238
|
capture_output=True, |
|
239
|
text=True, |
|
240
|
timeout=10, |
|
241
|
) |
|
242
|
return result.stdout.strip() if result.returncode == 0 else "" |
|
243
|
except (FileNotFoundError, subprocess.TimeoutExpired): |
|
244
|
return "" |
|
245
|
|
|
246
|
def wiki_commit(self, repo_path: Path, page_name: str, content: str, user: str = "") -> bool: |
|
247
|
"""Create or update a wiki page. Pipes content to fossil wiki commit.""" |
|
248
|
cmd = [self.binary, "wiki", "commit", page_name, "-R", str(repo_path)] |
|
249
|
if user: |
|
250
|
cmd.extend(["--technote-user", user]) |
|
251
|
result = subprocess.run(cmd, input=content, capture_output=True, text=True, timeout=30, env=self._env) |
|
252
|
return result.returncode == 0 |
|
253
|
|
|
254
|
def wiki_create(self, repo_path: Path, page_name: str, content: str) -> bool: |
|
255
|
"""Create a new wiki page.""" |
|
256
|
cmd = [self.binary, "wiki", "create", page_name, "-R", str(repo_path)] |
|
257
|
result = subprocess.run(cmd, input=content, capture_output=True, text=True, timeout=30, env=self._env) |
|
258
|
return result.returncode == 0 |
|
259
|
|
|
260
|
def ticket_add(self, repo_path: Path, fields: dict) -> bool: |
|
261
|
"""Add a new ticket. Fields dict maps field names to values.""" |
|
262
|
cmd = [self.binary, "ticket", "add", "-R", str(repo_path)] |
|
263
|
for key, value in fields.items(): |
|
264
|
cmd.append(f"{key}") |
|
265
|
cmd.append(f"{value}") |
|
266
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, env=self._env) |
|
267
|
return result.returncode == 0 |
|
268
|
|
|
269
|
def ticket_change(self, repo_path: Path, uuid: str, fields: dict) -> bool: |
|
270
|
"""Update an existing ticket.""" |
|
271
|
cmd = [self.binary, "ticket", "change", uuid, "-R", str(repo_path)] |
|
272
|
for key, value in fields.items(): |
|
273
|
cmd.append(f"{key}") |
|
274
|
cmd.append(f"{value}") |
|
275
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, env=self._env) |
|
276
|
return result.returncode == 0 |
|
277
|
|
|
278
|
def technote_create(self, repo_path: Path, title: str, body: str, timestamp: str | None = None, user: str = "") -> bool: |
|
279
|
"""Create a new technote. |
|
280
|
|
|
281
|
Uses: fossil wiki create --technote <timestamp> <title> -R <repo> |
|
282
|
with the body piped via stdin. |
|
283
|
""" |
|
284
|
if not timestamp: |
|
285
|
from datetime import UTC, datetime |
|
286
|
|
|
287
|
timestamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") |
|
288
|
|
|
289
|
cmd = [self.binary, "wiki", "create", title, "--technote", timestamp, "-R", str(repo_path)] |
|
290
|
if user: |
|
291
|
cmd.extend(["--technote-user", user]) |
|
292
|
result = subprocess.run(cmd, input=body, capture_output=True, text=True, timeout=30, env=self._env) |
|
293
|
return result.returncode == 0 |
|
294
|
|
|
295
|
def technote_edit(self, repo_path: Path, technote_id: str, body: str, user: str = "") -> bool: |
|
296
|
"""Edit an existing technote body. |
|
297
|
|
|
298
|
Uses: fossil wiki commit <comment> --technote <technote_id> -R <repo> |
|
299
|
with the new body piped via stdin. |
|
300
|
""" |
|
301
|
cmd = [self.binary, "wiki", "commit", "update", "--technote", technote_id, "-R", str(repo_path)] |
|
302
|
if user: |
|
303
|
cmd.extend(["--technote-user", user]) |
|
304
|
result = subprocess.run(cmd, input=body, capture_output=True, text=True, timeout=30, env=self._env) |
|
305
|
return result.returncode == 0 |
|
306
|
|
|
307
|
def uv_add(self, repo_path: Path, name: str, filepath: Path) -> bool: |
|
308
|
"""Add an unversioned file: fossil uv add <filepath> --as <name> -R <repo>.""" |
|
309
|
cmd = [self.binary, "uv", "add", str(filepath), "--as", name, "-R", str(repo_path)] |
|
310
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60, env=self._env) |
|
311
|
return result.returncode == 0 |
|
312
|
|
|
313
|
def uv_cat(self, repo_path: Path, name: str) -> bytes: |
|
314
|
"""Get unversioned file content: fossil uv cat <name> -R <repo>. |
|
315
|
|
|
316
|
Returns raw bytes. Raises FileNotFoundError if the file doesn't exist |
|
317
|
or the command fails. |
|
318
|
""" |
|
319
|
cmd = [self.binary, "uv", "cat", name, "-R", str(repo_path)] |
|
320
|
result = subprocess.run(cmd, capture_output=True, timeout=60, env=self._env) |
|
321
|
if result.returncode != 0: |
|
322
|
raise FileNotFoundError(f"Unversioned file not found: {name}") |
|
323
|
return result.stdout |
|
324
|
|
|
325
|
def git_export(self, repo_path: Path, mirror_dir: Path, autopush_url: str = "", auth_token: str = "") -> dict: |
|
326
|
"""Export Fossil repo to a Git mirror directory. Incremental. |
|
327
|
|
|
328
|
When auth_token is provided, credentials are passed via Git environment |
|
329
|
variables instead of being embedded in the URL (avoids exposure in |
|
330
|
process args and command output). |
|
331
|
|
|
332
|
Returns {success, message}. |
|
333
|
""" |
|
334
|
mirror_dir.mkdir(parents=True, exist_ok=True) |
|
335
|
cmd = [self.binary, "git", "export", str(mirror_dir), "-R", str(repo_path)] |
|
336
|
|
|
337
|
env = dict(self._env) |
|
338
|
|
|
339
|
temp_paths = [] |
|
340
|
if autopush_url: |
|
341
|
cmd.extend(["--autopush", autopush_url]) |
|
342
|
if auth_token: |
|
343
|
env["GIT_TERMINAL_PROMPT"] = "0" |
|
344
|
# Use a temporary askpass script instead of a shell credential |
|
345
|
# helper to avoid command injection via token metacharacters. |
|
346
|
# The token is stored in a separate file so it never appears |
|
347
|
# in shell syntax. |
|
348
|
import stat |
|
349
|
import tempfile |
|
350
|
|
|
351
|
token_fd, token_path = tempfile.mkstemp(suffix=".tok") |
|
352
|
with os.fdopen(token_fd, "w") as f: |
|
353
|
f.write(auth_token) |
|
354
|
os.chmod(token_path, stat.S_IRUSR) |
|
355
|
temp_paths.append(token_path) |
|
356
|
|
|
357
|
askpass_fd, askpass_path = tempfile.mkstemp(suffix=".sh") |
|
358
|
with os.fdopen(askpass_fd, "w") as f: |
|
359
|
f.write(f"#!/bin/sh\ncat '{token_path}'\n") |
|
360
|
os.chmod(askpass_path, stat.S_IRWXU) |
|
361
|
temp_paths.append(askpass_path) |
|
362
|
env["GIT_ASKPASS"] = askpass_path |
|
363
|
|
|
364
|
try: |
|
365
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300, env=env) |
|
366
|
output = (result.stdout + result.stderr).strip() |
|
367
|
if auth_token: |
|
368
|
output = output.replace(auth_token, "[REDACTED]") |
|
369
|
return {"success": result.returncode == 0, "message": output} |
|
370
|
except subprocess.TimeoutExpired: |
|
371
|
return {"success": False, "message": "Export timed out after 5 minutes"} |
|
372
|
finally: |
|
373
|
for p in temp_paths: |
|
374
|
os.unlink(p) |
|
375
|
|
|
376
|
def generate_ssh_key(self, key_path: Path, comment: str = "fossilrepo") -> dict: |
|
377
|
"""Generate an SSH key pair for Git authentication. |
|
378
|
|
|
379
|
Returns {success, public_key, fingerprint}. |
|
380
|
""" |
|
381
|
import os |
|
382
|
|
|
383
|
try: |
|
384
|
key_path.parent.mkdir(parents=True, exist_ok=True) |
|
385
|
result = subprocess.run( |
|
386
|
["ssh-keygen", "-t", "ed25519", "-f", str(key_path), "-N", "", "-C", comment], |
|
387
|
capture_output=True, |
|
388
|
text=True, |
|
389
|
timeout=10, |
|
390
|
env={**os.environ}, |
|
391
|
) |
|
392
|
if result.returncode == 0: |
|
393
|
pub_key = key_path.with_suffix(".pub").read_text().strip() |
|
394
|
# Get fingerprint |
|
395
|
fp_result = subprocess.run( |
|
396
|
["ssh-keygen", "-lf", str(key_path.with_suffix(".pub"))], |
|
397
|
capture_output=True, |
|
398
|
text=True, |
|
399
|
timeout=10, |
|
400
|
) |
|
401
|
fingerprint = fp_result.stdout.strip().split()[1] if fp_result.returncode == 0 else "" |
|
402
|
return {"success": True, "public_key": pub_key, "fingerprint": fingerprint} |
|
403
|
except Exception as e: |
|
404
|
return {"success": False, "public_key": "", "fingerprint": "", "error": str(e)} |
|
405
|
return {"success": False, "public_key": "", "fingerprint": ""} |
|
406
|
|
|
407
|
def http_proxy(self, repo_path: Path, request_body: bytes, content_type: str = "", localauth: bool = True) -> tuple[bytes, str]: |
|
408
|
"""Proxy a single Fossil HTTP sync request. |
|
409
|
|
|
410
|
Runs ``fossil http <repo_path>`` with a full HTTP request on stdin. |
|
411
|
Fossil reads the HTTP method line + headers + body from stdin and |
|
412
|
writes a full HTTP response (headers + body) to stdout. |
|
413
|
|
|
414
|
When *localauth* is True, ``--localauth`` grants full push permissions. |
|
415
|
When False, only anonymous pull/clone is allowed (for public repos). |
|
416
|
""" |
|
417
|
import os |
|
418
|
|
|
419
|
env = { |
|
420
|
**os.environ, |
|
421
|
**{k: v for k, v in self._env.items() if k not in os.environ or k == "USER"}, |
|
422
|
} |
|
423
|
# Ensure GATEWAY_INTERFACE is NOT set — it triggers CGI auto-detect |
|
424
|
# which conflicts with the explicit "http" subcommand. |
|
425
|
env.pop("GATEWAY_INTERFACE", None) |
|
426
|
|
|
427
|
# Build a raw HTTP request for fossil http's stdin |
|
428
|
http_request = ( |
|
429
|
f"POST /xfer HTTP/1.1\r\n" |
|
430
|
f"Host: localhost\r\n" |
|
431
|
f"Content-Type: {content_type or 'application/x-fossil'}\r\n" |
|
432
|
f"Content-Length: {len(request_body)}\r\n" |
|
433
|
f"\r\n" |
|
434
|
).encode() + request_body |
|
435
|
|
|
436
|
cmd = [self.binary, "http", str(repo_path)] |
|
437
|
if localauth: |
|
438
|
cmd.append("--localauth") |
|
439
|
|
|
440
|
try: |
|
441
|
result = subprocess.run( |
|
442
|
cmd, |
|
443
|
input=http_request, |
|
444
|
capture_output=True, |
|
445
|
timeout=120, |
|
446
|
env=env, |
|
447
|
) |
|
448
|
except subprocess.TimeoutExpired: |
|
449
|
logger.error("fossil http timed out for %s", repo_path) |
|
450
|
raise |
|
451
|
except FileNotFoundError: |
|
452
|
logger.error("fossil binary not found at %s", self.binary) |
|
453
|
raise |
|
454
|
|
|
455
|
if result.returncode != 0: |
|
456
|
stderr_text = result.stderr.decode("utf-8", errors="replace") |
|
457
|
logger.warning("fossil http exited %d for %s: %s", result.returncode, repo_path, stderr_text) |
|
458
|
|
|
459
|
raw = result.stdout |
|
460
|
|
|
461
|
# Fossil CGI output: HTTP headers separated from body by a blank line. |
|
462
|
# Try \r\n\r\n first (standard HTTP), fall back to \n\n. |
|
463
|
separator = b"\r\n\r\n" |
|
464
|
sep_idx = raw.find(separator) |
|
465
|
if sep_idx == -1: |
|
466
|
separator = b"\n\n" |
|
467
|
sep_idx = raw.find(separator) |
|
468
|
|
|
469
|
if sep_idx == -1: |
|
470
|
# No header/body separator found — treat the entire output as body. |
|
471
|
return raw, "application/x-fossil" |
|
472
|
|
|
473
|
header_block = raw[:sep_idx] |
|
474
|
body = raw[sep_idx + len(separator) :] |
|
475
|
|
|
476
|
# Parse Content-Type from the CGI headers. |
|
477
|
response_content_type = "application/x-fossil" |
|
478
|
for line in header_block.split(b"\r\n" if b"\r\n" in header_block else b"\n"): |
|
479
|
if line.lower().startswith(b"content-type:"): |
|
480
|
response_content_type = line.split(b":", 1)[1].strip().decode("utf-8", errors="replace") |
|
481
|
break |
|
482
|
|
|
483
|
return body, response_content_type |
|
484
|
|
|
485
|
def shun(self, repo_path: Path, artifact_uuid: str, reason: str = "") -> dict: |
|
486
|
"""Shun (permanently remove) an artifact from the repo. |
|
487
|
|
|
488
|
This is IRREVERSIBLE. The artifact is permanently expunged from the repository. |
|
489
|
""" |
|
490
|
cmd = [self.binary, "shun", artifact_uuid, "-R", str(repo_path)] |
|
491
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, env=self._env) |
|
492
|
return {"success": result.returncode == 0, "message": (result.stdout + result.stderr).strip()} |
|
493
|
|
|
494
|
def shun_list(self, repo_path: Path) -> list[str]: |
|
495
|
"""List currently shunned artifact UUIDs.""" |
|
496
|
cmd = [self.binary, "shun", "--list", "-R", str(repo_path)] |
|
497
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, env=self._env) |
|
498
|
if result.returncode == 0: |
|
499
|
return [line.strip() for line in result.stdout.strip().splitlines() if line.strip()] |
|
500
|
return [] |
|
501
|
|
|
502
|
def bundle_export(self, repo_path: Path, branch: str = "", checkin: str = "") -> bytes: |
|
503
|
"""Export a Fossil bundle. Returns raw bytes of the .bundle file.""" |
|
504
|
import tempfile |
|
505
|
|
|
506
|
with tempfile.NamedTemporaryFile(suffix=".bundle", delete=False) as tmp: |
|
507
|
tmp_path = Path(tmp.name) |
|
508
|
try: |
|
509
|
cmd = [self.binary, "bundle", "export", str(tmp_path), "-R", str(repo_path)] |
|
510
|
if branch: |
|
511
|
cmd += ["--branch", branch] |
|
512
|
elif checkin: |
|
513
|
cmd += ["--checkin", checkin] |
|
514
|
else: |
|
515
|
cmd += ["--branch", "trunk"] |
|
516
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, env=self._env) |
|
517
|
if result.returncode != 0: |
|
518
|
return b"" |
|
519
|
return tmp_path.read_bytes() |
|
520
|
finally: |
|
521
|
tmp_path.unlink(missing_ok=True) |
|
522
|
|
|
523
|
def bundle_import(self, repo_path: Path, bundle_bytes: bytes, publish: bool = False) -> bool: |
|
524
|
"""Import a Fossil bundle into the repo. Returns True on success.""" |
|
525
|
import tempfile |
|
526
|
|
|
527
|
with tempfile.NamedTemporaryFile(suffix=".bundle", delete=False) as tmp: |
|
528
|
tmp.write(bundle_bytes) |
|
529
|
tmp_path = Path(tmp.name) |
|
530
|
try: |
|
531
|
cmd = [self.binary, "bundle", "import", str(tmp_path), "-R", str(repo_path)] |
|
532
|
if publish: |
|
533
|
cmd.append("--publish") |
|
534
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, env=self._env) |
|
535
|
return result.returncode == 0 |
|
536
|
finally: |
|
537
|
tmp_path.unlink(missing_ok=True) |
|
538
|
|