Coverage for node / src / stigmem_node / models / federation.py: 96%

1"""Federation and conflict-resolution models.""" 

2 

3from __future__ import annotations 

4 

5from pydantic import BaseModel, Field, field_validator 

6 

7from .constants import VALID_SCOPES 

8from .facts import FactRecord, FactValue 

9 

10 

11class PeerRegisterRequest(BaseModel): 

12 node_url: str = Field(..., min_length=1) 

13 node_id: str = Field(..., min_length=1) 

14 federation_pubkey: str = Field(..., min_length=1) 

15 allowed_scopes: list[str] 

16 declaration_sig: str = Field(..., min_length=1) 

17 signed_at: str = Field(..., min_length=1) 

18 

19 @field_validator("allowed_scopes") 

20 @classmethod 

21 def check_scopes(cls, scopes: list[str]) -> list[str]: 

22 invalid = set(scopes) - VALID_SCOPES 

23 if invalid: 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 raise ValueError(f"invalid scopes: {invalid}") 

25 return scopes 

26 

27 

28class PeerRecord(BaseModel): 

29 peer_id: str 

30 node_id: str 

31 node_url: str 

32 status: str 

33 allowed_scopes: list[str] 

34 established_at: str | None 

35 

36 

37class PeerRegisterResponse(BaseModel): 

38 peer_id: str 

39 status: str 

40 verified_at: str | None 

41 

42 

43class PeerApprovalRequest(BaseModel): 

44 pubkey_fingerprint: str = Field(..., min_length=1) 

45 

46 

47class PeerApprovalResponse(BaseModel): 

48 peer_id: str 

49 node_id: str 

50 status: str 

51 approved_at: str 

52 

53 

54class FederationFactsResponse(BaseModel): 

55 facts: list[FactRecord] 

56 cursor: str | None 

57 has_more: bool 

58 

59 

60class AuditEntry(BaseModel): 

61 id: str 

62 peer_id: str 

63 event_type: str 

64 detail: str | None 

65 ts: str 

66 

67 

68class ConflictResolveRequest(BaseModel): 

69 """Request body for POST /v1/conflicts/:id/resolve (Spec-15-Fact-Semantics).""" 

70 

71 winning_fact_id: str | None = None 

72 resolution_note: str = "" 

73 new_value: FactValue | None = None