diff --git a/CHANGELOG.md b/CHANGELOG.md
index e1d5d874..6571d03a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
  
 ### Added
+- Support for a DSN in `tarantool.dbapi.connect()`. The expected format
+  is `[scheme://][user[:password]@]host:port[?option=value&...]`, where
+  a Unix socket address is either `unix/:path` or an absolute path, and
+  an IPv6 address is enclosed in `[]`. The scheme, if any, is ignored,
+  as Tarantool itself ignores it. Allowed options are the Tarantool URI
+  parameters: `transport`, `ssl_key_file`, `ssl_cert_file`,
+  `ssl_ca_file`, `ssl_ciphers`, `ssl_password`, `ssl_password_file` and
+  `auth_type`. Parameters set explicitly take precedence over the ones
+  from the DSN (PR #351).
  
 ### Changed
  
diff --git a/tarantool/dbapi.py b/tarantool/dbapi.py
index 3b4d1013..cd580e5b 100644
--- a/tarantool/dbapi.py
+++ b/tarantool/dbapi.py
@@ -6,8 +6,11 @@
 # pylint: disable=fixme,unused-import,bad-option-value,no-self-use
 # flake8: noqa: F401
 
+import typing
+
 from tarantool.connection import Connection as BaseConnection
 from tarantool.error import (
+    ConfigurationError,
     Error,
     InterfaceError,
     DatabaseError,
@@ -17,6 +20,7 @@
     ProgrammingError,
     NotSupportedError,
 )
+from tarantool.utils import parse_dsn
 
 Warning = Warning  # pylint: disable=redefined-builtin,self-assigning-variable
 
@@ -400,9 +404,10 @@ def connect(dsn=None, host=None, port=None,
     """
     Constructor for creating a connection to the database.
 
-    :param dsn: **Not implemented**. Tarantool server URI:
-        ``[[[username[:password]@]host:]port``.
-    :type dsn: :obj:`str`
+    :param dsn: Tarantool server DSN, refer to
+        :func:`~tarantool.utils.parse_dsn`. Parameters set explicitly
+        take precedence over the ones from the DSN.
+    :type dsn: :obj:`str`, optional
 
     :param host: Refer to :paramref:`~tarantool.Connection.params.host`.
 
@@ -415,14 +420,18 @@ def connect(dsn=None, host=None, port=None,
 
     :rtype: :class:`~tarantool.Connection`
 
-    :raise:  :exc:`~NotImplementedError`,
+    :raise: :exc:`~tarantool.error.InterfaceError`,
         :class:`~tarantool.Connection` exceptions
     """
 
+    params: typing.Dict[str, typing.Any] = {}
+
     if dsn:
-        raise NotImplementedError("dsn param is not implemented in"
-                                  "this version of dbapi module")
-    params = {}
+        try:
+            params = parse_dsn(dsn)
+        except ConfigurationError as exc:
+            raise InterfaceError(str(exc)) from exc
+
     if host:
         params["host"] = host
     if port:
@@ -432,6 +441,6 @@ def connect(dsn=None, host=None, port=None,
     if password:
         params["password"] = password
 
-    kwargs.update(params)
+    params.update(kwargs)
 
-    return Connection(**kwargs)
+    return Connection(**params)
diff --git a/tarantool/utils.py b/tarantool/utils.py
index b7ab436d..b576db8b 100644
--- a/tarantool/utils.py
+++ b/tarantool/utils.py
@@ -4,11 +4,47 @@
 
 from base64 import decodebytes as base64_decode
 from dataclasses import dataclass
+import re
 import typing
 import uuid
+import socket
+
+from tarantool.error import ConfigurationError
 
 ENCODING_DEFAULT = "utf-8"
 
+DSN_DEFAULT_HOST = "127.0.0.1"
+"""
+Host to use if a DSN consists of a port only.
+"""
+
+DSN_UNIX_PREFIX = "unix/:"
+"""
+Prefix of a Unix socket address in a DSN.
+"""
+
+DSN_SCHEME_RE = re.compile(r'[A-Za-z][A-Za-z0-9+.\-]*://')
+"""
+Scheme of a DSN. Only a scheme at the very beginning of a DSN is
+recognized as such, the same way Tarantool does it.
+
+:meta private:
+"""
+
+DSN_OPTIONS: typing.Tuple[str, ...] = (
+    'transport',
+    'ssl_key_file',
+    'ssl_cert_file',
+    'ssl_ca_file',
+    'ssl_ciphers',
+    'ssl_password',
+    'ssl_password_file',
+    'auth_type',
+)
+"""
+Tarantool URI parameters allowed in a DSN query.
+"""
+
 
 def strxor(rhs, lhs):
     """
@@ -146,3 +182,245 @@ def greeting_decode(greeting_buf):
     except ValueError as exc:
         print('exx', exc)
         raise ValueError("Invalid greeting: " + str(greeting_buf)) from exc
+
+
+def _dsn_error(dsn: str, msg: str) -> ConfigurationError:
+    """
+    Build a DSN parse error.
+
+    :param dsn: Source DSN.
+    :type dsn: :obj:`str`
+
+    :param msg: Error description.
+    :type msg: :obj:`str`
+
+    :rtype: :exc:`~tarantool.error.ConfigurationError`
+
+    :meta private:
+    """
+
+    return ConfigurationError(f'DSN "{dsn}": {msg}')
+
+
+def _is_dsn_port(port_str: str) -> bool:
+    """
+    Check whether a DSN substring is a port.
+
+    :param port_str: Port substring.
+    :type port_str: :obj:`str`
+
+    :rtype: :obj:`bool`
+
+    :meta private:
+    """
+
+    return port_str.isascii() and port_str.isdigit()
+
+
+def _is_dsn_userinfo(userinfo: str) -> bool:
+    """
+    Check whether a DSN substring before an ``@`` is a user name with
+    a password rather than a part of a Unix socket path. Tarantool
+    allows neither ``/`` nor more than one ``:`` in a user name and a
+    password, so ``/tmp/tt@1.sock`` is a socket path, not a user name.
+
+    :param userinfo: Substring before the first ``@``.
+    :type userinfo: :obj:`str`
+
+    :rtype: :obj:`bool`
+
+    :meta private:
+    """
+
+    return '/' not in userinfo
+
+
+def _parse_dsn_port(dsn: str, port_str: str) -> int:
+    """
+    Parse the port of a DSN.
+
+    :param dsn: Source DSN.
+    :type dsn: :obj:`str`
+
+    :param port_str: Port substring.
+    :type port_str: :obj:`str`
+
+    :rtype: :obj:`int`
+
+    :raise: :exc:`~tarantool.error.ConfigurationError`
+
+    :meta private:
+    """
+
+    if not _is_dsn_port(port_str):
+        raise _dsn_error(dsn, f'port "{port_str}" is not a number')
+    port = int(port_str)
+    if port < 1 or port > 65535:
+        raise _dsn_error(dsn, 'port must be in range [1, 65535]')
+    return port
+
+
+def _parse_dsn_address(
+        dsn: str,
+        address: str) -> typing.Tuple[typing.Optional[str],
+                                      typing.Union[int, str]]:
+    """
+    Parse the address of a DSN. For a Unix socket address, host is
+    ``None`` and port is a socket path.
+
+    :param dsn: Source DSN.
+    :type dsn: :obj:`str`
+
+    :param address: Address substring.
+    :type address: :obj:`str`
+
+    :return: `(host, port)` pair.
+    :rtype: :obj:`tuple`
+
+    :raise: :exc:`~tarantool.error.ConfigurationError`
+
+    :meta private:
+    """
+    # pylint: disable=too-many-return-statements,too-many-branches
+
+    if address.startswith(DSN_UNIX_PREFIX):
+        path = address[len(DSN_UNIX_PREFIX):]
+        if not path:
+            raise _dsn_error(dsn, 'Unix socket path is empty')
+        if not path.startswith(('/', './')):
+            raise _dsn_error(dsn, f'Unix socket path "{path}" is neither '
+                                  'absolute nor started with "./"')
+        return None, path
+
+    if address.startswith(('/', './')):
+        return None, address
+
+    if '/' in address:
+        raise _dsn_error(dsn, f'address "{address}" is neither a host with '
+                              'a port nor a Unix socket path')
+
+    if address.startswith('['):
+        delim = address.find(']')
+        if delim == -1:
+            raise _dsn_error(dsn, 'IPv6 address is not closed with "]"')
+        host, tail = address[1:delim], address[delim + 1:]
+        try:
+            socket.inet_pton(socket.AF_INET6, host)
+        except (OSError, ValueError):
+            raise _dsn_error(dsn, f'"{host}" is not an IPv6 address') from None
+        if not tail.startswith(':'):
+            raise _dsn_error(dsn, 'port is not specified')
+        return host, _parse_dsn_port(dsn, tail[1:])
+
+    if ':' in address:
+        host, port_str = address.rsplit(':', 1)
+        if not host:
+            raise _dsn_error(dsn, 'host value is empty')
+        if ':' in host:
+            raise _dsn_error(dsn, 'IPv6 address must be enclosed in "[]"')
+        return host, _parse_dsn_port(dsn, port_str)
+
+    if _is_dsn_port(address):
+        return DSN_DEFAULT_HOST, _parse_dsn_port(dsn, address)
+
+    raise _dsn_error(dsn, 'port is not specified')
+
+
+def _parse_dsn_options(dsn: str, query: str) -> typing.Dict[str, str]:
+    """
+    Parse the query of a DSN into connection options.
+
+    :param dsn: Source DSN.
+    :type dsn: :obj:`str`
+
+    :param query: Query substring.
+    :type query: :obj:`str`
+
+    :rtype: :obj:`dict`
+
+    :raise: :exc:`~tarantool.error.ConfigurationError`
+
+    :meta private:
+    """
+
+    options: typing.Dict[str, str] = {}
+    for option_str in query.split('&'):
+        if not option_str:
+            continue
+        name, delim, value = option_str.partition('=')
+        if not delim:
+            raise _dsn_error(dsn, f'option "{name}" has no value')
+        if name not in DSN_OPTIONS:
+            raise _dsn_error(dsn, f'unknown option "{name}"')
+        options[name] = value
+    return options
+
+
+def parse_dsn(dsn: str) -> typing.Dict[str, typing.Any]:
+    """
+    Parse a Tarantool DSN string into :class:`~tarantool.Connection`
+    parameters.
+
+    Expected format is
+    ``[scheme://][user[:password]@]host:port[?option=value&...]``.
+    A Unix socket address is either ``unix/:path`` or a path itself,
+    absolute or started with ``./``; an IPv6 address must be enclosed
+    in ``[]``. The scheme, if any, is ignored, as Tarantool itself
+    ignores it, but it is recognized only at the very beginning of a
+    DSN. Values are not percent-decoded, so neither ``@`` nor ``/``
+    is allowed in a user name or a password. Whitespace is not
+    allowed anywhere. Allowed options are the Tarantool URI parameters
+    listed in :data:`~tarantool.utils.DSN_OPTIONS`.
+
+    :param dsn: Tarantool server DSN.
+    :type dsn: :obj:`str`
+
+    :return: Keyword arguments for :class:`~tarantool.Connection`.
+        Only the parameters explicitly set in the DSN are present.
+    :rtype: :obj:`dict`
+
+    :raise: :exc:`~tarantool.error.ConfigurationError`
+    """
+
+    if not isinstance(dsn, str):
+        raise ConfigurationError('DSN should be of a string type')
+
+    source = dsn
+    if not dsn:
+        raise ConfigurationError('DSN should not be an empty string')
+
+    if any(char.isspace() for char in dsn):
+        raise _dsn_error(source, 'whitespace is not allowed')
+
+    params: typing.Dict[str, typing.Any] = {}
+
+    scheme = DSN_SCHEME_RE.match(dsn)
+    if scheme:
+        dsn = dsn[scheme.end():]
+
+    query = ''
+    if '?' in dsn:
+        dsn, query = dsn.split('?', 1)
+
+    userinfo, delim, tail = dsn.partition('@')
+    if delim and _is_dsn_userinfo(userinfo):
+        dsn = tail
+        if '@' in dsn:
+            raise _dsn_error(source, '"@" is not allowed in a user name '
+                                     'or a password')
+        user, delim, password = userinfo.partition(':')
+        if not user:
+            raise _dsn_error(source, 'user value is empty')
+        if ':' in password:
+            raise _dsn_error(source, '":" is not allowed in a password')
+        params['user'] = user
+        if delim:
+            params['password'] = password
+
+    if not dsn:
+        raise _dsn_error(source, 'address is not specified')
+
+    params['host'], params['port'] = _parse_dsn_address(source, dsn)
+    params.update(_parse_dsn_options(source, query))
+
+    return params
diff --git a/test/suites/__init__.py b/test/suites/__init__.py
index 7d092585..89ba1527 100644
--- a/test/suites/__init__.py
+++ b/test/suites/__init__.py
@@ -26,6 +26,8 @@
 from .test_push import TestSuitePush
 from .test_connection import TestSuiteConnection
 from .test_crud import TestSuiteCrud
+from .test_dsn import TestSuiteDsnParse
+from .test_dsn import TestSuiteDsnConnect
 
 test_cases = (TestSuiteSchemaUnicodeConnection,
               TestSuiteSchemaBinaryConnection,
@@ -34,7 +36,8 @@
               TestSuiteEncoding, TestSuitePool, TestSuiteSsl,
               TestSuiteDecimal, TestSuiteUUID, TestSuiteDatetime,
               TestSuiteInterval, TestSuitePackage, TestSuiteErrorExt,
-              TestSuitePush, TestSuiteConnection, TestSuiteCrud, TestSuiteSocketFD)
+              TestSuitePush, TestSuiteConnection, TestSuiteCrud, TestSuiteSocketFD,
+              TestSuiteDsnParse, TestSuiteDsnConnect)
 
 
 def load_tests(loader, tests, pattern):
diff --git a/test/suites/test_dsn.py b/test/suites/test_dsn.py
new file mode 100644
index 00000000..58081d43
--- /dev/null
+++ b/test/suites/test_dsn.py
@@ -0,0 +1,379 @@
+"""
+This module tests DSN parsing and connecting with a DSN.
+"""
+# pylint: disable=missing-class-docstring,missing-function-docstring,duplicate-code
+# pylint: disable=protected-access,too-many-public-methods
+
+import sys
+import unittest
+
+from tarantool import dbapi
+from tarantool.error import ConfigurationError, InterfaceError
+from tarantool.utils import parse_dsn
+
+from .lib.tarantool_server import TarantoolServer
+from .utils import assert_admin_success
+
+
+class TestSuiteDsnParse(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        print(' DSN PARSE '.center(70, '='), file=sys.stderr)
+        print('-' * 70, file=sys.stderr)
+
+    def _assert_parsed(self, cases):
+        for dsn, expected in cases.items():
+            with self.subTest(dsn=dsn):
+                self.assertEqual(parse_dsn(dsn), expected)
+
+    def _assert_rejected(self, cases):
+        for dsn in cases:
+            with self.subTest(dsn=dsn):
+                with self.assertRaises(ConfigurationError):
+                    parse_dsn(dsn)
+
+    def test_address(self):
+        self._assert_parsed({
+            '3301':
+                {'host': '127.0.0.1', 'port': 3301},
+            'localhost:3301':
+                {'host': 'localhost', 'port': 3301},
+            '192.168.10.10:3301':
+                {'host': '192.168.10.10', 'port': 3301},
+            'server001.example.com:3301':
+                {'host': 'server001.example.com', 'port': 3301},
+        })
+
+    def test_ipv6_address(self):
+        self._assert_parsed({
+            '[::1]:3301':
+                {'host': '::1', 'port': 3301},
+            '[2a00:1148:b0ba:2016::10]:3301':
+                {'host': '2a00:1148:b0ba:2016::10', 'port': 3301},
+            '[::ffff:127.0.0.1]:3301':
+                {'host': '::ffff:127.0.0.1', 'port': 3301},
+        })
+
+    def test_ipv6_address_requires_brackets(self):
+        self._assert_rejected(['::1:3301', '2a00:1148:b0ba:2016::10:3301'])
+
+    def test_invalid_ipv6_address(self):
+        # Tarantool allows an IPv6 address only inside the brackets:
+        # neither a host name nor a zone index is accepted there.
+        self._assert_rejected([
+            '[localhost]:3301',
+            '[fe80::1%eth0]:3301',
+            '[]:3301',
+            '[::1:3301',
+            # inet_pton() raises a ValueError, not an OSError, on it.
+            '[\x00::1]:3301',
+        ])
+
+    def test_unix_socket(self):
+        self._assert_parsed({
+            'unix/:/tmp/tt.sock':
+                {'host': None, 'port': '/tmp/tt.sock'},
+            'unix/:./var/run/tt.iproto':
+                {'host': None, 'port': './var/run/tt.iproto'},
+            '/tmp/unix_domain_socket.sock':
+                {'host': None, 'port': '/tmp/unix_domain_socket.sock'},
+            'user:pass@unix/:/tmp/tt.sock':
+                {'user': 'user', 'password': 'pass',
+                 'host': None, 'port': '/tmp/tt.sock'},
+            'unix/:/tmp/tt.sock?transport=ssl':
+                {'host': None, 'port': '/tmp/tt.sock', 'transport': 'ssl'},
+        })
+
+    def test_unix_socket_path_with_at_sign(self):
+        # Tarantool reads a "@" as a part of a socket path, since a
+        # user name may contain no "/".
+        self._assert_parsed({
+            '/var/run/tt@1.sock':
+                {'host': None, 'port': '/var/run/tt@1.sock'},
+            'unix/:/tmp/a@b.sock':
+                {'host': None, 'port': '/tmp/a@b.sock'},
+        })
+
+    def test_empty_unix_socket_path(self):
+        self._assert_rejected(['unix/:'])
+
+    def test_relative_unix_socket_path(self):
+        # Tarantool takes a path as a socket address only if it is
+        # absolute or started with "./".
+        self._assert_rejected(['unix/:x.sock', 'unix/:relative/x.sock'])
+
+    def test_user_password(self):
+        self._assert_parsed({
+            'user@localhost:3301':
+                {'user': 'user', 'host': 'localhost', 'port': 3301},
+            'user:pass@localhost:3301':
+                {'user': 'user', 'password': 'pass',
+                 'host': 'localhost', 'port': 3301},
+            'user:@localhost:3301':
+                {'user': 'user', 'password': '',
+                 'host': 'localhost', 'port': 3301},
+        })
+
+    def test_values_are_not_percent_decoded(self):
+        # Tarantool uri.parse() does not decode percent-encoded values.
+        self._assert_parsed({
+            'user:p%40ss@localhost:3301':
+                {'user': 'user', 'password': 'p%40ss',
+                 'host': 'localhost', 'port': 3301},
+        })
+
+    def test_at_sign_is_not_allowed_in_userinfo(self):
+        self._assert_rejected(['user:p@ss@localhost:3301'])
+
+    def test_empty_user(self):
+        self._assert_rejected(['@localhost:3301', ':pass@localhost:3301'])
+
+    def test_colon_is_not_allowed_in_userinfo(self):
+        self._assert_rejected(['user:pa:ss@localhost:3301'])
+
+    def test_slash_is_not_allowed_in_userinfo(self):
+        # Tarantool reinterprets such a DSN as a host with a service,
+        # silently connecting to a wrong address.
+        self._assert_rejected(['user:pa/ss@localhost:3301'])
+
+    def test_scheme_is_ignored(self):
+        self._assert_parsed({
+            'tcp://localhost:3301':
+                {'host': 'localhost', 'port': 3301},
+            'TCP://localhost:3301':
+                {'host': 'localhost', 'port': 3301},
+            'tcp+ssl://localhost:3301':
+                {'host': 'localhost', 'port': 3301},
+            'tarantool://user:pass@localhost:3301':
+                {'user': 'user', 'password': 'pass',
+                 'host': 'localhost', 'port': 3301},
+        })
+
+    def test_scheme_is_recognized_only_at_the_beginning(self):
+        # A "://" in the middle of a DSN is not a scheme delimiter,
+        # so a query value may contain an URL of its own.
+        self._assert_parsed({
+            'localhost:3301?ssl_ca_file=https://ca.example.com/ca.crt':
+                {'host': 'localhost', 'port': 3301,
+                 'ssl_ca_file': 'https://ca.example.com/ca.crt'},
+        })
+        self._assert_rejected([
+            '://localhost:3301',
+            '1tcp://localhost:3301',
+            'user:pass://localhost:3301',
+        ])
+
+    def test_whitespace_is_not_allowed(self):
+        self._assert_rejected([
+            ' localhost:3301',
+            'localhost:3301 ',
+            'localhost: 3301',
+            'ho st:3301',
+            'localhost:33\n01',
+            'localhost:3301?transport=s sl',
+        ])
+
+    def test_options(self):
+        dsn = ('localhost:3301?transport=ssl'
+               '&ssl_key_file=k.key'
+               '&ssl_cert_file=c.crt'
+               '&ssl_ca_file=ca.crt'
+               '&ssl_ciphers=ECDHE-RSA-AES256-GCM-SHA384'
+               '&ssl_password=secret'
+               '&ssl_password_file=/etc/pw.txt'
+               '&auth_type=pap-sha256')
+        self._assert_parsed({
+            dsn: {
+                'host': 'localhost',
+                'port': 3301,
+                'transport': 'ssl',
+                'ssl_key_file': 'k.key',
+                'ssl_cert_file': 'c.crt',
+                'ssl_ca_file': 'ca.crt',
+                'ssl_ciphers': 'ECDHE-RSA-AES256-GCM-SHA384',
+                'ssl_password': 'secret',
+                'ssl_password_file': '/etc/pw.txt',
+                'auth_type': 'pap-sha256',
+            },
+        })
+
+    def test_last_option_value_wins(self):
+        self._assert_parsed({
+            'localhost:3301?transport=ssl&transport=plain':
+                {'host': 'localhost', 'port': 3301, 'transport': 'plain'},
+        })
+
+    def test_invalid_options(self):
+        self._assert_rejected([
+            'localhost:3301?fetch_schema=false',
+            'localhost:3301?unknown=1',
+            'localhost:3301?transport',
+        ])
+
+    def test_no_port(self):
+        self._assert_rejected(['localhost', 'localhost:', '[::1]'])
+
+    def test_invalid_port(self):
+        self._assert_rejected([
+            'localhost:notaport',
+            'localhost:0',
+            'localhost:99999',
+        ])
+
+    def test_port_is_ascii_digits_only(self):
+        # int() is more permissive than Tarantool: it also takes a
+        # sign, underscore separators and non-ASCII digits.
+        self._assert_rejected([
+            'localhost:+3301',
+            'localhost:-3301',
+            'localhost:3_301',
+            'localhost:0x10',
+            'localhost:٠٣٣٠١',
+            '٠٣٣٠١',
+        ])
+        self._assert_parsed({
+            'localhost:03301': {'host': 'localhost', 'port': 3301},
+        })
+
+    def test_empty_host(self):
+        self._assert_rejected([':3301'])
+
+    def test_empty_dsn(self):
+        self._assert_rejected([''])
+
+    def test_dsn_is_not_a_string(self):
+        for dsn in (3301, None, ['localhost:3301']):
+            with self.subTest(dsn=dsn):
+                with self.assertRaises(ConfigurationError):
+                    parse_dsn(dsn)
+
+
+class TestSuiteDsnConnect(unittest.TestCase):
+    EVAL_USER = "return box.session.user()"
+
+    @classmethod
+    def setUpClass(cls):
+        print(' DSN CONNECT '.center(70, '='), file=sys.stderr)
+        print('-' * 70, file=sys.stderr)
+
+        cls.srv = TarantoolServer()
+        cls.srv.script = 'test/suites/box.lua'
+        cls.srv.start()
+
+        resp = cls.srv.admin("""
+            box.schema.user.create('test', {password = 'test', if_not_exists = true})
+            box.schema.user.grant('test', 'read,write,execute', 'universe',
+                                  nil, {if_not_exists = true})
+
+            return true
+        """)
+        assert_admin_success(resp)
+
+        if sys.platform.startswith("win"):
+            cls.sock_srv = None
+        else:
+            cls.sock_srv = TarantoolServer(create_unix_socket=True)
+            cls.sock_srv.script = 'test/suites/box.lua'
+            cls.sock_srv.start()
+
+    def setUp(self):
+        # prevent a remote tarantool from clean our session
+        if self.srv.is_started():
+            self.srv.touch_lock()
+
+    @property
+    def address(self):
+        return f'{self.srv.host}:{self.srv.args["primary"]}'
+
+    def test_connect_host_port(self):
+        conn = dbapi.connect(dsn=self.address)
+        try:
+            self.assertEqual(conn.ping(notime=True), "Success")
+        finally:
+            conn.close()
+
+    def test_connect_user_password(self):
+        conn = dbapi.connect(dsn=f'test:test@{self.address}')
+        try:
+            self.assertSequenceEqual(conn.eval(self.EVAL_USER), ["test"])
+        finally:
+            conn.close()
+
+    def test_connect_scheme_is_ignored(self):
+        conn = dbapi.connect(dsn=f'tcp://test:test@{self.address}')
+        try:
+            self.assertSequenceEqual(conn.eval(self.EVAL_USER), ["test"])
+        finally:
+            conn.close()
+
+    @unittest.skipIf(sys.platform.startswith("win"),
+                     'Unix sockets are not supported on Windows')
+    def test_connect_unix_socket(self):
+        conn = dbapi.connect(dsn=f'unix/:{self.sock_srv.args["primary"]}')
+        try:
+            self.assertEqual(conn.ping(notime=True), "Success")
+        finally:
+            conn.close()
+
+    @unittest.skipIf(sys.platform.startswith("win"),
+                     'Unix sockets are not supported on Windows')
+    def test_connect_unix_socket_bare_path(self):
+        conn = dbapi.connect(dsn=str(self.sock_srv.args["primary"]))
+        try:
+            self.assertEqual(conn.ping(notime=True), "Success")
+        finally:
+            conn.close()
+
+    def test_explicit_args_take_precedence(self):
+        dsn = f'wronguser:wrongpass@{self.address}'
+        conn = dbapi.connect(dsn=dsn, user='test', password='test')
+        try:
+            self.assertSequenceEqual(conn.eval(self.EVAL_USER), ["test"])
+        finally:
+            conn.close()
+
+    def test_explicit_kwargs_take_precedence(self):
+        conn = dbapi.connect(dsn=f'{self.address}?transport=ssl',
+                             transport='', connect_now=False)
+        self.assertEqual(conn.transport, '')
+
+    def test_dsn_options_are_applied(self):
+        dsn = (f'{self.address}?transport=ssl'
+               '&ssl_key_file=k.key'
+               '&ssl_cert_file=c.crt'
+               '&ssl_ca_file=ca.crt'
+               '&ssl_ciphers=ECDHE-RSA-AES256-GCM-SHA384'
+               '&ssl_password=secret'
+               '&ssl_password_file=/etc/pw.txt'
+               '&auth_type=chap-sha1')
+        conn = dbapi.connect(dsn=dsn, connect_now=False)
+        self.assertEqual(conn.transport, 'ssl')
+        self.assertEqual(conn.ssl_key_file, 'k.key')
+        self.assertEqual(conn.ssl_cert_file, 'c.crt')
+        self.assertEqual(conn.ssl_ca_file, 'ca.crt')
+        self.assertEqual(conn.ssl_ciphers, 'ECDHE-RSA-AES256-GCM-SHA384')
+        self.assertEqual(conn.ssl_password, 'secret')
+        self.assertEqual(conn.ssl_password_file, '/etc/pw.txt')
+        self.assertEqual(conn._client_auth_type, 'chap-sha1')
+
+    def test_dsn_unix_socket_params(self):
+        conn = dbapi.connect(dsn='unix/:/tmp/tt.sock', connect_now=False)
+        self.assertIsNone(conn.host)
+        self.assertEqual(conn.port, '/tmp/tt.sock')
+
+    def test_invalid_dsn(self):
+        # PEP-249 requires the connect() errors to be a part of the
+        # DB-API exception hierarchy.
+        for dsn in ('localhost', 'localhost:notaport', 'localhost:3301?unknown=1'):
+            with self.subTest(dsn=dsn):
+                with self.assertRaises(InterfaceError):
+                    dbapi.connect(dsn=dsn)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.srv.stop()
+        cls.srv.clean()
+
+        if cls.sock_srv is not None:
+            cls.sock_srv.stop()
+            cls.sock_srv.clean()