Coverage for integrations / social / realtime_acl.py: 0.0%

1""" 

2HevolveSocial — shared topic-shape parser used by both the 

3publish-side gate (realtime._authorize_topic_for_user_id) and the 

4subscribe-side gate (tenant_acl.authorize_subscribe). 

5 

6Plan reference: sunny-gliding-eich.md, Part E.13. 

7 

8Review M2 fix (post Pass-5 + WAMP ACL ship): the two gates were 

9each parsing topic strings independently — same split-by-`.`, same 

10`parts[2]` check, same special-cases for `conv` / `user`. The 

11risk surfaced as Pass-2 N-NEW-4 (substring vs segment match) on 

12the publish side; if a future fix landed only in publish, the 

13subscribe gate would silently diverge. This module is the single 

14source of truth. 

15 

16Topic shape canon: 

17 tenant.<tid>.<scope>.<id>[.<event>]+ 

18 

19 scope ∈ {conv, user, community, call, ...} 

20 

21Returns ParsedTopic (NamedTuple) with `tid`, `scope`, `id`, and 

22`event_suffix` so callers don't have to re-split. 

23""" 

24 

25from __future__ import annotations 

26 

27from collections import namedtuple 

28from typing import Optional 

29 

30 

31ParsedTopic = namedtuple( 

32 'ParsedTopic', 

33 ['is_tenant_scoped', 'tid', 'scope', 'id', 'event_suffix']) 

34 

35 

36def parse_topic(topic: Optional[str]) -> ParsedTopic: 

37 """Parse a `tenant.<tid>.<scope>.<id>[.<event>]+` topic into 

38 its named components. Non-tenant topics return 

39 ParsedTopic(is_tenant_scoped=False, ...) with all fields None. 

40 

41 Returns a stable shape regardless of topic validity — caller 

42 inspects `is_tenant_scoped` before trusting the other fields. 

43 Never raises. 

44 

45 Examples: 

46 'tenant.t1.conv.c1.message' → 

47 (True, 't1', 'conv', 'c1', 'message') 

48 'tenant.t1.user.alice' → 

49 (True, 't1', 'user', 'alice', '') 

50 'community.feed' → 

51 (False, None, None, None, None) 

52 '' → 

53 (False, None, None, None, None) 

54 """ 

55 if not topic or not isinstance(topic, str): 

56 return ParsedTopic(False, None, None, None, None) 

57 

58 if not topic.startswith('tenant.'): 

59 return ParsedTopic(False, None, None, None, None) 

60 

61 parts = topic.split('.') 

62 # Need at least: ['tenant', tid, scope, id] 

63 if len(parts) < 4: 

64 return ParsedTopic(False, None, None, None, None) 

65 

66 return ParsedTopic( 

67 is_tenant_scoped=True, 

68 tid=parts[1], 

69 scope=parts[2], 

70 id=parts[3], 

71 event_suffix='.'.join(parts[4:]), 

72 ) 

73 

74 

75__all__ = ['parse_topic', 'ParsedTopic']