}
}

BOOST_ASIO_DECL int open(const char* path, int flags,
    boost::system::error_code& ec);

BOOST_ASIO_DECL int close(int d, state_type& state,
    boost::system::error_code& ec);


            

Reported by FlawFinder.

                }

  // Open a new socket implementation.
  boost::system::error_code open(implementation_type&,
      const protocol_type&, boost::system::error_code& ec)
  {
    ec = boost::asio::error::operation_not_supported;
    return ec;
  }

            

Reported by FlawFinder.

                }

  // Open the serial port using the specified device name.
  BOOST_ASIO_DECL boost::system::error_code open(implementation_type& impl,
      const std::string& device, boost::system::error_code& ec);

  // Assign a native descriptor to a serial port implementation.
  boost::system::error_code assign(implementation_type& impl,
      const native_handle_type& native_descriptor,

            

Reported by FlawFinder.

                }

  // Open a new socket implementation.
  boost::system::error_code open(implementation_type& impl,
      const protocol_type& protocol, boost::system::error_code& ec)
  {
    if (!do_open(impl, protocol.family(),
          protocol.type(), protocol.protocol(), ec))
      impl.protocol_ = protocol;

            

Reported by FlawFinder.

              import textwrap
import yaml

from ..resmokelib.utils import default_if_none


# Setup to preserve order in yaml.dump, see https://stackoverflow.com/a/8661021
def _represent_dict_order(self, data):
    return self.represent_mapping("tag:yaml.org,2002:map", list(data.items()))

            

Reported by Pylint.

              
            fh.seek(0)
            text = fh.read()
            assert text[byte_offset] == '0'
            fh.seek(0)
            fh.write(text[:byte_offset])
            fh.write(str(next(next_code_generator)))
            fh.write(text[byte_offset + 1:])
            fh.seek(0)

            

Reported by Bandit.

                      :param tests_by_task: Dictionary of tasks to run with tests to run in each.
        """
        json_text = self.create_generate_tasks_configuration(tests_by_task)
        assert self.generate_tasks_file is not None
        if self.generate_tasks_file:
            write_file(self.generate_tasks_file, json_text)


# pylint: disable=too-many-arguments

            

Reported by Bandit.

                                  implicit_multiversion = run_vars.get("implicit_multiversion", False)

                    if implicit_multiversion:
                        assert suite is not None
                        is_sharded = self.multiversion_util.is_suite_sharded(suite)
                        version_list = get_version_configs(is_sharded)

                    if is_jstestfuzz:
                        fuzzer_params = self.task_def_to_fuzzer_params(

            

Reported by Bandit.

              
def list_commands_for_api(api_version: str, mongod_or_mongos: str, install_dir: str) -> Set[str]:
    """Get a list of commands in a given API version by calling listCommands."""
    assert mongod_or_mongos in ("mongod", "mongos")
    logging.info("Calling listCommands on %s", mongod_or_mongos)
    dbpath = TemporaryDirectory()
    fixturelib = FixtureLib()
    mongod_executable = os.path.join(install_dir, "mongod")
    mongos_executable = os.path.join(install_dir, "mongos")

            

Reported by Bandit.

                              ctxt.add_new_namespace_incompatible_error(old_cmd.command_name, old_namespace,
                                                          new_namespace, new_idl_file_path)
    else:
        assert False, 'unrecognized namespace option'


def check_error_reply(old_basic_types_path: str, new_basic_types_path: str,
                      import_directories: List[str]) -> IDLCompatibilityErrorCollection:
    """Check IDL compatibility between old and new ErrorReply."""

            

Reported by Bandit.