pyrox.dev
1# Test printing via CUPS.
2
3import ./make-test-python.nix ({pkgs, ... }:
4let
5 printingServer = startWhenNeeded: {
6 services.printing.enable = true;
7 services.printing.stateless = true;
8 services.printing.startWhenNeeded = startWhenNeeded;
9 services.printing.listenAddresses = [ "*:631" ];
10 services.printing.defaultShared = true;
11 services.printing.extraConf = ''
12 <Location />
13 Order allow,deny
14 Allow from all
15 </Location>
16 '';
17 networking.firewall.allowedTCPPorts = [ 631 ];
18 # Add a HP Deskjet printer connected via USB to the server.
19 hardware.printers.ensurePrinters = [{
20 name = "DeskjetLocal";
21 deviceUri = "usb://foobar/printers/foobar";
22 model = "drv:///sample.drv/deskjet.ppd";
23 }];
24 };
25 printingClient = startWhenNeeded: {
26 services.printing.enable = true;
27 services.printing.startWhenNeeded = startWhenNeeded;
28 # Add printer to the client as well, via IPP.
29 hardware.printers.ensurePrinters = [{
30 name = "DeskjetRemote";
31 deviceUri = "ipp://${if startWhenNeeded then "socketActivatedServer" else "serviceServer"}/printers/DeskjetLocal";
32 model = "drv:///sample.drv/deskjet.ppd";
33 }];
34 hardware.printers.ensureDefaultPrinter = "DeskjetRemote";
35 };
36
37in {
38 name = "printing";
39 meta = with pkgs.lib.maintainers; {
40 maintainers = [ domenkozar eelco matthewbauer ];
41 };
42
43 nodes = {
44 socketActivatedServer = { ... }: (printingServer true);
45 serviceServer = { ... }: (printingServer false);
46
47 socketActivatedClient = { ... }: (printingClient true);
48 serviceClient = { ... }: (printingClient false);
49 };
50
51 testScript = ''
52 import os
53 import re
54
55 start_all()
56
57 with subtest("Make sure that cups is up on both sides and printers are set up"):
58 serviceServer.wait_for_unit("cups.service")
59 serviceClient.wait_for_unit("cups.service")
60 socketActivatedClient.wait_for_unit("ensure-printers.service")
61
62
63 def test_printing(client, server):
64 assert "scheduler is running" in client.succeed("lpstat -r")
65
66 with subtest("UNIX socket is used for connections"):
67 assert "/var/run/cups/cups.sock" in client.succeed("lpstat -H")
68 with subtest("HTTP server is available too"):
69 client.succeed("curl --fail http://localhost:631/")
70 client.succeed(f"curl --fail http://{server.name}:631/")
71 server.fail(f"curl --fail --connect-timeout 2 http://{client.name}:631/")
72
73 with subtest("LP status checks"):
74 assert "DeskjetRemote accepting requests" in client.succeed("lpstat -a")
75 assert "DeskjetLocal accepting requests" in client.succeed(
76 f"lpstat -h {server.name}:631 -a"
77 )
78 client.succeed("cupsdisable DeskjetRemote")
79 out = client.succeed("lpq")
80 print(out)
81 assert re.search(
82 "DeskjetRemote is not ready.*no entries",
83 client.succeed("lpq"),
84 flags=re.DOTALL,
85 )
86 client.succeed("cupsenable DeskjetRemote")
87 assert re.match(
88 "DeskjetRemote is ready.*no entries", client.succeed("lpq"), flags=re.DOTALL
89 )
90
91 # Test printing various file types.
92 for file in [
93 "${pkgs.groff.doc}/share/doc/*/examples/mom/penguin.pdf",
94 "${pkgs.groff.doc}/share/doc/*/meref.ps",
95 "${pkgs.cups.out}/share/doc/cups/images/cups.png",
96 "${pkgs.pcre.doc}/share/doc/pcre/pcre.txt",
97 ]:
98 file_name = os.path.basename(file)
99 with subtest(f"print {file_name}"):
100 # Print the file on the client.
101 print(client.succeed("lpq"))
102 client.succeed(f"lp {file}")
103 client.wait_until_succeeds(
104 f"lpq; lpq | grep -q -E 'active.*root.*{file_name}'"
105 )
106
107 # Ensure that a raw PCL file appeared in the server's queue
108 # (showing that the right filters have been applied). Of
109 # course, since there is no actual USB printer attached, the
110 # file will stay in the queue forever.
111 server.wait_for_file("/var/spool/cups/d*-001")
112 server.wait_until_succeeds(f"lpq -a | grep -q -E '{file_name}'")
113
114 # Delete the job on the client. It should disappear on the
115 # server as well.
116 client.succeed("lprm")
117 client.wait_until_succeeds("lpq -a | grep -q -E 'no entries'")
118
119 retry(lambda _: "no entries" in server.succeed("lpq -a"))
120
121 # The queue is empty already, so this should be safe.
122 # Otherwise, pairs of "c*"-"d*-001" files might persist.
123 server.execute("rm /var/spool/cups/*")
124
125
126 test_printing(serviceClient, serviceServer)
127 test_printing(socketActivatedClient, socketActivatedServer)
128 '';
129})