pyrox.dev
1import ./make-test-python.nix (
2 { pkgs, ... }: {
3 name = "nagios";
4 meta = with pkgs.lib.maintainers; {
5 maintainers = [ symphorien ];
6 };
7
8 nodes.machine = { lib, ... }: let
9 writer = pkgs.writeShellScript "write" ''
10 set -x
11 echo "$@" >> /tmp/notifications
12 '';
13 in
14 {
15 # tested service
16 services.sshd.enable = true;
17 # nagios
18 services.nagios = {
19 enable = true;
20 # make state transitions faster
21 extraConfig.interval_length = "5";
22 objectDefs =
23 (map (x: "${pkgs.nagios}/etc/objects/${x}.cfg") [ "templates" "timeperiods" "commands" ]) ++ [
24 (
25 pkgs.writeText "objects.cfg" ''
26 # notifications are written to /tmp/notifications
27 define command {
28 command_name notify-host-by-file
29 command_line ${writer} "$HOSTNAME is $HOSTSTATE$"
30 }
31 define command {
32 command_name notify-service-by-file
33 command_line ${writer} "$SERVICEDESC$ is $SERVICESTATE$"
34 }
35
36 # nagios boilerplate
37 define contact {
38 contact_name alice
39 alias alice
40 host_notifications_enabled 1
41 service_notifications_enabled 1
42 service_notification_period 24x7
43 host_notification_period 24x7
44 service_notification_options w,u,c,r,f,s
45 host_notification_options d,u,r,f,s
46 service_notification_commands notify-service-by-file
47 host_notification_commands notify-host-by-file
48 email foo@example.com
49 }
50 define contactgroup {
51 contactgroup_name admins
52 alias Admins
53 members alice
54 }
55 define hostgroup{
56 hostgroup_name allhosts
57 alias All hosts
58 }
59
60 # monitored objects
61 define host {
62 use generic-host
63 host_name localhost
64 alias localhost
65 address localhost
66 hostgroups allhosts
67 contact_groups admins
68 # make state transitions faster.
69 max_check_attempts 2
70 check_interval 1
71 retry_interval 1
72 }
73 define service {
74 use generic-service
75 host_name localhost
76 service_description ssh
77 check_command check_ssh
78 # make state transitions faster.
79 max_check_attempts 2
80 check_interval 1
81 retry_interval 1
82 }
83 ''
84 )
85 ];
86 };
87 };
88
89 testScript = { ... }: ''
90 with subtest("ensure sshd starts"):
91 machine.wait_for_unit("sshd.service")
92
93
94 with subtest("ensure nagios starts"):
95 machine.wait_for_file("/var/log/nagios/current")
96
97
98 def assert_notify(text):
99 machine.wait_for_file("/tmp/notifications")
100 real = machine.succeed("cat /tmp/notifications").strip()
101 print(f"got {real!r}, expected {text!r}")
102 assert text == real
103
104
105 with subtest("ensure we get a notification when sshd is down"):
106 machine.succeed("systemctl stop sshd")
107 assert_notify("ssh is CRITICAL")
108
109
110 with subtest("ensure tests can succeed"):
111 machine.succeed("systemctl start sshd")
112 machine.succeed("rm /tmp/notifications")
113 assert_notify("ssh is OK")
114 '';
115 }
116)