Coverage for netbox_data_import/inference_connection_test.py: 100.00%

1# SPDX-License-Identifier: Apache-2.0 

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

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

4 

5It resolves the configured reference, calls Chat Completions, and reports one typed category. Model 

6discovery is optional. The result never returns a secret value, provider response text, or a Vault 

7response body. 

8""" 

9 

10from contextlib import suppress 

11from dataclasses import dataclass 

12 

13from .inference_adapter import InferenceBackendError, InferenceRequest 

14from .inference_backend import NoActiveInferenceBackend, adapter_for_backend, resolve_backend_by_id 

15from .inference_credentials import CredentialFailure, credential_backend_for 

16from .inference_settings import VAULT_SETTING, InvalidInferenceConfiguration 

17from .inference_transport import WallClockDeadline, WallClockDeadlineExceeded 

18 

19CONNECTION_TEST_CATEGORIES = ( 

20 "ok", 

21 "credential_unavailable", 

22 "credential_denied", 

23 "invalid_credential_reference", 

24 "invalid_secret_material", 

25 "invalid_configuration", 

26 "transport_failure", 

27 "timeout", 

28 "authentication_failure", 

29 "rate_limit", 

30 "malformed_envelope", 

31 "invalid_response", 

32) 

33 

34 

35@dataclass(frozen=True) 

36class ConnectionTestResult: 

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

38 

39 category: str 

40 detail: str 

41 backend_key: str | None = None 

42 backend_source: str | None = None 

43 models: tuple[str, ...] = () 

44 

45 

46def _connection_request(response_mode: str) -> InferenceRequest: 

47 """Return the small request that verifies the configured model and Chat Completions route.""" 

48 return InferenceRequest( 

49 system_instruction="Return one small JSON object with the key ok and the value true.", 

50 user_payload_json='{"connection_test":true}', 

51 requested_response_mode=response_mode, 

52 ) 

53 

54 

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

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

57 try: 

58 backend = resolve_backend_by_id(pk) 

59 # Model discovery is optional, so the whole foreground test gets one normal-call budget. 

60 deadline = WallClockDeadline.after(backend.connect_timeout + backend.read_timeout) 

61 adapter = adapter_for_backend(backend, deadline=deadline) 

62 except NoActiveInferenceBackend: 

63 return ConnectionTestResult( 

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

65 ) 

66 except InvalidInferenceConfiguration as exc: 

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

68 except CredentialFailure as exc: 

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

70 

71 from .inference_backend import plugin_settings 

72 

73 try: 

74 with credential_backend_for( 

75 backend.credential_reference, 

76 plugin_settings().get(VAULT_SETTING, {}), 

77 deadline=deadline, 

78 ) as store: 

79 api_key = store.resolve(backend.credential_reference) 

80 except WallClockDeadlineExceeded: 

81 return ConnectionTestResult( 

82 "timeout", 

83 "The connection test exceeded its overall time limit.", 

84 backend_key, 

85 backend.source, 

86 ) 

87 except CredentialFailure as exc: 

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

89 

90 models: tuple[str, ...] = () 

91 try: 

92 # Model discovery is optional. The real completion call decides connection success. 

93 completion_error: InferenceBackendError | None 

94 try: 

95 completion = adapter.complete(_connection_request(backend.response_mode), api_key) 

96 except InferenceBackendError as completion_exc: 

97 completion_error = completion_exc 

98 else: 

99 completion_error = None 

100 with suppress(InferenceBackendError): 

101 models = adapter.discover_models(api_key) 

102 if completion_error is not None: 

103 detail = str(completion_error) 

104 if models and completion_error.diagnostic.status_code == 400: 

105 detail += " Select one of the available models below, save the backend, and run the test again." 

106 return ConnectionTestResult(completion_error.category, detail, backend_key, backend.source, models) 

107 if completion.is_refusal or not (completion.content_text or "").strip(): 

108 return ConnectionTestResult( 

109 "invalid_response", 

110 "The backend completed the test but did not return an answer.", 

111 backend_key, 

112 backend.source, 

113 models, 

114 ) 

115 if completion.diagnostic.redacted: 

116 return ConnectionTestResult( 

117 "invalid_response", 

118 "The backend echoed credential material. Its response was discarded.", 

119 backend_key, 

120 backend.source, 

121 models, 

122 ) 

123 finally: 

124 api_key = "" 

125 

126 if models: 

127 model_word = "model" if len(models) == 1 else "models" 

128 detail = ( 

129 f"The credential resolved, and the API completed a test request. " 

130 f"The endpoint offered {len(models)} {model_word}." 

131 ) 

132 else: 

133 detail = ( 

134 "The credential resolved, and the API completed a test request. " 

135 "The endpoint did not provide a compatible model list. Enter the model id manually." 

136 ) 

137 return ConnectionTestResult( 

138 "ok", 

139 detail, 

140 backend_key, 

141 backend.source, 

142 models, 

143 ) 

144 

145 

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