nixpkgs/nixos/tests/printing.nix

# Test printing via CUPS.

import ./make-test-python.nix (
  {
    pkgs,
    socket ? true, # whether to use socket activation
    listenTcp ? true, # whether to open port 631 on client
    ...
  }:

  let
    inherit (pkgs) lib;
  in

  {
    name = "printing";
    meta = with lib.maintainers; {
      maintainers = [
        domenkozar
        matthewbauer
      ];
    };

    nodes.server =
      { ... }:
      {
        services.printing = {
          enable = true;
          stateless = true;
          startWhenNeeded = socket;
          listenAddresses = [ "*:631" ];
          defaultShared = true;
          openFirewall = true;
          extraConf = ''
            <Location />
              Order allow,deny
              Allow from all
            </Location>
          '';
        };
        # Add a HP Deskjet printer connected via USB to the server.
        hardware.printers.ensurePrinters = [
          {
            name = "DeskjetLocal";
            deviceUri = "usb://foobar/printers/foobar";
            model = "drv:///sample.drv/deskjet.ppd";
          }
        ];
      };

    nodes.client =
      { lib, ... }:
      {
        services.printing.enable = true;
        services.printing.startWhenNeeded = socket;
        services.printing.listenAddresses = lib.mkIf (!listenTcp) [ ];
        # Add printer to the client as well, via IPP.
        hardware.printers.ensurePrinters = [
          {
            name = "DeskjetRemote";
            deviceUri = "ipp://server/printers/DeskjetLocal";
            model = "drv:///sample.drv/deskjet.ppd";
          }
        ];
        hardware.printers.ensureDefaultPrinter = "DeskjetRemote";
      };

    testScript = ''
      import os
      import re

      start_all()

      with subtest("Make sure that cups is up on both sides and printers are set up"):
          server.wait_for_unit("ensure-printers.service")
          client.wait_for_unit("ensure-printers.service")

      assert "scheduler is running" in client.succeed("lpstat -r")

      with subtest("UNIX socket is used for connections"):
          assert "/var/run/cups/cups.sock" in client.succeed("lpstat -H")

      with subtest("HTTP server is available too"):
          ${lib.optionalString listenTcp ''client.succeed("curl --fail http://localhost:631/")''}
          client.succeed(f"curl --fail http://{server.name}:631/")
          server.fail(f"curl --fail --connect-timeout 2 http://{client.name}:631/")

      with subtest("LP status checks"):
          assert "DeskjetRemote accepting requests" in client.succeed("lpstat -a")
          assert "DeskjetLocal accepting requests" in client.succeed(
              f"lpstat -h {server.name}:631 -a"
          )
          client.succeed("cupsdisable DeskjetRemote")
          out = client.succeed("lpq")
          print(out)
          assert re.search(
              "DeskjetRemote is not ready.*no entries",
              client.succeed("lpq"),
              flags=re.DOTALL,
          )
          client.succeed("cupsenable DeskjetRemote")
          assert re.match(
              "DeskjetRemote is ready.*no entries", client.succeed("lpq"), flags=re.DOTALL
          )

      # Test printing various file types.
      for file in [
          "${pkgs.groff.doc}/share/doc/*/examples/mom/penguin.pdf",
          "${pkgs.groff.doc}/share/doc/*/meref.ps",
          "${pkgs.cups.out}/share/doc/cups/images/cups.png",
          "${pkgs.pcre.doc}/share/doc/pcre/pcre.txt",
      ]:
          file_name = os.path.basename(file)
          with subtest(f"print {file_name}"):
              # Print the file on the client.
              print(client.succeed("lpq"))
              client.succeed(f"lp {file}")
              client.wait_until_succeeds(
                  f"lpq; lpq | grep -q -E 'active.*root.*{file_name}'"
              )

              # Ensure that a raw PCL file appeared in the server's queue
              # (showing that the right filters have been applied).  Of
              # course, since there is no actual USB printer attached, the
              # file will stay in the queue forever.
              server.wait_for_file("/var/spool/cups/d*-001")
              server.wait_until_succeeds(f"lpq -a | grep -q -E '{file_name}'")

              # Delete the job on the client.  It should disappear on the
              # server as well.
              client.succeed("lprm")
              client.wait_until_succeeds("lpq -a | grep -q -E 'no entries'")

              retry(lambda _: "no entries" in server.succeed("lpq -a"))

              # The queue is empty already, so this should be safe.
              # Otherwise, pairs of "c*"-"d*-001" files might persist.
              server.execute("rm /var/spool/cups/*")
    '';
  }
)