|
13 | 13 | # You should have received a copy of the GNU Affero General Public License |
14 | 14 | # along with this program. If not, see <https://www.gnu.org/licenses/>. |
15 | 15 |
|
16 | | -"""Tests for the HOT-classification registry.""" |
| 16 | +"""Test settings-override classification, resolution, and PATCH preservation.""" |
17 | 17 |
|
18 | 18 | from string import Template |
19 | 19 | from typing import ClassVar |
20 | 20 |
|
21 | 21 | import pytest |
22 | | -from pydantic import BaseModel, SecretStr |
| 22 | +from pydantic import BaseModel, HttpUrl, SecretStr |
23 | 23 | from sqlmodel.ext.asyncio.session import AsyncSession |
24 | 24 |
|
25 | 25 | from app.core.alerts.config import AlertSettings |
|
43 | 43 | MaterializerPurpose, |
44 | 44 | nested_overridable_field_names, |
45 | 45 | override_rows_for_key, |
| 46 | + preserve_credential_urls_in_model_payload, |
46 | 47 | preserve_patch_credential_url_value, |
| 48 | + preserve_secrets_in_model_payload, |
47 | 49 | ReloadClassification, |
48 | 50 | resolve_nested_field_metadata, |
49 | 51 | SECRET_STR_MASK, |
50 | 52 | unwrap_secrets_for_storage, |
51 | 53 | ) |
| 54 | +from app.core.utils.fields import CredentialHttpUrl, redact_credential_url |
52 | 55 | from app.core.utils.pydantic import field_with_metadata |
53 | 56 | from app.inventory.config import InventorySettings |
54 | 57 | from app.sep.config import SEPSettings |
@@ -200,6 +203,60 @@ def test_preserve_patch_credential_url_value_for_materializer_payload() -> None: |
200 | 203 | assert preserved["endpoint"] == current["endpoint"] |
201 | 204 |
|
202 | 205 |
|
| 206 | +class _CredentialUrlModel(BaseModel): |
| 207 | + """Represent a model with an optional credential-bearing endpoint.""" |
| 208 | + |
| 209 | + endpoint: CredentialHttpUrl | None = HttpUrl.build( |
| 210 | + scheme="https", |
| 211 | + username="test-user", |
| 212 | + host="service.test", |
| 213 | + password="synthetic-test-value", |
| 214 | + ) |
| 215 | + |
| 216 | + |
| 217 | +def test_preserve_credential_urls_with_none_current_leaf() -> None: |
| 218 | + """Leave a masked URL unchanged when no live password exists to restore.""" |
| 219 | + current = _CredentialUrlModel(endpoint=None) |
| 220 | + incoming = _CredentialUrlModel().model_dump(mode="json") |
| 221 | + |
| 222 | + preserved = preserve_credential_urls_in_model_payload( |
| 223 | + _CredentialUrlModel, current, incoming |
| 224 | + ) |
| 225 | + |
| 226 | + assert preserved == incoming |
| 227 | + assert preserved is not incoming |
| 228 | + |
| 229 | + |
| 230 | +def test_preserve_patch_credential_url_value_recurses_into_nested_model() -> None: |
| 231 | + """Restore a nested URL password while retaining an outer-field PATCH.""" |
| 232 | + |
| 233 | + class _ConnectionGroup(BaseModel): |
| 234 | + connection: _CredentialUrlModel |
| 235 | + label: str |
| 236 | + |
| 237 | + class _ConnectionSettings(BaseModel): |
| 238 | + group: _ConnectionGroup |
| 239 | + |
| 240 | + current = _ConnectionGroup( |
| 241 | + connection=_CredentialUrlModel(), |
| 242 | + label="original", |
| 243 | + ) |
| 244 | + incoming = current.model_dump(mode="json") |
| 245 | + incoming["label"] = "updated" |
| 246 | + |
| 247 | + preserved = preserve_patch_credential_url_value( |
| 248 | + _ConnectionSettings.model_fields["group"], current, incoming |
| 249 | + ) |
| 250 | + |
| 251 | + assert preserved == { |
| 252 | + "connection": {"endpoint": str(current.connection.endpoint)}, |
| 253 | + "label": "updated", |
| 254 | + } |
| 255 | + assert incoming["connection"]["endpoint"] == redact_credential_url( |
| 256 | + str(current.connection.endpoint) |
| 257 | + ) |
| 258 | + |
| 259 | + |
203 | 260 | class _SecretLeafModel(BaseModel): |
204 | 261 | """Nested model with a scalar SecretStr leaf (PMM-shaped).""" |
205 | 262 |
|
@@ -229,6 +286,131 @@ class _DictSecretSettings(BaseModel): |
229 | 286 | ) |
230 | 287 |
|
231 | 288 |
|
| 289 | +def test_preserve_secrets_in_model_payload_with_secret_dict() -> None: |
| 290 | + """Restore masked dictionary values without replacing an explicit new secret.""" |
| 291 | + current = _DictSecretSettings() |
| 292 | + incoming = current.model_dump(mode="json") |
| 293 | + incoming["secrets"]["token"] = "replacement-token" |
| 294 | + |
| 295 | + preserved = preserve_secrets_in_model_payload( |
| 296 | + _DictSecretSettings, current, incoming |
| 297 | + ) |
| 298 | + |
| 299 | + assert preserved == { |
| 300 | + "secrets": { |
| 301 | + "api_key": current.secrets["api_key"].get_secret_value(), |
| 302 | + "token": "replacement-token", |
| 303 | + } |
| 304 | + } |
| 305 | + assert incoming["secrets"]["api_key"] == SECRET_STR_MASK |
| 306 | + |
| 307 | + |
| 308 | +def test_preserve_secrets_in_model_payload_with_secret_list() -> None: |
| 309 | + """Restore masked list elements by position and keep explicitly replaced secrets.""" |
| 310 | + |
| 311 | + class _ListOfSecretsSettings(BaseModel): |
| 312 | + tokens: list[SecretStr] |
| 313 | + |
| 314 | + current = _ListOfSecretsSettings( |
| 315 | + tokens=[SecretStr("keep-first"), SecretStr("replace-second")] |
| 316 | + ) |
| 317 | + incoming = current.model_dump(mode="json") |
| 318 | + incoming["tokens"][1] = "new-second" |
| 319 | + |
| 320 | + preserved = preserve_secrets_in_model_payload( |
| 321 | + _ListOfSecretsSettings, current, incoming |
| 322 | + ) |
| 323 | + |
| 324 | + assert preserved == {"tokens": ["keep-first", "new-second"]} |
| 325 | + assert incoming["tokens"][0] == SECRET_STR_MASK |
| 326 | + |
| 327 | + |
| 328 | +@pytest.mark.parametrize( |
| 329 | + "order", [(2, 0, 1), (0, 2, 1)], ids=["last-first", "first-then-last"] |
| 330 | +) |
| 331 | +def test_preserve_secrets_in_model_payload_matches_reordered_models( |
| 332 | + order: tuple[int, ...], |
| 333 | +) -> None: |
| 334 | + """Pair homogeneous model items by their public values rather than PATCH position.""" |
| 335 | + |
| 336 | + class _ListSecretSettings(BaseModel): |
| 337 | + items: list[_SecretLeafModel] |
| 338 | + |
| 339 | + current = _ListSecretSettings( |
| 340 | + items=[ |
| 341 | + _SecretLeafModel(api_key=SecretStr("secret-a"), label="a"), |
| 342 | + _SecretLeafModel(api_key=SecretStr("secret-b"), label="b"), |
| 343 | + _SecretLeafModel(api_key=SecretStr("secret-c"), label="c"), |
| 344 | + ] |
| 345 | + ) |
| 346 | + incoming = { |
| 347 | + "items": [current.items[index].model_dump(mode="json") for index in order] |
| 348 | + } |
| 349 | + |
| 350 | + preserved = preserve_secrets_in_model_payload( |
| 351 | + _ListSecretSettings, current, incoming |
| 352 | + ) |
| 353 | + |
| 354 | + assert preserved == { |
| 355 | + "items": [ |
| 356 | + { |
| 357 | + "api_key": current.items[index].api_key.get_secret_value(), |
| 358 | + "label": current.items[index].label, |
| 359 | + } |
| 360 | + for index in order |
| 361 | + ] |
| 362 | + } |
| 363 | + |
| 364 | + |
| 365 | +@pytest.mark.parametrize( |
| 366 | + ("order", "expected_secrets"), |
| 367 | + [ |
| 368 | + ((1,), ["secret-b", "secret-a", "secret-c"]), |
| 369 | + ((2, 0), ["secret-c", "secret-a", "secret-b"]), |
| 370 | + ], |
| 371 | + ids=["field-overlap-then-position", "sole-unused-item"], |
| 372 | +) |
| 373 | +def test_preserve_secrets_in_model_payload_with_unrecognized_discriminator( |
| 374 | + order: tuple[int, ...], expected_secrets: list[str] |
| 375 | +) -> None: |
| 376 | + """Restore masks through fallback matching without reusing a stored item.""" |
| 377 | + |
| 378 | + class _ProviderSecretLeaf(_SecretLeafModel): |
| 379 | + provider: str = "external" |
| 380 | + |
| 381 | + class _ListSecretSettings(BaseModel): |
| 382 | + items: list[_ProviderSecretLeaf] |
| 383 | + |
| 384 | + current = _ListSecretSettings( |
| 385 | + items=[ |
| 386 | + _ProviderSecretLeaf(api_key=SecretStr("secret-a"), label="a"), |
| 387 | + _ProviderSecretLeaf(api_key=SecretStr("secret-b"), label="b"), |
| 388 | + _ProviderSecretLeaf(api_key=SecretStr("secret-c"), label="c"), |
| 389 | + ] |
| 390 | + ) |
| 391 | + incoming_items = [ |
| 392 | + current.items[index].model_dump(mode="json", exclude={"provider"}) |
| 393 | + for index in order |
| 394 | + ] |
| 395 | + # This discriminator matches no class name; omit labels to force fallback pairing. |
| 396 | + fallback = current.items[0].model_dump(mode="json", exclude={"label"}) |
| 397 | + incoming_items.extend(fallback.copy() for _ in range(4 - len(incoming_items))) |
| 398 | + incoming = {"items": incoming_items} |
| 399 | + |
| 400 | + preserved = preserve_secrets_in_model_payload( |
| 401 | + _ListSecretSettings, current, incoming |
| 402 | + ) |
| 403 | + |
| 404 | + assert preserved == { |
| 405 | + "items": [ |
| 406 | + {**item, "api_key": secret} |
| 407 | + for item, secret in zip( |
| 408 | + incoming_items, [*expected_secrets, SECRET_STR_MASK], strict=True |
| 409 | + ) |
| 410 | + ] |
| 411 | + } |
| 412 | + |
| 413 | + |
232 | 414 | def test_preserve_patch_secret_value_for_top_level_field() -> None: |
233 | 415 | """Assert a masked top-level SecretStr PATCH restores the stored secret.""" |
234 | 416 | field = _TopLevelSecretSettings.model_fields["TOKEN"] |
|
0 commit comments