nixos/tests/nats.nix

   1 let
   2
   3   port = 4222;
   4   username = "client";
   5   password = "password";
   6   topic = "foo.bar";
   7
   8 in import ./make-test-python.nix ({ pkgs, lib, ... }: {
   9   name = "nats";
  10   meta = with pkgs.lib; { maintainers = with maintainers; [ c0deaddict ]; };
  11
  12   nodes = let
  13     client = { pkgs, ... }: {
  14       environment.systemPackages = with pkgs; [ natscli ];
  15     };
  16   in {
  17     server = { pkgs, ... }: {
  18       networking.firewall.allowedTCPPorts = [ port ];
  19       services.nats = {
  20         inherit port;
  21         enable = true;
  22         settings = {
  23           authorization = {
  24             users = [{
  25               user = username;
  26               inherit password;
  27             }];
  28           };
  29         };
  30       };
  31     };
  32
  33     client1 = client;
  34     client2 = client;
  35   };
  36
  37   testScript = let file = "/tmp/msg";
  38   in ''
  39     def nats_cmd(*args):
  40         return (
  41             "nats "
  42             "--server=nats://server:${toString port} "
  43             "--user=${username} "
  44             "--password=${password} "
  45             "{}"
  46         ).format(" ".join(args))
  47
  48     def parallel(*fns):
  49         from threading import Thread
  50         threads = [ Thread(target=fn) for fn in fns ]
  51         for t in threads: t.start()
  52         for t in threads: t.join()
  53
  54     start_all()
  55     server.wait_for_unit("nats.service")
  56
  57     with subtest("pub sub"):
  58         parallel(
  59             lambda: client1.succeed(nats_cmd("sub", "--count", "1", "${topic}")),
  60             lambda: client2.succeed("sleep 2 && {}".format(nats_cmd("pub", "${topic}", "hello"))),
  61         )
  62   '';
  63 })