Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/topics/utils.py: 53%

1from pathlib import Path 

2from typing import List 

3 

4from opal_common.paths import PathUtils 

5 

6POLICY_PREFIX = "policy:" 

7 

8 

9def policy_topics(paths: List[Path]) -> List[str]: 

10 """Prefixes a list of directories with the policy topic prefix.""" 

11 return ["{}{}".format(POLICY_PREFIX, str(path)) for path in paths] 

12 

13 

14def remove_prefix(topic: str, prefix: str = POLICY_PREFIX): 

15 """Removes the policy topic prefix to get the path (directory) encoded in 

16 the topic.""" 

17 if topic.startswith(prefix): 

18 return topic[len(prefix) :] 

19 return topic 

20 

21 

22def pubsub_topics_from_directories(dirs: List[str]) -> List[str]: 

23 """Converts a list of directories on the policy repository that the client 

24 wants to subscribe to into a list of topics. 

25 

26 this method also ensures the client only subscribes to non- 

27 intersecting directories by dedupping directories that are 

28 descendents of one another. 

29 """ 

30 policy_directories = PathUtils.non_intersecting_directories([Path(d) for d in dirs]) 

31 return policy_topics(policy_directories)