Coverage for node / src / stigmem_node / routes / facts / assertion.py: 100%

1"""POST /v1/facts route wrapper and plugin hook dispatch.""" 

2 

3from __future__ import annotations 

4 

5import uuid 

6from typing import Annotated 

7 

8from fastapi import Depends, Header, HTTPException, status 

9 

10from ...auth import Identity, resolve_identity 

11from ...models.facts import AssertRequest, FactRecord 

12from ...plugins import Deny, Failure, Success, TenantContext, get_registry 

13from ...tracing import start_span 

14from .._facts_assert import assert_fact_impl as _assert_fact_impl 

15from .common import router 

16 

17 

18@router.post("", response_model=FactRecord, status_code=status.HTTP_201_CREATED) 

19def assert_fact( 

20 req: AssertRequest, 

21 identity: Annotated[Identity, Depends(resolve_identity)], 

22 session_id: Annotated[str | None, Header(alias="Stigmem-Session")] = None, 

23) -> FactRecord: 

24 """Assert a fact into the fabric. 

25 

26 Covered by Spec-03-HTTP-API and Spec-01-Fact-Model. Normalizes 

27 entity/source URIs on ingest. 

28 """ 

29 request_id = str(uuid.uuid4()) 

30 tenant = TenantContext( 

31 tenant_id=identity.tenant_id, 

32 metadata={"tenant_context_source": "hook"}, 

33 ) 

34 registry = get_registry() 

35 with start_span( 

36 "stigmem.assert_fact", 

37 **{"stigmem.tenant": identity.tenant_id, "stigmem.principal": identity.entity_uri}, 

38 ) as _span: 

39 decision = registry.fire_voting( 

40 "pre_assert_authorize", 

41 req=req, 

42 identity=identity, 

43 tenant=tenant, 

44 request_id=request_id, 

45 ) 

46 if isinstance(decision, Deny): 

47 registry.fire_fire_and_forget( 

48 "post_assert_audit", 

49 fact=None, 

50 req=req, 

51 identity=identity, 

52 tenant=tenant, 

53 request_id=request_id, 

54 outcome=Failure(reason=decision.reason), 

55 ) 

56 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=decision.reason) 

57 

58 decision = registry.fire_voting( 

59 "pre_assert_validate", 

60 req=req, 

61 identity=identity, 

62 tenant=tenant, 

63 request_id=request_id, 

64 ) 

65 if isinstance(decision, Deny): 

66 registry.fire_fire_and_forget( 

67 "post_assert_audit", 

68 fact=None, 

69 req=req, 

70 identity=identity, 

71 tenant=tenant, 

72 request_id=request_id, 

73 outcome=Failure(reason=decision.reason), 

74 ) 

75 raise HTTPException( 

76 status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=decision.reason 

77 ) 

78 

79 req = registry.fire_filter_chain( 

80 "pre_assert_transform", 

81 req, 

82 identity=identity, 

83 tenant=tenant, 

84 request_id=request_id, 

85 ) 

86 try: 

87 fact = _assert_fact_impl( 

88 req, 

89 identity, 

90 _span, 

91 request_id=request_id, 

92 tenant=tenant, 

93 session_id=session_id, 

94 ) 

95 registry.fire_fire_and_forget( 

96 "post_assert_propagate", 

97 fact=fact, 

98 identity=identity, 

99 tenant=tenant, 

100 request_id=request_id, 

101 ) 

102 registry.fire_fire_and_forget( 

103 "post_assert_audit", 

104 fact=fact, 

105 req=req, 

106 identity=identity, 

107 tenant=tenant, 

108 request_id=request_id, 

109 outcome=Success(), 

110 ) 

111 return fact 

112 except Exception as exc: 

113 registry.fire_fire_and_forget( 

114 "post_assert_audit", 

115 fact=None, 

116 req=req, 

117 identity=identity, 

118 tenant=tenant, 

119 request_id=request_id, 

120 outcome=Failure(reason=str(exc), exception_type=type(exc).__name__), 

121 ) 

122 raise