Coverage for tests/test_authentication.py: 38%

1import unittest 

2 

3from lsst.daf.butler.tests.utils import mock_env 

4 

5try: 

6 from lsst.daf.butler.remote_butler import RemoteButler 

7 from lsst.daf.butler.remote_butler.authentication import cadc 

8 from lsst.daf.butler.remote_butler.authentication.rubin import ( 

9 _EXPLICIT_BUTLER_ACCESS_TOKEN_ENVIRONMENT_KEY, 

10 _RSP_JUPYTER_ACCESS_TOKEN_ENVIRONMENT_KEY, 

11 RubinAuthenticationProvider, 

12 _get_authentication_token_from_environment, 

13 ) 

14except ImportError: 

15 RemoteButler = None 

16 

17 

18@unittest.skipIf( 

19 RemoteButler is None, "RemoteButler could not be imported, optional dependencies may not be installed" 

20) 

21class TestButlerClientAuthentication(unittest.TestCase): 

22 """Test access-token logic""" 

23 

24 def test_explicit_butler_token(self): 

25 with mock_env( 

26 { 

27 _EXPLICIT_BUTLER_ACCESS_TOKEN_ENVIRONMENT_KEY: "token1", 

28 _RSP_JUPYTER_ACCESS_TOKEN_ENVIRONMENT_KEY: "not-this-token", 

29 } 

30 ): 

31 token = _get_authentication_token_from_environment("https://untrustedserver.com") 

32 self.assertEqual(token, "token1") 

33 

34 def test_jupyter_token_with_safe_server(self): 

35 with mock_env({_RSP_JUPYTER_ACCESS_TOKEN_ENVIRONMENT_KEY: "token2"}): 

36 token = _get_authentication_token_from_environment("https://data.LSST.cloud/butler") 

37 self.assertEqual(token, "token2") 

38 

39 def test_jupyter_token_with_unsafe_server(self): 

40 with mock_env({_RSP_JUPYTER_ACCESS_TOKEN_ENVIRONMENT_KEY: "token2"}): 

41 token = _get_authentication_token_from_environment("https://untrustedserver.com/butler") 

42 self.assertIsNone(token) 

43 

44 def test_missing_token(self): 

45 with mock_env({}): 

46 token = _get_authentication_token_from_environment("https://data.lsst.cloud/butler") 

47 self.assertIsNone(token) 

48 

49 def test_header_generation(self): 

50 auth = RubinAuthenticationProvider("tokendata") 

51 self.assertEqual(auth.get_server_headers(), {"Authorization": "Bearer tokendata"}) 

52 # At the Rubin Science Platform, the server sends pre-signed URLs that 

53 # do not require authentication. 

54 self.assertEqual(auth.get_datastore_headers(), {}) 

55 

56 def test_cadc_auth(self): 

57 auth = cadc.CadcAuthenticationProvider("tokendata") 

58 self.assertEqual(auth.get_server_headers(), {}) 

59 self.assertEqual(auth.get_datastore_headers(), {"Authorization": "Bearer tokendata"}) 

60 with mock_env({cadc._CADC_TOKEN_ENVIRONMENT_KEY: "tokendata"}): 

61 token = cadc._get_authentication_token_from_environment("https://www.canfar.net/butler") 

62 self.assertEqual(token, "tokendata")