Coverage for /usr/local/lib/python3.10/site-packages/opal_server-0.0.0-py3.10.egg/opal_server/scopes/loader.py: 68%

1from opal_common.http_utils import redact_url 

2from opal_common.logger import logger 

3from opal_common.schemas.policy_source import ( 

4 GitPolicyScopeSource, 

5 NoAuthData, 

6 SSHAuthData, 

7) 

8from opal_common.schemas.scopes import Scope 

9from opal_server.config import ServerRole, opal_server_config 

10from opal_server.scopes.scope_repository import ScopeRepository 

11 

12DEFAULT_SCOPE_ID = "default" 

13 

14 

15async def load_scopes(repo: ScopeRepository): 

16 logger.info("Server is primary, loading default scope.") 

17 await _load_env_scope(repo) 

18 

19 

20async def _load_env_scope(repo: ScopeRepository): 

21 # backwards compatible opal scope 

22 if opal_server_config.POLICY_REPO_URL is not None: 22 ↛ exitline 22 didn't return from function '_load_env_scope' because the condition on line 22 was always true

23 logger.info( 

24 "Adding default scope from env: {url}", 

25 url=redact_url(opal_server_config.POLICY_REPO_URL), 

26 ) 

27 

28 auth = NoAuthData() 

29 

30 if opal_server_config.POLICY_REPO_SSH_KEY is not None: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

31 private_ssh_key = opal_server_config.POLICY_REPO_SSH_KEY 

32 private_ssh_key = private_ssh_key.replace("_", "\n") 

33 

34 if not private_ssh_key.endswith("\n"): 

35 private_ssh_key += "\n" 

36 

37 auth = SSHAuthData(username="git", private_key=private_ssh_key) 

38 

39 scope = Scope( 

40 scope_id=DEFAULT_SCOPE_ID, 

41 policy=GitPolicyScopeSource( 

42 source_type=opal_server_config.POLICY_SOURCE_TYPE.lower(), 

43 url=opal_server_config.POLICY_REPO_URL, 

44 manifest=opal_server_config.POLICY_REPO_MANIFEST_PATH, 

45 branch=opal_server_config.POLICY_REPO_MAIN_BRANCH, 

46 auth=auth, 

47 ), 

48 ) 

49 

50 await repo.put(scope)