"""Tests for ifcurl.service — POST /preview endpoint and caching.""" from __future__ import annotations import json from pathlib import Path from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from ifcurl.service import _t2_cache, _t3_cache, app client = TestClient(app) # Fake values used across tests FAKE_HEXSHA = "a" * 40 FAKE_PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64 MUTABLE_URL = "ifc://example.com/org/repo@heads/main?path=model.ifc" IMMUTABLE_URL = f"ifc://example.com/org/repo@{FAKE_HEXSHA}?path=model.ifc" # Mutable ref so tier-4 is never checked, keeping tier-3 tests hermetic SELECTOR_URL = "ifc://example.com/org/repo@heads/main?path=model.ifc&selector=IfcWall" @pytest.fixture(autouse=True) def clear_caches(): """Reset in-memory caches and tier-4 filesystem cache between tests.""" _t2_cache.clear() _t3_cache.clear() yield _t2_cache.clear() _t3_cache.clear() @pytest.fixture() def tmp_t4_cache(tmp_path, monkeypatch): """Redirect tier-4 PNG cache to a temp directory.""" monkeypatch.setattr("ifcurl.service.user_cache_dir", lambda *a, **kw: str(tmp_path)) return tmp_path def _mock_fetch(ifc_bytes: bytes): return lambda ifc_url, token=None: (FAKE_HEXSHA, ifc_bytes) # --------------------------------------------------------------------------- # Input validation # --------------------------------------------------------------------------- class TestValidation: def test_invalid_url_scheme_returns_400(self): r = client.post("/preview", json={"url": "https://example.com/model.ifc"}) assert r.status_code == 400 def test_missing_url_field_returns_422(self): r = client.post("/preview", json={}) assert r.status_code == 422 def test_url_without_path_param_returns_400(self): r = client.post("/preview", json={"url": "ifc://example.com/org/repo@heads/main"}) assert r.status_code == 400 assert "path" in r.json()["detail"] def test_malformed_url_returns_400(self): r = client.post("/preview", json={"url": "ifc://example.com/repo_no_ref?path=m.ifc"}) assert r.status_code == 400 # --------------------------------------------------------------------------- # Successful render # --------------------------------------------------------------------------- class TestPreviewRender: def test_returns_png_content_type(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): r = client.post("/preview", json={"url": MUTABLE_URL}) assert r.status_code == 200 assert r.headers["content-type"] == "image/png" assert r.content == FAKE_PNG def test_fetch_error_returns_404(self): with patch("ifcurl.service.fetch_ifc", side_effect=ValueError("Ref not found")): r = client.post("/preview", json={"url": MUTABLE_URL}) assert r.status_code == 404 def test_render_error_returns_422(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", side_effect=ValueError("No geometry")): r = client.post("/preview", json={"url": MUTABLE_URL}) assert r.status_code == 422 # --------------------------------------------------------------------------- # Tier 2: byte cache # --------------------------------------------------------------------------- class TestTier2Cache: def test_bytes_cached_after_first_request(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)) as mock_fetch, \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": MUTABLE_URL}) assert (FAKE_HEXSHA, "model.ifc") in _t2_cache def test_second_request_uses_cached_bytes(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() call_count = 0 def counting_fetch(ifc_url, token=None): nonlocal call_count call_count += 1 return FAKE_HEXSHA, ifc_bytes with patch("ifcurl.service.fetch_ifc", counting_fetch), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": MUTABLE_URL}) client.post("/preview", json={"url": MUTABLE_URL}) # fetch_ifc still called both times (for the commit hexsha), # but _t2_cache hit means ifc_bytes comes from cache on 2nd call assert call_count == 2 assert (FAKE_HEXSHA, "model.ifc") in _t2_cache # --------------------------------------------------------------------------- # Tier 3: GUID/selector cache # --------------------------------------------------------------------------- class TestTier3Cache: def test_guids_cached_after_selector_request(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": SELECTOR_URL}) assert (FAKE_HEXSHA, "model.ifc", "IfcWall") in _t3_cache def test_cached_guids_are_frozenset(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": SELECTOR_URL}) cached = _t3_cache[(FAKE_HEXSHA, "model.ifc", "IfcWall")] assert isinstance(cached, frozenset) def test_second_selector_request_skips_filter_elements(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() filter_call_count = 0 real_filter = __import__("ifcopenshell.util.selector", fromlist=["filter_elements"]).filter_elements def counting_filter(model, selector): nonlocal filter_call_count filter_call_count += 1 return real_filter(model, selector) with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG), \ patch("ifcopenshell.util.selector.filter_elements", counting_filter): client.post("/preview", json={"url": SELECTOR_URL}) client.post("/preview", json={"url": SELECTOR_URL}) assert filter_call_count == 1 # only called on first request # --------------------------------------------------------------------------- # Tier 4: PNG filesystem cache # --------------------------------------------------------------------------- class TestTier4Cache: def test_png_cached_for_immutable_ref(self, tmp_t4_cache, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() render_call_count = 0 def counting_render(*args, **kwargs): nonlocal render_call_count render_call_count += 1 return FAKE_PNG with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", counting_render): r1 = client.post("/preview", json={"url": IMMUTABLE_URL}) r2 = client.post("/preview", json={"url": IMMUTABLE_URL}) assert r1.content == FAKE_PNG assert r2.content == FAKE_PNG assert render_call_count == 1 # render only called once def test_png_not_cached_for_mutable_ref(self, tmp_t4_cache, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() render_call_count = 0 def counting_render(*args, **kwargs): nonlocal render_call_count render_call_count += 1 return FAKE_PNG with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", counting_render): client.post("/preview", json={"url": MUTABLE_URL}) client.post("/preview", json={"url": MUTABLE_URL}) assert render_call_count == 2 # no tier-4 cache for mutable ref def test_tier4_cache_file_exists_on_disk(self, tmp_t4_cache, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() with patch("ifcurl.service.fetch_ifc", _mock_fetch(ifc_bytes)), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": IMMUTABLE_URL}) png_files = list(tmp_t4_cache.rglob("*.png")) assert len(png_files) == 1 assert png_files[0].read_bytes() == FAKE_PNG # --------------------------------------------------------------------------- # Authentication # --------------------------------------------------------------------------- class TestAuthentication: def test_request_token_passed_to_fetch_ifc(self, model_with_geometry): ifc_bytes = model_with_geometry.to_string().encode() received_token = [] def capturing_fetch(ifc_url, token=None): received_token.append(token) return FAKE_HEXSHA, ifc_bytes with patch("ifcurl.service.fetch_ifc", capturing_fetch), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": MUTABLE_URL, "token": "mytoken123"}) assert received_token == ["mytoken123"] def test_config_token_used_when_no_request_token(self, model_with_geometry, monkeypatch): ifc_bytes = model_with_geometry.to_string().encode() received_token = [] def capturing_fetch(ifc_url, token=None): received_token.append(token) return FAKE_HEXSHA, ifc_bytes monkeypatch.setattr("ifcurl.service.get_token_for_host", lambda host: "config_token") with patch("ifcurl.service.fetch_ifc", capturing_fetch), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": MUTABLE_URL}) assert received_token == ["config_token"] def test_request_token_overrides_config_token(self, model_with_geometry, monkeypatch): ifc_bytes = model_with_geometry.to_string().encode() received_token = [] def capturing_fetch(ifc_url, token=None): received_token.append(token) return FAKE_HEXSHA, ifc_bytes monkeypatch.setattr("ifcurl.service.get_token_for_host", lambda host: "config_token") with patch("ifcurl.service.fetch_ifc", capturing_fetch), \ patch("ifcurl.service.render_mod.render", return_value=FAKE_PNG): client.post("/preview", json={"url": MUTABLE_URL, "token": "request_token"}) assert received_token == ["request_token"]