|
69 | 69 | TEST_PASSWORD = "testing_psswd" |
70 | 70 |
|
71 | 71 |
|
| 72 | +def _install_compliance_catalog_test_cache() -> None: |
| 73 | + """Memoize the heavy SDK catalog loaders for the whole test session. |
| 74 | +
|
| 75 | + ``get_bulk_compliance_frameworks_universal`` re-reads and Pydantic-validates |
| 76 | + ~100 compliance JSONs (≈20 MB) and ``CheckMetadata.get_bulk`` re-reads ~1k |
| 77 | + check metadata files on *every* call. Production amortizes this through the |
| 78 | + per-process lazy caches (``PROWLER_CHECKS`` / ``PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE``) |
| 79 | + and ``warm_compliance_caches``, but the test suite parametrizes over every |
| 80 | + provider and deliberately resets the API-level caches, so the same catalogs |
| 81 | + were re-parsed dozens of times across the suite (≈3s/call locally, ≈19s under |
| 82 | + coverage in CI). |
| 83 | +
|
| 84 | + The catalog files are immutable during a run and callers treat the parsed |
| 85 | + objects as read-only, so caching the result per provider is safe. This is the |
| 86 | + test-only equivalent of an ``lru_cache`` on the SDK functions, without |
| 87 | + changing SDK behavior in production. |
| 88 | +
|
| 89 | + A second, lower-level cache memoizes ``load_compliance_framework_universal`` |
| 90 | + **per file path**. ``get_bulk_compliance_frameworks_universal`` parses *every* |
| 91 | + compliance JSON and only then filters by provider, so a per-provider cache |
| 92 | + still re-parses all ~100 files on the first load of each provider. The |
| 93 | + per-path cache makes the first provider parse the files once and every other |
| 94 | + provider/test reuse the already-parsed ``ComplianceFramework`` objects (only |
| 95 | + the cheap ``listdir`` + filtering re-runs). ``_load_jsons_from_dir`` calls |
| 96 | + ``load_compliance_framework_universal`` as a module global, so patching the |
| 97 | + attribute is picked up without touching the SDK. |
| 98 | +
|
| 99 | + Installed at conftest import time (before test modules are collected) so that |
| 100 | + even ``from ... import get_bulk_compliance_frameworks_universal`` bindings in |
| 101 | + the test modules resolve to the cached wrapper. |
| 102 | + """ |
| 103 | + import prowler.lib.check.compliance_models as compliance_models |
| 104 | + from prowler.lib.check.models import CheckMetadata |
| 105 | + |
| 106 | + original_bulk_frameworks = ( |
| 107 | + compliance_models.get_bulk_compliance_frameworks_universal |
| 108 | + ) |
| 109 | + original_get_bulk = CheckMetadata.get_bulk |
| 110 | + original_load = compliance_models.load_compliance_framework_universal |
| 111 | + |
| 112 | + def cached_bulk_frameworks(provider): |
| 113 | + if provider not in _COMPLIANCE_FRAMEWORK_CACHE: |
| 114 | + _COMPLIANCE_FRAMEWORK_CACHE[provider] = original_bulk_frameworks(provider) |
| 115 | + return _COMPLIANCE_FRAMEWORK_CACHE[provider] |
| 116 | + |
| 117 | + def cached_get_bulk(provider): |
| 118 | + if provider not in _COMPLIANCE_CHECKS_CACHE: |
| 119 | + _COMPLIANCE_CHECKS_CACHE[provider] = original_get_bulk(provider) |
| 120 | + return _COMPLIANCE_CHECKS_CACHE[provider] |
| 121 | + |
| 122 | + def cached_load(path): |
| 123 | + if path not in _COMPLIANCE_PATH_CACHE: |
| 124 | + _COMPLIANCE_PATH_CACHE[path] = original_load(path) |
| 125 | + return _COMPLIANCE_PATH_CACHE[path] |
| 126 | + |
| 127 | + compliance_models.get_bulk_compliance_frameworks_universal = cached_bulk_frameworks |
| 128 | + compliance_models.load_compliance_framework_universal = cached_load |
| 129 | + CheckMetadata.get_bulk = staticmethod(cached_get_bulk) |
| 130 | + |
| 131 | + # ``api.compliance`` does ``from ... import get_bulk_compliance_frameworks_universal`` |
| 132 | + # so it holds its own binding; patch it too in case it was imported first. |
| 133 | + import api.compliance as api_compliance |
| 134 | + |
| 135 | + api_compliance.get_bulk_compliance_frameworks_universal = cached_bulk_frameworks |
| 136 | + |
| 137 | + |
| 138 | +# Module-scoped so the ``_compliance_cache_guard`` fixture below can reset them. |
| 139 | +# Keeping them out of ``_install_compliance_catalog_test_cache``'s local scope is |
| 140 | +# what makes the caches resettable between tests; the wrappers above close over |
| 141 | +# these names, and the original loaders stay referenced so patched behaviour is |
| 142 | +# still honoured. |
| 143 | +_COMPLIANCE_FRAMEWORK_CACHE: dict[str, dict] = {} |
| 144 | +_COMPLIANCE_CHECKS_CACHE: dict[str, dict] = {} |
| 145 | +_COMPLIANCE_PATH_CACHE: dict[str, object] = {} |
| 146 | + |
| 147 | + |
| 148 | +_install_compliance_catalog_test_cache() |
| 149 | + |
| 150 | + |
| 151 | +@pytest.fixture(autouse=True) |
| 152 | +def _compliance_cache_guard(request): |
| 153 | + """Reset the compliance catalog caches after any test that used ``monkeypatch``. |
| 154 | +
|
| 155 | + The session-wide caches in ``_install_compliance_catalog_test_cache`` let the |
| 156 | + read-only, parametrized compliance tests parse the ~100 catalog JSONs once |
| 157 | + instead of dozens of times. A test that swaps a loader (or mutates a returned |
| 158 | + object) could otherwise leak that state into later tests through the shared |
| 159 | + dicts. Using ``monkeypatch`` as the opt-in signal keeps the full speed-up for |
| 160 | + catalog-reading tests while giving patching tests a clean slate afterwards; |
| 161 | + the next test simply repopulates the caches from disk. |
| 162 | + """ |
| 163 | + yield |
| 164 | + if "monkeypatch" in request.fixturenames: |
| 165 | + _COMPLIANCE_FRAMEWORK_CACHE.clear() |
| 166 | + _COMPLIANCE_CHECKS_CACHE.clear() |
| 167 | + _COMPLIANCE_PATH_CACHE.clear() |
| 168 | + import api.compliance as api_compliance |
| 169 | + |
| 170 | + api_compliance.AVAILABLE_COMPLIANCE_FRAMEWORKS.clear() |
| 171 | + |
| 172 | + |
72 | 173 | def today_after_n_days(n_days: int) -> str: |
73 | 174 | return datetime.strftime( |
74 | 175 | datetime.today().date() + timedelta(days=n_days), "%Y-%m-%d" |
|
0 commit comments