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