peter/infix
1
0
Fork 0
mirror of https://github.com/kernelkit/infix.git synced 2026-07-22 09:13:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Furl into general curl

Browse Source 
Signed-off-by: Ejub Sabic <ejub1946@outlook.com>
This commit is contained in:
Ejub Sabic
2025-11-03 15:10:09 +01:00
parent 26c005f498
commit 51d5ae4540
11 changed files with 54 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/case/infix_containers/basic/test.py
+4 -3
View File
@@ -8,13 +8,14 @@ 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
from infamy.util import until, curl
def _verify(server):
# TODO: Should really use mDNS here....
url = infamy.Furl(f"http://[{server}]:91/index.html")
return url.check("It works")
url = f"http://[{server}]:91/index.html"
response = curl(url)
return response is not None and "It works" in response
with infamy.Test() as test:
test/case/infix_containers/bridge/test.py
+2 -3
View File
@@ -11,7 +11,7 @@ port accessed from the host.
"""
import base64
import infamy
from infamy.util import until, to_binary
from infamy.util import until, to_binary, curl
with infamy.Test() as test:
NAME = "web-docker0"
@@ -81,7 +81,6 @@ with infamy.Test() as test:
until(lambda: c.running(NAME), attempts=60)
_, hport = env.ltop.xlate("host", "data")
url = infamy.Furl(URL)
with infamy.IsolatedMacVlan(hport) as ns:
ns.addip(OURIP)
@@ -90,6 +89,6 @@ with infamy.Test() as test:
ns.must_reach(DUTIP)
with test.step("Verify container is reachable on http://10.0.0.2:8080"):
until(lambda: url.nscheck(ns, MESG), attempts=10)
until(lambda: MESG in ns.call(lambda: curl(URL)), attempts=10)
test.succeed()
test/case/infix_containers/environment/test.py
+3 -3
View File
@@ -10,7 +10,7 @@ and are available inside the running container.
3. Verify served content against environment variables
"""
import infamy
from infamy.util import until, to_binary
from infamy.util import until, to_binary, curl
with infamy.Test() as test:
@@ -23,7 +23,7 @@ with infamy.Test() as test:
NAME = "web-env"
DUTIP = "10.0.0.2"
OURIP = "10.0.0.1"
url = infamy.Furl(f"http://{DUTIP}:8080/cgi-bin/env.cgi")
URL =f"http://{DUTIP}:8080/cgi-bin/env.cgi"
with test.step("Set up topology and attach to target DUT"):
env = infamy.Env()
@@ -103,6 +103,6 @@ with infamy.Test() as test:
for var in ENV_VARS:
expected_strings.append(f'{var["key"]}={var["value"]}')
until(lambda: url.nscheck(ns, expected_strings))
until(lambda: all(string in ns.call(lambda: curl(URL)) for string in expected_strings))
test.succeed()
test/case/infix_containers/firewall_basic/test.py
+5 -6
View File
@@ -26,7 +26,7 @@ external port 8080 to ensure the web page, with a known key phrase, is
only reachable from the public interface `ext0`, on 192.168.0.1:8080.
"""
import infamy
from infamy.util import until, to_binary
from infamy.util import until, to_binary, curl
with infamy.Test() as test:
@@ -181,15 +181,14 @@ table ip nat {
until(lambda: c.running(WEBNM), attempts=60)
with infamy.IsolatedMacVlan(hport) as ns:
NEEDLE = "tiny web server from the curiOS docker"
MESG = "tiny web server from the curiOS docker"
ns.addip(OURIP)
with test.step("Verify connectivity, host can reach target:ext0"):
ns.must_reach(EXTIP)
with test.step("Verify 'web' is NOT reachable on http://container-host.local:91"):
url = infamy.Furl(BAD_URL)
until(lambda: not url.nscheck(ns, NEEDLE))
until(lambda: not ns.call(lambda: curl(BAD_URL)))
with test.step("Verify 'web' is reachable on http://container-host.local:8080"):
url = infamy.Furl(GOOD_URL)
until(lambda: url.nscheck(ns, NEEDLE))
until(lambda: MESG in ns.call(lambda: curl(GOOD_URL)))
test.succeed()
test/case/infix_containers/phys/test.py
+6 -6
View File
@@ -6,14 +6,15 @@ Verify connectivity with a simple web server container that's been
given a physical interface instead of an end of a VETH pair.
"""
import infamy
from infamy.util import until, to_binary
from infamy.util import until, to_binary, curl
with infamy.Test() as test:
NAME = "web-phys"
DUTIP = "10.0.0.2"
OURIP = "10.0.0.1"
MESG = "Kilroy was here"
BODY = f"<html><body><p>{MESG}</p></body></html>"
MESG1 = "It works"
MESG2 = "Kilroy was here"
BODY = f"<html><body><p>{MESG2}</p></body></html>"
URL = f"http://{DUTIP}:91/index.html"
with test.step("Set up topology and attach to target DUT"):
@@ -60,7 +61,6 @@ with infamy.Test() as test:
until(lambda: c.running(NAME), attempts=60)
_, hport = env.ltop.xlate("host", "data")
url = infamy.Furl(URL)
with infamy.IsolatedMacVlan(hport) as ns:
ns.addip(OURIP)
@@ -68,7 +68,7 @@ with infamy.Test() as test:
ns.must_reach(DUTIP)
with test.step("Verify container is reachable on http://10.0.0.2:91"):
until(lambda: url.nscheck(ns, "It works"), attempts=10)
until(lambda: MESG1 in ns.call(lambda: curl(URL)), attempts=10)
with test.step("Add a content mount, overriding index.html"):
# Verify modifying a running container takes, issue #930
@@ -88,6 +88,6 @@ with infamy.Test() as test:
})
with test.step("Verify server is restarted and returns new content"):
until(lambda: url.nscheck(ns, MESG), attempts=60)
until(lambda: MESG2 in ns.call(lambda: curl(URL)), attempts=60)
test.succeed()
test/case/infix_containers/veth/test.py
+3 -3
View File
@@ -18,13 +18,14 @@ regular bridge, a VETH pair connects the container to the bridge.
"""
import base64
import infamy
from infamy.util import until
from infamy.util import until, curl
with infamy.Test() as test:
NAME = "web-br0-veth"
DUTIP = "10.0.0.2"
OURIP = "10.0.0.1"
URL = f"http://{DUTIP}:91/index.html"
MESG = "It works"
with test.step("Set up topology and attach to target DUT"):
env = infamy.Env()
@@ -97,13 +98,12 @@ with infamy.Test() as test:
until(lambda: c.running(NAME), attempts=60)
_, hport = env.ltop.xlate("host", "data")
url = infamy.Furl(URL)
with infamy.IsolatedMacVlan(hport) as ns:
ns.addip(OURIP)
with test.step("Verify basic DUT connectivity, host:data can ping DUT 10.0.0.2"):
ns.must_reach(DUTIP)
with test.step("Verify container 'web-br0-veth' is reachable on http://10.0.0.2:91"):
until(lambda: url.nscheck(ns, "It works"), attempts=10)
until(lambda: MESG in ns.call(lambda: curl(URL)), attempts=10)
test.succeed()
test/case/infix_containers/volume/test.py
+4 -4
View File
@@ -7,7 +7,7 @@ content in the volume on restart.
"""
import infamy
from infamy.util import until
from infamy.util import until, curl
with infamy.Test() as test:
NAME = "web-volume"
@@ -19,7 +19,7 @@ with infamy.Test() as test:
target = env.attach("target", "mgmt")
tgtssh = env.attach("target", "mgmt", "ssh")
addr = target.get_mgmt_ip()
url = infamy.Furl(f"http://[{addr}]:{PORT}/index.html")
URL = f"http://[{addr}]:{PORT}/index.html"
if not target.has_model("infix-containers"):
test.skip()
@@ -53,7 +53,7 @@ with infamy.Test() as test:
tgtssh.runsh(cmd)
with test.step("Verify container volume content"):
until(lambda: url.check(MESG), attempts=10)
until(lambda: MESG in curl(URL), attempts=10)
with test.step("Upgrade container"):
out = tgtssh.runsh(f"sudo container upgrade {NAME}")
@@ -66,6 +66,6 @@ with infamy.Test() as test:
# print(f"Container {NAME} upgraded: {out.stdout}")
with test.step("Verify container volume content survived upgrade"):
until(lambda: url.check(MESG), attempts=60)
until(lambda: MESG in curl(URL), attempts=60)
test.succeed()
test/case/use_case/ospf_container/test.py
+7 -7
View File
@@ -58,7 +58,6 @@ image::internal-network.svg[Internal networks]
import infamy
import infamy.util as util
import infamy.route as route
from infamy.furl import Furl
BODY = "<html><body><p>Router responding</p></body></html>"
@@ -647,12 +646,13 @@ with infamy.Test() as test:
ns.addroute("0.0.0.0/0", "192.168.100.1")
# breakpoint()
with test.step("Verify ABR:data can access container A on R1 (10.1.1.101)"):
furl = Furl("http://10.1.1.101:8080")
util.until(lambda: furl.nscheck(ns, BODY))
URL = "http://10.1.1.101:8080"
util.until(lambda: BODY in ns.call(lambda: util.curl(URL)))
with test.step("Verify ABR:data can access container A on R2 (10.1.2.101)"):
furl = Furl("http://10.1.2.101:8080")
util.until(lambda: furl.nscheck(ns, BODY))
URL = "http://10.1.2.101:8080"
util.until(lambda: BODY in ns.call(lambda: util.curl(URL)))
with test.step("Verify ABR:data can access container A on R3 (10.1.3.101)"):
furl = Furl("http://10.1.3.101:8080")
util.until(lambda: furl.nscheck(ns, BODY))
URL = "http://10.1.3.101:8080"
util.until(lambda: BODY in ns.call(lambda: util.curl(URL)))
test.succeed()
test/infamy/__init__.py
-1
View File
@@ -5,7 +5,6 @@ from .env import Env
from .env import ArgumentParser
from .env import test_argument
from .firewall import Firewall
from .furl import Furl
from .netns import IsolatedMacVlan,IsolatedMacVlans
from .portscanner import PortScanner
from .sniffer import Sniffer
test/infamy/furl.py
-72
View File
@@ -1,72 +0,0 @@
"""Simple HTTP URL fetcher and content checker
This module provides the a lightweight wrapper around urllib for testing
HTTP endpoints. It's designed specifically for test scenarios where you
need to:
- Fetch HTTP content and verify it contains expected strings
- Perform checks from within network namespaces
- Validate multiple content patterns in a single request
- Handle network errors gracefully in test environments
Example usage:
# Single needle check
url = Furl("http://example.com/api")
if url.check("success"):
print("API returned success")
# Multiple needle check (all must match)
if url.check(["user: alice", "status: active", "role: admin"]):
print("All user details found")
# Check from network namespace
with IsolatedMacVlan("eth0") as ns:
if url.nscheck(ns, "expected content"):
print("Content verified from namespace")
"""
import urllib.error
import urllib.parse
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, needles, timeout=10):
"""Connect to web server URL, fetch body and check for needle(s)
Args:
needles: String or list of strings to search for in response
timeout: Request timeout in seconds
Returns:
bool: True if all needles found in response, False otherwise
"""
# Backwards compat, make needles a list for uniform processing
if isinstance(needles, str):
needles = [needles]
try:
with urllib.request.urlopen(self.url, timeout=timeout) as response:
text = response.read().decode('utf-8')
return all(needle in text for needle in needles)
except urllib.error.URLError:
return False
except ConnectionResetError:
return False
def nscheck(self, netns, needles):
""""Call check() from netns
Args:
netns: Network namespace to call from
needles: String or list of strings to search for in response
Returns:
bool: True if all needles found in response, False otherwise
"""
return netns.call(lambda: self.check(needles))
test/infamy/util.py
+20
View File
@@ -5,6 +5,7 @@ import threading
import infamy.neigh
from infamy import netconf
from infamy import restconf
import urllib.request
class ParallelFn(threading.Thread):
@@ -97,3 +98,22 @@ def warn(msg):
YELLOW = "\033[93m"
RST = "\033[0m"
print(f"{YELLOW}warn - {msg}{RST}")
def curl(url, timeout=10):
"""Fetch a URL and return its response body as a UTF-8 string.
Args:
url (str): The full URL to fetch.
timeout (int): Request timeout in seconds.
Returns:
str | None: Response body as text, or None if the request failed.
"""
url = urllib.parse.quote(url, safe='/:')
try:
with urllib.request.urlopen(url, timeout=timeout) as response:
return response.read().decode('utf-8', errors='replace')
except (urllib.error.URLError, ConnectionResetError, UnicodeEncodeError) as e:
print(f"[WARN] curl: failed to fetch {url}: {e}")
return ""