test/case: new test, basic container test

- New helper class for container testing
 - New helper class to urllib, Furl

Due to extremely weak Python-fu in the undersigned, this patch changes
the __init__.py file to add new helper classes for container tests.

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>

test/case/all.yaml

@@ -16,6 +16,9 @@
 - name:  ietf-routing
   suite: ietf_routing/all.yaml
 
+- name:  infix-containers
+  suite: infix_containers/all.yaml
+
 - name:  infix-dhcp
   suite: infix_dhcp/all.yaml
 

test/case/infix_containers/all.yaml

@@ -0,0 +1,3 @@
+---
+
+- case: container_basic.py

test/case/infix_containers/container_basic.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+#
+# Verify that a simple web server container can be configured to run
+# with host networking, on port 80.  Operation is verified using a
+# simple GET request for index.html and checking for a key phrase.
+#
+# The RPC actions: stop + start, and restart are also verified.
+#
+
+import infamy
+from   infamy.util import until
+
+def _verify(server):
+    url = infamy.Furl(f"http://[{server}]/index.html")
+    return url.check("It works")
+
+with infamy.Test() as test:
+    NAME  = "web"
+    IMAGE = "curios-httpd-edge.tar.gz"
+
+    with test.step("Set up topology and attach to target DUT"):
+        env = infamy.Env(infamy.std_topology("1x2"))
+        target = env.attach("target", "mgmt")
+        addr = target.address()
+
+    with test.step(f"Create {NAME} container from bundled OCI image"):
+        target.put_config_dict("infix-containers", {
+            "containers": {
+                "container": [
+                    {
+                        "name": f"{NAME}",
+                        "image": f"oci-archive:{IMAGE}",
+                        "network": {
+                            "host": True
+                        }
+                    }
+                ]
+            }
+        })
+
+    with test.step(f"Verify {NAME} continer has started"):
+        c = infamy.Container(target)
+        until(lambda: c.running(NAME), attempts=10)
+
+    with test.step(f"Verify {NAME} container responds"):
+        until(lambda: _verify(addr), attempts=10)
+
+    with test.step(f"Verify {NAME} container can be stopped and restarted"):
+        c = infamy.Container(target)
+        c.action(NAME, "stop")
+        until(lambda: not c.running(NAME), attempts=10)
+        c.action(NAME, "restart")
+
+    with test.step(f"Verify {NAME} container still responds"):
+        # Wait for it to restart and respond, or fail
+        until(lambda: _verify(addr), attempts=10)
+
+    test.succeed()

test/infamy/__init__.py

@@ -1,6 +1,8 @@
 import os
 
+from .container import Container
 from .env import Env
+from .furl import Furl
 from .netns import IsolatedMacVlan
 from .sniffer import Sniffer
 from .tap import Test

test/infamy/container.py

@@ -0,0 +1,44 @@
+"""Manage Infix containers"""
+
+class Container:
+    """Helper methods"""
+
+    def __init__(self, target):
+        self.system = target
+
+    def _find(self, name):
+        oper = self.system.get_data("/infix-containers:containers/container")
+        if not oper:
+            return None
+
+        for container in oper["containers"]["container"]:
+            if container["name"] == name:
+                return container
+        return None
+
+    def exists(self, name):
+        """Check if container {name} runs on target."""
+        container = self._find(name)
+        if not container:
+            return False
+        return True
+
+    def running(self, name):
+        """Check if container {name} exists and is running."""
+        container = self._find(name)
+        if container and container["running"]:
+            return True
+        return False
+
+    def action(self, name, act):
+        """Call NETCONF action 'type' on container 'name'"""
+        return self.system.call_action_dict("infix-containers", {
+            "containers": {
+                "container": [
+                    {
+                        "name": f"{name}",
+                        f"{act}": {}
+                    }
+                ]
+            }
+        })

test/infamy/furl.py

@@ -0,0 +1,24 @@
+"""Fugly URL fetcher"""
+
+import urllib.error
+import urllib.request
+
+class Furl:
+    """Furl wraps urllib in a way similar to curl"""
+    def __init__(self, url):
+        """Create new URL checker"""
+        self.url = urllib.parse.quote(url, safe='/:')
+
+    def check(self, needle):
+        """Connect to web server URL, fetch body and check for needle"""
+        try:
+            with urllib.request.urlopen(self.url) as response:
+                text = response.read().decode('utf-8')
+                #print(text)
+                return needle in text
+        except urllib.error.URLError as _:
+            return False
+
+    def nscheck(self, netns, needle):
+        """"Call check() from netns"""
+        return netns.call(lambda: self.check(needle))