Blame for editor/admin.py

"""Do the update, don't just report it.

`fleet.py` reads every mesh's database and says what is out of date. Nothing
could act on that: installing or updating an extension writes files into the
site, and Joomla exposes no REST endpoint for it — `installer/install` is 404.
The only interface that can is the one a human uses, `/administrator`. So this
drives that, headless, with Playwright.

Division of labour, and it is the whole trick: **the browser acts, the database
reads.** Clicking "Check for Updates" is what fills `#__updates`; once it is
filled, `fleet.survey` already knows how to read it exactly. So nothing here
scrapes a version out of HTML — the only things parsed from the page are the
checkbox rows we are about to tick and the alerts Joomla shows afterwards.
That keeps this file small and immune to every admin-template change that
isn't the update form itself.

Writes to a live production site. `run(dry=True)` is the default: it signs in,
refreshes the update cache (which is itself only a read on the site's part) and
reports what it *would* have ticked, without ticking it.
"""
import os
import re
import time
from contextlib import contextmanager

import fleet
import mesh

TIMEOUT = 60_000
# Joomla's own update view. Selecting everything and pressing Update is
# precisely the bulk operation, so there is no per-extension loop to write.
UPDATE_URL = "/administrator/index.php?option=com_installer&view=update"
# Core rides a different, multi-step component. Out of scope deliberately: it
# takes the site offline mid-run and wants a backup first.
CORE_URL = "/administrator/index.php?option=com_joomlaupdate"


class AdminError(RuntimeError):
    pass


def credentials(site, username=None, password=None):
    """Typed-in credentials win; otherwise the site's own record, otherwise the
    process env. Nothing new is stored — an admin password is not written to
    the registry by this module."""
    env = mesh.env(site) if site is not None else os.environ
    return (username or env.get("ADMIN_USER") or env.get("TEST_JOOMLA_USERNAME") or "",
            password or env.get("ADMIN_PASS") or env.get("TEST_JOOMLA_PASSWORD") or "")


@contextmanager
def browser(headless=True):
    # Imported here, not at module scope: a deployment without a browser (the
    # HF Space, which has neither the room nor a writable cache) must lose the
    # update run and nothing else.
    try:
        from playwright.sync_api import sync_playwright
    except ImportError:
        raise AdminError(
            "This deployment has no browser. Install it where the app runs: "
            "pip install playwright && python -m playwright install chromium")

    with sync_playwright() as pw:
        b = pw.chromium.launch(headless=headless)
        page = b.new_page()
        page.set_default_timeout(TIMEOUT)
        try:
            yield page
        finally:
            b.close()


def sign_in(page, base, username, password):
    """The admin login form has been the same three fields since Joomla 3."""
    if not username or not password:
        raise AdminError("No administrator credentials for this site.")
    page.goto(base.rstrip("/") + "/administrator/", wait_until="domcontentloaded")
    if page.locator("#mod-login-username").count() == 0:
        return  # already signed in on this context
    page.fill("#mod-login-username", username)
    page.fill("#mod-login-password", password)
    page.press("#mod-login-password", "Enter")
    page.wait_for_load_state("domcontentloaded")
    # The form disappearing is the only reliable success signal — the dashboard
    # itself is a template and every site restyles it.
    if page.locator("#mod-login-username").count():
        why = page.locator(".alert").first
        raise AdminError("Administrator sign-in refused"
                         + (f": {why.inner_text().strip()[:120]}" if why.count() else "."))


def _toolbar(page, task):
    """Joomla 4/5 wraps toolbar buttons in <joomla-toolbar-button task="...">;
    Joomla 3 renders a plain onclick button. Try both, prefer the modern one."""
    for sel in (f'joomla-toolbar-button[task="{task}"] button',
                f'joomla-toolbar-button[task="{task}"]',
                f'button[onclick*="{task}"]',
                f'#toolbar-{task.split(".")[-1]} button'):
        el = page.locator(sel).first
        if el.count():
            el.click()
            return True
    return False


def find_updates(page, base):
    """Press "Check for Updates". This is the run that fills `#__updates` —
    without it the cache is whatever it was, which is what `fleet.stale` warns
    about."""
    page.goto(base.rstrip("/") + UPDATE_URL, wait_until="domcontentloaded")
    if not _toolbar(page, "update.find"):
        raise AdminError("No 'Check for Updates' button on the update view.")
    page.wait_for_load_state("networkidle")


def pending_rows(page):
    """The rows Joomla is offering, straight off the form we are about to
    submit. Read from the page, not the database, because these carry the
    checkbox values — the update ids, which nothing else knows."""
    rows = []
    for cb in page.locator('input[name="cid[]"]').all():
        tr = cb.locator("xpath=ancestor::tr[1]")
        # First line of the name cell. The whole row carries the update-site
        # URL and the download link too, which is noise in a log.
        cells = tr.locator("td, th")
        text = cells.nth(1).inner_text() if cells.count() > 1 else tr.inner_text()
        label = next((ln.strip() for ln in text.splitlines() if ln.strip()), "?")
        rows.append({"cid": cb.get_attribute("value"), "label": label[:120]})
    return rows


def apply_updates(page, only=None):
    """Tick everything (or the rows matching `only`) and press Update."""
    rows = pending_rows(page)
    if only:
        want = [o.lower() for o in only]
        rows = [r for r in rows if any(w in r["label"].lower() for w in want)]
    if not rows:
        return rows, []
    for r in rows:
        page.locator(f'input[name="cid[]"][value="{r["cid"]}"]').first.check()
    if not _toolbar(page, "update.update"):
        raise AdminError("No 'Update' button on the update view.")
    # An update unpacks and copies files; it is slow and it is not one request.
    page.wait_for_load_state("networkidle", timeout=TIMEOUT * 5)
    messages = [" ".join(a.inner_text().split())[:300]
                for a in page.locator("#system-message-container .alert").all()]
    return rows, messages


INSTALL_URL = "/administrator/index.php?option=com_installer&view=install"


def _alerts(page):
    return [" ".join(a.inner_text().split())[:300]
            for a in page.locator("#system-message-container .alert").all()]


def install_package(page, base, path=None, url=None):
    """Put one package onto the site through Joomla's own installer.

    Two doors, and the file is preferred. A vendor's download link is usually
    tied to the session or the download id that fetched it, so handing that URL
    to the *site* to fetch fails in a way that looks like the package is bad. We
    already have the file; uploading it is the path that does not depend on
    whether the vendor lets a stranger through.

    Nothing is parsed out of the page except Joomla's own alerts. The verdict
    on whether it worked comes from the database afterwards, as everywhere else
    here.
    """
    if not path and not url:
        raise AdminError("nothing to install — a package file or a URL is needed")

    page.goto(base.rstrip("/") + INSTALL_URL, wait_until="domcontentloaded")

    if path:
        chooser = page.locator('input[type="file"]#install_package').first
        if not chooser.count():
            chooser = page.locator('input[type="file"]').first
        if not chooser.count():
            raise AdminError("no upload field on the install view")
        chooser.set_input_files(str(path))
        # The upload tab's own button, not a toolbar task.
        for sel in ("#installbutton_package", 'button:has-text("Upload & Install")',
                    'input[type="submit"][value*="Install"]'):
            button = page.locator(sel).first
            if button.count():
                button.click()
                break
        else:
            raise AdminError("no 'Upload & Install' button on the install view")
    else:
        tab = page.locator('button:has-text("Install from URL"), a:has-text("Install from URL")').first
        if tab.count():
            tab.click()
        field = page.locator("#install_url").first
        if not field.count():
            raise AdminError("no 'Install from URL' field on the install view")
        field.fill(url)
        button = page.locator("#installbutton_url").first
        if not button.count():
            raise AdminError("no install button beside the URL field")
        button.click()

    # Unpacking and copying files is slow and is not one request.
    page.wait_for_load_state("networkidle", timeout=TIMEOUT * 5)
    messages = _alerts(page)
    bad = [m for m in messages if re.search(r"\berror\b|\bfail|\bwarning\b", m, re.I)]
    if bad:
        raise AdminError("Joomla refused the package: " + " | ".join(bad)[:400])
    return messages


def install_run(site, path=None, url=None, element="", username=None, password=None,
                headless=True, actor="", skip_backup=False):
    """One site: back up, install, then say what actually changed.

    The backup is not optional and not best-effort. An extension install runs
    the vendor's migration SQL against a live production database, and that is
    the one step nothing here can undo — so a site that cannot be backed up
    does not get installed onto. `skip_backup` exists for the operator who has
    just taken one by hand, and it is recorded as having been skipped.
    """
    import backup
    import runlog

    base = mesh.env(site)["JOOMLA_URL"]
    user, pw = credentials(site, username, password)
    started = time.time()
    mesh.CURRENT.set(site or "")

    run_id = runlog.start("install", element or (path and os.path.basename(str(path))) or url,
                          actor=actor, detail={"site": site, "url": url, "file": str(path or "")},
                          site=site)
    yield {"site": site, "step": "start", "run": run_id, "element": element}

    try:
        if skip_backup:
            snapshot = None
            yield {"site": site, "step": "backup", "skipped": True}
        else:
            yield {"site": site, "step": "backup", "note": "snapshotting the database first"}
            snapshot = backup.take(site, reason=f"install {element or 'package'}")
            yield {"site": site, "step": "backed-up", "file": snapshot["file"],
                   "bytes": snapshot["bytes"]}

        before = fleet.survey(site)

        with browser(headless=headless) as page:
            sign_in(page, base, user, pw)
            yield {"site": site, "step": "signed-in"}
            messages = install_package(page, base, path=path, url=url)
            yield {"site": site, "step": "installed", "messages": messages}

        after = fleet.survey(site)
        result = {"site": site, "step": "done",
                  "outdated_before": before["outdated_count"],
                  "outdated_after": after["outdated_count"],
                  "backup": (snapshot or {}).get("file"),
                  "messages": messages,
                  "seconds": round(time.time() - started, 1)}
        runlog.finish(run_id, runlog.DONE, note="; ".join(messages)[:400],
                      detail=result, site=site)
        yield result

    except Exception as e:
        why = f"{type(e).__name__}: {e}"
        runlog.finish(run_id, runlog.FAILED, note=why[:400], site=site)
        # The handler must not raise. Whatever went wrong upstream is the
        # thing the operator needs; an exception in here would replace it with
        # a worse one and lose the only useful message.
        try:
            snap = locals().get("snapshot")
            restore = backup.restore_command(snap) if snap else ""
        except Exception:
            restore = ""
        yield {"site": site, "step": "error", "run": run_id, "error": why,
               "restore": restore}


def install_fleet(element, sites=None, headless=True, actor="", skip_backup=False):
    """The captured package onto every mesh that still needs it.

    One site at a time, and one site failing does not stop the rest — but each
    gets its own backup and its own record, so a bad release is contained to
    the sites it reached rather than to all of them.
    """
    import packages

    entry = packages.known().get(element)
    if not entry:
        raise AdminError(f"no captured package for {element} — capture it first")
    plan = packages.targets(element)
    names = sites or [s["site"] for s in plan["sites"] if "error" not in s]

    yield {"step": "fleet", "element": element, "version": plan["version"],
           "sites": names, "package": entry.get("file")}
    for name in names:
        try:
            for event in install_run(name, path=entry.get("file"), url=entry.get("url"),
                                     element=element, headless=headless, actor=actor,
                                     skip_backup=skip_backup):
                yield event
        except Exception as e:
            yield {"site": name, "step": "error", "error": f"{type(e).__name__}: {e}"}
    yield {"step": "fleet-done", "element": element}


def run(site=None, username=None, password=None, dry=True, only=None, headless=True):
    """One mesh, start to finish, as a stream of events.

    Yields dicts so a console can show it happening — an update run is minutes
    long, and a call that only returns at the end is indistinguishable from a
    hang.
    """
    base = mesh.env(site)["JOOMLA_URL"]
    user, pw = credentials(site, username, password)
    started = time.time()
    yield {"site": site, "step": "start", "url": base, "dry": dry, "user": user}

    before = fleet.survey(site)
    yield {"site": site, "step": "before", "outdated": before["outdated_count"],
           "checked": before["checked"], "stale": before["stale"]}

    with browser(headless=headless) as page:
        sign_in(page, base, user, pw)
        yield {"site": site, "step": "signed-in"}

        find_updates(page, base)
        rows = pending_rows(page)
        yield {"site": site, "step": "found", "count": len(rows),
               "rows": [r["label"] for r in rows]}

        if dry:
            yield {"site": site, "step": "done", "dry": True, "updated": 0,
                   "would_update": len(rows), "seconds": round(time.time() - started, 1)}
            return

        picked, messages = apply_updates(page, only)
        yield {"site": site, "step": "applied", "count": len(picked),
               "messages": messages}

    # The site's own database is the verdict, not the page we just left.
    after = fleet.survey(site)
    yield {"site": site, "step": "done", "dry": False,
           "outdated_before": before["outdated_count"],
           "outdated_after": after["outdated_count"],
           "updated": len(picked), "messages": messages,
           "seconds": round(time.time() - started, 1)}


def run_fleet(dry=True, only=None, sites=None, username=None, password=None):
    """Every mesh in turn. One site failing must not stop the rest — a fleet
    run that aborts on the first bad password is useless at fleet scale."""
    names = sites or list(mesh.sites())
    yield {"step": "fleet", "sites": names, "dry": dry}
    for name in names:
        try:
            for event in run(name, username=username, password=password,
                             dry=dry, only=only):
                yield event
        except Exception as e:
            yield {"site": name, "step": "error",
                   "error": f"{type(e).__name__}: {str(e)[:300]}"}
    yield {"step": "fleet-done"}