Add Peertube test

services/flake-part.nix

@@ -9,6 +9,7 @@
       checks = {
         mastodon = import ./tests/mastodon.nix { inherit self pkgs; };
         pixelfed-garage = import ./tests/pixelfed-garage.nix { inherit self pkgs; };
+        peertube = import ./tests/peertube.nix { inherit self pkgs; };
       };
     };
 }

services/tests/peertube.nix

@@ -0,0 +1,233 @@
+## This file is a basic test of Peertube functionalities.
+
+{ pkgs, self }:
+
+let
+  lib = pkgs.lib;
+
+  testVideo = pkgs.copyPathToStore ./green.mp4;
+  testVideoColour = "#00FF00";
+
+  postVideoInBrowser =
+    pkgs.writers.writePython3Bin "post-video-in-browser"
+      {
+        libraries = with pkgs.python3Packages; [ selenium ];
+        flakeIgnore = [ "E501" ]; # welcome to the 21st century
+      }
+      ''
+        import sys
+        from urllib.parse import urlparse
+
+        from selenium import webdriver
+        from selenium.webdriver.common.by import By
+        from selenium.webdriver.firefox.options import Options
+        from selenium.webdriver.support.ui import WebDriverWait
+        from selenium.common.exceptions import NoSuchElementException
+
+        options = Options()
+        print("########################################", file=sys.stderr)
+        print("A", file=sys.stderr)
+        options.add_argument("--headless")
+        print("########################################", file=sys.stderr)
+        print("B", file=sys.stderr)
+        service = webdriver.FirefoxService(executable_path="${lib.getExe pkgs.geckodriver}")
+        print("########################################", file=sys.stderr)
+        print("C", file=sys.stderr)
+        driver = webdriver.Firefox(options=options, service=service)
+        print("########################################", file=sys.stderr)
+        print("D", file=sys.stderr)
+        driver.set_window_size(4096, 2160)
+        print("########################################", file=sys.stderr)
+        print("E", file=sys.stderr)
+        driver.implicitly_wait(360)
+        print("########################################", file=sys.stderr)
+        print("F", file=sys.stderr)
+        wait = WebDriverWait(driver, timeout=360, poll_frequency=10)
+        print("########################################", file=sys.stderr)
+
+        ############################################################
+        # Login
+
+
+        def load(driver, page):
+            print(f"Loading page {page}...", file=sys.stderr)
+            driver.get(page)
+
+            print("Waiting until page is loaded...", file=sys.stderr)
+            wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
+
+            if urlparse(driver.current_url).path == "/login":
+                print("Hit a login page.", file=sys.stderr)
+
+                print("Enter username...", file=sys.stderr)
+                driver.find_element(By.ID, "username").send_keys("root")
+
+                print("Enter password...", file=sys.stderr)
+                driver.find_element(By.ID, "password").send_keys(sys.argv[1])
+
+                print("Click “Login” button...", file=sys.stderr)
+                driver.find_element(By.XPATH, "//input[@value='Login']").click()
+
+                print("Waiting until we are logged-in...", file=sys.stderr)
+                wait.until(lambda d: urlparse(d.current_url).path == urlparse(page).path)
+
+                print("Waiting until page is loaded...", file=sys.stderr)
+                wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
+
+            print("Clicking the annoying setup wizard away...", file=sys.stderr)
+            try:
+                driver.find_element(By.XPATH, "//input[@value='Remind me later']").click()
+            except NoSuchElementException:
+                # Somehow, sometimes, the wizard just does not show up; then we
+                # ignore the error and carry on like nothing happened.
+                print("Setup wizard did not show up.", file=sys.stderr)
+
+            print(f"Done loading page {page}.", file=sys.stderr)
+
+
+        ############################################################
+        # Upload video and take a screenshot
+
+        print("Go to the upload page...", file=sys.stderr)
+        load(driver, "http://peertube.localhost/videos/upload")
+
+        print("Submit video file...", file=sys.stderr)
+        driver.find_element(By.XPATH, "//input[@type='file']").send_keys("${testVideo}")
+
+        print("Wait for file to upload, then publish it...", file=sys.stderr)
+        publish_button = driver.find_element(By.XPATH, "//button[.//span[normalize-space()='Publish']]")
+        wait.until(lambda _d: "disabled" not in publish_button.get_attribute("class"))
+        publish_button.click()
+
+        print("Waiting until we are redirected...", file=sys.stderr)
+        wait.until(lambda d: urlparse(d.current_url).path != "/videos/upload")
+        wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
+
+        # FIXME: The video cannot play and we get “Failed to play video”. I
+        # believe it is a codec problem, and it possibly has to do with video
+        # codecs enabled in Firefox in headless mode -- after all, who would
+        # want to play a video? The following is a list of things that I have
+        # tried without success. Maybe one day we can manage?
+        #
+        # video = driver.find_element(By.XPATH, "//video")
+        # wait.until(lambda _d: not video.get_attribute("src").startswith("blob:"))
+        # wait.until(lambda d: d.execute_script("return arguments[0].readyState", video) == 4)
+        # driver.find_element(By.XPATH, "//*[contains(text(), 'Failed to play video')]")
+        #
+        # def detect_image_in_screen(d):
+        #     print("Taking a screenshot...", file=sys.stderr)
+        #     d.save_screenshot("/screenshot.png")
+        #     print("Checking it...", file=sys.stderr)
+        #     displayed_colours = subprocess.run(
+        #         [
+        #             "magick",
+        #             "/screenshot.png",
+        #             "-define",
+        #             "histogram:unique-colors=true",
+        #             "-format",
+        #             "%c",
+        #             "histogram:info:",
+        #         ],
+        #         capture_output=True,
+        #         text=True,
+        #         check=True,
+        #     ).stdout
+        #     return bool(re.match(".*#${testVideoColour}.*", displayed_colours, re.S))
+        #
+        # print("Wait until the image shows in screen...", file=sys.stderr)
+        # wait.until(detect_image_in_screen)
+
+        ############################################################
+
+        print("Done; bye!", file=sys.stderr)
+        driver.close()
+      '';
+
+  acquireRootPassword = pkgs.writeShellScriptBin "acquire-root-password" ''
+    readonly retry=30
+    readonly wait=60
+
+    for _ in $(seq $retry); do
+      password=$(journalctl -u peertube | perl -ne '/password: (.*)/ && print $1')
+      if [ -n "$password" ]; then
+        echo "$password"
+        exit 0
+      fi
+      sleep $wait
+    done
+
+    echo "Could not acquire root password in $((retry * wait))s."
+    exit 1
+  '';
+in
+
+pkgs.nixosTest {
+  name = "peertube";
+
+  nodes = {
+    server =
+      { config, ... }:
+      {
+        imports = with self.nixosModules; [
+          fediversity
+          ../vm/garage-vm.nix
+          ../vm/peertube-vm.nix
+          ../vm/interactive-vm.nix
+        ];
+
+        virtualisation = {
+          memorySize = lib.mkVMOverride 8192;
+          cores = 8;
+        };
+
+        environment.systemPackages = with pkgs; [
+          python3
+          firefox-unwrapped
+          geckodriver
+          toot
+          acquireRootPassword
+          postVideoInBrowser
+          imagemagick
+          ffmpeg # to identify videos
+          expect
+        ];
+
+        ## FIXME: The CI is very slow, so the default timeout of 120s is not
+        ## good enough. We bump it drastically.
+        systemd.services.postgresql.serviceConfig.TimeoutSec = lib.mkForce 3600;
+
+        environment.variables = {
+          AWS_ACCESS_KEY_ID = config.services.garage.ensureKeys.peertube.id;
+          AWS_SECRET_ACCESS_KEY = config.services.garage.ensureKeys.peertube.secret;
+          PT_INITIAL_ROOT_PASSWORD = "testtest";
+        };
+      };
+  };
+
+  testScript =
+    { nodes, ... }:
+    ''
+      server.start()
+
+      # FIXME: I think this trick to look for a password can be replaced by
+      # services.peertube.serviceEnvironmentFile.PT_INITIAL_ROOT_PASSWORD=testtest
+
+      with subtest("Peertube starts"):
+        server.wait_for_unit("peertube.service")
+        root_password = server.succeed("acquire-root-password").rstrip()
+
+      with subtest("Post a video in the browser"):
+        server.succeed(f"post-video-in-browser {root_password}")
+
+      with subtest("Find video in garage"):
+        server.succeed("mc alias set garage ${nodes.server.fediversity.internal.garage.api.url} --api s3v4 --path off $AWS_ACCESS_KEY_ID $AWS_SECRET_ACCESS_KEY")
+        video = server.succeed("mc find garage --regex '\\.mp4'").rstrip()
+        if video == "":
+          raise Exception("Could not find any .mp4 video stored in Garage")
+        server.succeed(f"mc cat {video} > /video.mp4")
+        garage_hash = server.succeed("identify -quiet -format '%#' /video.mp4")
+        hash = server.succeed("identify -quiet -format '%#' ${testVideo}")
+        if garage_hash != hash:
+          raise Exception("The video stored in Garage does not correspond to the original one.")
+    '';
+}