Repository: lektor/lektor

Issue:
Github Pages: Change default branch from "master" -> "main"
The hard-coded default branch name for pushing to github pages (github.io) is `master`. This doesn't work out of the box anymore, as both github and git itself default to `main`. The code in question is in [publisher.py:609](https://github.com/lektor/lektor/blob/0a4bd3b0bca1033c406ffa898fef71b9210303ae/lektor/publisher.py#L609).

Would it be ok to simply change it in future versions? Would a PR be welcome then?

If not, would it be possible to at least give the option to configure the branch in `.lektorproject`? I could try a PR for this, if it were welcome.

Discussion:
yagebu: For ghpages, I think it's preferable to force-push an orphaned commit instead of having some history of deployed pages. Since that would be consistent with the current documentation (but not the implementation), I think changing it to work like that would be fine.
dairiki: > For ghpages, I think it's preferable to force-push an orphaned commit instead of having some history of deployed pages. Since that would be consistent with the current documentation (but not the implementation), I think changing it to work like that would be fine.

I tend to agree that force-push should be the preferred behavior, however, I worry about changing the default behavior without warning (other than in the changelog).  In the rare cases where someone does want to preserve the history of the deployed HTML, changing to force-push will delete any such accumulated history.

The safer way to make the change would be to add a `preserve_history=false` option that would enable the force-push behavior. The default behavior would be left unchanged, but a warning would be issued that the default behavior will be changed in future and that users should add an explicit `preserver_history=true` if they want to keep the non-force-push behavior.

Is it worth the effort to do that?


Patch:
diff --git a/lektor/compat.py b/lektor/compat.py
new file mode 100644
index 0000000..2c1ee37
--- /dev/null
+++ b/lektor/compat.py
@@ -0,0 +1,53 @@
+import os
+import stat
+import sys
+import tempfile
+from functools import partial
+from itertools import chain
+
+
+def _ensure_tree_writeable(path: str) -> None:
+    """Attempt to ensure that all files in the tree rooted at path are writeable."""
+    dirscans = []
+
+    def fix_mode(path, statfunc):
+        try:
+            # paranoia regarding symlink attacks
+            current_mode = statfunc(follow_symlinks=False).st_mode
+            if not stat.S_ISLNK(current_mode):
+                isdir = stat.S_ISDIR(current_mode)
+                fixed_mode = current_mode | (0o700 if isdir else 0o200)
+                if current_mode != fixed_mode:
+                    os.chmod(path, fixed_mode)
+                if isdir:
+                    dirscans.append(os.scandir(path))
+        except FileNotFoundError:
+            pass
+
+    fix_mode(path, partial(os.stat, path))
+    for entry in chain.from_iterable(dirscans):
+        fix_mode(entry.path, entry.stat)
+
+
+class FixedTemporaryDirectory(tempfile.TemporaryDirectory):
+    """A version of tempfile.TemporaryDirectory that works if dir contains read-only files.
+
+    On python < 3.8 under Windows, if any read-only files are created
+    in a TemporaryDirectory, TemporaryDirectory will throw an
+    exception when it tries to remove them on cleanup. See
+    https://bugs.python.org/issue26660
+
+    This can create issues, e.g., with temporary git repositories since
+    git creates read-only files in its object store.
+
+    """
+
+    def cleanup(self) -> None:
+        _ensure_tree_writeable(self.name)
+        super().cleanup()
+
+
+if sys.version_info >= (3, 8):
+    TemporaryDirectory = tempfile.TemporaryDirectory
+else:
+    TemporaryDirectory = FixedTemporaryDirectory
diff --git a/lektor/publisher.py b/lektor/publisher.py
index 61ed3cf..e934430 100644
--- a/lektor/publisher.py
+++ b/lektor/publisher.py
@@ -3,172 +3,287 @@ import hashlib
 import io
 import os
 import posixpath
-import shutil
-import subprocess
-import tempfile
 from contextlib import contextmanager
+from contextlib import ExitStack
+from contextlib import suppress
 from ftplib import Error as FTPError
+from inspect import cleandoc
+from pathlib import Path
+from subprocess import CalledProcessError
+from subprocess import CompletedProcess
+from subprocess import DEVNULL
+from subprocess import PIPE
+from subprocess import STDOUT
+from typing import Any
+from typing import Callable
+from typing import ContextManager
+from typing import Dict
+from typing import Generator
+from typing import Iterable
+from typing import Iterator
+from typing import Mapping
+from typing import NoReturn
+from typing import Optional
+from typing import Sequence
+from typing import Tuple
+from typing import TYPE_CHECKING
+from warnings import warn
 
 from werkzeug import urls
 
+from lektor.compat import TemporaryDirectory
 from lektor.exception import LektorException
+from lektor.utils import bool_from_string
 from lektor.utils import locate_executable
 from lektor.utils import portable_popen
 
+if TYPE_CHECKING:  # pragma: no cover
+    from _typeshed import StrOrBytesPath
+    from _typeshed import StrPath
+    from lektor.environment import Environment
 
-def _patch_git_env(env_overrides, ssh_command=None):
-    env = dict(os.environ)
-    env.update(env_overrides or ())
-
-    keys = [
-        ("GIT_COMMITTER_NAME", "GIT_AUTHOR_NAME", "Lektor Bot"),
-        ("GIT_COMMITTER_EMAIL", "GIT_AUTHOR_EMAIL", "bot@getlektor.com"),
-    ]
-
-    for key_a, key_b, default in keys:
-        value_a = env.get(key_a)
-        value_b = env.get(key_b)
-        if value_a:
-            if not value_b:
-                env[key_b] = value_a
-        elif value_b:
-            if not value_a:
-                env[key_a] = value_b
-        else:
-            env[key_a] = default
-            env[key_b] = default
-
-    if ssh_command is not None and not env.get("GIT_SSH_COMMAND"):
-        env["GIT_SSH_COMMAND"] = ssh_command
-
-    return env
 
-
-def _write_ssh_key_file(temp_fn, credentials):
-    if credentials:
-        key_file = credentials.get("key_file")
-        if key_file is not None:
-            return key_file
-        key = credentials.get("key")
-        if key:
-            parts = key.split(":", 1)
-            if len(parts) == 1:
-                kt = "RSA"
+@contextmanager
+def _ssh_key_file(
+    credentials: Optional[Mapping[str, str]]
+) -> Iterator[Optional["StrPath"]]:
+    with ExitStack() as stack:
+        key_file: Optional["StrPath"]
+        key_file = credentials.get("key_file") if credentials else None
+        key = credentials.get("key") if credentials else None
+        if not key_file and key:
+            if ":" in key:
+                key_type, _, key = key.partition(":")
+                key_type = key_type.upper()
             else:
-                kt, key = parts
-            with open(temp_fn, "w", encoding="utf-8") as f:
-                f.write("-----BEGIN %s PRIVATE KEY-----\n" % kt.upper())
-                for x in range(0, len(key), 64):
-                    f.write(key[x : x + 64] + "\n")
-                f.write("-----END %s PRIVATE KEY-----\n" % kt.upper())
-            os.chmod(temp_fn, 0o600)
-            return temp_fn
-    return None
-
-
-def _get_ssh_cmd(port=None, keyfile=None):
-    ssh_args = []
-    if port:
-        ssh_args.append("-p %s" % port)
-    if keyfile:
-        ssh_args.append('-i "%s"' % keyfile)
-    return "ssh %s" % " ".join(ssh_args)
+                key_type = "RSA"
+            key_file = Path(stack.enter_context(TemporaryDirectory()), "keyfile")
+            with key_file.open("w", encoding="utf-8") as f:
+                f.write(f"-----BEGIN {key_type} PRIVATE KEY-----\n")
+                f.writelines(key[x : x + 64] + "\n" for x in range(0, len(key), 64))
+                f.write(f"-----END {key_type} PRIVATE KEY-----\n")
+        yield key_file
 
 
 @contextmanager
-def _temporary_folder(env):
-    base = env.temp_path
-    try:
-        os.makedirs(base)
-    except OSError:
-        pass
-
-    folder = tempfile.mkdtemp(prefix=".deploytemp", dir=base)
-    scratch = os.path.join(folder, "scratch")
-    os.mkdir(scratch)
-    os.chmod(scratch, 0o755)
-    try:
-        yield scratch
-    finally:
-        try:
-            shutil.rmtree(folder)
-        except (IOError, OSError):
-            pass
+def _ssh_command(
+    credentials: Optional[Mapping[str, str]], port: Optional[int] = None
+) -> Iterator[Optional[str]]:
+    with _ssh_key_file(credentials) as key_file:
+        args = []
+        if port:
+            args.append(f" -p {port}")
+        if key_file:
+            args.append(f' -i "{key_file}" -o IdentitiesOnly=yes')
+        if args:
+            yield "ssh" + " ".join(args)
+        else:
+            yield None
 
 
 class PublishError(LektorException):
     """Raised by publishers if something goes wrong."""
 
 
-class Command:
-    def __init__(self, argline, cwd=None, env=None, capture=True, silent=False):
-        environ = dict(os.environ)
+class Command(ContextManager["Command"]):
+    """A wrapper around subprocess.Popen to facilitate streaming output via generator.
+
+    :param argline: Command with arguments to execute.
+    :param cwd: Optional. Directory in which to execute command.
+    :param env: Optional. Environment with which to run command.
+    :param capture: Default `True`. Whether to capture stdout and stderr.
+    :param silent: Default `False`. Discard output altogether.
+    :param check: Default `False`.
+        If set, raise ``CalledProcessError`` on non-zero return code.
+    :param input: Optional. A string to feed to the subprocess via stdin.
+    :param capture_stdout: Default `False`. Capture stdout and
+        return in ``CompletedProcess.stdout``.
+
+    Basic Usage
+    ===========
+
+    To run a command, returning any output on stdout or stderr to the caller
+    as an iterable (generator), while checking the return code from the command:
+
+        def run_command(argline):
+            # This passes the output
+            rv = yield from Command(argline)
+            if rv.returncode != 0:
+                raise RuntimeError("Command failed!")
+
+    This could be called as follows:
+
+        for outline in run_command(('ls')):
+            print(outline.rstrip())
+
+    Supplying input via stdin, Capturing stdout
+    ===========================================
+
+    The following example shows how input may be fed to a subprocess via stdin,
+    and how stdout may be captured for further processing.
+
+        def run_wc(input):
+            rv = yield from Command(('wc'), check=True, input=input, capture_stdout=True)
+            lines, words, chars = rv.stdout.split()
+            print(f"{words} words, {chars} chars")
+
+        stderr_lines = list(run_wc("a few words"))
+        # prints "3 words, 11 chars"
+
+    Note that ``check=True`` will cause a ``CalledProcessError`` to be raised if the
+    ``wc`` subprocess returns a non-zero return code.
+
+    """
+
+    def __init__(
+        self,
+        argline: Iterable[str],
+        cwd: Optional["StrOrBytesPath"] = None,
+        env: Optional[Mapping[str, str]] = None,
+        capture: bool = True,
+        silent: bool = False,
+        check: bool = False,
+        input: Optional[str] = None,
+        capture_stdout: bool = False,
+    ) -> None:
+        kwargs: Dict[str, Any] = {"cwd": cwd}
         if env:
-            environ.update(env)
-        kwargs = {"cwd": cwd, "env": environ}
+            kwargs["env"] = {**os.environ, **env}
         if silent:
-            kwargs["stdout"] = subprocess.DEVNULL
-            kwargs["stderr"] = subprocess.DEVNULL
+            kwargs["stdout"] = DEVNULL
+            kwargs["stderr"] = DEVNULL
             capture = False
+        if input is not None:
+            kwargs["stdin"] = PIPE
+        if capture or capture_stdout:
+            kwargs["stdout"] = PIPE
         if capture:
-            kwargs["stdout"] = subprocess.PIPE
-            kwargs["stderr"] = subprocess.STDOUT
-            # Since 3.7, Python has sane encoding defaults in the case that the system is
-            # (likely mis-)configured to use ASCII as the default encoding (PEP538).
-            # It also provides a way for the user to force the use of UTF-8 (PEP540).
-            kwargs["text"] = True
-            kwargs["errors"] = "replace"
-        self.capture = capture
-        self._cmd = portable_popen(argline, **kwargs)
-
-    def wait(self):
-        returncode = self._cmd.wait()
-        if self._cmd.stdout is not None:
-            self._cmd.stdout.close()
-        return returncode
+            kwargs["stderr"] = STDOUT if not capture_stdout else PIPE
+
+        # Python >= 3.7 has sane encoding defaults in the case that the system is
+        # (likely mis-)configured to use ASCII as the default encoding (PEP538).
+        # It also provides a way for the user to force the use of UTF-8 (PEP540).
+        kwargs["text"] = True
+        kwargs["errors"] = "replace"
+
+        self.capture = capture  # b/c - unused
+        self.check = check
+        self._stdout = None
+
+        with ExitStack() as stack:
+            self._cmd = stack.enter_context(portable_popen(list(argline), **kwargs))
+            self._closer: Optional[Callable[[], None]] = stack.pop_all().close
+
+        if input is not None or capture_stdout:
+            self._output = self._communicate(input, capture_stdout, capture)
+        elif capture:
+            self._output = self._cmd.stdout
+        else:
+            self._output = None
+
+    def _communicate(
+        self, input: Optional[str], capture_stdout: bool, capture: bool
+    ) -> Optional[Iterator[str]]:
+        proc = self._cmd
+        try:
+            if capture_stdout:
+                self._stdout, errout = proc.communicate(input)
+            else:
+                errout, _ = proc.communicate(input)
+        except BaseException:
+            proc.kill()
+            with suppress(CalledProcessError):
+                self.close()
+            raise
+        if capture:
+            return iter(errout.splitlines())
+        return None
+
+    def close(self) -> None:
+        """Wait for subprocess to complete.
+
+        If check=True was passed to the constructor, raises ``CalledProcessError``
+        if the subprocess returns a non-zero status code.
+        """
+        closer, self._closer = self._closer, None
+        if closer:
+            # This waits for process and closes standard file descriptors
+            closer()
+        if self.check:
+            rc = self._cmd.poll()
+            if rc != 0:
+                raise CalledProcessError(rc, self._cmd.args, self._stdout)
+
+    def wait(self) -> int:
+        """Wait for subprocess to complete. Return status code."""
+        self._cmd.wait()
+        self.close()
+        return self._cmd.returncode
+
+    def result(self) -> "CompletedProcess[str]":
+        """Wait for subprocess to complete.  Return ``CompletedProcess`` instance.
+
+        If ``capture_stdout=True`` was passed to the constructor, the output
+        captured from stdout will be available on the ``.stdout`` attribute
+        of the return value.
+        """
+        return CompletedProcess(self._cmd.args, self.wait(), self._stdout)
 
     @property
-    def returncode(self):
+    def returncode(self) -> Optional[int]:
+        """Return exit status of the subprocess.
+
+        Or ``None`` if the subprocess is still alive.
+        """
         return self._cmd.returncode
 
-    def __enter__(self):
-        return self
+    def __exit__(self, *__: Any) -> None:
+        self.close()
 
-    def __exit__(self, exc_type, exc_value, tb):
-        self.wait()
+    def __iter__(self) -> Generator[str, None, "CompletedProcess[str]"]:
+        """A generator with yields any captured output and returns a ``CompletedProcess``.
 
-    def __iter__(self):
-        if not self.capture:
-            raise RuntimeError("Not capturing")
+        If ``capture`` is ``True`` (the default).  Both stdout and stderr are available
+        in the iterator output.
 
-        for line in self._cmd.stdout:
+        If ``capture_stdout`` is set, stdout is captured to a string which is made
+        available via ``CompletedProcess.stdout`` attribute of the return value. Stderr
+        output is available via the iterator output, as normal.
+        """
+        if self._output is None:
+            raise RuntimeError("Not capturing")
+        for line in self._output:
             yield line.rstrip()
+        return self.result()
 
-    def safe_iter(self):
-        with self:
-            for line in self:
-                yield line
+    safe_iter = __iter__  # b/c - deprecated
 
     @property
-    def output(self):
+    def output(self) -> Iterator[str]:  # b/c - deprecated
         return self.safe_iter()
 
 
 class Publisher:
-    def __init__(self, env, output_path):
+    def __init__(self, env: "Environment", output_path: str) -> None:
         self.env = env
         self.output_path = os.path.abspath(output_path)
 
-    def fail(self, message):
+    def fail(self, message: str) -> NoReturn:
         # pylint: disable=no-self-use
         raise PublishError(message)
 
-    def publish(self, target_url, credentials=None, **extra):
+    def publish(
+        self,
+        target_url: urls.URL,
+        credentials: Optional[Mapping[str, str]] = None,
+        **extra: Any,
+    ) -> Iterator[str]:
         raise NotImplementedError()
 
 
 class RsyncPublisher(Publisher):
-    def get_command(self, target_url, tempdir, credentials):
+    @contextmanager
+    def get_command(self, target_url, credentials):
         credentials = credentials or {}
         argline = ["rsync", "-rclzv", "--exclude=.lektor"]
         target = []
@@ -183,33 +298,26 @@ class RsyncPublisher(Publisher):
         if delete:
             argline.append("--delete-after")
 
-        keyfile = _write_ssh_key_file(
-            os.path.join(tempdir, "ssh-auth-key"), credentials
-        )
-
-        if target_url.port is not None or keyfile is not None:
-            argline.append("-e")
-            argline.append(_get_ssh_cmd(target_url.port, keyfile))
+        with _ssh_command(credentials, target_url.port) as ssh_command:
+            if ssh_command:
+                argline.extend(("-e", ssh_command))
 
-        username = credentials.get("username") or target_url.username
-        if username:
-            target.append(username + "@")
+            username = credentials.get("username") or target_url.username
+            if username:
+                target.append(username + "@")
 
-        if target_url.ascii_host is not None:
-            target.append(target_url.ascii_host)
-            target.append(":")
-        target.append(target_url.path.rstrip("/") + "/")
+            if target_url.ascii_host is not None:
+                target.append(target_url.ascii_host)
+                target.append(":")
+            target.append(target_url.path.rstrip("/") + "/")
 
-        argline.append(self.output_path.rstrip("/\\") + "/")
-        argline.append("".join(target))
-        return Command(argline, env=env)
+            argline.append(self.output_path.rstrip("/\\") + "/")
+            argline.append("".join(target))
+            yield Command(argline, env=env)
 
     def publish(self, target_url, credentials=None, **extra):
-        with _temporary_folder(self.env) as tempdir:
-            client = self.get_command(target_url, tempdir, credentials)
-            with client:
-                for line in client:
-                    yield line
+        with self.get_command(target_url, credentials) as client:
+            yield from client
 
 
 class FtpConnection:
@@ -514,130 +622,272 @@ class FtpTlsPublisher(FtpPublisher):
     connection_class = FtpTlsConnection
 
 
-class GithubPagesPublisher(Publisher):
-    @staticmethod
-    def get_credentials(url, credentials=None):
-        credentials = credentials or {}
-        username = credentials.get("username") or url.username
-        password = credentials.get("password") or url.password
-        rv = username
-        if username and password:
-            rv += ":" + password
-        return rv if rv else None
-
-    def update_git_config(self, repo, url, branch, credentials=None):
-        ssh_command = None
-        path = url.host + "/" + url.path.strip("/")
-        cred = None
-        if url.scheme in ("ghpages", "ghpages+ssh"):
-            push_url = "git@github.com:%s.git" % path
-            keyfile = _write_ssh_key_file(
-                os.path.join(repo, ".git", "ssh-auth-key"), credentials
+class GitRepo(ContextManager["GitRepo"]):
+    """A temporary git repository.
+
+    This class provides some lower-level utility methods which may be
+    externally useful, but the main use case is:
+
+        def publish(html_output):
+            gitrepo = GitRepo(html_output)
+            yield from gitrepo.publish_ghpages(
+                push_url="git@github.com:owner/repo.git",
+                branch="gh-pages"
             )
-            if keyfile or url.port:
-                ssh_command = _get_ssh_cmd(url.port, keyfile)
-        else:
-            push_url = "https://github.com/%s.git" % path
-            cred = self.get_credentials(url, credentials)
-
-        with open(os.path.join(repo, ".git", "config"), "a", encoding="utf-8") as f:
-            f.write(
-                '[remote "origin"]\nurl = %s\n'
-                "fetch = +refs/heads/%s:refs/remotes/origin/%s\n"
-                % (push_url, branch, branch)
+
+    :param work_tree: The work tree for the repository.
+    """
+
+    def __init__(self, work_tree: "StrPath") -> None:
+        environ = {**os.environ, "GIT_WORK_TREE": str(work_tree)}
+
+        for what, default in [("NAME", "Lektor Bot"), ("EMAIL", "bot@getlektor.com")]:
+            value = (
+                environ.get(f"GIT_AUTHOR_{what}")
+                or environ.get(f"GIT_COMMITTER_{what}")
+                or default
             )
-            if cred:
-                cred_path = os.path.join(repo, ".git", "credentials")
-                f.write('[credential]\nhelper = store --file "%s"\n' % cred_path)
-                with open(cred_path, "w", encoding="utf-8") as cf:
-                    cf.write("https://%s@github.com\n" % cred)
+            for key in f"GIT_AUTHOR_{what}", f"GIT_COMMITTER_{what}":
+                environ[key] = environ.get(key) or value
+
+        with ExitStack() as stack:
+            environ["GIT_DIR"] = stack.enter_context(TemporaryDirectory(suffix=".git"))
+            self.environ = environ
+            self.run("init", "--quiet")
+
+            self._exit_stack = stack.pop_all()
+
+    def __exit__(self, *__: Any) -> None:
+        self._exit_stack.close()
+
+    def _popen(self, args: Sequence[str], **kwargs: Any) -> Command:
+        cmd = ["git"]
+        cmd.extend(args)
+        return Command(cmd, env=self.environ, **kwargs)
+
+    def popen(
+        self,
+        *args: str,
+        check: bool = True,
+        input: Optional[str] = None,
+        capture_stdout: bool = False,
+    ) -> Command:
+        """Run a git subcommand."""
+        return self._popen(
+            args, check=check, input=input, capture_stdout=capture_stdout
+        )
 
-        return ssh_command
+    def run(
+        self,
+        *args: str,
+        check: bool = True,
+        input: Optional[str] = None,
+        capture_stdout: bool = False,
+    ) -> "CompletedProcess[str]":
+        """Run a git subcommand and wait for completion."""
+        return self._popen(
+            args, check=check, input=input, capture_stdout=capture_stdout, capture=False
+        ).result()
+
+    def set_ssh_credentials(self, credentials: Mapping[str, str]) -> None:
+        """Set up git ssh credentials.
+
+        This repository will be configured to used whatever SSH credentials
+        can found in ``credentials`` (if any).
+        """
+        stack = self._exit_stack
+        ssh_command = stack.enter_context(_ssh_command(credentials))
+        if ssh_command:
+            self.environ.setdefault("GIT_SSH_COMMAND", ssh_command)
 
-    def link_artifacts(self, path):
-        try:
-            link = os.link
-        except AttributeError:
-            link = shutil.copy
-
-        # Clean old
-        for filename in os.listdir(path):
-            if filename == ".git":
-                continue
-            filename = os.path.join(path, filename)
-            try:
-                os.remove(filename)
-            except OSError:
-                shutil.rmtree(filename)
+    def set_https_credentials(self, credentials: Mapping[str, str]) -> None:
+        """Set up git http(s) credentials.
 
-        # Add new
-        for dirpath, dirnames, filenames in os.walk(self.output_path):
-            dirnames[:] = [x for x in dirnames if x != ".lektor"]
-            for filename in filenames:
-                full_path = os.path.join(self.output_path, dirpath, filename)
-                dst = os.path.join(
-                    path,
-                    full_path[len(self.output_path) :]
-                    .lstrip(os.path.sep)
-                    .lstrip(os.path.altsep or ""),
-                )
-                try:
-                    os.makedirs(os.path.dirname(dst))
-                except (OSError, IOError):
-                    pass
-                try:
-                    link(full_path, dst)
-                except OSError:  # Different Filesystems
-                    shutil.copy(full_path, dst)
+        This repository will be configured to used whatever HTTP credentials
+        can found in ``credentials`` (if any).
+        """
+        username = credentials.get("username", "")
+        password = credentials.get("password")
+        if username or password:
+            userpass = f"{username}:{password}" if password else username
+            git_dir = self.environ["GIT_DIR"]
+            cred_file = Path(git_dir, "lektor_cred_file")
+            # pylint: disable=unspecified-encoding
+            cred_file.write_text(f"https://{userpass}@github.com\n")
+            self.run("config", "credential.helper", f'store --file "{cred_file}"')
+
+    def add_to_index(self, filename: str, content: str) -> None:
+        """Create a file in the index.
+
+        This creates file named ``filename`` with content ``content`` in the git
+        index.
+        """
+        oid = self.run(
+            "hash-object", "-w", "--stdin", input=content, capture_stdout=True
+        ).stdout.strip()
+        self.run("update-index", "--add", "--cacheinfo", "100644", oid, filename)
+
+    def publish_ghpages(
+        self,
+        push_url: str,
+        branch: str,
+        cname: Optional[str] = None,
+        preserve_history: bool = True,
+    ) -> Iterator[str]:
+        """Publish the contents of the work tree to GitHub pages.
+
+        :param push_url: The URL to push to.
+        :param branch: The branch to push to
+        :param cname: Optional. Create a top-level ``CNAME`` with given contents.
+        """
+        refspec = f"refs/heads/{branch}"
+        if preserve_history:
+            yield "Fetching existing head"
+            fetch_cmd = self.popen("fetch", "--depth=1", push_url, refspec, check=False)
+            yield from _prefix_output(fetch_cmd)
+            if fetch_cmd.returncode == 0:
+                # If fetch was succesful, reset HEAD to remote head
+                yield from _prefix_output(self.popen("reset", "--soft", "FETCH_HEAD"))
+            else:
+                # otherwise assume remote branch does not exist
+                yield f"Creating new branch {branch}"
 
-    @staticmethod
-    def write_cname(path, target_url):
-        params = target_url.decode_query()
-        cname = params.get("cname")
+        # At this point, the index is still empty. Add all but .lektor dir to index
+        yield from _prefix_output(
+            self.popen("add", "--force", "--all", "--", ".", ":(exclude).lektor")
+        )
         if cname is not None:
-            with open(os.path.join(path, "CNAME"), "w", encoding="utf-8") as f:
-                f.write("%s\n" % cname)
-
-    @staticmethod
-    def detect_target_branch(target_url):
-        # When pushing to the username.github.io repo we need to push to
-        # master, otherwise to gh-pages
-        if target_url.host.lower() + ".github.io" == target_url.path.strip("/").lower():
-            branch = "master"
+            self.add_to_index("CNAME", f"{cname}\n")
+
+        # Check for changes
+        diff_cmd = self.popen("diff", "--cached", "--exit-code", "--quiet", check=False)
+        yield from _prefix_output(diff_cmd)
+        if diff_cmd.returncode == 0:
+            yield "No changes to publish☺"
+        elif diff_cmd.returncode == 1:
+            yield "Creating commit"
+            yield from _prefix_output(
+                self.popen("commit", "--quiet", "--message", "Synchronized build")
+            )
+            push_cmd = ["push", push_url, f"HEAD:{refspec}"]
+            if not preserve_history:
+                push_cmd.insert(1, "--force")
+            yield "Pushing to github"
+            yield from _prefix_output(self.popen(*push_cmd))
+            yield "Success!"
         else:
-            branch = "gh-pages"
-        return branch
+            diff_cmd.result().check_returncode()  # raise error
 
-    def publish(self, target_url, credentials=None, **extra):
+
+def _prefix_output(lines: Iterable[str], prefix: str = "> ") -> Iterator[str]:
+    """Add prefix to lines."""
+    return (f"{prefix}{line}" for line in lines)
+
+
+class GithubPagesPublisher(Publisher):
+    """Publish to GitHub pages."""
+
+    def publish(
+        self,
+        target_url: urls.URL,
+        credentials: Optional[Mapping[str, str]] = None,
+        **extra: Any,
+    ) -> Iterator[str]:
         if not locate_executable("git"):
             self.fail("git executable not found; cannot deploy.")
 
-        branch = self.detect_target_branch(target_url)
+        push_url, branch, cname, preserve_history, warnings = self._parse_url(
+            target_url
+        )
+        creds = self._parse_credentials(credentials, target_url)
+
+        yield from iter(warnings)
 
-        with _temporary_folder(self.env) as path:
-            ssh_command = None
+        with GitRepo(self.output_path) as repo:
+            if push_url.startswith("https:"):
+                repo.set_https_credentials(creds)
+            else:
+                repo.set_ssh_credentials(creds)
+            yield from repo.publish_ghpages(push_url, branch, cname, preserve_history)
+
+    def _parse_url(
+        self, target_url: urls.URL
+    ) -> Tuple[str, str, Optional[str], bool, Sequence[str]]:
+        if not target_url.host:
+            self.fail("github owner missing from target URL")
+        gh_owner = target_url.host.lower()
+        gh_project = target_url.path.strip("/").lower()
+        if not gh_project:
+            self.fail("github project missing from target URL")
 
-            def git(args, **kwargs):
-                kwargs["env"] = _patch_git_env(kwargs.pop("env", None), ssh_command)
-                return Command(["git"] + args, cwd=path, **kwargs)
+        params = target_url.decode_query()
+        cname = params.get("cname")
+        branch = params.get("branch")
+        preserve_history = bool_from_string(params.get("preserve_history"), True)
 
-            for line in git(["init"]).output:
-                yield line
-            ssh_command = self.update_git_config(path, target_url, branch, credentials)
-            for line in git(["remote", "update"]).output:
-                yield line
+        warnings = []
 
-            if git(["checkout", "-q", branch], silent=True).wait() != 0:
-                git(["checkout", "-qb", branch], silent=True).wait()
+        if not branch:
+            if gh_project == f"{gh_owner}.github.io":
+                warnings.extend(
+                    cleandoc(self._EXPLICIT_BRANCH_SUGGESTED_MSG).splitlines()
+                )
+                warn(
+                    " ".join(
+                        cleandoc(self._DEFAULT_BRANCH_DEPRECATION_MSG).splitlines()
+                    ),
+                    category=DeprecationWarning,
+                )
+                branch = "master"
+            else:
+                branch = "gh-pages"
+
+        if target_url.scheme in ("ghpages", "ghpages+ssh"):
+            push_url = f"ssh://git@github.com/{gh_owner}/{gh_project}.git"
+            default_port = 22
+        else:
+            push_url = f"https://github.com/{gh_owner}/{gh_project}.git"
+            default_port = 443
+        if target_url.port and target_url.port != default_port:
+            self.fail("github does not support pushing to non-standard ports")
+
+        return push_url, branch, cname, preserve_history, warnings
+
+    _EXPLICIT_BRANCH_SUGGESTED_MSG = """
+    ================================================================
+    WARNING!!! You should explicitly set the name of the published
+    branch of your GitHub pages repository.
+
+    The default branch for new GitHub pages repositories has changed
+    to 'main', but Lektor still defaults to the old value, 'master'.
+    In a future version of Lektor, the default branch name will
+    changed to match the new GitHub default.
+
+    For details, see
+    https://getlektor.com/docs/deployment/ghpages/#pushing-to-an-explicit-branch
+    ================================================================
+    """
+
+    _DEFAULT_BRANCH_DEPRECATION_MSG = """
+    Currently, by default, Lektor pushes to the 'master' branch when
+    deploying to GitHub pages repositories.  In a future version of
+    Lektor, the default branch will GitHub's new default, 'main'.
+    It is suggest that you explicitly set which branch to push to.
+    """
 
-            self.link_artifacts(path)
-            self.write_cname(path, target_url)
-            for line in git(["add", "-f", "--all", "."]).output:
-                yield line
-            for line in git(["commit", "-qm", "Synchronized build"]).output:
-                yield line
-            for line in git(["push", "origin", branch]).output:
-                yield line
+    @staticmethod
+    def _parse_credentials(
+        credentials: Optional[Mapping[str, str]], target_url: urls.URL
+    ) -> Mapping[str, str]:
+        creds = dict(credentials or {})
+        # Fill in default username/password from target url
+        for key, default in [
+            ("username", target_url.username),
+            ("password", target_url.password),
+        ]:
+            if not creds.get(key) and default:
+                creds[key] = default
+        return creds
 
 
 builtin_publishers = {


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 173a497..262998f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -16,6 +16,7 @@ from lektor.environment import Environment
 from lektor.environment.expressions import Expression
 from lektor.project import Project
 from lektor.reporter import BufferReporter
+from lektor.utils import locate_executable
 
 
 @pytest.fixture(scope="session")
@@ -231,3 +232,15 @@ def project_cli_runner(isolated_cli_runner, project, save_sys_path):
         else:
             shutil.copy2(entry_path, entry)
     return isolated_cli_runner
+
+
+@pytest.fixture
+def no_utils(monkeypatch):
+    """Monkeypatch $PATH to hide any installed external utilities
+    (e.g. git, imagemagick)."""
+    monkeypatch.setitem(os.environ, "PATH", "/dev/null")
+    locate_executable.cache_clear()
+    try:
+        yield
+    finally:
+        locate_executable.cache_clear()
diff --git a/tests/test_compat.py b/tests/test_compat.py
new file mode 100644
index 0000000..03a3eb4
--- /dev/null
+++ b/tests/test_compat.py
@@ -0,0 +1,32 @@
+import os
+from pathlib import Path
+
+import pytest
+
+from lektor.compat import _ensure_tree_writeable
+from lektor.compat import FixedTemporaryDirectory
+from lektor.compat import TemporaryDirectory
+
+
+def test_ensure_tree_writeable(tmp_path):
+    topdir = tmp_path / "topdir"
+    subdir = topdir / "subdir"
+    regfile = subdir / "regfile"
+    subdir.mkdir(parents=True)
+    regfile.touch(mode=0)
+    subdir.chmod(0)
+    topdir.chmod(0)
+
+    _ensure_tree_writeable(topdir)
+
+    for p in topdir, subdir, regfile:
+        assert os.access(p, os.W_OK)
+
+
+@pytest.mark.parametrize("tmpdir_class", [FixedTemporaryDirectory, TemporaryDirectory])
+def test_TemporaryDirectory(tmpdir_class):
+    with tmpdir_class() as tmpdir:
+        file = Path(tmpdir, "test-file")
+        file.touch(mode=0)
+        os.chmod(tmpdir, 0)
+    assert not os.path.exists(tmpdir)
diff --git a/tests/test_deploy.py b/tests/test_deploy.py
index c73e96f..3627abb 100644
--- a/tests/test_deploy.py
+++ b/tests/test_deploy.py
@@ -1,11 +1,9 @@
-import textwrap
 from os.path import dirname
 from os.path import join
 
 import pytest
 from werkzeug.urls import url_parse
 
-from lektor.publisher import GithubPagesPublisher
 from lektor.publisher import RsyncPublisher
 
 
@@ -17,131 +15,26 @@ def test_get_server(env):
     assert server.extra == {"extra_field": "extra_value"}
 
 
-def test_ghpages_update_git_config(tmpdir, env):
-    output_path = tmpdir.mkdir("output")
-    publisher = GithubPagesPublisher(env, str(output_path))
-    repo_path = tmpdir.mkdir("repo")
-    repo_config = repo_path.mkdir(".git").join("config").ensure(file=True)
-    target_url = url_parse("ghpages://user/repo")
-    branch = "master"
-    publisher.update_git_config(str(repo_path), target_url, branch)
-    expected = textwrap.dedent(
-        """
-        [remote "origin"]
-        url = git@github.com:user/repo.git
-        fetch = +refs/heads/master:refs/remotes/origin/master
-    """
-    ).strip()
-    assert repo_config.read().strip() == expected
-
-
-def test_ghpages_update_git_config_https(tmpdir, env):
-    output_path = tmpdir.mkdir("output")
-    publisher = GithubPagesPublisher(env, str(output_path))
-    repo_path = tmpdir.mkdir("repo")
-    repo_config = repo_path.mkdir(".git").join("config").ensure(file=True)
-    target_url = url_parse("ghpages+https://pybee/pybee.github.io?cname=pybee.org")
-    branch = "lektor"
-    publisher.update_git_config(str(repo_path), target_url, branch)
-    expected = textwrap.dedent(
-        """
-        [remote "origin"]
-        url = https://github.com/pybee/pybee.github.io.git
-        fetch = +refs/heads/lektor:refs/remotes/origin/lektor
-    """
-    ).strip()
-    assert repo_config.read().strip() == expected
-
-
-def test_ghpages_update_git_config_https_credentials(tmpdir, env):
-    output_path = tmpdir.mkdir("output")
-    publisher = GithubPagesPublisher(env, str(output_path))
-    repo_path = tmpdir.mkdir("repo")
-    repo_config = repo_path.mkdir(".git").join("config").ensure(file=True)
-    target_url = url_parse("ghpages+https://pybee/pybee.github.io?cname=pybee.org")
-    branch = "lektor"
-    credentials_file = repo_path.join(".git", "credentials")
-    credentials = {
-        "username": "fakeuser",
-        "password": "fakepass",
-    }
-    publisher.update_git_config(
-        str(repo_path), target_url, branch, credentials=credentials
-    )
-    expected = (
-        textwrap.dedent(
-            """
-        [remote "origin"]
-        url = https://github.com/pybee/pybee.github.io.git
-        fetch = +refs/heads/lektor:refs/remotes/origin/lektor
-        [credential]
-        helper = store --file "{}"
-    """
-        )
-        .format(credentials_file)
-        .strip()
-    )
-    assert repo_config.read().strip() == expected
-
-    assert (
-        credentials_file.read().strip()
-        == "https://fakeuser:fakepass@github.com".strip()
-    )
-
-
-def test_ghpages_write_cname(tmpdir, env):
-    output_path = tmpdir.mkdir("output")
-    publisher = GithubPagesPublisher(env, str(output_path))
-    target_url = url_parse("ghpages+https://pybee/pybee.github.io?cname=pybee.org")
-    publisher.write_cname(str(output_path), target_url)
-    assert (output_path / "CNAME").read() == "pybee.org\n"
-
-
-def test_ghpages_detect_branch_username(tmpdir, env):
-    output_path = tmpdir.mkdir("output")
-    publisher = GithubPagesPublisher(env, str(output_path))
-    target_url = url_parse("ghpages+https://MacDownApp/MacDownApp.github.io")
-    branch = publisher.detect_target_branch(target_url)
-    assert branch == "master"
-
-
-def test_ghpages_detect_branch_username_case_insensitive(tmpdir, env):
-    output_path = tmpdir.mkdir("output")
-    publisher = GithubPagesPublisher(env, str(output_path))
-    target_url = url_parse("ghpages+https://MacDownApp/macdownapp.github.io")
-    branch = publisher.detect_target_branch(target_url)
-    assert branch == "master"
-
-
-def test_ghpages_detect_branch_project(tmpdir, env):
-    output_path = tmpdir.mkdir("output")
-    publisher = GithubPagesPublisher(env, str(output_path))
-    target_url = url_parse("ghpages+https://MacDownApp/MacDownApp.github.io/macdown")
-    branch = publisher.detect_target_branch(target_url)
-    assert branch == "gh-pages"
-
-
 def test_rsync_command_credentials(tmpdir, mocker, env):
     output_path = tmpdir.mkdir("output")
     publisher = RsyncPublisher(env, str(output_path))
     target_url = url_parse("http://example.com")
-    ssh_path = tmpdir.mkdir("ssh")
     credentials = {
         "username": "fakeuser",
         "password": "fakepass",
     }
     mock_popen = mocker.patch("lektor.publisher.portable_popen")
-    publisher.get_command(target_url, str(ssh_path), credentials)
-    assert mock_popen.called
-    assert mock_popen.call_args[0] == (
-        [
-            "rsync",
-            "-rclzv",
-            "--exclude=.lektor",
-            str(output_path) + "/",
-            "fakeuser@example.com:/",
-        ],
-    )
+    with publisher.get_command(target_url, credentials):
+        assert mock_popen.called
+        assert mock_popen.call_args[0] == (
+            [
+                "rsync",
+                "-rclzv",
+                "--exclude=.lektor",
+                str(output_path) + "/",
+                "fakeuser@example.com:/",
+            ],
+        )
 
 
 output_path = join(dirname(__file__), "OUTPUT_PATH")
@@ -267,8 +160,7 @@ output_path = join(dirname(__file__), "OUTPUT_PATH")
 def test_rsync_publisher(target_url, called_command, tmpdir, mocker, env):
     publisher = RsyncPublisher(env, str(output_path))
     target_url = url_parse(target_url)
-    ssh_path = join(output_path, "ssh")
     mock_popen = mocker.patch("lektor.publisher.portable_popen")
-    publisher.get_command(target_url, str(ssh_path), credentials=None)
-    assert mock_popen.called
-    assert mock_popen.call_args[0] == (called_command,)
+    with publisher.get_command(target_url, credentials=None):
+        assert mock_popen.called
+        assert mock_popen.call_args[0] == (called_command,)
diff --git a/tests/test_publisher.py b/tests/test_publisher.py
index c4cca1a..0c4ff68 100644
--- a/tests/test_publisher.py
+++ b/tests/test_publisher.py
@@ -1,14 +1,172 @@
 import gc
 import os
+import re
+import signal
 import sys
 import warnings
 import weakref
+from contextlib import ExitStack
+from itertools import chain
+from pathlib import Path
 from shutil import which
+from subprocess import CalledProcessError
+from subprocess import DEVNULL
+from subprocess import PIPE
+from subprocess import run
 
 import pytest
+from werkzeug.urls import url_parse
 
+from lektor.publisher import _ssh_command
+from lektor.publisher import _ssh_key_file
 from lektor.publisher import Command
+from lektor.publisher import GithubPagesPublisher
+from lektor.publisher import GitRepo
 from lektor.publisher import publish
+from lektor.publisher import PublishError
+from lektor.utils import locate_executable
+
+
+def test_ssh_key_file():
+    credentials = {"key_file": "/some/id", "key": "rsa:ignored"}
+    with _ssh_key_file(credentials) as key_file:
+        assert key_file == "/some/id"
+
+
+def test_ssh_key_file_returns_none():
+    with _ssh_key_file({}) as key_file:
+        assert key_file is None
+
+
+@pytest.mark.parametrize("prefix, key_type", [("test:", "TEST"), ("", "RSA")])
+def test_ssh_key_file_creates_key(prefix, key_type):
+    credentials = {"key": prefix + "".join(["1234567890abcdef"] * 7)}
+    with _ssh_key_file(credentials) as key_file:
+        assert Path(key_file).read_text() == (  # pylint: disable=unspecified-encoding
+            f"-----BEGIN {key_type} PRIVATE KEY-----\n"
+            "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef\n"
+            "1234567890abcdef1234567890abcdef1234567890abcdef\n"
+            f"-----END {key_type} PRIVATE KEY-----\n"
+        )
+    assert not Path(key_file).exists()
+
+
+@pytest.mark.parametrize(
+    "credentials, port, expected",
+    [
+        ({}, None, None),
+        ({}, 42, "ssh -p 42"),
+        ({"key_file": "/keyfile"}, None, 'ssh -i "/keyfile" -o IdentitiesOnly=yes'),
+    ],
+)
+def test_ssh_command(credentials, port, expected):
+    with _ssh_command(credentials, port) as key_file:
+        assert key_file == expected
+
+
+def _pyscript(script):
+    return (sys.executable, "-c", script)
+
+
+@pytest.mark.parametrize(
+    "cmd, input, expect_rc, expected",
+    [
+        ("print('xxx')", None, 0, ["xxx"]),
+        ("print('yyy', file=sys.stderr)", None, 0, ["yyy"]),
+        ("exit(42)", None, 42, []),
+        (
+            "for _ in sorted(sys.stdin): print(_.strip())",
+            "foo\nbar\n",
+            0,
+            ["bar", "foo"],
+        ),
+    ],
+)
+def test_Command_capture(cmd, input, expect_rc, expected):
+    command = Command(_pyscript("import sys\n" + cmd), input=input)
+
+    def iter_output():
+        proc = yield from command
+        assert proc.returncode == expect_rc
+        assert proc.stdout is None
+
+    assert list(iter_output()) == expected
+
+
+def test_Command_env():
+    command = Command(
+        _pyscript("import os; print(os.environ['TESTVAR'])"),
+        env={"TESTVAR": "boo"},
+    )
+    assert list(command) == ["boo"]
+
+
+def test_Command_raises_CalledProcessError():
+    command = Command(_pyscript("print('x'); exit(42)"), check=True)
+
+    def iter_output():
+        with pytest.raises(CalledProcessError) as exc_info:
+            yield from command
+        assert exc_info.value.returncode == 42
+
+    assert list(iter_output()) == ["x"]
+
+
+@pytest.mark.parametrize(
+    "capture_stdout, silent, stdout, out_err",
+    [
+        (False, False, None, ("foo\n", "bar\n")),
+        (True, False, "foo\n", ("", "bar\n")),
+        (False, True, None, ("", "")),
+        (True, True, "foo\n", ("", "")),
+    ],
+)
+def test_Command_no_capture(capture_stdout, silent, stdout, out_err, capfd):
+    command = Command(
+        _pyscript("import sys; print('foo'); print('bar', file=sys.stderr)"),
+        capture=False,
+        silent=silent,
+        capture_stdout=capture_stdout,
+    )
+    assert command.wait() == 0
+    assert command.result().stdout == stdout
+    captured = tuple(_.replace(os.linesep, "\n") for _ in capfd.readouterr())
+    assert captured == out_err
+
+
+def test_Command_iter_raises_if_not_capturing():
+    command = Command(_pyscript("exit(0)"), capture=False)
+    with pytest.raises(RuntimeError) as exc_info:
+        next(iter(command))
+    assert "Not capturing" in str(exc_info.value)
+    assert command.wait() == 0
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Windows")
+# XXX: Can not figure out how to reliably test KeyboardInterrupt handling under Windows.
+# Response to signal.SIGINT and/or signal.CTRL_C_EVENT varies depending on what shell
+# the test is run under.
+def test_Command_handles_keyboard_interrupt():
+    pid = os.getpid()
+    sig = signal.SIGINT
+    with pytest.raises(KeyboardInterrupt):
+        command = Command(
+            (sys.executable, "-c", f"import os; os.kill({pid}, {sig})"),
+            capture_stdout=True,
+        )
+        list(command)
+
+
+def test_Command_returncode():
+    command = Command((sys.executable, "-c", "exit(42)"))
+    assert command.returncode is None
+    list(command)
+    assert command.returncode == 42
+
+
+def test_Command_output():
+    command = Command(_pyscript("print('foo')"))
+    assert list(command.output) == ["foo"]
 
 
 def test_Command_triggers_no_warnings():
@@ -19,7 +177,7 @@ def test_Command_triggers_no_warnings():
         warnings.simplefilter("error")
 
         # This is essentially how RsyncPublisher runs rsync.
-        with Command([sys.executable, "-c", "print()"]) as client:
+        with Command(_pyscript("print()")) as client:
             for _ in client:
                 pass
 
@@ -64,3 +222,408 @@ def test_RsyncPublisher_integration(env, tmp_path, delete):
         for _ in target_path.iterdir()
     }
     assert target_files == files
+
+
+@pytest.fixture
+def work_tree(tmp_path):
+    work_tree = tmp_path / "work_tree"
+    work_tree.mkdir()
+    return work_tree
+
+
+@pytest.fixture
+def clean_git_environ(monkeypatch):
+    """Delete any GIT_ configuration in the current environment.
+
+    This unsets any existing GIT_SSH_COMMAND, among other things.
+
+    Of note, if one runs tests using git rebase exec, git sets all kinds
+    of GIT_* environment variables.
+    """
+    for key in list(os.environ):
+        if key.startswith("GIT_"):
+            monkeypatch.delitem(os.environ, key)
+
+
+@pytest.fixture
+def gitrepo(work_tree, clean_git_environ):
+    if not locate_executable("git"):
+        pytest.skip("no git")
+    with GitRepo(work_tree) as repo:
+        yield repo
+
+
+def test_GitRepo_popen(gitrepo):
+    assert "No commits yet" in list(gitrepo.popen("status"))
+
+
+def test_GitRepo_run(gitrepo):
+    proc = gitrepo.run("status", capture_stdout=True)
+    assert "No commits yet" in proc.stdout
+
+
+@pytest.mark.usefixtures("clean_git_environ")
+def test_GitRepo_set_ssh_credentials(gitrepo):
+    gitrepo.set_ssh_credentials({"key_file": "/id_foo"})
+    assert gitrepo.environ["GIT_SSH_COMMAND"].startswith('ssh -i "/id_foo"')
+
+
+@pytest.mark.usefixtures("clean_git_environ")
+def test_GitRepo_set_ssh_credentials_no_creds(gitrepo):
+    gitrepo.set_ssh_credentials({})
+    assert "GIT_SSH_COMMAND" not in gitrepo.environ
+
+
+def test_GitRepo_set_https_credentials(gitrepo):
+    gitrepo.set_https_credentials({"username": "user", "password": "pw"})
+    with gitrepo.popen("config", "--local", "--list") as output:
+        for line in output:
+            m = re.match(r'credential\.helper=store\s+--file\s+"(\S+)"', line)
+            if m:
+                cred_file = m.group(1)
+                break
+    # pylint: disable=unspecified-encoding
+    assert Path(cred_file).read_text().rstrip() == "https://user:pw@github.com"
+
+
+def test_GitRepo_set_https_credentials_no_cred(gitrepo):
+    gitrepo.set_https_credentials({})
+    assert not any(
+        line.startswith("credential.helper")
+        for line in gitrepo.popen("config", "--local", "--list")
+    )
+
+
+def test_GitRepo_add_to_index(gitrepo):
+    gitrepo.add_to_index("testfile", "testcontent\n")
+    proc = gitrepo.run("diff", "--cached", capture_stdout=True)
+    assert "+testcontent" in proc.stdout
+
+
+class DummyUpstreamRepo:
+    def __init__(self, git_dir):
+        run(("git", "init", "--bare", git_dir), check=True, stdout=DEVNULL)
+        self.git_dir = git_dir
+
+    @property
+    def url(self):
+        return self.git_dir
+
+    def run(self, args, **kwargs):
+        cmd = ("git", "--bare", "--git-dir", self.git_dir) + args
+        # XXX: Use text=True instead of encoding for py>3.6
+        return run(cmd, stdout=PIPE, encoding="utf-8", check=False)
+
+    def count_commits(self, branch):
+        proc = self.run(("log", "--pretty=oneline", branch))
+        return len(proc.stdout.splitlines())
+
+    def ls_files(self, treeish):
+        return set(self.iter_files(treeish))
+
+    def iter_files(self, treeish, prefix=""):
+        # Currently only lists top-level files
+        for line in self.run(("ls-tree", treeish)).stdout.splitlines():
+            _, typ, oid, name = line.split(maxsplit=3)
+            if typ == "blob":
+                yield prefix + name
+            else:
+                assert typ == "tree"
+                yield from self.iter_files(oid, prefix=f"{prefix}{name}/")
+
+
+@pytest.fixture
+def upstream_repo(tmp_path, clean_git_environ):
+    git_dir = tmp_path / "upstream.git"
+    return DummyUpstreamRepo(git_dir.__fspath__())
+
+
+@pytest.fixture
+def no_scary_output(capsys):
+    # Check that no scary messages are output to stdout
+    def is_scary(line):
+        return line.startswith("fatal")
+
+    yield
+    captured = capsys.readouterr()
+    scary_output = list(
+        filter(is_scary, chain(captured.out.splitlines(), captured.err.splitlines()))
+    )
+    assert len(scary_output) == 0
+
+
+@pytest.fixture
+def publish_ghpages(work_tree, clean_git_environ, no_scary_output):
+    # Run publish_ghpages on a fresh GitRepo instance
+    if not locate_executable("git"):
+        pytest.skip("no git")
+
+    def GitRepo_publish_ghpages(*args, **kwargs):
+        with GitRepo(work_tree) as repo:
+            rv = yield from repo.publish_ghpages(*args, **kwargs)
+            return rv
+
+    return GitRepo_publish_ghpages
+
+
+def test_GitRepo_publish_ghpages(publish_ghpages, upstream_repo, work_tree):
+    test_file = work_tree / "new-file"
+    test_file.write_text("test\n")
+    for line in publish_ghpages(upstream_repo.url, "gh-pages"):
+        print(line)
+    assert upstream_repo.count_commits("gh-pages") == 1
+    assert upstream_repo.ls_files("gh-pages") == {"new-file"}
+
+    test_file.rename(work_tree / "renamed-file")
+    for line in publish_ghpages(upstream_repo.url, "gh-pages"):
+        print(line)
+    assert upstream_repo.count_commits("gh-pages") == 2
+    assert upstream_repo.ls_files("gh-pages") == {"renamed-file"}
+
+
+def test_GitRepo_publish_ghpages_cname(publish_ghpages, upstream_repo, work_tree):
+    for line in publish_ghpages(upstream_repo.url, "gh-pages", "example.org"):
+        print(line)
+    assert upstream_repo.count_commits("gh-pages") == 1
+    assert upstream_repo.ls_files("gh-pages") == {"CNAME"}
+
+    for line in publish_ghpages(upstream_repo.url, "gh-pages"):
+        print(line)
+    assert upstream_repo.count_commits("gh-pages") == 2
+    assert upstream_repo.ls_files("gh-pages") == set()
+
+
+def test_GitRepo_publish_ghpages_ignores_lektor_dir(
+    publish_ghpages, upstream_repo, work_tree
+):
+    lektor_dir = work_tree / ".lektor"
+    lektor_dir.mkdir()
+    lektor_dir.joinpath("ignored").write_text("should not be published")
+    work_tree.joinpath("included").write_text("should be published")
+
+    for line in publish_ghpages(upstream_repo.url, "gh-pages"):
+        print(line)
+    assert upstream_repo.count_commits("gh-pages") == 1
+    assert upstream_repo.ls_files("gh-pages") == {"included"}
+
+
+def test_GitRepo_publish_ghpages_no_changes(publish_ghpages, upstream_repo):
+    for line in publish_ghpages(upstream_repo.url, "gh-pages", "example.com"):
+        print(line)
+    assert upstream_repo.ls_files("gh-pages") == {"CNAME"}
+    assert any(
+        "No changes" in line
+        for line in publish_ghpages(upstream_repo.url, "gh-pages", "example.com")
+    )
+
+
+def test_GitRepo_publish_ghpages_discard_history(
+    publish_ghpages, upstream_repo, work_tree
+):
+    for line in publish_ghpages(upstream_repo.url, "gh-pages", "example.com"):
+        print(line)
+    assert upstream_repo.ls_files("gh-pages") == {"CNAME"}
+
+    work_tree.joinpath("myfile").write_text("contents")
+    for line in publish_ghpages(upstream_repo.url, "gh-pages", None, False):
+        print(line)
+    assert upstream_repo.count_commits("gh-pages") == 1
+    assert upstream_repo.ls_files("gh-pages") == {"myfile"}
+
+
+@pytest.fixture
+def output_path(tmp_path):
+    return str(tmp_path / "output_path")
+
+
+@pytest.fixture
+def ghp_publisher(env, output_path):
+    return GithubPagesPublisher(env, output_path)
+
+
+@pytest.mark.parametrize(
+    "target_url, publish_args, warns",
+    [
+        (
+            "ghpages://owner/project?cname=example.com&preserve_history=no",
+            (
+                "ssh://git@github.com/owner/project.git",
+                "gh-pages",
+                "example.com",
+                False,
+            ),
+            False,
+        ),
+        (
+            "ghpages+https://owner/owner.github.io",
+            ("https://github.com/owner/owner.github.io.git", "master", None, True),
+            True,
+        ),
+    ],
+)
+@pytest.mark.skipif(not locate_executable("git"), reason="no git")
+def test_GithubPagesPublisher_publish(
+    ghp_publisher, output_path, mocker, target_url, publish_args, warns
+):
+    GitRepo = mocker.patch("lektor.publisher.GitRepo", spec_set=True)
+    repo = mocker.Mock(spec_set=GitRepo)
+    repo.publish_ghpages.return_value = iter(["Published!"])
+    GitRepo.return_value.__enter__.return_value = repo
+
+    url = url_parse(target_url)
+    with ExitStack() as stack:
+        if warns:
+            stack.enter_context(pytest.deprecated_call())
+        output = list(ghp_publisher.publish(url))
+
+    GitRepo.assert_called_once_with(output_path)
+    if "+https:" in target_url:
+        repo.set_https_credentials.assert_called_once_with({})
+    else:
+        repo.set_ssh_credentials.assert_called_once_with({})
+    repo.publish_ghpages.assert_called_once_with(*publish_args)
+    if warns:
+        assert "WARNING" in " ".join(output)
+        output = output[-1:]
+    assert output == ["Published!"]
+
+
+@pytest.mark.usefixtures("no_utils")
+def test_GithubPagesPublisher_publish_fails_if_no_git(ghp_publisher):
+    url = url_parse("ghpages://owner/project")
+    with pytest.raises(PublishError) as exc_info:
+        list(ghp_publisher.publish(url))
+    assert re.search(r"git.*not found", str(exc_info.value))
+
+
+@pytest.mark.parametrize(
+    "target_url, expected",
+    [
+        (
+            "ghpages://owner/project",
+            ("ssh://git@github.com/owner/project.git", "gh-pages", None, True, []),
+        ),
+        (
+            "ghpages://owner/owner.github.io?branch=main",
+            ("ssh://git@github.com/owner/owner.github.io.git", "main", None, True, []),
+        ),
+        (
+            "ghpages://owner/project?branch=brnch&cname=example.com",
+            (
+                "ssh://git@github.com/owner/project.git",
+                "brnch",
+                "example.com",
+                True,
+                [],
+            ),
+        ),
+        (
+            "ghpages://owner/project?preserve_history=no",
+            ("ssh://git@github.com/owner/project.git", "gh-pages", None, False, []),
+        ),
+        (
+            "ghpages://owner/project?preserve_history=yes",
+            ("ssh://git@github.com/owner/project.git", "gh-pages", None, True, []),
+        ),
+        (
+            "ghpages+ssh://own/proj",
+            ("ssh://git@github.com/own/proj.git", "gh-pages", None, True, []),
+        ),
+        (
+            "ghpages+https://own/proj",
+            ("https://github.com/own/proj.git", "gh-pages", None, True, []),
+        ),
+        (
+            "ghpages+ssh://own:22/proj",
+            ("ssh://git@github.com/own/proj.git", "gh-pages", None, True, []),
+        ),
+        (
+            "ghpages+https://own:443/proj",
+            ("https://github.com/own/proj.git", "gh-pages", None, True, []),
+        ),
+    ],
+)
+def test_GithubPagesPublisher_parse_url(ghp_publisher, target_url, expected):
+    url = url_parse(target_url)
+    assert ghp_publisher._parse_url(url) == expected
+
+
+def test_GithubPagesPublisher_parse_url_warns_on_default_master_branch(ghp_publisher):
+    url = url_parse("ghpages://owner/owner.github.io")
+    with pytest.deprecated_call():
+        push_url, branch, cname, preserve_history, warnings = ghp_publisher._parse_url(
+            url
+        )
+    assert (push_url, branch, cname, preserve_history) == (
+        "ssh://git@github.com/owner/owner.github.io.git",
+        "master",
+        None,
+        True,
+    )
+    assert len(warnings) > 0
+    assert re.search(r"WARNING.*set.*branch", " ".join(warnings))
+
+
+@pytest.mark.parametrize(
+    "target_url, expected",
+    [
+        ("ghpages:///project", "owner missing"),
+        ("ghpages://owner", "project missing"),
+        ("ghpages://owner:2222/project", "non-standard port"),
+        ("ghpages+https://owner:4430/project", "non-standard port"),
+    ],
+)
+def test_GithubPagesPublisher_parse_url_failures(ghp_publisher, target_url, expected):
+    url = url_parse(target_url)
+    with pytest.raises(PublishError) as exc_info:
+        ghp_publisher._parse_url(url)
+    assert expected in str(exc_info.value)
+
+
+@pytest.mark.parametrize(
+    "target_url, credentials, expected",
+    [
+        ("ghpages://owner/proj", None, {}),
+        (
+            "ghpages://owner/proj",
+            {"username": "user", "password": "pw"},
+            {"username": "user", "password": "pw"},
+        ),
+        ("ghpages://user:pw@owner/proj", None, {"username": "user", "password": "pw"}),
+        (
+            "ghpages://user:pw@owner/proj",
+            {"arbi": "trary"},
+            {"username": "user", "password": "pw", "arbi": "trary"},
+        ),
+        ("ghpages://user:pw@owner/proj", None, {"username": "user", "password": "pw"}),
+        (
+            "ghpages://user:pw@owner/proj",
+            {"password": "secret"},
+            {"username": "user", "password": "secret"},
+        ),
+    ],
+)
+def test_GithubPagesPublisher_parse_credentials(
+    ghp_publisher, credentials, target_url, expected
+):
+    url = url_parse(target_url)
+    assert ghp_publisher._parse_credentials(credentials, url) == expected
+
+
+def test_publish(env, output_path, mocker):
+    Publisher = mocker.Mock()
+    env.add_publisher("publishtest", Publisher)
+    credentials = {"foo": "bar"}
+    rv = publish(env, "publishtest://host/path", output_path, credentials)
+    url = url_parse("publishtest://host/path")
+    assert Publisher.mock_calls == [
+        mocker.call(env, output_path),
+        mocker.call().publish(url, credentials),
+    ]
+    assert rv is Publisher.return_value.publish.return_value
+
+
+def test_publish_unknown_scheme(env, output_path):
+    with pytest.raises(PublishError) as exc_info:
+        publish(env, "unknownscheme://host/path", output_path)
+    assert "unknown scheme" in str(exc_info.value)
diff --git a/tests/test_quickstart.py b/tests/test_quickstart.py
index a7acd0d..36dda8d 100644
--- a/tests/test_quickstart.py
+++ b/tests/test_quickstart.py
@@ -18,13 +18,6 @@ class struct_passwd(NamedTuple):
     pw_shell: str = "/bin/false"
 
 
-@pytest.fixture
-def no_utils(monkeypatch):
-    """Monkeypatch $PATH to hide any installed external utilities (e.g. git)."""
-    monkeypatch.setitem(os.environ, "PATH", "/dev/null")
-    locate_executable.cache_clear()
-
-
 @pytest.fixture
 def git_config_file(tmp_path, monkeypatch):
     """Create a temporary git config file, and monkeypatch $GIT_CONFIG to point to it."""