update pytests: add huggingface and end-to-end functional tests

Author: deruyter92

pyproject.toml

@@ -70,6 +70,7 @@ line_length = 100
 [tool.pytest.ini_options]
 markers = [
     "functional: marks tests that require pretrained weights (deselect with '-m \"not functional\"')",
+    "network: marks tests that may need internet access on first run (deselect with '-m \"not network\"')",
 ]
 
 [tool.codespell]

tests/__init__.py

@@ -0,0 +1,97 @@
+"""
+Shared fixtures, markers, and skip-helpers for the ``fmpose3d_api`` test suite.
+
+Skip logic
+----------
+* **weights_ready(filename)** – ``True`` when the HuggingFace-cached file
+  already exists on disk *or* we can reach ``huggingface.co`` so that
+  ``hf_hub_download`` will succeed.
+* **has_internet** – evaluated once at collection time via a quick TCP probe.
+* **HF_HUB_OFFLINE** – if set to ``"1"`` in the environment the network
+  check is skipped entirely (consistent with how ``huggingface_hub``
+  itself behaves).
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# HuggingFace repo & filenames (must match fmpose3d.fmpose3d._HF_REPO_ID)
+# ---------------------------------------------------------------------------
+
+HF_REPO_ID: str = "deruyter92/fmpose_temp"
+
+HUMAN_WEIGHTS_FILENAME: str = "fmpose3d_humans.pth"
+ANIMAL_WEIGHTS_FILENAME: str = "fmpose3d_animals.pth"
+
+# ---------------------------------------------------------------------------
+# Connectivity helpers
+# ---------------------------------------------------------------------------
+
+
+def _has_internet(host: str = "huggingface.co", port: int = 443, timeout: float = 3) -> bool:
+    """Return ``True`` if *host* is reachable via TCP."""
+    if os.environ.get("HF_HUB_OFFLINE", "0") == "1":
+        return False
+    try:
+        socket.create_connection((host, port), timeout=timeout)
+        return True
+    except OSError:
+        return False
+
+
+def _weights_cached(filename: str) -> bool:
+    """Return ``True`` if *filename* already lives in the local HF cache."""
+    try:
+        from huggingface_hub import try_to_load_from_cache
+
+        result = try_to_load_from_cache(HF_REPO_ID, filename)
+        return isinstance(result, str)
+    except Exception:
+        return False
+
+
+def weights_ready(filename: str) -> bool:
+    """``True`` when we can obtain *filename* — either from cache or network."""
+    return _weights_cached(filename) or _has_internet()
+
+
+# Evaluate once at collection time.
+HAS_INTERNET: bool = _has_internet()
+HUMAN_WEIGHTS_READY: bool = weights_ready(HUMAN_WEIGHTS_FILENAME)
+ANIMAL_WEIGHTS_READY: bool = weights_ready(ANIMAL_WEIGHTS_FILENAME)
+
+try:
+    import deeplabcut  # noqa: F401
+
+    DLC_AVAILABLE: bool = True
+except ImportError:
+    DLC_AVAILABLE = False
+
+# ---------------------------------------------------------------------------
+# Reusable skip markers
+# ---------------------------------------------------------------------------
+
+requires_network = pytest.mark.skipif(
+    not HAS_INTERNET,
+    reason="No internet connection (cannot reach huggingface.co)",
+)
+
+requires_human_weights = pytest.mark.skipif(
+    not HUMAN_WEIGHTS_READY,
+    reason="Human weights not cached and no internet connection",
+)
+
+requires_animal_weights = pytest.mark.skipif(
+    not ANIMAL_WEIGHTS_READY,
+    reason="Animal weights not cached and no internet connection",
+)
+
+requires_dlc = pytest.mark.skipif(
+    not DLC_AVAILABLE,
+    reason="DeepLabCut is not installed",
+)

tests/fmpose3d_api/__init__.py

@@ -6,12 +6,7 @@
 by Ti Wang, Xiaohang Yu, and Mackenzie Weygandt Mathis
 Licensed under Apache 2.0
 
-Tests for fmpose3d/fmpose3d.py — the high-level inference API.
-
-Organised into:
-  * Unit tests for individual components (no pretrained weights needed).
-  * Functional integration tests for the full human and animal pipelines
-    (require pretrained weights on disk; marked ``@pytest.mark.functional``).
+Unit tests for fmpose3d/fmpose3d.py — the high-level inference API.
 """
 
 from __future__ import annotations
@@ -42,50 +37,6 @@
 )
 from fmpose3d.common.config import FMPose3DConfig, InferenceConfig
 
-# ---------------------------------------------------------------------------
-# Paths
-# ---------------------------------------------------------------------------
-
-PROJECT_ROOT = Path(__file__).resolve().parent.parent
-
-def _find_first(candidates: list[Path]) -> Path | None:
-    """Return the first path that exists, or ``None``."""
-    for p in candidates:
-        if p.exists():
-            return p
-    return None
-
-
-HUMAN_WEIGHTS = _find_first([
-    PROJECT_ROOT / "pre_trained_models" / "fmpose3d_h36m" / "FMpose3D_pretrained_weights.pth",
-])
-ANIMAL_WEIGHTS = _find_first([
-    PROJECT_ROOT / "animals" / "pre_trained_models" / "fmpose3d_animals" / "fmpose3d_animals_pretrained_weights.pth",
-    PROJECT_ROOT / "pre_trained_models" / "fmpose3d_animals" / "fmpose3d_animals_pretrained_weights.pth",
-])
-HUMAN_TEST_IMAGE = _find_first([
-    PROJECT_ROOT / "demo" / "images" / "running.png",
-])
-ANIMAL_TEST_IMAGE = _find_first([
-    PROJECT_ROOT / "animals" / "demo" / "images" / "dog.JPEG",
-    PROJECT_ROOT / "animals" / "demo" / "images" / "dog.jpeg",
-    PROJECT_ROOT / "animals" / "demo" / "images" / "dog.jpg",
-])
-
-# Evaluated at collection time — files either exist or they don't.
-_human_pipeline_available = HUMAN_WEIGHTS is not None and HUMAN_TEST_IMAGE is not None
-_animal_pipeline_available = ANIMAL_WEIGHTS is not None and ANIMAL_TEST_IMAGE is not None
-
-try:
-    import deeplabcut  # noqa: F401
-
-    _dlc_available = True
-except ImportError:
-    _dlc_available = False
-
-_animal_pipeline_available = _animal_pipeline_available and _dlc_available
-
-
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
@@ -747,174 +698,3 @@ def test_predict_maps_valid_bodyparts(self):
         assert scores.shape == (1, 1, 26)
         # target[24] ← source[0] → (0*3, 0*3+1) = (0.0, 1.0)
         np.testing.assert_allclose(kpts[0, 0, 24], fake_bp[0, 0, :2])
-
-
-# =========================================================================
-# Functional tests — human pipeline
-# =========================================================================
-
-
-@pytest.mark.functional
-@pytest.mark.skipif(
-    not _human_pipeline_available,
-    reason="Human pretrained weights or test image not found",
-)
-class TestHumanPipeline:
-    """End-to-end integration test for the human (17-joint H36M) pipeline."""
-
-    @pytest.fixture(scope="class")
-    def pipeline(self):
-        """Run the full pipeline once and cache all results."""
-        from PIL import Image
-
-        api = FMPose3DInference(
-            model_weights_path=str(HUMAN_WEIGHTS),
-            device="cpu",
-        )
-
-        # 2D estimation
-        result_2d = api.prepare_2d(source=str(HUMAN_TEST_IMAGE))
-
-        # Image size
-        img = Image.open(str(HUMAN_TEST_IMAGE))
-        w, h = img.size
-        image_size = (h, w)
-
-        # 3D lifting — two runs with the same seed for reproducibility check
-        result_3d_a = api.pose_3d(
-            result_2d.keypoints, image_size=image_size, seed=42,
-        )
-        result_3d_b = api.pose_3d(
-            result_2d.keypoints, image_size=image_size, seed=42,
-        )
-
-        return {
-            "result_2d": result_2d,
-            "image_size": image_size,
-            "result_3d_a": result_3d_a,
-            "result_3d_b": result_3d_b,
-        }
-
-    def test_2d_keypoints_shape(self, pipeline):
-        r2d = pipeline["result_2d"]
-        P, F, J, C = r2d.keypoints.shape
-        assert J == 17
-        assert C == 2
-        assert F >= 1
-
-    def test_2d_scores_shape(self, pipeline):
-        r2d = pipeline["result_2d"]
-        assert r2d.scores.ndim == 3
-        assert r2d.scores.shape[2] == 17
-
-    def test_2d_image_size(self, pipeline):
-        r2d = pipeline["result_2d"]
-        h, w = pipeline["image_size"]
-        assert r2d.image_size == (h, w)
-
-    def test_3d_poses_shape(self, pipeline):
-        r3d = pipeline["result_3d_a"]
-        F = pipeline["result_2d"].keypoints.shape[1]
-        assert r3d.poses_3d.shape == (F, 17, 3)
-        assert r3d.poses_3d_world.shape == (F, 17, 3)
-
-    def test_root_joint_zeroed(self, pipeline):
-        r3d = pipeline["result_3d_a"]
-        np.testing.assert_allclose(r3d.poses_3d[:, 0, :], 0.0, atol=1e-6)
-
-    def test_world_z_floor(self, pipeline):
-        r3d = pipeline["result_3d_a"]
-        assert np.min(r3d.poses_3d_world[:, :, 2]) >= -1e-6
-
-    def test_poses_finite(self, pipeline):
-        r3d = pipeline["result_3d_a"]
-        assert np.all(np.isfinite(r3d.poses_3d))
-        assert np.all(np.isfinite(r3d.poses_3d_world))
-
-    def test_reproducibility(self, pipeline):
-        r3d_a = pipeline["result_3d_a"]
-        r3d_b = pipeline["result_3d_b"]
-        np.testing.assert_allclose(r3d_a.poses_3d, r3d_b.poses_3d, atol=1e-6)
-        np.testing.assert_allclose(
-            r3d_a.poses_3d_world, r3d_b.poses_3d_world, atol=1e-6,
-        )
-
-
-# =========================================================================
-# Functional tests — animal pipeline
-# =========================================================================
-
-
-@pytest.mark.functional
-@pytest.mark.skipif(
-    not _animal_pipeline_available,
-    reason="Animal pretrained weights, test image, or DeepLabCut not available",
-)
-class TestAnimalPipeline:
-    """End-to-end integration test for the animal (26-joint Animal3D) pipeline."""
-
-    @pytest.fixture(scope="class")
-    def pipeline(self):
-        """Run the full pipeline once and cache all results."""
-        from PIL import Image
-
-        api = FMPose3DInference.for_animals(
-            model_weights_path=str(ANIMAL_WEIGHTS),
-            device="cpu",
-        )
-
-        result_2d = api.prepare_2d(source=str(ANIMAL_TEST_IMAGE))
-
-        img = Image.open(str(ANIMAL_TEST_IMAGE))
-        w, h = img.size
-        image_size = (h, w)
-
-        result_3d_a = api.pose_3d(
-            result_2d.keypoints, image_size=image_size, seed=42,
-        )
-        result_3d_b = api.pose_3d(
-            result_2d.keypoints, image_size=image_size, seed=42,
-        )
-
-        return {
-            "result_2d": result_2d,
-            "image_size": image_size,
-            "result_3d_a": result_3d_a,
-            "result_3d_b": result_3d_b,
-        }
-
-    def test_2d_keypoints_shape(self, pipeline):
-        r2d = pipeline["result_2d"]
-        _, F, J, C = r2d.keypoints.shape
-        assert J == 26
-        assert C == 2
-
-    def test_2d_scores_shape(self, pipeline):
-        r2d = pipeline["result_2d"]
-        assert r2d.scores.ndim == 3
-        assert r2d.scores.shape[2] == 26
-
-    def test_3d_poses_shape(self, pipeline):
-        r3d = pipeline["result_3d_a"]
-        F = pipeline["result_2d"].keypoints.shape[1]
-        assert r3d.poses_3d.shape == (F, 26, 3)
-        assert r3d.poses_3d_world.shape == (F, 26, 3)
-
-    def test_poses_finite(self, pipeline):
-        r3d = pipeline["result_3d_a"]
-        assert np.all(np.isfinite(r3d.poses_3d))
-        assert np.all(np.isfinite(r3d.poses_3d_world))
-
-    def test_poses_reasonable_magnitude(self, pipeline):
-        """Poses should not be excessively large (basic sanity)."""
-        r3d = pipeline["result_3d_a"]
-        assert np.max(np.abs(r3d.poses_3d)) < 1e4
-        assert np.max(np.abs(r3d.poses_3d_world)) < 1e4
-
-    def test_reproducibility(self, pipeline):
-        r3d_a = pipeline["result_3d_a"]
-        r3d_b = pipeline["result_3d_b"]
-        np.testing.assert_allclose(r3d_a.poses_3d, r3d_b.poses_3d, atol=1e-6)
-        np.testing.assert_allclose(
-            r3d_a.poses_3d_world, r3d_b.poses_3d_world, atol=1e-6,
-        )