tests.py 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # Licensed to Cloudera, Inc. under one
  4. # or more contributor license agreements. See the NOTICE file
  5. # distributed with this work for additional information
  6. # regarding copyright ownership. Cloudera, Inc. licenses this file
  7. # to you under the Apache License, Version 2.0 (the
  8. # "License"); you may not use this file except in compliance
  9. # with the License. You may obtain a copy of the License at
  10. #
  11. # http://www.apache.org/licenses/LICENSE-2.0
  12. #
  13. # Unless required by applicable law or agreed to in writing, software
  14. # distributed under the License is distributed on an "AS IS" BASIS,
  15. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16. # See the License for the specific language governing permissions and
  17. # limitations under the License.
  18. import logging
  19. import tempfile
  20. from desktop.log import get_audit_logger, AuditHandler
  21. from desktop.conf import AUDIT_EVENT_LOG_DIR, AUDIT_LOG_MAX_FILE_SIZE
  22. def test_one_audit():
  23. with tempfile.NamedTemporaryFile("w+t") as log_tmp:
  24. # KB
  25. reset = [
  26. AUDIT_EVENT_LOG_DIR.set_for_testing(log_tmp.name),
  27. AUDIT_LOG_MAX_FILE_SIZE.set_for_testing('25KB')
  28. ]
  29. audit_logger = get_audit_logger()
  30. audit_handler = audit_logger.handlers[0]
  31. assert 25 * 1024 ** 1 == audit_handler.maxBytes
  32. assert len(audit_logger.handlers) == 1, audit_logger.handlers
  33. assert isinstance(audit_handler, AuditHandler), audit_logger.handlers
  34. audit_logger = get_audit_logger()
  35. assert len(audit_logger.handlers) == 1, audit_logger.handlers # Not adding handler twice
  36. # Cleanup
  37. audit_logger.removeHandler(audit_handler)
  38. for r in reset:
  39. r()
  40. # MB
  41. reset = [
  42. AUDIT_EVENT_LOG_DIR.set_for_testing(log_tmp.name),
  43. AUDIT_LOG_MAX_FILE_SIZE.set_for_testing('25MB')
  44. ]
  45. audit_logger = get_audit_logger()
  46. audit_handler = audit_logger.handlers[0]
  47. assert 25 * 1024 ** 2 == audit_handler.maxBytes
  48. assert len(audit_logger.handlers) == 1, audit_logger.handlers
  49. assert isinstance(audit_handler, AuditHandler), audit_logger.handlers
  50. audit_logger = get_audit_logger()
  51. assert len(audit_logger.handlers) == 1, audit_logger.handlers # Not adding handler twice
  52. # Cleanup
  53. audit_logger.removeHandler(audit_handler)
  54. for r in reset:
  55. r()
  56. # GB
  57. reset = [
  58. AUDIT_EVENT_LOG_DIR.set_for_testing(log_tmp.name),
  59. AUDIT_LOG_MAX_FILE_SIZE.set_for_testing('25GB')
  60. ]
  61. audit_logger = get_audit_logger()
  62. audit_handler = audit_logger.handlers[0]
  63. assert 25 * 1024 ** 3 == audit_handler.maxBytes
  64. assert len(audit_logger.handlers) == 1, audit_logger.handlers
  65. assert isinstance(audit_handler, AuditHandler), audit_logger.handlers
  66. audit_logger = get_audit_logger()
  67. assert len(audit_logger.handlers) == 1, audit_logger.handlers # Not adding handler twice
  68. # Cleanup
  69. audit_logger.removeHandler(audit_handler)
  70. for r in reset:
  71. r()