content: #
  # Copyright 2015 Hewlett-Packard Development Company, L.P.
  #
  # SPDX-License-Identifier: Apache-2.0
  import os
  from unittest import mock
  
  import fixtures
  import testtools
  
  from bandit.core import config
  from bandit.core import constants
  from bandit.core import issue
  from bandit.core import manager
  
  
  class ManagerTests(testtools.TestCase):
      def _get_issue_instance(
          self,
          sev=constants.MEDIUM,
          cwe=issue.Cwe.MULTIPLE_BINDS,
          conf=constants.MEDIUM,
      ):
          new_issue = issue.Issue(sev, cwe, conf, "Test issue")
          new_issue.fname = "code.py"
          new_issue.test = "bandit_plugin"
          new_issue.lineno = 1
          return new_issue
  
      def setUp(self):
          super().setUp()
          self.profile = {}
          self.profile["include"] = {
              "any_other_function_with_shell_equals_true",
              "assert_used",
          }
  
          self.config = config.BanditConfig()
          self.manager = manager.BanditManager(
              config=self.config, agg_type="file", debug=False, verbose=False
          )
  
      def test_create_manager(self):
          # make sure we can create a manager
          self.assertEqual(False, self.manager.debug)
          self.assertEqual(False, self.manager.verbose)
          self.assertEqual("file", self.manager.agg_type)
  
      def test_create_manager_with_profile(self):
          # make sure we can create a manager
          m = manager.BanditManager(
              config=self.config,
              agg_type="file",
              debug=False,
              verbose=False,
              profile=self.profile,
          )
  
          self.assertEqual(False, m.debug)
          self.assertEqual(False, m.verbose)
          self.assertEqual("file", m.agg_type)
  
      def test_matches_globlist(self):
          self.assertTrue(manager._matches_glob_list("test", ["*tes*"]))
          self.assertFalse(manager._matches_glob_list("test", ["*fes*"]))
  
      def test_is_file_included(self):
          a = manager._is_file_included(
              path="a.py",
              included_globs=["*.py"],
              excluded_path_strings=[],
              enforce_glob=True,
          )
  
          b = manager._is_file_included(
              path="a.dd",
              included_globs=["*.py"],
              excluded_path_strings=[],
              enforce_glob=False,
          )
  
          c = manager._is_file_included(
              path="a.py",
              included_globs=["*.py"],
              excluded_path_strings=["a.py"],
              enforce_glob=True,
          )
  
          d = manager._is_file_included(
              path="a.dd",
              included_globs=["*.py"],
              excluded_path_strings=[],
              enforce_glob=True,
          )
  
          e = manager._is_file_included(
              path="x_a.py",
              included_globs=["*.py"],
              excluded_path_strings=["x_*.py"],
              enforce_glob=True,
          )
  
          f = manager._is_file_included(
              path="x.py",
              included_globs=["*.py"],
              excluded_path_strings=["x_*.py"],
              enforce_glob=True,
          )
          self.assertTrue(a)
          self.assertTrue(b)
          self.assertFalse(c)
          self.assertFalse(d)
          self.assertFalse(e)
          self.assertTrue(f)
  
      @mock.patch("os.walk")
      def test_get_files_from_dir(self, os_walk):
          os_walk.return_value = [
              ("/", ("a"), ()),
              ("/a", (), ("a.py", "b.py", "c.ww")),
          ]
  
          inc, exc = manager._get_files_from_dir(
              files_dir="", included_globs=["*.py"], excluded_path_strings=None
          )
  
          self.assertEqual({"/a/c.ww"}, exc)
          self.assertEqual({"/a/a.py", "/a/b.py"}, inc)
  
      def test_populate_baseline_success(self):
          # Test populate_baseline with valid JSON
          baseline_data = """{
              "results": [
                  {
                      "code": "test code",
                      "filename": "example_file.py",
                      "issue_severity": "low",
                      "issue_cwe": {
                          "id": 605,
                          "link": "%s"
                      },
                      "issue_confidence": "low",
                      "issue_text": "test issue",
                      "test_name": "some_test",
                      "test_id": "x",
                      "line_number": "n",
                      "line_range": "n-m"
                  }
              ]
          }
          """ % (
              "https://cwe.mitre.org/data/definitions/605.html"
          )
          issue_dictionary = {
              "code": "test code",
              "filename": "example_file.py",
              "issue_severity": "low",
              "issue_cwe": issue.Cwe(issue.Cwe.MULTIPLE_BINDS).as_dict(),
              "issue_confidence": "low",
              "issue_text": "test issue",
              "test_name": "some_test",
              "test_id": "x",
              "line_number": "n",
              "line_range": "n-m",
          }
          baseline_items = [issue.issue_from_dict(issue_dictionary)]
          self.manager.populate_baseline(baseline_data)
          self.assertEqual(baseline_items, self.manager.baseline)
  
      @mock.patch("logging.Logger.warning")
      def test_populate_baseline_invalid_json(self, mock_logger_warning):
          # Test populate_baseline with invalid JSON content
          baseline_data = """{"data": "bad"}"""
          self.manager.populate_baseline(baseline_data)
          # Default value for manager.baseline is []
          self.assertEqual([], self.manager.baseline)
          self.assertTrue(mock_logger_warning.called)
  
      def test_results_count(self):
          levels = [constants.LOW, constants.MEDIUM, constants.HIGH]
          self.manager.results = [
              issue.Issue(
                  severity=level, cwe=issue.Cwe.MULTIPLE_BINDS, confidence=level
              )
              for level in levels
          ]
  
          r = [
              self.manager.results_count(sev_filter=level, conf_filter=level)
              for level in levels
          ]
  
          self.assertEqual([3, 2, 1], r)
  
      def test_output_results_invalid_format(self):
          # Test that output_results succeeds given an invalid format
          temp_directory = self.useFixture(fixtures.TempDir()).path
          lines = 5
          sev_level = constants.LOW
          conf_level = constants.LOW
          output_filename = os.path.join(temp_directory, "_temp_output")
          output_format = "invalid"
          with open(output_filename, "w") as tmp_file:
              self.manager.output_results(
                  lines, sev_level, conf_level, tmp_file, output_format
              )
          self.assertTrue(os.path.isfile(output_filename))
  
      def test_output_results_valid_format(self):
          # Test that output_results succeeds given a valid format
          temp_directory = self.useFixture(fixtures.TempDir()).path
          lines = 5
          sev_level = constants.LOW
          conf_level = constants.LOW
          output_filename = os.path.join(temp_directory, "_temp_output.txt")
          output_format = "txt"
          with open(output_filename, "w") as tmp_file:
              self.manager.output_results(
                  lines, sev_level, conf_level, tmp_file, output_format
              )
          self.assertTrue(os.path.isfile(output_filename))
  
      @mock.patch("os.path.isdir")
      def test_discover_files_recurse_skip(self, isdir):
          isdir.return_value = True
          self.manager.discover_files(["thing"], False)
          self.assertEqual([], self.manager.files_list)
          self.assertEqual([], self.manager.excluded_files)
  
      @mock.patch("os.path.isdir")
      def test_discover_files_recurse_files(self, isdir):
          isdir.return_value = True
          with mock.patch.object(manager, "_get_files_from_dir") as m:
              m.return_value = ({"files"}, {"excluded"})
              self.manager.discover_files(["thing"], True)
              self.assertEqual(["files"], self.manager.files_list)
              self.assertEqual(["excluded"], self.manager.excluded_files)
  
      @mock.patch("os.path.isdir")
      def test_discover_files_exclude(self, isdir):
          isdir.return_value = False
          with mock.patch.object(manager, "_is_file_included") as m:
              m.return_value = False
              self.manager.discover_files(["thing"], True)
              self.assertEqual([], self.manager.files_list)
              self.assertEqual(["thing"], self.manager.excluded_files)
  
      @mock.patch("os.path.isdir")
      def test_discover_files_exclude_dir(self, isdir):
          isdir.return_value = False
  
          # Test exclude dir using wildcard
          self.manager.discover_files(["./x/y.py"], True, "./x/*")
          self.assertEqual([], self.manager.files_list)
          self.assertEqual(["./x/y.py"], self.manager.excluded_files)
  
          # Test exclude dir without wildcard
          isdir.side_effect = [True, False]
          self.manager.discover_files(["./x/y.py"], True, "./x/")
          self.assertEqual([], self.manager.files_list)
          self.assertEqual(["./x/y.py"], self.manager.excluded_files)
  
          # Test exclude dir without wildcard or trailing slash
          isdir.side_effect = [True, False]
          self.manager.discover_files(["./x/y.py"], True, "./x")
          self.assertEqual([], self.manager.files_list)
          self.assertEqual(["./x/y.py"], self.manager.excluded_files)
  
          # Test exclude dir without prefix or suffix
          isdir.side_effect = [False, False]
          self.manager.discover_files(["./x/y/z.py"], True, "y")
          self.assertEqual([], self.manager.files_list)
          self.assertEqual(["./x/y/z.py"], self.manager.excluded_files)
  
      @mock.patch("os.path.isdir")
      def test_discover_files_exclude_cmdline(self, isdir):
          isdir.return_value = False
          with mock.patch.object(manager, "_is_file_included") as m:
              self.manager.discover_files(
                  ["a", "b", "c"], True, excluded_paths="a,b"
              )
              m.assert_called_with(
                  "c", ["*.py", "*.pyw"], ["a", "b"], enforce_glob=False
              )
  
      @mock.patch("os.path.isdir")
      def test_discover_files_exclude_glob(self, isdir):
          isdir.return_value = False
          self.manager.discover_files(
              ["a.py", "test_a.py", "test.py"], True, excluded_paths="test_*.py"
          )
          self.assertEqual(["./a.py", "./test.py"], self.manager.files_list)
          self.assertEqual(["test_a.py"], self.manager.excluded_files)
  
      @mock.patch("os.path.isdir")
      def test_discover_files_include(self, isdir):
          isdir.return_value = False
          with mock.patch.object(manager, "_is_file_included") as m:
              m.return_value = True
              self.manager.discover_files(["thing"], True)
              self.assertEqual(["./thing"], self.manager.files_list)
              self.assertEqual([], self.manager.excluded_files)
  
      def test_run_tests_keyboardinterrupt(self):
          # Test that bandit manager exits when there is a keyboard interrupt
          temp_directory = self.useFixture(fixtures.TempDir()).path
          some_file = os.path.join(temp_directory, "some_code_file.py")
          with open(some_file, "w") as fd:
              fd.write("some_code = x + 1")
          self.manager.files_list = [some_file]
          with mock.patch(
              "bandit.core.metrics.Metrics.count_issues"
          ) as mock_count_issues:
              mock_count_issues.side_effect = KeyboardInterrupt
              # assert a SystemExit with code 2
              self.assertRaisesRegex(SystemExit, "2", self.manager.run_tests)
  
      def test_run_tests_ioerror(self):
          # Test that a file name is skipped and added to the manager.skipped
          # list when there is an IOError attempting to open/read the file
          temp_directory = self.useFixture(fixtures.TempDir()).path
          no_such_file = os.path.join(temp_directory, "no_such_file.py")
          self.manager.files_list = [no_such_file]
          self.manager.run_tests()
          # since the file name and the IOError.strerror text are added to
          # manager.skipped, we convert skipped to str to find just the file name
          # since IOError is not constant
          self.assertIn(no_such_file, str(self.manager.skipped))
  
      def test_compare_baseline(self):
          issue_a = self._get_issue_instance()
          issue_a.fname = "file1.py"
  
          issue_b = self._get_issue_instance()
          issue_b.fname = "file2.py"
  
          issue_c = self._get_issue_instance(sev=constants.HIGH)
          issue_c.fname = "file1.py"
  
          # issue c is in results, not in baseline
          self.assertEqual(
              [issue_c],
              manager._compare_baseline_results(
                  [issue_a, issue_b], [issue_a, issue_b, issue_c]
              ),
          )
  
          # baseline and results are the same
          self.assertEqual(
              [],
              manager._compare_baseline_results(
                  [issue_a, issue_b, issue_c], [issue_a, issue_b, issue_c]
              ),
          )
  
          # results are better than baseline
          self.assertEqual(
              [],
              manager._compare_baseline_results(
                  [issue_a, issue_b, issue_c], [issue_a, issue_b]
              ),
          )
  
      def test_find_candidate_matches(self):
          issue_a = self._get_issue_instance()
          issue_b = self._get_issue_instance()
  
          issue_c = self._get_issue_instance()
          issue_c.fname = "file1.py"
  
          # issue a and b are the same, both should be returned as candidates
          self.assertEqual(
              {issue_a: [issue_a, issue_b]},
              manager._find_candidate_matches([issue_a], [issue_a, issue_b]),
          )
  
          # issue a and c are different, only a should be returned
          self.assertEqual(
              {issue_a: [issue_a]},
              manager._find_candidate_matches([issue_a], [issue_a, issue_c]),
          )
  
          # c doesn't match a, empty list should be returned
          self.assertEqual(
              {issue_a: []},
              manager._find_candidate_matches([issue_a], [issue_c]),
          )
  
          # a and b match, a and b should both return a and b candidates
          self.assertEqual(
              {issue_a: [issue_a, issue_b], issue_b: [issue_a, issue_b]},
              manager._find_candidate_matches(
                  [issue_a, issue_b], [issue_a, issue_b, issue_c]
              ),
          )
  