Coverage for netbox_data_import/inference_connection_test.py: 97%

1# SPDX-License-Identifier: Apache-2.0 

2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com> 

3"""The Inference Backend connection test (specification 8.6). 

4 

5It runs on the worker queue, on the same secret boundary as a proposal job, so no web process ever 

6resolves a credential. It resolves the configured reference and reports one typed category. It 

7never returns a secret value and never a Vault response body. 

8""" 

9 

10from dataclasses import dataclass 

11 

12from .inference_backend import NoActiveInferenceBackend, resolve_backend_by_id 

13from .inference_credentials import CredentialFailure, credential_backend_for 

14from .inference_settings import VAULT_SETTING, InvalidInferenceConfiguration 

15 

16CONNECTION_TEST_CATEGORIES = ( 

17 "ok", 

18 "credential_unavailable", 

19 "credential_denied", 

20 "invalid_credential_reference", 

21 "invalid_secret_material", 

22 "invalid_configuration", 

23) 

24 

25 

26@dataclass(frozen=True) 

27class ConnectionTestResult: 

28 """One typed connection-test outcome: no secret, no Vault response body.""" 

29 

30 category: str 

31 detail: str 

32 backend_key: str | None = None 

33 backend_source: str | None = None 

34 

35 def as_dict(self) -> dict[str, str | None]: 

36 """Return the result as job data. It carries no credential reference.""" 

37 return { 

38 "category": self.category, 

39 "detail": self.detail, 

40 "backend_key": self.backend_key, 

41 "backend_source": self.backend_source, 

42 } 

43 

44 

45def run_connection_test(pk: int, backend_key: str) -> ConnectionTestResult: 

46 """Resolve the authorized row's credential and report with its queued display key.""" 

47 try: 

48 backend = resolve_backend_by_id(pk) 

49 except NoActiveInferenceBackend: 

50 return ConnectionTestResult( 

51 "invalid_configuration", f"Inference Backend '{backend_key}' no longer exists.", backend_key 

52 ) 

53 except InvalidInferenceConfiguration as exc: 

54 return ConnectionTestResult("invalid_configuration", str(exc), backend_key) 

55 except CredentialFailure as exc: 

56 return ConnectionTestResult(exc.category, str(exc), backend_key) 

57 

58 from .inference_backend import plugin_settings 

59 

60 try: 

61 with credential_backend_for(backend.credential_reference, plugin_settings().get(VAULT_SETTING, {})) as store: 

62 store.resolve(backend.credential_reference) 

63 except CredentialFailure as exc: 

64 return ConnectionTestResult(exc.category, str(exc), backend_key, backend.source) 

65 return ConnectionTestResult( 

66 "ok", 

67 "The credential resolved and holds usable material.", 

68 backend_key, 

69 backend.source, 

70 ) 

71 

72 

73__all__ = ("CONNECTION_TEST_CATEGORIES", "ConnectionTestResult", "run_connection_test")