test/asf/test_http_static.py

   1 from config import config
   2 from asfframework import VppAsfTestCase, VppTestRunner
   3 import unittest
   4 import subprocess
   5 import tempfile
   6 from vpp_qemu_utils import (
   7     create_host_interface,
   8     delete_host_interfaces,
   9     create_namespace,
  10     delete_namespace,
  11 )
  12
  13
  14 @unittest.skipIf(
  15     "http_static" in config.excluded_plugins, "Exclude HTTP Static Server plugin tests"
  16 )
  17 @unittest.skipIf(config.skip_netns_tests, "netns not available or disabled from cli")
  18 class TestHttpStaticVapi(VppAsfTestCase):
  19     """enable the http static server and send requests [VAPI]"""
  20
  21     @classmethod
  22     def setUpClass(cls):
  23         super(TestHttpStaticVapi, cls).setUpClass()
  24         # 2 temp files to improve coverage of http_cache.c
  25         cls.temp = tempfile.NamedTemporaryFile()
  26         cls.temp.write(b"Hello world")
  27
  28         cls.temp2 = tempfile.NamedTemporaryFile()
  29         cls.temp2.write(b"Hello world2")
  30
  31         try:
  32             create_namespace("HttpStatic")
  33         except Exception:
  34             cls.logger.warning("Unable to create a namespace, retrying.")
  35             delete_namespace("HttpStatic")
  36             create_namespace("HttpStatic")
  37
  38         create_host_interface("vppHost", "vppOut", "HttpStatic", "10.10.1.1/24")
  39
  40         cls.vapi.cli("create host-interface name vppOut")
  41         cls.vapi.cli("set int state host-vppOut up")
  42         cls.vapi.cli("set int ip address host-vppOut 10.10.1.2/24")
  43
  44     @classmethod
  45     def tearDownClass(cls):
  46         delete_namespace("HttpStatic")
  47         delete_host_interfaces("vppHost")
  48         cls.temp.close()
  49         cls.temp2.close()
  50         super(TestHttpStaticVapi, cls).tearDownClass()
  51
  52     def test_http_static_vapi(self):
  53         self.vapi.http_static_enable(
  54             www_root="/tmp",
  55             uri="tcp://0.0.0.0/80",
  56         )
  57         # move file pointer to the beginning
  58         self.temp.seek(0)
  59         process = subprocess.run(
  60             [
  61                 "ip",
  62                 "netns",
  63                 "exec",
  64                 "HttpStatic",
  65                 "curl",
  66                 f"10.10.1.2/{self.temp.name[5:]}",
  67             ],
  68             capture_output=True,
  69         )
  70         self.assertIn(b"Hello world", process.stdout)
  71
  72         self.temp2.seek(0)
  73         process = subprocess.run(
  74             [
  75                 "ip",
  76                 "netns",
  77                 "exec",
  78                 "HttpStatic",
  79                 "curl",
  80                 f"10.10.1.2/{self.temp2.name[5:]}",
  81             ],
  82             capture_output=True,
  83         )
  84         self.assertIn(b"Hello world2", process.stdout)
  85
  86
  87 @unittest.skipIf(
  88     "http_static" in config.excluded_plugins, "Exclude HTTP Static Server plugin tests"
  89 )
  90 @unittest.skipIf(config.skip_netns_tests, "netns not available or disabled from cli")
  91 class TestHttpStaticCli(VppAsfTestCase):
  92     """enable the static http server and send requests [CLI]"""
  93
  94     @classmethod
  95     def setUpClass(cls):
  96         super(TestHttpStaticCli, cls).setUpClass()
  97         # 2 temp files to improve coverage of http_cache.c
  98         cls.temp = tempfile.NamedTemporaryFile()
  99         cls.temp.write(b"Hello world")
 100
 101         cls.temp2 = tempfile.NamedTemporaryFile()
 102         cls.temp2.write(b"Hello world2")
 103
 104         try:
 105             create_namespace("HttpStatic2")
 106         except Exception:
 107             cls.logger.warning("Unable to create namespace, retrying.")
 108             delete_namespace("HttpStatic2")
 109             create_namespace("HttpStatic2")
 110
 111         create_host_interface("vppHost2", "vppOut2", "HttpStatic2", "10.10.1.1/24")
 112
 113         cls.vapi.cli("create host-interface name vppOut2")
 114         cls.vapi.cli("set int state host-vppOut2 up")
 115         cls.vapi.cli("set int ip address host-vppOut2 10.10.1.2/24")
 116
 117     @classmethod
 118     def tearDownClass(cls):
 119         delete_namespace("HttpStatic2")
 120         delete_host_interfaces("vppHost2")
 121         cls.temp.close()
 122         cls.temp2.close()
 123         super(TestHttpStaticCli, cls).tearDownClass()
 124
 125     def test_http_static_cli(self):
 126         self.vapi.cli(
 127             "http static server www-root /tmp uri tcp://0.0.0.0/80 cache-size 2m"
 128         )
 129         # move file pointer to the beginning
 130         self.temp.seek(0)
 131         process = subprocess.run(
 132             [
 133                 "ip",
 134                 "netns",
 135                 "exec",
 136                 "HttpStatic2",
 137                 "curl",
 138                 f"10.10.1.2/{self.temp.name[5:]}",
 139             ],
 140             capture_output=True,
 141         )
 142         self.assertIn(b"Hello world", process.stdout)
 143
 144         self.temp2.seek(0)
 145         process = subprocess.run(
 146             [
 147                 "ip",
 148                 "netns",
 149                 "exec",
 150                 "HttpStatic2",
 151                 "curl",
 152                 f"10.10.1.2/{self.temp2.name[5:]}",
 153             ],
 154             capture_output=True,
 155         )
 156         self.assertIn(b"Hello world2", process.stdout)
 157
 158         self.vapi.cli("show http static server cache")
 159         self.vapi.cli("clear http static cache")
 160         self.vapi.cli("show http static server sessions")
 161
 162
 163 if __name__ == "__main__":
 164     unittest.main(testRunner=VppTestRunner)