movefasta/nixpkgs
1
0
Fork 0
mirror of https://github.com/NixOS/nixpkgs.git synced 2025-06-12 04:35:41 +03:00
Code Issues Projects Releases Packages Wiki Activity

nixosTests.printing-{socket,service,socket-notcp,service-notcp}: migrate to runTest

Browse source 
Part of #386873
This commit is contained in:
Piotr Kwiecinski 2025-04-16 14:55:07 +02:00
parent 8d74a87fa1
commit 46ddd0d906
No known key found for this signature in database
GPG key ID: EC0DE1CB9D5258B4
2 changed files with 135 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

28
nixos/tests/all-tests.nix
View file

@ -1091,21 +1091,25 @@ in
pretalx = runTest ./web-apps/pretalx.nix; pretalx = runTest ./web-apps/pretalx.nix;
prefect = runTest ./prefect.nix; prefect = runTest ./prefect.nix;
pretix = runTest ./web-apps/pretix.nix; pretix = runTest ./web-apps/pretix.nix;
printing-socket = handleTest ./printing.nix { printing-socket = runTest {
socket = true; imports = [ ./printing.nix ];
listenTcp = true; _module.args.socket = true;
_module.args.listenTcp = true;
}; };
printing-service = handleTest ./printing.nix { printing-service = runTest {
socket = false; imports = [ ./printing.nix ];
listenTcp = true; _module.args.socket = false;
_module.args.listenTcp = true;
}; };
printing-socket-notcp = handleTest ./printing.nix { printing-socket-notcp = runTest {
socket = true; imports = [ ./printing.nix ];
listenTcp = false; _module.args.socket = true;
_module.args.listenTcp = false;
}; };
printing-service-notcp = handleTest ./printing.nix { printing-service-notcp = runTest {
socket = false; imports = [ ./printing.nix ];
listenTcp = false; _module.args.socket = false;
_module.args.listenTcp = false;
}; };
private-gpt = handleTest ./private-gpt.nix { }; private-gpt = handleTest ./private-gpt.nix { };
privatebin = runTest ./privatebin.nix; privatebin = runTest ./privatebin.nix;

241
nixos/tests/printing.nix
View file

@ -1,141 +1,138 @@
# Test printing via CUPS. # Test printing via CUPS.
{
pkgs,
socket ? true, # whether to use socket activation
listenTcp ? true, # whether to open port 631 on client
...
}:
import ./make-test-python.nix ( let
{ inherit (pkgs) lib;
pkgs, in
socket ? true, # whether to use socket activation
listenTcp ? true, # whether to open port 631 on client
...
}:
let {
inherit (pkgs) lib; name = "printing";
in meta = with lib.maintainers; {
maintainers = [
domenkozar
matthewbauer
];
};
{ nodes.server =
name = "printing"; { ... }:
meta = with lib.maintainers; { {
maintainers = [ services.printing = {
domenkozar enable = true;
matthewbauer 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.server = nodes.client =
{ ... }: { lib, ... }:
{ {
services.printing = { services.printing.enable = true;
enable = true; services.printing.startWhenNeeded = socket;
stateless = true; services.printing.listenAddresses = lib.mkIf (!listenTcp) [ ];
startWhenNeeded = socket; # Add printer to the client as well, via IPP.
listenAddresses = [ "*:631" ]; hardware.printers.ensurePrinters = [
defaultShared = true; {
openFirewall = true; name = "DeskjetRemote";
extraConf = '' deviceUri = "ipp://server/printers/DeskjetLocal";
<Location /> model = "drv:///sample.drv/deskjet.ppd";
Order allow,deny }
Allow from all ];
</Location> hardware.printers.ensureDefaultPrinter = "DeskjetRemote";
''; };
};
# 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 = testScript = ''
{ lib, ... }: import os
{ import re
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 = '' start_all()
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")
with subtest("Make sure that cups is up on both sides and printers are set up"): assert "scheduler is running" in client.succeed("lpstat -r")
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("UNIX socket is used for connections"): with subtest("HTTP server is available too"):
assert "/var/run/cups/cups.sock" in client.succeed("lpstat -H") ${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("HTTP server is available too"): with subtest("LP status checks"):
${lib.optionalString listenTcp ''client.succeed("curl --fail http://localhost:631/")''} assert "DeskjetRemote accepting requests" in client.succeed("lpstat -a")
client.succeed(f"curl --fail http://{server.name}:631/") assert "DeskjetLocal accepting requests" in client.succeed(
server.fail(f"curl --fail --connect-timeout 2 http://{client.name}:631/") 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
)
with subtest("LP status checks"): # Test printing various file types.
assert "DeskjetRemote accepting requests" in client.succeed("lpstat -a") for file in [
assert "DeskjetLocal accepting requests" in client.succeed( "${pkgs.groff.doc}/share/doc/*/examples/mom/penguin.pdf",
f"lpstat -h {server.name}:631 -a" "${pkgs.groff.doc}/share/doc/*/meref.ps",
) "${pkgs.cups.out}/share/doc/cups/images/cups.png",
client.succeed("cupsdisable DeskjetRemote") "${pkgs.pcre.doc}/share/doc/pcre/pcre.txt",
out = client.succeed("lpq") ]:
print(out) file_name = os.path.basename(file)
assert re.search( with subtest(f"print {file_name}"):
"DeskjetRemote is not ready.*no entries", # Print the file on the client.
client.succeed("lpq"), print(client.succeed("lpq"))
flags=re.DOTALL, client.succeed(f"lp {file}")
) client.wait_until_succeeds(
client.succeed("cupsenable DeskjetRemote") f"lpq; lpq | grep -q -E 'active.*root.*{file_name}'"
assert re.match( )
"DeskjetRemote is ready.*no entries", client.succeed("lpq"), flags=re.DOTALL
)
# Test printing various file types. # Ensure that a raw PCL file appeared in the server's queue
for file in [ # (showing that the right filters have been applied). Of
"${pkgs.groff.doc}/share/doc/*/examples/mom/penguin.pdf", # course, since there is no actual USB printer attached, the
"${pkgs.groff.doc}/share/doc/*/meref.ps", # file will stay in the queue forever.
"${pkgs.cups.out}/share/doc/cups/images/cups.png", server.wait_for_file("/var/spool/cups/d*-001")
"${pkgs.pcre.doc}/share/doc/pcre/pcre.txt", server.wait_until_succeeds(f"lpq -a | grep -q -E '{file_name}'")
]:
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 # Delete the job on the client. It should disappear on the
# (showing that the right filters have been applied). Of # server as well.
# course, since there is no actual USB printer attached, the client.succeed("lprm")
# file will stay in the queue forever. client.wait_until_succeeds("lpq -a | grep -q -E 'no entries'")
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 retry(lambda _: "no entries" in server.succeed("lpq -a"))
# 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.
# The queue is empty already, so this should be safe. server.execute("rm /var/spool/cups/*")
# Otherwise, pairs of "c*"-"d*-001" files might persist. '';
server.execute("rm /var/spool/cups/*") }
'';
}
)