Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/tests/integration/security/test_kerberos.py: 0%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17 

18from __future__ import annotations 

19 

20import logging 

21import os 

22from contextlib import nullcontext 

23from unittest import mock 

24 

25import pytest 

26 

27from airflow.security import kerberos 

28from airflow.security.kerberos import renew_from_kt 

29from tests.test_utils.config import conf_vars 

30 

31 

32@pytest.mark.integration("kerberos") 

33class TestKerberosIntegration: 

34 @classmethod 

35 def setup_class(cls): 

36 assert "KRB5_KTNAME" in os.environ, "Missing KRB5_KTNAME environment variable" 

37 cls.keytab = os.environ["KRB5_KTNAME"] 

38 

39 @pytest.mark.parametrize( 

40 "kerberos_config", 

41 [ 

42 pytest.param({}, id="default-config"), 

43 pytest.param({("kerberos", "include_ip"): "True"}, id="explicit-include-ip"), 

44 pytest.param({("kerberos", "include_ip"): "False"}, id="explicit-not-include-ip"), 

45 pytest.param({("kerberos", "forwardable"): "True"}, id="explicit-forwardable"), 

46 pytest.param({("kerberos", "forwardable"): "False"}, id="explicit-not-forwardable"), 

47 ], 

48 ) 

49 def test_renew_from_kt(self, kerberos_config): 

50 """We expect return 0 (exit code) and successful run.""" 

51 with conf_vars(kerberos_config): 

52 assert renew_from_kt(principal=None, keytab=self.keytab) == 0 

53 

54 @pytest.mark.parametrize( 

55 "exit_on_fail, expected_context", 

56 [ 

57 pytest.param(True, pytest.raises(SystemExit), id="exit-on-fail"), 

58 pytest.param(False, nullcontext(), id="return-code-of-fail"), 

59 ], 

60 ) 

61 def test_args_from_cli(self, exit_on_fail, expected_context, caplog): 

62 """Test exit code if keytab not exist.""" 

63 keytab = "/not/exists/keytab" 

64 result = None 

65 

66 with mock.patch.dict(os.environ, KRB5_KTNAME=keytab), conf_vars({("kerberos", "keytab"): keytab}): 

67 with expected_context as ctx: 

68 with caplog.at_level(logging.ERROR, logger=kerberos.log.name): 

69 caplog.clear() 

70 result = renew_from_kt(principal=None, keytab=keytab, exit_on_fail=exit_on_fail) 

71 

72 # If `exit_on_fail` set to True than exit code in exception, otherwise in function return 

73 exit_code = ctx.value.code if exit_on_fail else result 

74 assert exit_code == 1 

75 assert caplog.record_tuples