diff mbox series

[3/4] lib/tests/fetch2: add tests for curl method

Message ID 20260305-add_alt_fetch_method_curl-v1-3-0d0220e5fa59@se.com
State New
Headers show
Series fetch2: add alternative fetch method based on curl | expand

Commit Message

Pascal Eberhard via B4 Relay March 5, 2026, 3:32 p.m. UTC 
From: Pascal Eberhard <pascal.eberhard@se.com>

Duplicate wget fetch method test cases to test curl fetch method.

Signed-off-by: Pascal Eberhard <pascal.eberhard@se.com>
---
 lib/bb/tests/fetch.py | 68 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 68 insertions(+)
diff mbox series

Patch

diff --git a/lib/bb/tests/fetch.py b/lib/bb/tests/fetch.py
index 74eb73472..5c587d7b4 100644
--- a/lib/bb/tests/fetch.py
+++ b/lib/bb/tests/fetch.py
@@ -1550,6 +1550,7 @@  class FetchLatestVersionTest(FetcherTest):
 
                     url = "http://127.0.0.1:%s/%s" % (port, data.path)
                     ud = bb.fetch2.FetchData(url, self.d)
+                    self.assertEqual(ud.method.__class__.__name__, "Wget")
                     pupver = ud.method.latest_versionstring(ud, self.d)
                     verstring = pupver[0]
                     self.assertTrue(verstring, msg="Could not find upstream version for %s" % data.pn)
@@ -1558,6 +1559,35 @@  class FetchLatestVersionTest(FetcherTest):
         finally:
             server.stop()
 
+    def test_curl_latest_versionstring(self):
+        self.d.setVar("BB_FETCH_METHOD_HTTP", "curl")
+        testdata = os.path.dirname(os.path.abspath(__file__)) + "/fetch-testdata"
+        server = HTTPService(testdata, host="127.0.0.1")
+        server.start()
+        port = server.port
+        try:
+            for data, v in self.test_wget_uris.items():
+                with self.subTest(pn=data.pn):
+                    self.d.setVar("PN", data.pn)
+                    self.d.setVar("PV", data.pv)
+                    if data.check_uri:
+                        checkuri = "http://127.0.0.1:%s/%s" % (port, data.check_uri)
+                        self.d.setVar("UPSTREAM_CHECK_URI", checkuri)
+                    if data.check_regex:
+                        self.d.setVar("UPSTREAM_CHECK_REGEX", data.check_regex)
+
+                    url = "http://127.0.0.1:%s/%s" % (port, data.path)
+                    ud = bb.fetch2.FetchData(url, self.d)
+                    self.assertEqual(ud.method.__class__.__name__, "Curl")
+                    pupver = ud.method.latest_versionstring(ud, self.d)
+                    verstring = pupver[0]
+                    self.assertTrue(verstring, msg="Could not find upstream version for %s" % data.pn)
+                    r = bb.utils.vercmp_string(v, verstring)
+                    self.assertTrue(r == -1 or r == 0, msg="Package %s, version: %s <= %s" % (data.pn, v, verstring))
+        finally:
+            server.stop()
+        self.d.delVar("BB_FETCH_METHOD_HTTP")
+
     @skipIfNoNetwork()
     def test_crate_latest_versionstring(self):
         for k, v in self.test_crate_uris.items():
@@ -1590,6 +1620,7 @@  class FetchCheckStatusTest(FetcherTest):
             with self.subTest(url=u):
                 ud = fetch.ud[u]
                 m = ud.method
+                self.assertEqual(ud.method.__class__.__name__, "Wget")
                 ret = m.checkstatus(fetch, ud, self.d)
                 self.assertTrue(ret, msg="URI %s, can't check status" % (u))
 
@@ -1605,10 +1636,47 @@  class FetchCheckStatusTest(FetcherTest):
             with self.subTest(url=u):
                 ud = fetch.ud[u]
                 m = ud.method
+                self.assertEqual(ud.method.__class__.__name__, "Wget")
+                ret = m.checkstatus(fetch, ud, self.d)
+                self.assertTrue(ret, msg="URI %s, can't check status" % (u))
+
+        connection_cache.close_connections()
+
+    @skipIfNoNetwork()
+    def test_curl_checkstatus(self):
+        self.d.setVar("BB_FETCH_METHOD_HTTP", "curl")
+
+        fetch = bb.fetch2.Fetch(self.test_wget_uris, self.d)
+        for u in self.test_wget_uris:
+            with self.subTest(url=u):
+                ud = fetch.ud[u]
+                m = ud.method
+                self.assertEqual(ud.method.__class__.__name__, "Curl")
+                ret = m.checkstatus(fetch, ud, self.d)
+                self.assertTrue(ret, msg="URI %s, can't check status" % (u))
+
+        self.d.delVar("BB_FETCH_METHOD_HTTP")
+
+    @skipIfNoNetwork()
+    def test_curl_checkstatus_connection_cache(self):
+        from bb.fetch2 import FetchConnectionCache
+
+        self.d.setVar("BB_FETCH_METHOD_HTTP", "curl")
+
+        connection_cache = FetchConnectionCache()
+        fetch = bb.fetch2.Fetch(self.test_wget_uris, self.d,
+                    connection_cache = connection_cache)
+
+        for u in self.test_wget_uris:
+            with self.subTest(url=u):
+                ud = fetch.ud[u]
+                m = ud.method
+                self.assertEqual(ud.method.__class__.__name__, "Curl")
                 ret = m.checkstatus(fetch, ud, self.d)
                 self.assertTrue(ret, msg="URI %s, can't check status" % (u))
 
         connection_cache.close_connections()
+        self.d.delVar("BB_FETCH_METHOD_HTTP")
 
 
 class GitMakeShallowTest(FetcherTest):