Merge branch 'migration' of https://github.com/geetu040/openml-python into flow-migration-stacked

Author: Omswastik-11

openml/testing.py

@@ -11,10 +11,13 @@
 import unittest
 from pathlib import Path
 from typing import ClassVar
+from urllib.parse import urljoin
 
 import requests
 
 import openml
+from openml._api.clients import HTTPCache, HTTPClient
+from openml._api.config import RetryPolicy
 from openml.exceptions import OpenMLServerException
 from openml.tasks import TaskType
 
@@ -276,6 +279,91 @@ def _check_fold_timing_evaluations(  # noqa: PLR0913
                         assert evaluation <= max_val
 
 
+class TestAPIBase(unittest.TestCase):
+    server: str
+    base_url: str
+    api_key: str
+    timeout: int
+    retries: int
+    retry_policy: RetryPolicy
+    dir: str
+    ttl: int
+    cache: HTTPCache
+    http_client: HTTPClient
+
+    def setUp(self) -> None:
+        self.server = "https://test.openml.org/"
+        self.base_url = "api/v1/xml"
+        self.api_key = "normaluser"
+        self.timeout = 10
+        self.retries = 3
+        self.retry_policy = RetryPolicy.HUMAN
+        self.dir = "test_cache"
+        self.ttl = 60 * 60 * 24 * 7
+
+        self.cache = self._get_http_cache(
+            path=Path(self.dir),
+            ttl=self.ttl,
+        )
+        self.http_client = self._get_http_client(
+            server=self.server,
+            base_url=self.base_url,
+            api_key=self.api_key,
+            timeout=self.timeout,
+            retries=self.retries,
+            retry_policy=self.retry_policy,
+            cache=self.cache,
+        )
+
+        if self.cache.path.exists():
+            shutil.rmtree(self.cache.path)
+
+    def tearDown(self) -> None:
+        if self.cache.path.exists():
+            shutil.rmtree(self.cache.path)
+
+    def _get_http_cache(
+        self,
+        path: Path,
+        ttl: int,
+    ) -> HTTPCache:
+        return HTTPCache(
+            path=path,
+            ttl=ttl,
+        )
+
+    def _get_http_client(  # noqa: PLR0913
+        self,
+        server: str,
+        base_url: str,
+        api_key: str,
+        timeout: int,
+        retries: int,
+        retry_policy: RetryPolicy,
+        cache: HTTPCache | None = None,
+    ) -> HTTPClient:
+        return HTTPClient(
+            server=server,
+            base_url=base_url,
+            api_key=api_key,
+            timeout=timeout,
+            retries=retries,
+            retry_policy=retry_policy,
+            cache=cache,
+        )
+
+    def _get_url(
+        self,
+        server: str | None = None,
+        base_url: str | None = None,
+        path: str | None = None,
+    ) -> str:
+        server = server if server else self.server
+        base_url = base_url if base_url else self.base_url
+        path = path if path else ""
+        return urljoin(self.server, urljoin(self.base_url, path))
+
+
 def check_task_existence(
     task_type: TaskType,
     dataset_id: int,

tests/test_api/test_http.py

@@ -0,0 +1,140 @@
+from requests import Response, Request
+import time
+import xmltodict
+import pytest
+from openml.testing import TestAPIBase
+
+
+class TestHTTPClient(TestAPIBase):
+    def test_cache(self):
+        url = self._get_url(path="task/31")
+        params = {"param1": "value1", "param2": "value2"}
+
+        key = self.cache.get_key(url, params)
+
+        # validate key
+        self.assertEqual(
+            key,
+            "org/openml/test/api/v1/task/31/param1=value1&param2=value2",
+        )
+
+        # create fake response
+        req = Request("GET", url).prepare()
+        response = Response()
+        response.status_code = 200
+        response.url = url
+        response.reason = "OK"
+        response._content = b"<xml>test</xml>"
+        response.headers = {"Content-Type": "text/xml"}
+        response.encoding = "utf-8"
+        response.request = req
+        response.elapsed = type("Elapsed", (), {"total_seconds": lambda self: 0.1})()
+
+        # save to cache
+        self.cache.save(key, response)
+
+        # load from cache
+        cached_response = self.cache.load(key)
+
+        # validate loaded response
+        self.assertEqual(cached_response.status_code, 200)
+        self.assertEqual(cached_response.url, url)
+        self.assertEqual(cached_response.content, b"<xml>test</xml>")
+        self.assertEqual(
+            cached_response.headers["Content-Type"], "text/xml"
+        )
+
+    @pytest.mark.uses_test_server()
+    def test_get(self):
+        response = self.http_client.get("task/1")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"<oml:task", response.content)
+
+    @pytest.mark.uses_test_server()
+    def test_get_with_cache_creates_cache(self):
+        response = self.http_client.get("task/1", use_cache=True)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(self.cache.path.exists())
+
+        # verify cache directory structure exists
+        cache_key = self.cache.get_key(
+            self._get_url(path="task/1"),
+            {},
+        )
+        cache_path = self.cache._key_to_path(cache_key)
+
+        self.assertTrue((cache_path / "meta.json").exists())
+        self.assertTrue((cache_path / "headers.json").exists())
+        self.assertTrue((cache_path / "body.bin").exists())
+
+    @pytest.mark.uses_test_server()
+    def test_get_uses_cached_response(self):
+        # first request populates cache
+        response1 = self.http_client.get("task/1", use_cache=True)
+
+        # second request should load from cache
+        response2 = self.http_client.get("task/1", use_cache=True)
+
+        self.assertEqual(response1.content, response2.content)
+        self.assertEqual(response1.status_code, response2.status_code)
+
+    @pytest.mark.uses_test_server()
+    def test_get_cache_expires(self):
+        # force short TTL
+        self.cache.ttl = 1
+        path = "task/1"
+
+        url = self._get_url(path=path)
+        key = self.cache.get_key(url, {})
+        cache_path = self.cache._key_to_path(key) / "meta.json"
+
+        response1 = self.http_client.get(path, use_cache=True)
+        response1_cache_time_stamp = cache_path.stat().st_ctime
+
+        time.sleep(2)
+
+        response2 = self.http_client.get(path, use_cache=True)
+        response2_cache_time_stamp = cache_path.stat().st_ctime
+
+        # cache expired -> new request
+        self.assertNotEqual(response1_cache_time_stamp, response2_cache_time_stamp)
+        self.assertEqual(response2.status_code, 200)
+        self.assertEqual(response1.content, response2.content)
+
+    @pytest.mark.uses_test_server()
+    def test_post_and_delete(self):
+        task_xml = """
+        <oml:task_inputs xmlns:oml="http://openml.org/openml">
+            <oml:task_type_id>5</oml:task_type_id>
+            <oml:input name="source_data">193</oml:input>
+            <oml:input name="estimation_procedure">17</oml:input>
+        </oml:task_inputs>
+        """
+
+        task_id = None
+        try:
+            # POST the task
+            post_response = self.http_client.post(
+                "task",
+                files={"description": task_xml},
+            )
+            self.assertEqual(post_response.status_code, 200)
+            xml_resp = xmltodict.parse(post_response.content)
+            task_id = int(xml_resp["oml:upload_task"]["oml:id"])
+
+            # GET the task to verify it exists
+            get_response = self.http_client.get(f"task/{task_id}")
+            self.assertEqual(get_response.status_code, 200)
+
+        finally:
+            # DELETE the task if it was created
+            if task_id is not None:
+                try:
+                    del_response = self.http_client.delete(f"task/{task_id}")
+                    # optional: verify delete
+                    if del_response.status_code != 200:
+                        print(f"Warning: delete failed for task {task_id}")
+                except Exception as e:
+                    print(f"Warning: failed to delete task {task_id}: {e}")