Coverage for src/lektor_ng/publisher.py: 59%

1from __future__ import annotations 

2 

3import errno 

4import hashlib 

5import io 

6import os 

7import posixpath 

8import urllib.parse 

9from collections.abc import Callable, Generator, Iterable, Iterator, Mapping, Sequence 

10from contextlib import AbstractContextManager, ExitStack, contextmanager, suppress 

11from ftplib import Error as FTPError 

12from inspect import cleandoc 

13from pathlib import Path 

14from subprocess import DEVNULL, PIPE, STDOUT, CalledProcessError, CompletedProcess 

15from tempfile import TemporaryDirectory 

16from typing import TYPE_CHECKING, Any, NoReturn 

17from urllib.parse import urlsplit 

18from warnings import warn 

19 

20from werkzeug.datastructures import MultiDict 

21 

22from lektor_ng.compat import werkzeug_urls_URL 

23from lektor_ng.exception import LektorException 

24from lektor_ng.utils import bool_from_string, locate_executable, portable_popen 

25 

26if TYPE_CHECKING: # pragma: no cover 

27 from _typeshed import StrOrBytesPath, StrPath 

28 from lektor.environment import Environment 

29 

30 

31def _parse_query(query: str, **kwargs: Any) -> MultiDict: 

32 return MultiDict(urllib.parse.parse_qsl(query, **kwargs)) 

33 

34 

35def _ascii_host(host: str) -> str: 

36 """Translate internationalized domain name to IDNA-encoded ASCII.""" 

37 return host.encode("idna").decode("ascii") 

38 

39 

40@contextmanager 

41def _ssh_key_file(credentials: Mapping[str, str] | None) -> Iterator[StrPath | None]: 

42 with ExitStack() as stack: 

43 key_file: StrPath | None 

44 key_file = credentials.get("key_file") if credentials else None 

45 key = credentials.get("key") if credentials else None 

46 if not key_file and key: 

47 if ":" in key: 

48 key_type, _, key = key.partition(":") 

49 key_type = key_type.upper() 

50 else: 

51 key_type = "RSA" 

52 key_file = Path(stack.enter_context(TemporaryDirectory()), "keyfile") 

53 with key_file.open("w", encoding="utf-8") as f: 

54 f.write(f"-----BEGIN {key_type} PRIVATE KEY-----\n") 

55 f.writelines(key[x : x + 64] + "\n" for x in range(0, len(key), 64)) 

56 f.write(f"-----END {key_type} PRIVATE KEY-----\n") 

57 yield key_file 

58 

59 

60@contextmanager 

61def _ssh_command(credentials: Mapping[str, str] | None, port: int | None = None) -> Iterator[str | None]: 

62 with _ssh_key_file(credentials) as key_file: 

63 args = [] 

64 if port: 

65 args.append(f" -p {port}") 

66 if key_file: 

67 args.append(f' -i "{key_file}" -o IdentitiesOnly=yes') 

68 if args: 

69 ssh_command = "ssh" + " ".join(args) 

70 else: 

71 ssh_command = None 

72 

73 yield ssh_command 

74 

75 

76class PublishError(LektorException): 

77 """Raised by publishers if something goes wrong.""" 

78 

79 

80class Command(AbstractContextManager["Command"]): 

81 """A wrapper around subprocess.Popen to facilitate streaming output via generator. 

82 

83 :param argline: Command with arguments to execute. 

84 :param cwd: Optional. Directory in which to execute command. 

85 :param env: Optional. Environment with which to run command. 

86 :param capture: Default `True`. Whether to capture stdout and stderr. 

87 :param silent: Default `False`. Discard output altogether. 

88 :param check: Default `False`. 

89 If set, raise ``CalledProcessError`` on non-zero return code. 

90 :param input: Optional. A string to feed to the subprocess via stdin. 

91 :param capture_stdout: Default `False`. Capture stdout and 

92 return in ``CompletedProcess.stdout``. 

93 

94 Basic Usage 

95 =========== 

96 

97 To run a command, returning any output on stdout or stderr to the caller 

98 as an iterable (generator), while checking the return code from the command: 

99 

100 def run_command(argline): 

101 # This passes the output 

102 rv = yield from Command(argline) 

103 if rv.returncode != 0: 

104 raise RuntimeError("Command failed!") 

105 

106 This could be called as follows: 

107 

108 for outline in run_command(('ls')): 

109 print(outline.rstrip()) 

110 

111 Supplying input via stdin, Capturing stdout 

112 =========================================== 

113 

114 The following example shows how input may be fed to a subprocess via stdin, 

115 and how stdout may be captured for further processing. 

116 

117 def run_wc(input): 

118 rv = yield from Command( 

119 ('wc'), check=True, input=input, capture_stdout=True 

120 ) 

121 lines, words, chars = rv.stdout.split() 

122 print(f"{words} words, {chars} chars") 

123 

124 stderr_lines = list(run_wc("a few words")) 

125 # prints "3 words, 11 chars" 

126 

127 Note that ``check=True`` will cause a ``CalledProcessError`` to be raised if the 

128 ``wc`` subprocess returns a non-zero return code. 

129 

130 """ 

131 

132 def __init__( 

133 self, 

134 argline: Iterable[str], 

135 *, 

136 cwd: StrOrBytesPath | None = None, 

137 env: Mapping[str, str] | None = None, 

138 capture: bool = True, 

139 silent: bool = False, 

140 check: bool = False, 

141 input: str | None = None, 

142 capture_stdout: bool = False, 

143 ) -> None: 

144 kwargs: dict[str, Any] = {"cwd": cwd} 

145 if env: 

146 kwargs["env"] = {**os.environ, **env} 

147 if silent: 

148 kwargs["stdout"] = DEVNULL 

149 kwargs["stderr"] = DEVNULL 

150 capture = False 

151 if input is not None: 

152 kwargs["stdin"] = PIPE 

153 if capture or capture_stdout: 

154 kwargs["stdout"] = PIPE 

155 if capture: 

156 kwargs["stderr"] = STDOUT if not capture_stdout else PIPE 

157 

158 # Python >= 3.7 has sane encoding defaults in the case that the system is 

159 # (likely mis-)configured to use ASCII as the default encoding (PEP538). 

160 # It also provides a way for the user to force the use of UTF-8 (PEP540). 

161 kwargs["text"] = True 

162 kwargs["errors"] = "replace" 

163 

164 self.capture = capture # b/c - unused 

165 self.check = check 

166 self._stdout = None 

167 

168 with ExitStack() as stack: 

169 self._cmd = stack.enter_context(portable_popen(list(argline), **kwargs)) 

170 self._closer: Callable[[], None] | None = stack.pop_all().close 

171 

172 if input is not None or capture_stdout: 

173 self._output = self._communicate(input, capture_stdout, capture) 

174 elif capture: 

175 self._output = self._cmd.stdout 

176 else: 

177 self._output = None 

178 

179 def _communicate(self, input: str | None, capture_stdout: bool, capture: bool) -> Iterator[str] | None: 

180 proc = self._cmd 

181 try: 

182 if capture_stdout: 

183 self._stdout, errout = proc.communicate(input) 

184 else: 

185 errout, _ = proc.communicate(input) 

186 except BaseException: 

187 proc.kill() 

188 with suppress(CalledProcessError): 

189 self.close() 

190 raise 

191 if capture: 

192 return iter(errout.splitlines()) 

193 return None 

194 

195 def close(self) -> None: 

196 """Wait for subprocess to complete. 

197 

198 If check=True was passed to the constructor, raises ``CalledProcessError`` 

199 if the subprocess returns a non-zero status code. 

200 """ 

201 closer, self._closer = self._closer, None 

202 if closer: 

203 # This waits for process and closes standard file descriptors 

204 closer() 

205 if self.check: 

206 rc = self._cmd.poll() 

207 if rc != 0: 

208 raise CalledProcessError(rc, self._cmd.args, self._stdout) 

209 

210 def wait(self) -> int: 

211 """Wait for subprocess to complete. Return status code.""" 

212 self._cmd.wait() 

213 self.close() 

214 return self._cmd.returncode 

215 

216 def result(self) -> CompletedProcess[str]: 

217 """Wait for subprocess to complete. Return ``CompletedProcess`` instance. 

218 

219 If ``capture_stdout=True`` was passed to the constructor, the output 

220 captured from stdout will be available on the ``.stdout`` attribute 

221 of the return value. 

222 """ 

223 return CompletedProcess(self._cmd.args, self.wait(), self._stdout) 

224 

225 @property 

226 def returncode(self) -> int | None: 

227 """Return exit status of the subprocess. 

228 

229 Or ``None`` if the subprocess is still alive. 

230 """ 

231 return self._cmd.returncode 

232 

233 def __exit__(self, *__: object) -> None: 

234 self.close() 

235 

236 def __iter__(self) -> Generator[str, None, CompletedProcess[str]]: 

237 """A generator with yields any captured output and returns a 

238 ``CompletedProcess``. 

239 

240 If ``capture`` is ``True`` (the default). Both stdout and stderr are available 

241 in the iterator output. 

242 

243 If ``capture_stdout`` is set, stdout is captured to a string which is made 

244 available via ``CompletedProcess.stdout`` attribute of the return value. Stderr 

245 output is available via the iterator output, as normal. 

246 """ 

247 if self._output is None: 

248 raise RuntimeError("Not capturing") 

249 for line in self._output: 

250 yield line.rstrip() 

251 return self.result() 

252 

253 safe_iter = __iter__ # b/c - deprecated 

254 

255 @property 

256 def output(self) -> Iterator[str]: # b/c - deprecated 

257 return self.safe_iter() 

258 

259 

260class Publisher: 

261 def __init__(self, env: Environment, output_path: str) -> None: 

262 self.env = env 

263 self.output_path = os.path.abspath(output_path) 

264 

265 def fail(self, message: str) -> NoReturn: 

266 # pylint: disable=no-self-use 

267 raise PublishError(message) 

268 

269 def publish( 

270 self, 

271 target_url: str, 

272 credentials: Mapping[str, str] | None = None, 

273 **extra: Any, 

274 ) -> Iterator[str]: 

275 raise NotImplementedError() 

276 

277 

278class RsyncPublisher(Publisher): 

279 @contextmanager 

280 def get_command(self, target_url, credentials): 

281 credentials = credentials or {} 

282 argline = ["rsync", "-rclzv", "--exclude=.lektor"] 

283 target = [] 

284 env = {} 

285 

286 url = urlsplit(target_url) 

287 options = _parse_query(url.query, keep_blank_values=True) 

288 exclude = options.getlist("exclude") 

289 for file in exclude: 

290 argline.extend(("--exclude", file)) 

291 

292 delete = options.get("delete", False) in ("", "on", "yes", "true", "1", None) 

293 if delete: 

294 argline.append("--delete-after") 

295 

296 with _ssh_command(credentials, url.port) as ssh_command: 

297 if ssh_command: 

298 argline.extend(("-e", ssh_command)) 

299 

300 username = credentials.get("username") or url.username 

301 if username: 

302 target.append(username + "@") 

303 

304 if url.hostname is not None: 

305 target.append(_ascii_host(url.hostname)) 

306 target.append(":") 

307 target.append(url.path.rstrip("/") + "/") 

308 

309 argline.append(self.output_path.rstrip("/\\") + "/") 

310 argline.append("".join(target)) 

311 yield Command(argline, env=env) 

312 

313 def publish(self, target_url, credentials=None, **extra): 

314 with self.get_command(target_url, credentials) as client: 

315 yield from client 

316 

317 

318class FtpConnection: 

319 def __init__(self, target_url, credentials=None): 

320 credentials = credentials or {} 

321 url = urlsplit(target_url) 

322 if url.hostname is None: 

323 raise PublishError("No host name was specified in the target URL ({target_url})") 

324 self.con = self.make_connection() 

325 self.url = url 

326 self.username = credentials.get("username") or url.username 

327 self.password = credentials.get("password") or url.password 

328 self.log_buffer = [] 

329 self._known_folders = set() 

330 

331 @staticmethod 

332 def make_connection(): 

333 # pylint: disable=import-outside-toplevel 

334 from ftplib import FTP 

335 

336 return FTP() 

337 

338 def drain_log(self): 

339 log = self.log_buffer[:] 

340 del self.log_buffer[:] 

341 for chunk in log: 

342 for line in chunk.splitlines(): 

343 if not isinstance(line, str): 

344 line = line.decode("utf-8", "replace") 

345 yield line.rstrip() 

346 

347 def connect(self): 

348 options = _parse_query(self.url.query, keep_blank_values=True) 

349 assert self.url.hostname is not None 

350 host = _ascii_host(self.url.hostname) 

351 port = self.url.port or 21 

352 

353 log = self.log_buffer 

354 log.append("000 Connecting to server ...") 

355 try: 

356 log.append(self.con.connect(host, port)) 

357 except Exception as e: 

358 log.append("000 Could not connect.") 

359 log.append(str(e)) 

360 return False 

361 

362 try: 

363 credentials = {} 

364 if self.username: 

365 credentials["user"] = self.username 

366 if self.password: 

367 credentials["passwd"] = self.password 

368 log.append(self.con.login(**credentials)) 

369 

370 except Exception as e: 

371 log.append("000 Could not authenticate.") 

372 log.append(str(e)) 

373 return False 

374 

375 passive = options.get("passive") in ("on", "yes", "true", "1", None) 

376 log.append("000 Using passive mode: %s" % (passive and "yes" or "no")) 

377 self.con.set_pasv(passive) 

378 

379 try: 

380 log.append(self.con.cwd(self.url.path)) 

381 except Exception as e: 

382 log.append(str(e)) 

383 return False 

384 

385 log.append("000 Connected!") 

386 return True 

387 

388 def mkdir(self, path, recursive=True): 

389 if not isinstance(path, str): 

390 path = path.decode("utf-8") 

391 if path in self._known_folders: 

392 return 

393 dirname, _ = posixpath.split(path) 

394 if dirname and recursive: 

395 self.mkdir(dirname) 

396 try: 

397 self.con.mkd(path) 

398 except FTPError as e: 

399 msg = str(e) 

400 if msg[:4] != "550 ": 

401 self.log_buffer.append(str(e)) 

402 return 

403 self._known_folders.add(path) 

404 

405 def append(self, filename, data): 

406 if not isinstance(filename, str): 

407 filename = filename.decode("utf-8") 

408 

409 input = io.BytesIO(data.encode("utf-8")) 

410 

411 try: 

412 self.con.storbinary("APPE " + filename, input) 

413 except FTPError as e: 

414 self.log_buffer.append(str(e)) 

415 return False 

416 return True 

417 

418 def get_file(self, filename, out=None): 

419 if not isinstance(filename, str): 

420 filename = filename.decode("utf-8") 

421 getvalue = False 

422 if out is None: 

423 out = io.BytesIO() 

424 getvalue = True 

425 try: 

426 self.con.retrbinary("RETR " + filename, out.write) 

427 except FTPError as e: 

428 msg = str(e) 

429 if msg[:4] != "550 ": 

430 self.log_buffer.append(e) 

431 return None 

432 if getvalue: 

433 return out.getvalue().decode("utf-8") 

434 return out 

435 

436 def upload_file(self, filename, src, mkdir=False): 

437 if isinstance(src, str): 

438 src = io.BytesIO(src.encode("utf-8")) 

439 if mkdir: 

440 directory = posixpath.dirname(filename) 

441 if directory: 

442 self.mkdir(directory, recursive=True) 

443 if not isinstance(filename, str): 

444 filename = filename.decode("utf-8") 

445 try: 

446 self.con.storbinary("STOR " + filename, src, blocksize=32768) 

447 except FTPError as e: 

448 self.log_buffer.append(str(e)) 

449 return False 

450 return True 

451 

452 def rename_file(self, src, dst): 

453 try: 

454 self.con.rename(src, dst) 

455 except FTPError as e: 

456 self.log_buffer.append(str(e)) 

457 try: 

458 self.con.delete(dst) 

459 except Exception as e: 

460 self.log_buffer.append(str(e)) 

461 try: 

462 self.con.rename(src, dst) 

463 except Exception as e: 

464 self.log_buffer.append(str(e)) 

465 

466 def delete_file(self, filename): 

467 if isinstance(filename, str): 

468 filename = filename.encode("utf-8") 

469 try: 

470 self.con.delete(filename) 

471 except Exception as e: 

472 self.log_buffer.append(str(e)) 

473 

474 def delete_folder(self, filename): 

475 if isinstance(filename, str): 

476 filename = filename.encode("utf-8") 

477 try: 

478 self.con.rmd(filename) 

479 except Exception as e: 

480 self.log_buffer.append(str(e)) 

481 self._known_folders.discard(filename) 

482 

483 

484class FtpTlsConnection(FtpConnection): 

485 @staticmethod 

486 def make_connection(): 

487 # pylint: disable=import-outside-toplevel 

488 from ftplib import FTP_TLS 

489 

490 return FTP_TLS() 

491 

492 def connect(self): 

493 connected = super().connect() 

494 if connected: 

495 # Upgrade data connection to TLS. 

496 self.con.prot_p() # pylint: disable=no-member 

497 return connected 

498 

499 

500class FtpPublisher(Publisher): 

501 connection_class = FtpConnection 

502 

503 @staticmethod 

504 def read_existing_artifacts(con): 

505 contents = con.get_file(".lektor/listing") 

506 if not contents: 

507 return {}, set() 

508 duplicates = set() 

509 rv = {} 

510 # Later records override earlier ones. There can be duplicate 

511 # entries if the file was not compressed. 

512 for line in contents.splitlines(): 

513 items = line.split("|") 

514 if len(items) == 2: 

515 if not isinstance(items[0], str): 

516 artifact_name = items[0].decode("utf-8") 

517 else: 

518 artifact_name = items[0] 

519 if artifact_name in rv: 

520 duplicates.add(artifact_name) 

521 rv[artifact_name] = items[1] 

522 return rv, duplicates 

523 

524 def iter_artifacts(self): 

525 """Iterates over all artifacts in the build folder and yields the 

526 artifacts. 

527 """ 

528 for dirpath, dirnames, filenames in os.walk(self.output_path): 

529 dirnames[:] = [x for x in dirnames if not self.env.is_ignored_artifact(x)] 

530 for filename in filenames: 

531 if self.env.is_ignored_artifact(filename): 

532 continue 

533 full_path = os.path.join(self.output_path, dirpath, filename) 

534 local_path = full_path[len(self.output_path) :].lstrip(os.path.sep) 

535 if os.path.altsep: 

536 local_path = local_path.lstrip(os.path.altsep) 

537 h = hashlib.sha1() 

538 try: 

539 with open(full_path, "rb") as f: 

540 while 1: 

541 item = f.read(4096) 

542 if not item: 

543 break 

544 h.update(item) 

545 except OSError as e: 

546 if e.errno != errno.ENOENT: 

547 raise 

548 yield ( 

549 local_path.replace(os.path.sep, "/"), 

550 full_path, 

551 h.hexdigest(), 

552 ) 

553 

554 @staticmethod 

555 def get_temp_filename(filename): 

556 dirname, basename = posixpath.split(filename) 

557 return posixpath.join(dirname, "." + basename + ".tmp") 

558 

559 def upload_artifact(self, con, artifact_name, source_file, checksum): 

560 with open(source_file, "rb") as source: 

561 tmp_dst = self.get_temp_filename(artifact_name) 

562 con.log_buffer.append(f"000 Updating {artifact_name}") 

563 con.upload_file(tmp_dst, source, mkdir=True) 

564 con.rename_file(tmp_dst, artifact_name) 

565 con.append(".lektor/listing", f"{artifact_name}|{checksum}\n") 

566 

567 def consolidate_listing(self, con, current_artifacts): 

568 server_artifacts, duplicates = self.read_existing_artifacts(con) 

569 known_folders = set() 

570 for artifact_name in current_artifacts.keys(): 

571 known_folders.add(posixpath.dirname(artifact_name)) 

572 

573 for artifact_name in server_artifacts: 

574 if artifact_name not in current_artifacts: 

575 con.log_buffer.append(f"000 Deleting {artifact_name}") 

576 con.delete_file(artifact_name) 

577 folder = posixpath.dirname(artifact_name) 

578 if folder not in known_folders: 

579 con.log_buffer.append(f"000 Deleting {folder}") 

580 con.delete_folder(folder) 

581 

582 if duplicates or server_artifacts != current_artifacts: 

583 listing = [] 

584 for artifact_name, checksum in current_artifacts.items(): 

585 listing.append(f"{artifact_name}|{checksum}\n") 

586 listing.sort() 

587 con.upload_file(".lektor/.listing.tmp", "".join(listing)) 

588 con.rename_file(".lektor/.listing.tmp", ".lektor/listing") 

589 

590 def publish(self, target_url, credentials=None, **extra): 

591 con = self.connection_class(target_url, credentials) 

592 connected = con.connect() 

593 yield from con.drain_log() 

594 if not connected: 

595 return 

596 

597 yield "000 Reading server state ..." 

598 con.mkdir(".lektor") 

599 committed_artifacts, _ = self.read_existing_artifacts(con) 

600 yield from con.drain_log() 

601 

602 yield "000 Begin sync ..." 

603 current_artifacts = {} 

604 for artifact_name, filename, checksum in self.iter_artifacts(): 

605 current_artifacts[artifact_name] = checksum 

606 if checksum != committed_artifacts.get(artifact_name): 

607 self.upload_artifact(con, artifact_name, filename, checksum) 

608 yield from con.drain_log() 

609 yield "000 Sync done!" 

610 

611 yield "000 Consolidating server state ..." 

612 self.consolidate_listing(con, current_artifacts) 

613 yield from con.drain_log() 

614 

615 yield "000 All done!" 

616 

617 

618class FtpTlsPublisher(FtpPublisher): 

619 connection_class = FtpTlsConnection 

620 

621 

622class GitRepo(AbstractContextManager["GitRepo"]): 

623 """A temporary git repository. 

624 

625 This class provides some lower-level utility methods which may be 

626 externally useful, but the main use case is: 

627 

628 def publish(html_output): 

629 gitrepo = GitRepo(html_output) 

630 yield from gitrepo.publish_ghpages( 

631 push_url="git@github.com:owner/repo.git", 

632 branch="gh-pages" 

633 ) 

634 

635 :param work_tree: The work tree for the repository. 

636 """ 

637 

638 def __init__(self, work_tree: StrPath) -> None: 

639 environ = {**os.environ, "GIT_WORK_TREE": str(work_tree)} 

640 

641 for what, default in [("NAME", "Lektor Bot"), ("EMAIL", "bot@getlektor.com")]: 

642 value = environ.get(f"GIT_AUTHOR_{what}") or environ.get(f"GIT_COMMITTER_{what}") or default 

643 for key in f"GIT_AUTHOR_{what}", f"GIT_COMMITTER_{what}": 

644 environ[key] = environ.get(key) or value 

645 

646 with ExitStack() as stack: 

647 environ["GIT_DIR"] = stack.enter_context(TemporaryDirectory(suffix=".git")) 

648 self.environ = environ 

649 self.run("init", "--quiet") 

650 

651 self._exit_stack = stack.pop_all() 

652 

653 def __exit__(self, *__: object) -> None: 

654 self._exit_stack.close() 

655 

656 def _popen(self, args: Sequence[str], **kwargs: Any) -> Command: 

657 cmd = ["git"] 

658 cmd.extend(args) 

659 return Command(cmd, env=self.environ, **kwargs) 

660 

661 def popen( 

662 self, 

663 *args: str, 

664 check: bool = True, 

665 input: str | None = None, 

666 capture_stdout: bool = False, 

667 ) -> Command: 

668 """Run a git subcommand.""" 

669 return self._popen(args, check=check, input=input, capture_stdout=capture_stdout) 

670 

671 def run( 

672 self, 

673 *args: str, 

674 check: bool = True, 

675 input: str | None = None, 

676 capture_stdout: bool = False, 

677 ) -> CompletedProcess[str]: 

678 """Run a git subcommand and wait for completion.""" 

679 return self._popen(args, check=check, input=input, capture_stdout=capture_stdout, capture=False).result() 

680 

681 def set_ssh_credentials(self, credentials: Mapping[str, str]) -> None: 

682 """Set up git ssh credentials. 

683 

684 This repository will be configured to used whatever SSH credentials 

685 can found in ``credentials`` (if any). 

686 """ 

687 stack = self._exit_stack 

688 ssh_command = stack.enter_context(_ssh_command(credentials)) 

689 if ssh_command: 

690 self.environ.setdefault("GIT_SSH_COMMAND", ssh_command) 

691 

692 def set_https_credentials(self, credentials: Mapping[str, str]) -> None: 

693 """Set up git http(s) credentials. 

694 

695 This repository will be configured to used whatever HTTP credentials 

696 can found in ``credentials`` (if any). 

697 """ 

698 username = credentials.get("username", "") 

699 password = credentials.get("password") 

700 if username or password: 

701 userpass = f"{username}:{password}" if password else username 

702 git_dir = self.environ["GIT_DIR"] 

703 cred_file = Path(git_dir, "lektor_cred_file") 

704 # pylint: disable=unspecified-encoding 

705 cred_file.write_text(f"https://{userpass}@github.com\n") 

706 self.run("config", "credential.helper", f'store --file "{cred_file}"') 

707 

708 def add_to_index(self, filename: str, content: str) -> None: 

709 """Create a file in the index. 

710 

711 This creates file named ``filename`` with content ``content`` in the git 

712 index. 

713 """ 

714 oid = self.run("hash-object", "-w", "--stdin", input=content, capture_stdout=True).stdout.strip() 

715 self.run("update-index", "--add", "--cacheinfo", "100644", oid, filename) 

716 

717 def publish_ghpages( 

718 self, 

719 push_url: str, 

720 branch: str, 

721 cname: str | None = None, 

722 preserve_history: bool = True, 

723 ) -> Iterator[str]: 

724 """Publish the contents of the work tree to GitHub pages. 

725 

726 :param push_url: The URL to push to. 

727 :param branch: The branch to push to 

728 :param cname: Optional. Create a top-level ``CNAME`` with given contents. 

729 """ 

730 refspec = f"refs/heads/{branch}" 

731 if preserve_history: 

732 yield "Fetching existing head" 

733 fetch_cmd = self.popen("fetch", "--depth=1", push_url, refspec, check=False) 

734 yield from _prefix_output(fetch_cmd) 

735 if fetch_cmd.returncode == 0: 

736 # If fetch was succesful, reset HEAD to remote head 

737 yield from _prefix_output(self.popen("reset", "--soft", "FETCH_HEAD")) 

738 else: 

739 # otherwise assume remote branch does not exist 

740 yield f"Creating new branch {branch}" 

741 

742 # At this point, the index is still empty. Add all but .lektor dir to index 

743 yield from _prefix_output(self.popen("add", "--force", "--all", "--", ".", ":(exclude).lektor")) 

744 if cname is not None: 

745 self.add_to_index("CNAME", f"{cname}\n") 

746 

747 # Check for changes 

748 diff_cmd = self.popen("diff", "--cached", "--no-renames", "--exit-code", "--quiet", check=False) 

749 yield from _prefix_output(diff_cmd) 

750 if diff_cmd.returncode == 0: 

751 yield "No changes to publish☺" 

752 elif diff_cmd.returncode == 1: 

753 yield "Creating commit" 

754 yield from _prefix_output(self.popen("commit", "--quiet", "--message", "Synchronized build")) 

755 push_cmd = ["push", push_url, f"HEAD:{refspec}"] 

756 if not preserve_history: 

757 push_cmd.insert(1, "--force") 

758 yield "Pushing to github" 

759 yield from _prefix_output(self.popen(*push_cmd)) 

760 yield "Success!" 

761 else: 

762 diff_cmd.result().check_returncode() # raise error 

763 

764 

765def _prefix_output(lines: Iterable[str], prefix: str = "> ") -> Iterator[str]: