Classroom_Desktop/tests/test-classroom-presentations.sh

#!/bin/bash
# The Presentations library: what it imports, what it refuses, what it
# keeps and what it is never allowed to delete.
#
# No LibreOffice and no poppler here. Those are exercised by the lab and
# by the release validator against a real Debian; this is the part that
# has to be right whatever is installed — when a download counts as
# finished, whose files may be touched, what a teacher's choice of
# thumbnail does and does not change, and which of the two retention
# clocks is running.
set -uo pipefail

ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

PYTHONPATH="$ROOT/lib" python3 - "$ROOT" <<'PY'
import os
import errno
import importlib.util
import shutil
import sys
import tempfile
import time
from datetime import timedelta
from importlib.machinery import SourceFileLoader
from pathlib import Path

root = Path(sys.argv[1])
sys.path.insert(0, str(root / "lib"))

import classroom_presentations as library

loader = SourceFileLoader(
    "classroom_presentation_import",
    str(root / "scripts" / "classroom-presentation-import"),
)
spec = importlib.util.spec_from_loader(loader.name, loader)
importer = importlib.util.module_from_spec(spec)
loader.exec_module(importer)

passed = failed = 0


def check(label, condition, detail=""):
    global passed, failed
    if condition:
        print(f"  ok   {label}")
        passed += 1
    else:
        print(f"  FAIL {label}")
        if detail:
            print(f"       {detail}")
        failed += 1


work = Path(tempfile.mkdtemp(prefix="classroom-presentations-"))
downloads = work / "Downloads"
wonderink = downloads / "Wonder Ink"
shelf = work / "library"
wonderink.mkdir(parents=True)
shelf.mkdir(parents=True)


def download(name, contents="x" * 400, age=60):
    """A file sitting in Downloads, finished `age` seconds ago."""
    path = downloads / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(contents)
    when = time.time() - age
    os.utime(path, (when, when))
    return path


def entry(**data):
    """A presentation in the library, without running an import."""
    directory = shelf / data.get("id", "one")
    directory.mkdir(parents=True, exist_ok=True)
    base = {
        "id": directory.name,
        "state": "ready",
        "title": "A session",
        "pages": 10,
        "thumbnail_page": 1,
        "imported_at": library.stamp(),
        "last_presented_at": None,
        "times_presented": 0,
    }
    base.update(data)
    presentation = library.Presentation(directory, base)
    presentation.save()
    (directory / "slides.pdf").write_text("%PDF-1.4")
    return presentation


print("1. a download that has finished, and ones that have not")
done = download("Session 7 - Romans.pptx")
check("a settled PowerPoint is importable", library.importable(done))

arriving = download("Still arriving.pptx")
partial = downloads / "Still arriving.pptx.crdownload"
partial.write_text("half")
check("one with a .crdownload beside it is not",
      not library.importable(arriving),
      "Chrome renames the .crdownload away when the transfer completes")

check("and the .crdownload itself is never a candidate",
      not library.importable(partial))

fresh = download("Just landed.pdf", age=0)
check("one still being written is not",
      not library.importable(fresh),
      "a file copied from a USB stick has no .crdownload to go by")

check("something that is not a presentation is not",
      not library.importable(download("notes.txt")))

empty = download("Nothing.pdf", contents="", age=60)
check("a nought-byte file is not", not library.importable(empty))

print("2. Wonder Ink's worksheets are not presentations")
worksheet = wonderink / "Colouring page.pdf"
worksheet.write_text("%PDF-1.4" + "x" * 400)
os.utime(worksheet, (time.time() - 60, time.time() - 60))
check("a file in the Wonder Ink folder is recognised as Wonder Ink's",
      library.within_wonderink(worksheet, downloads))
check("and one beside it in Downloads is not",
      not library.within_wonderink(done, downloads))

names = [p.name for p in library.candidates(downloads)]
check("so it is not offered for import", "Colouring page.pdf" not in names,
      str(names))
check("the finished session is", "Session 7 - Romans.pptx" in names, str(names))
check("and nothing unfinished is",
      "Still arriving.pptx" not in names and "Just landed.pdf" not in names,
      str(names))

print("3. titles a teacher would recognise")
check("an ordinary name is kept",
      library.readable_title("Session 7 - Romans.pptx") == "Session 7 - Romans",
      library.readable_title("Session 7 - Romans.pptx"))
check("underscores become spaces",
      library.readable_title("Session_7_Romans.pdf") == "Session 7 Romans",
      library.readable_title("Session_7_Romans.pdf"))
check("a browser's duplicate counter goes",
      library.readable_title("Bible Study (1).pdf") == "Bible Study",
      library.readable_title("Bible Study (1).pdf"))
check("a percent-escaped space is decoded",
      library.readable_title("Session%207.pdf") == "Session 7",
      library.readable_title("Session%207.pdf"))
hashed = library.readable_title(
    "a4f39c2e91bb4d7e.pdf", "Session 7 of the autumn study\nmore")
check("a hash falls back to the document's own first words",
      hashed == "Session 7 of the autumn study", hashed)
check("and something with no name at all still gets one",
      library.readable_title("...pdf", "") == "Presentation",
      library.readable_title("...pdf", ""))

print("4. two downloads of the same name are two presentations")
first = entry(id="first", title="Session 7")
second = entry(id="second", title="Session 7")
# The bug this replaces: excluding the import being named BY TITLE also
# removed the other presentation's identical title from the set, so a
# PDF and a PPTX of one session both came out called "Session 7".
taken = library.titles_in_use(shelf, exclude="second")
check("the other presentation's title is still in the way",
      "Session 7" in taken, str(taken))
check("so the second gets a title of its own",
      library.unique_title("Session 7", taken) == "Session 7 (2)",
      library.unique_title("Session 7", taken))
check("and a name nothing else has is left alone",
      library.unique_title("Session 8", taken) == "Session 8")
shutil.rmtree(shelf / "first")
shutil.rmtree(shelf / "second")

print("5. the thumbnail is the teacher's choice, and only that")
# The first slide of a Bible study session is a generic opening shared by
# every session; the one that names the session is usually the second.
p = entry(id="thumb", pages=12)
before = dict(p.data)
p.set_thumbnail_page(2)
after = library.load(shelf / "thumb")
check("the chosen slide is remembered", after.thumbnail_page == 2,
      str(after.thumbnail_page))
check("it survives being read back from disk, so it survives a restart",
      library.load(shelf / "thumb").thumbnail_page == 2)
check("choosing it did not mark the session as presented",
      after.last_presented_at is None and after.times_presented == 0)
check("and did not move the retention clock",
      after.data["imported_at"] == before["imported_at"]
      and after.data.get("last_presented_at") is None)
check("a slide past the end is clamped rather than stored",
      entry(id="thumb2", pages=5, thumbnail_page=99).thumbnail_page == 5)
check("and a nonsense one does not crash the card",
      entry(id="thumb3", pages=5, thumbnail_page="banana").thumbnail_page == 1)

print("6. the two retention clocks")
old = library.now() - timedelta(days=181)
recent = library.now() - timedelta(days=100)

never = entry(id="never", imported_at=library.stamp(old))
check("never presented, imported over 180 days ago, is expired",
      never.expired(), str(never.expires_at()))

young = entry(id="young", imported_at=library.stamp(recent))
check("never presented but only 100 days old is not",
      not young.expired(), str(young.expires_at()))

shown = entry(id="shown",
              imported_at=library.stamp(old),
              last_presented_at=library.stamp(library.now() - timedelta(days=2)),
              times_presented=3)
check("presented two days ago is kept, however old the import is",
      not shown.expired(),
      "presenting has to start the 30-day clock and stop the 180-day one")

stale = entry(id="stale",
              imported_at=library.stamp(old),
              last_presented_at=library.stamp(library.now() - timedelta(days=31)),
              times_presented=1)
check("presented 31 days ago is expired", stale.expired())

print("7. presenting is the only thing that resets the clock")
p = entry(id="counted", pages=4)
check("a fresh import has never been presented",
      p.last_presented_at is None and p.times_presented == 0)
p.mark_presented()
reread = library.load(shelf / "counted")
check("presenting is recorded", reread.last_presented_at is not None)
check("and counted", reread.times_presented == 1)
first_due = reread.expires_at()
time.sleep(1.1)
reread.mark_presented()
again = library.load(shelf / "counted")
check("presenting again starts the thirty days over",
      again.expires_at() > first_due, f"{first_due} -> {again.expires_at()}")
check("and counts again", again.times_presented == 2)

print("8. an import that was interrupted is not shown to a teacher")
half = entry(id="half", state="importing")
check("it is in the library's own list",
      any(x.id == "half" for x in library.library(shelf)))
check("but it is not ready, so no card offers Present",
      not library.load(shelf / "half").ready)
broken = shelf / "broken"
broken.mkdir()
(broken / "presentation.json").write_text("{ this is not json")
check("a truncated metadata file is skipped rather than crashing the library",
      all(x.id != "broken" for x in library.library(shelf)),
      "a card a teacher cannot open is worse than no card")

print("9. the presenting lock cannot outlive a reboot")
check("it lives in the runtime directory",
      "/run" in str(library.presenting_lock())
      or str(library.presenting_lock()).startswith("/tmp"),
      str(library.presenting_lock()))

print("10. the ledger identifies a download transaction")
mark = library.fingerprint(done)
library.remember(mark, "some-id", shelf)
check("a fingerprint is path, size and time",
      mark is not None and str(done.resolve()) in mark, str(mark))
check("it is written down", library.ledger(shelf).get(mark) == "some-id")
check("so an interrupted move can be matched after a restart",
      mark in library.ledger(shelf))
touched = download("Session 7 - Romans.pptx", contents="y" * 900)
check("but re-downloading the same session is a new presentation",
      library.fingerprint(touched) != mark,
      "a different size or time is a different download")

print("11. a completed download is moved into the managed library")
original_render_pages = library.render_pages
original_render_thumbnail = library.render_thumbnail
original_first_text = library.first_text
original_convert = library.convert_to_pdf


def fake_render_pages(_pdf, directory, width=library.SLIDE_WIDTH,
                      progress=None):
    directory.mkdir(parents=True, exist_ok=True)
    (directory / "page-001.png").write_bytes(b"png")
    return 1, ""


def fake_render_thumbnail(_source, destination,
                          width=library.THUMBNAIL_WIDTH):
    destination.write_bytes(b"thumbnail")
    return True


library.render_pages = fake_render_pages
library.render_thumbnail = fake_render_thumbnail
library.first_text = lambda _pdf, pages=2: ""

moved = download("Move me.pdf", contents="%PDF-1.4 moved")
moved_contents = moved.read_bytes()
managed = importer.import_one(moved, shelf)
check("the source leaves Downloads after import", not moved.exists())
check("the one original is kept in the managed entry",
      managed is not None and managed.managed_copy.read_bytes() == moved_contents,
      str(managed.managed_copy if managed else None))
check("its PDF and page images are derived assets",
      managed is not None and managed.ready and managed.slides.exists()
      and managed.page_image(1).exists())

print("12. conversion failure does not lose the downloaded original")
library.convert_to_pdf = lambda _source, _destination: (False, "broken deck")
broken_source = download("Broken deck.pptx", contents="not really a deck")
failed_import = importer.import_one(broken_source, shelf)
check("the failed source was still moved out of Downloads",
      not broken_source.exists())
check("a failed card retains the managed original",
      failed_import is not None and failed_import.failed
      and failed_import.managed_copy.read_text() == "not really a deck")
check("the failure did not create a second original",
      failed_import is not None
      and list((failed_import.directory / "original").iterdir())
      == [failed_import.managed_copy])

print("13. a cross-filesystem move publishes before deleting")
cross_source = download("Across filesystems.pdf", contents="cross-device")
cross_destination = work / "other-device" / cross_source.name
real_replace = importer.os.replace
source_text = str(cross_source)
destination_text = str(cross_destination)


def cross_replace(source, destination):
    if os.fspath(source) == source_text and os.fspath(destination) == destination_text:
        raise OSError(errno.EXDEV, "different filesystems")
    return real_replace(source, destination)


importer.os.replace = cross_replace
try:
    importer.move_download(cross_source, cross_destination)
finally:
    importer.os.replace = real_replace

check("the cross-filesystem destination is complete",
      cross_destination.read_text() == "cross-device")
check("only the destination remains after it is durable",
      not cross_source.exists()
      and not cross_destination.with_name(cross_destination.name + ".moving").exists())

print("14. restart recovery never deletes the only moved original")
resume_dir = shelf / "resume"
resume_original = resume_dir / "original" / "Resume.pdf"
resume_original.parent.mkdir(parents=True)
resume_original.write_text("%PDF-1.4 resume")
resume_data = {
    "id": "resume",
    "state": "importing",
    "title": "Resume",
    "original_name": "Resume.pdf",
    "source_path": str(downloads / "Resume.pdf"),
    "managed_copy": "original/Resume.pdf",
    "format": "pdf",
    "imported_at": library.stamp(),
    "last_presented_at": None,
    "times_presented": 0,
    "thumbnail_page": 1,
    "pages": 0,
    "error": None,
}
library.Presentation(resume_dir, resume_data).save()
importer.sweep(shelf)
resumed = library.load(resume_dir)
check("an import interrupted after the move resumes in place",
      resumed is not None and resumed.ready)
check("and its original survives recovery", resume_original.exists())

print("15. removal is isolated and refuses an active presentation")
remove_one = entry(id="remove-one", title="Similar title")
keep_one = entry(id="keep-one", title="Similar title (2)")
(remove_one.directory / "original").mkdir()
(remove_one.directory / "original" / "one.pdf").write_text("one")
(keep_one.directory / "original").mkdir()
(keep_one.directory / "original" / "two.pdf").write_text("two")
check("removing one managed entry succeeds", library.remove(remove_one))
check("a similarly named presentation and its files are untouched",
      keep_one.directory.exists()
      and (keep_one.directory / "original" / "two.pdf").exists())

runtime = work / "runtime"
runtime.mkdir()
old_runtime = os.environ.get("XDG_RUNTIME_DIR")
os.environ["XDG_RUNTIME_DIR"] = str(runtime)
library.presenting_lock().write_text(keep_one.id)
check("an actively presented entry cannot be removed",
      not library.remove(keep_one) and keep_one.directory.exists())
library.presenting_lock().unlink()
if old_runtime is None:
    os.environ.pop("XDG_RUNTIME_DIR", None)
else:
    os.environ["XDG_RUNTIME_DIR"] = old_runtime

library.render_pages = original_render_pages
library.render_thumbnail = original_render_thumbnail
library.first_text = original_first_text
library.convert_to_pdf = original_convert

print("16. teacher-facing wording and compositor policy")
ui = (root / "scripts" / "classroom-presentations").read_text()
layout = (root / "config" / "plasma" / "lookandfeel" / "contents"
          / "layouts" / "org.kde.plasma.desktop-layout.js").read_text()
kwin = (root / "config" / "plasma" / "xdg" / "kwinrc").read_text()
rules = (root / "config" / "plasma" / "xdg" / "kwinrulesrc").read_text()
check("the empty state has one concise browser-neutral instruction",
      'label="No presentations yet\\n\\n"' in ui
      and "Download a PowerPoint or PDF to get started." in ui
      and "Download a presentation in Chrome" not in ui)
check("cover selection uses the requested plain wording",
      'label="Use as cover"' in ui and 'label="Current cover"' in ui
      and "this is the library picture" not in ui)
check("Remove no longer promises a copy in Chrome",
      "This will delete it from this computer." in ui
      and "Your download in Chrome" not in ui)
check("the presenter has a compositor-enforced fullscreen rule",
      "title=Presenting" in rules and "fullscreen=true" in rules
      and "fullscreenrule=2" in rules)
check("normal title bars contain only Close",
      "ButtonsOnRight=X" in kwin and "ButtonsOnRight=FX" not in kwin)
check("the unfinished Help launcher is absent from the panel",
      "ekklesia-manual.desktop" not in layout)

print("17. running out of memory is a state a teacher can act on, never a loop")
library.render_pages = fake_render_pages
library.render_thumbnail = fake_render_thumbnail
library.first_text = lambda _pdf, pages=2: ""
# The kernel killing LibreOffice: SIGKILL directly, or 137 through the
# soffice shell wrapper.
check("a SIGKILL is recognised either way",
      library.killed(-9) and library.killed(137)
      and not library.killed(1) and not library.killed(0))

library.convert_to_pdf = lambda _source, _destination: (False, library.KILLED)
heavy = download("Heavy deck.pptx", contents="photographs")
starved = importer.import_one(heavy, shelf)
check("a conversion killed for memory is a failed import",
      starved is not None and starved.failed, starved.data if starved else None)
check("of the kind a teacher can try again",
      starved.failure == "memory" and starved.data["error_kind"] == "memory")
check("and the downloaded original is kept",
      starved.managed_copy.read_text() == "photographs")
check("the attempt was counted before the work began",
      starved.data.get("attempts") == 1 and starved.data.get("attempt_started_at"))
check("a file that simply will not convert is a different failure",
      failed_import.failure == "unreadable")
check("and cannot be tried again", not library.request_retry(failed_import))

# A service killed in the middle of an import, found by the next start.
library.convert_to_pdf = original_convert


def interrupted(name, attempts):
    directory = shelf / name
    (directory / "original").mkdir(parents=True)
    (directory / "original" / f"{name}.pdf").write_text("%PDF-1.4 " + name)
    library.Presentation(directory, {
        "id": name, "state": "importing", "title": name,
        "original_name": f"{name}.pdf", "source_path": str(downloads / f"{name}.pdf"),
        "managed_copy": f"original/{name}.pdf", "format": "pdf",
        "imported_at": library.stamp(), "last_presented_at": None,
        "times_presented": 0, "thumbnail_page": 1, "pages": 0, "error": None,
        "attempts": attempts,
    }).save()
    return directory


once = interrupted("once", 1)
twice = interrupted("twice", 2)
importer.sweep(shelf)
check("an import interrupted once is resumed",
      library.load(once).ready)
gave_up = library.load(twice)
check("an import interrupted twice is not tried a third time",
      gave_up.failed and gave_up.failure == "memory", gave_up.data)
check("and its original is still there to try again from",
      (twice / "original" / "twice.pdf").exists())

check("Try again puts it back in the importer's hands",
      library.request_retry(gave_up) and library.load(twice).state == "retry")
check("which counts afresh", library.load(twice).data.get("attempts") == 0)
check("and the importer takes it up on its next pass",
      importer.retry_requested(shelf) == 1 and library.load(twice).ready)

# The importer not running at all: nothing will ever finish this one.
stuck = interrupted("stuck", 0)
entry_data = library.load(stuck).data
long_ago = library.now() - timedelta(seconds=library.STALLED_SECONDS + 60)
entry_data["attempt_started_at"] = library.stamp(long_ago)
library.Presentation(stuck, entry_data).save()
stuck_card = library.load(stuck)
check("an import nobody is working on stops saying Getting this ready",
      stuck_card.stalled and stuck_card.failure == "memory")
check("and offers Try again", library.request_retry(stuck_card))
check("a fresh import is not mistaken for a stalled one",
      not library.load(once).stalled and library.load(once).failure is None)

held = importer.hold_lock(shelf)
check("one importer holds the lock", held is not None)
check("a second importer is refused rather than run beside it",
      importer.hold_lock(shelf) is None)
held.close()
check("and it is free again once the first has gone",
      importer.hold_lock(shelf) is not None)
check("the lock file is not a presentation",
      all(p.id != ".importer.lock" for p in library.library(shelf)))

unit = (root / "systemd" / "user" / "ekklesia-classroom-presentations.service").read_text()
check("the service may use at most 40% of the machine's memory",
      "MemoryMax=40%" in unit)
check("a killed converter does not take the watcher with it",
      "OOMPolicy=continue" in unit)
check("and under pressure it goes before the teacher's windows",
      "OOMScoreAdjust=800" in unit)
check("the card says what to do, in a teacher's words",
      "Close other programs, then choose" in ui and 'label="Try again"' in ui)

library.render_pages = original_render_pages
library.render_thumbnail = original_render_thumbnail
library.first_text = original_first_text
library.convert_to_pdf = original_convert

shutil.rmtree(work, ignore_errors=True)

print()
print(f"{passed} passed, {failed} failed")
raise SystemExit(1 if failed else 0)
PY