Coverage for astrocyte/errors.py: 100%

1"""Astrocyte exception hierarchy.""" 

2 

3from __future__ import annotations 

4 

5 

6class AstrocyteError(Exception): 

7 """Base exception for all Astrocyte errors.""" 

8 

9 

10class ConfigError(AstrocyteError): 

11 """Configuration is invalid or missing.""" 

12 

13 

14class CapabilityNotSupported(AstrocyteError): 

15 """The provider does not support the requested capability.""" 

16 

17 def __init__(self, provider: str, capability: str) -> None: 

18 self.provider = provider 

19 self.capability = capability 

20 super().__init__(f"Provider '{provider}' does not support '{capability}'") 

21 

22 

23class AccessDenied(AstrocyteError): 

24 """Principal lacks required permission on bank.""" 

25 

26 def __init__(self, principal: str, bank_id: str, permission: str) -> None: 

27 self.principal = principal 

28 self.bank_id = bank_id 

29 self.permission = permission 

30 super().__init__(f"Principal '{principal}' denied '{permission}' on bank '{bank_id}'") 

31 

32 

33class AuthorizationError(AstrocyteError): 

34 """Caller's credential could not be resolved to a valid identity. 

35 

36 Raised by the JWT identity classifier (and any upstream middleware that 

37 invokes it) when a presented token is invalid, expired, or decodes to a 

38 shape Astrocyte cannot classify as either a delegated user token or a 

39 service account credential. Also raised when no credential is presented 

40 and anonymous access is disabled. 

41 

42 Must always fail closed — a credential presented and rejected must 

43 never silently fall through to a default bank, because cross-user data 

44 leakage is the failure mode being prevented. 

45 """ 

46 

47 def __init__(self, reason: str) -> None: 

48 self.reason = reason 

49 super().__init__(reason) 

50 

51 

52class RateLimited(AstrocyteError): 

53 """Request exceeds rate limit.""" 

54 

55 def __init__(self, bank_id: str, operation: str, retry_after_seconds: float | None = None) -> None: 

56 self.bank_id = bank_id 

57 self.operation = operation 

58 self.retry_after_seconds = retry_after_seconds 

59 msg = f"Rate limited: {operation} on bank '{bank_id}'" 

60 if retry_after_seconds is not None: 

61 msg += f" (retry after {retry_after_seconds:.1f}s)" 

62 super().__init__(msg) 

63 

64 

65class ProviderUnavailable(AstrocyteError): 

66 """Provider is unreachable or circuit breaker is open.""" 

67 

68 def __init__(self, provider: str, reason: str | None = None) -> None: 

69 self.provider = provider 

70 self.reason = reason 

71 msg = f"Provider '{provider}' unavailable" 

72 if reason: 

73 msg += f": {reason}" 

74 super().__init__(msg) 

75 

76 

77class PiiRejected(AstrocyteError): 

78 """Content rejected due to PII detection policy.""" 

79 

80 def __init__(self, pii_types: list[str]) -> None: 

81 self.pii_types = pii_types 

82 super().__init__(f"Content rejected: PII detected ({', '.join(pii_types)})") 

83 

84 

85class CrossBorderViolation(AstrocyteError): 

86 """Operation would violate data residency policy.""" 

87 

88 def __init__(self, from_zone: str, to_zone: str) -> None: 

89 self.from_zone = from_zone 

90 self.to_zone = to_zone 

91 super().__init__(f"Cross-border violation: {from_zone} → {to_zone}") 

92 

93 

94class MipRoutingError(AstrocyteError): 

95 """MIP routing configuration or evaluation error.""" 

96 

97 

98class IngestError(AstrocyteError): 

99 """Inbound ingest (webhook, stream, …) rejected a payload or configuration.""" 

100 

101 

102class LegalHoldActive(AstrocyteError): 

103 """Operation blocked because bank is under legal hold.""" 

104 

105 def __init__(self, bank_id: str, hold_id: str) -> None: 

106 self.bank_id = bank_id 

107 self.hold_id = hold_id 

108 super().__init__(f"Bank '{bank_id}' is under legal hold '{hold_id}'")