test/test_fuzz.py

   1 """ Send junk to the collector """
   2
   3 from random import Random
   4 from socket import getaddrinfo, socket, AF_INET, SOCK_STREAM
   5 from time import sleep
   6 from typing import Any
   7 import unittest
   8 from .common import send_and_drain, TestWithServers
   9
  10 REPEAT: int = 1000000
  11
  12
  13 class Fuzz(TestWithServers):
  14     def setUp(self, *args: str, **kwargs: Any) -> None:
  15         super().setUp("collector")
  16         self.rnd = Random()
  17         for fam, typ, pro, cnm, skadr in getaddrinfo(
  18             "127.0.0.1",
  19             self.conf.getint("collector", "port"),
  20             family=AF_INET,
  21             type=SOCK_STREAM,
  22         ):
  23             break  # Just take the first element
  24         self.sock = socket(AF_INET, SOCK_STREAM)
  25         self.sock.connect(skadr)
  26
  27     def tearDown(self) -> None:
  28         sleep(1)  # give collector some time
  29         send_and_drain(self.sock, None)
  30         self.sock.close()
  31         sleep(1)  # Let the server close their side
  32         super().tearDown()
  33
  34     def test_stream(self) -> None:
  35         for _ in range(REPEAT):
  36             size = self.rnd.randint(1, 5000)
  37             buf = self.rnd.randbytes(size)
  38             send_and_drain(self.sock, buf)
  39
  40     def test_msgs(self) -> None:
  41         for _ in range(REPEAT):
  42             size = self.rnd.randint(0, 300)
  43             buf = b"xx" + self.rnd.randbytes(size) + b"\r\n"
  44             send_and_drain(self.sock, buf)
  45
  46
  47 if __name__ == "__main__":
  48     unittest.main()