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"):
57 serviceServer.wait_for_unit("cups.service")
58 serviceClient.wait_for_unit("cups.service")
59
60 with subtest(
61 "Wait until cups is fully initialized and ensure-printers has "
62 "executed with 10s delay"
63 ):
64 serviceClient.sleep(20)
65 socketActivatedClient.wait_until_succeeds(
66 "systemctl status ensure-printers | grep -q -E 'code=exited, status=0/SUCCESS'"
67 )
68
69
70 def test_printing(client, server):
71 assert "scheduler is running" in client.succeed("lpstat -r")
72
73 with subtest("UNIX socket is used for connections"):
74 assert "/var/run/cups/cups.sock" in client.succeed("lpstat -H")
75 with subtest("HTTP server is available too"):
76 client.succeed("curl --fail http://localhost:631/")
77 client.succeed(f"curl --fail http://{server.name}:631/")
78 server.fail(f"curl --fail --connect-timeout 2 http://{client.name}:631/")
79
80 with subtest("LP status checks"):
81 assert "DeskjetRemote accepting requests" in client.succeed("lpstat -a")
82 assert "DeskjetLocal accepting requests" in client.succeed(
83 f"lpstat -h {server.name}:631 -a"
84 )
85 client.succeed("cupsdisable DeskjetRemote")
86 out = client.succeed("lpq")
87 print(out)
88 assert re.search(
89 "DeskjetRemote is not ready.*no entries",
90 client.succeed("lpq"),
91 flags=re.DOTALL,
92 )
93 client.succeed("cupsenable DeskjetRemote")
94 assert re.match(
95 "DeskjetRemote is ready.*no entries", client.succeed("lpq"), flags=re.DOTALL
96 )
97
98 # Test printing various file types.
99 for file in [
100 "${pkgs.groff.doc}/share/doc/*/examples/mom/penguin.pdf",
101 "${pkgs.groff.doc}/share/doc/*/meref.ps",
102 "${pkgs.cups.out}/share/doc/cups/images/cups.png",
103 "${pkgs.pcre.doc}/share/doc/pcre/pcre.txt",
104 ]:
105 file_name = os.path.basename(file)
106 with subtest(f"print {file_name}"):
107 # Print the file on the client.
108 print(client.succeed("lpq"))
109 client.succeed(f"lp {file}")
110 client.wait_until_succeeds(
111 f"lpq; lpq | grep -q -E 'active.*root.*{file_name}'"
112 )
113
114 # Ensure that a raw PCL file appeared in the server's queue
115 # (showing that the right filters have been applied). Of
116 # course, since there is no actual USB printer attached, the
117 # file will stay in the queue forever.
118 server.wait_for_file("/var/spool/cups/d*-001")
119 server.wait_until_succeeds(f"lpq -a | grep -q -E '{file_name}'")
120
121 # Delete the job on the client. It should disappear on the
122 # server as well.
123 client.succeed("lprm")
124 client.wait_until_succeeds("lpq -a | grep -q -E 'no entries'")
125
126 retry(lambda _: "no entries" in server.succeed("lpq -a"))
127
128 # The queue is empty already, so this should be safe.
129 # Otherwise, pairs of "c*"-"d*-001" files might persist.
130 server.execute("rm /var/spool/cups/*")
131
132
133 test_printing(serviceClient, serviceServer)
134 test_printing(socketActivatedClient, socketActivatedServer)
135 '';
136})