pyrox.dev
1import ./make-test-python.nix (
2 { lib, ... }:
3 {
4 name = "pufferpanel";
5 meta.maintainers = [ lib.maintainers.tie ];
6
7 nodes.machine =
8 { pkgs, ... }:
9 {
10 environment.systemPackages = [ pkgs.pufferpanel ];
11 services.pufferpanel = {
12 enable = true;
13 extraPackages = [ pkgs.netcat ];
14 environment = {
15 PUFFER_PANEL_REGISTRATIONENABLED = "false";
16 PUFFER_PANEL_SETTINGS_COMPANYNAME = "NixOS";
17 };
18 };
19 };
20
21 testScript = ''
22 import shlex
23 import json
24
25 curl = "curl --fail-with-body --silent"
26 baseURL = "http://localhost:8080"
27 adminName = "admin"
28 adminEmail = "admin@nixos.org"
29 adminPass = "admin"
30 adminCreds = json.dumps({
31 "email": adminEmail,
32 "password": adminPass,
33 })
34 stopCode = 9 # SIGKILL
35 serverPort = 1337
36 serverDefinition = json.dumps({
37 "name": "netcat",
38 "node": 0,
39 "users": [
40 adminName,
41 ],
42 "type": "netcat",
43 "run": {
44 "stopCode": stopCode,
45 "command": f"nc -l {serverPort}",
46 },
47 "environment": {
48 "type": "standard",
49 },
50 })
51
52 start_all()
53
54 machine.wait_for_unit("pufferpanel.service")
55 machine.wait_for_open_port(5657) # SFTP
56 machine.wait_for_open_port(8080) # HTTP
57
58 # Note that PufferPanel does not initialize database unless necessary.
59 # /api/config endpoint creates database file and triggers migrations.
60 # On success, we run a command to create administrator user that we use to
61 # interact with HTTP API.
62 resp = json.loads(machine.succeed(f"{curl} {baseURL}/api/config"))
63 assert resp["branding"]["name"] == "NixOS", "Invalid company name in configuration"
64 assert resp["registrationEnabled"] == False, "Expected registration to be disabled"
65
66 machine.succeed(f"pufferpanel --workDir /var/lib/pufferpanel user add --admin --name {adminName} --email {adminEmail} --password {adminPass}")
67
68 resp = json.loads(machine.succeed(f"{curl} -d '{adminCreds}' {baseURL}/auth/login"))
69 assert "servers.admin" in resp["scopes"], "User is not administrator"
70 token = resp["session"]
71 authHeader = shlex.quote(f"Authorization: Bearer {token}")
72
73 resp = json.loads(machine.succeed(f"{curl} -H {authHeader} -H 'Content-Type: application/json' -d '{serverDefinition}' {baseURL}/api/servers"))
74 serverID = resp["id"]
75 machine.succeed(f"{curl} -X POST -H {authHeader} {baseURL}/proxy/daemon/server/{serverID}/start")
76 machine.wait_for_open_port(serverPort)
77 '';
78 }
79)