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