anchor.diff:  # Test printing via CUPS.
   
  -import ./make-test.nix ({pkgs, ... }:
  +import ./make-test-python.nix ({pkgs, ... }:
   let
     printingServer = startWhenNeeded: {
       services.printing.enable = true;
       services.printing.startWhenNeeded = startWhenNeeded;
       services.printing.listenAddresses = [ "*:631" ];
       services.printing.defaultShared = true;
  -    services.printing.extraConf =
  -      ''
  -        <Location />
  -          Order allow,deny
  -          Allow from all
  -        </Location>
  -      '';
  +    services.printing.extraConf = ''
  +      <Location />
  +        Order allow,deny
  +        Allow from all
  +      </Location>
  +    '';
       networking.firewall.allowedTCPPorts = [ 631 ];
       # Add a HP Deskjet printer connected via USB to the server.
       hardware.printers.ensurePrinters = [{
  
positive.code:     hardware.printers.ensureDefaultPrinter = "DeskjetRemote";
    };
  
  in
  
  {
    name = "printing";
    meta = with pkgs.stdenv.lib.maintainers; {
      maintainers = [ domenkozar eelco matthewbauer ];
  
negative.code:     serviceClient = { ... }: (printingClient false);
    };
  
    testScript = ''
      import os
      import re
      import sys
  
      start_all()
  
      with subtest("Make sure that cups is up on both sides"):
          serviceServer.wait_for_unit("cups.service")
          serviceClient.wait_for_unit("cups.service")
  
      with subtest(
          "Wait until cups is fully initialized and ensure-printers has "
          "executed with 10s delay"
      ):
          serviceClient.sleep(20)
          socketActivatedClient.wait_until_succeeds(
              "systemctl status ensure-printers | grep -q -E 'code=exited, status=0/SUCCESS'"
          )
  
  
      def test_printing(client, server):
          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"):
              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/*")
  
  
      test_printing(serviceClient, serviceServer)
      test_printing(socketActivatedClient, socketActivatedServer)
    '';
  })
  