Repository: ClearcodeHQ/pytest-mysql (language: python)

Issue:
Access denied for user 'root'@'localhost
### What action do you want to perform
trying to run available test example 

cat tests/test_mysql.py
-------------------------
def test_mysql_test_without_cursor(mysql):
    """Run test without cursor and without fetching the data."""
    mysql.query("SELECT VERSION();")



cat    pytest.ini
-------------------------------------------------------------
[pytest]
mysql_dbname = testdb
xfail_strict = true
mysql_host = localhost
mysql_port = 33061
mysql_user = root
mysql_passwd = root
 

### What are the results
received 

E       MySQLdb._exceptions.OperationalError: (1045, "Access denied for user 'root'@'localhost' (using password: YES)")

--- tried to fix by running below in mysql  but did not help

ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'root';
FLUSH PRIVILEGES;

--- able to login as user: root pass:root on cmd line             mysql -u root -p

### What are the expected results
test should be successful.
mysql ficture factories code import from dbfixtures



Discussion / hints:
please post full stacktrace. And I'd appreciate also a test code
> please post full stacktrace. And I'd appreciate also a test code

I am using the same sample provided in the source code's test dir of pytest-mysql. specifically testing for test_mysql_test_without_cursor() sample.   pytest.ini added above.
Below is the backtrace. 

$ cd /home/ravi/cprog/pytest-mysql
$ pytest  tests/test_mysql.py
```
(common-venv-3.8.2) /home/ravi/cprog/pytest-mysql>pytest tests/test_mysql.py 
======================================================================================= test session starts ========================================================================================
platform linux -- Python 3.8.2, pytest-5.4.2, py-1.8.1, pluggy-0.13.1
rootdir: /home/ravi/cprog/pytest-mysql, inifile: pytest.ini
plugins: mysql-2.0.1, flask-1.0.0, postgresql-2.4.0
collected 2 items                                                                                                                                                                                  

tests/test_mysql.py .EE                                                                                                                                                                      [100%]

============================================================================================== ERRORS ==============================================================================================
_________________________________________________________________________ ERROR at setup of test_mysql_test_without_cursor _________________________________________________________________________

request = <SubRequest 'mysql2' for <Function test_mysql_test_without_cursor>>

    @pytest.fixture
    def mysql_fixture(request):
        """
        Client fixture for MySQL server.
    
        #. Get config.
        #. Try to import MySQLdb package.
        #. Connect to mysql server.
        #. Create database.
        #. Use proper database.
        #. Drop database after tests.
    
        :param FixtureRequest request: fixture request object
    
        :rtype: MySQLdb.connections.Connection
        :returns: connection to database
        """
        config = get_config(request)
        process = request.getfixturevalue(process_fixture_name)
        if not process.running():
            process.start()
    
        mysql_host = process.host
        mysql_user = 'root'
        mysql_passwd = passwd or config['passwd']
        mysql_db = db or config['dbname']
    
>       mysql_conn = MySQLdb.connect(
            host=mysql_host,
            unix_socket=process.unixsocket.strpath,
            user=mysql_user,
            passwd=mysql_passwd,
        )

../../.pyenv/versions/3.8.2/envs/common-venv-3.8.2/lib/python3.8/site-packages/pytest_mysql/factories/client.py:77: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../.pyenv/versions/common-venv-3.8.2/lib/python3.8/site-packages/MySQLdb/__init__.py:130: in Connect
    return Connection(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_mysql.connection open to '(null)' at 0x55d5c0424a60>, args = ()
kwargs = {'host': 'localhost', 'passwd': 'root', 'unix_socket': '/tmp/pytest-of-ravi/pytest-49/pytest-mysql1/mysql.3308.sock', 'user': 'root'}
CLIENT = <module 'MySQLdb.constants.CLIENT' from '/home/ravi/.pyenv/versions/common-venv-3.8.2/lib/python3.8/site-packages/MySQLdb/constants/CLIENT.py'>
FIELD_TYPE = <module 'MySQLdb.constants.FIELD_TYPE' from '/home/ravi/.pyenv/versions/common-venv-3.8.2/lib/python3.8/site-packages/MySQLdb/constants/FIELD_TYPE.py'>
conversions = {<class 'int'>: <function Thing2Str at 0x7f501953aee0>, <class 'float'>: <function Float2Str at 0x7f501953af70>, <class 'NoneType'>: <function None2NULL at 0x7f5019536040>, <class 'array.array'>: <function array2Str at 0x7f50195361f0>, ...}
_bytes_or_str = ((128, <class 'bytes'>), (None, <class 'str'>)), proxy = <built-in function proxy>
kwargs2 = {'client_flag': 196608, 'conv': {<class 'int'>: <function Thing2Str at 0x7f501953aee0>, <class 'float'>: <function Flo...6040>, <class 'array.array'>: <function array2Str at 0x7f50195361f0>, ...}, 'host': 'localhost', 'passwd': 'root', ...}
conv = {<class 'int'>: <function Thing2Str at 0x7f501953aee0>, <class 'float'>: <function Float2Str at 0x7f501953af70>, <class 'NoneType'>: <function None2NULL at 0x7f5019536040>, <class 'array.array'>: <function array2Str at 0x7f50195361f0>, ...}
conv2 = {<class 'int'>: <function Thing2Str at 0x7f501953aee0>, <class 'float'>: <function Float2Str at 0x7f501953af70>, <class 'NoneType'>: <function None2NULL at 0x7f5019536040>, <class 'array.array'>: <function array2Str at 0x7f50195361f0>, ...}
k = 245, v = <class 'bytes'>

    def __init__(self, *args, **kwargs):
        """
        Create a connection to the database. It is strongly recommended
        that you only use keyword parameters. Consult the MySQL C API
        documentation for more information.
    
        :param str host:        host to connect
        :param str user:        user to connect as
        :param str password:    password to use
        :param str passwd:      alias of password, for backward compatibility
        :param str database:    database to use
        :param str db:          alias of database, for backward compatibility
        :param int port:        TCP/IP port to connect to
        :param str unix_socket: location of unix_socket to use
        :param dict conv:       conversion dictionary, see MySQLdb.converters
        :param int connect_timeout:
            number of seconds to wait before the connection attempt fails.
    
        :param bool compress:   if set, compression is enabled
        :param str named_pipe:  if set, a named pipe is used to connect (Windows only)
        :param str init_command:
            command which is run once the connection is created
    
        :param str read_default_file:
            file from which default client values are read
    
        :param str read_default_group:
            configuration group to use from the default file
    
        :param type cursorclass:
            class object, used to create cursors (keyword only)
    
        :param bool use_unicode:
            If True, text-like columns are returned as unicode objects
            using the connection's character set. Otherwise, text-like
            columns are returned as bytes. Unicode objects will always
            be encoded to the connection's character set regardless of
            this setting.
            Default to True.
    
        :param str charset:
            If supplied, the connection character set will be changed
            to this character set.
    
        :param str auth_plugin:
            If supplied, the connection default authentication plugin will be
            changed to this value. Example values:
            `mysql_native_password` or `caching_sha2_password`
    
        :param str sql_mode:
            If supplied, the session SQL mode will be changed to this
            setting.
            For more details and legal values, see the MySQL documentation.
    
        :param int client_flag:
            flags to use or 0 (see MySQL docs or constants/CLIENTS.py)
    
        :param str ssl_mode:
            specify the security settings for connection to the server;
            see the MySQL documentation for more details
            (mysql_option(), MYSQL_OPT_SSL_MODE).
            Only one of 'DISABLED', 'PREFERRED', 'REQUIRED',
            'VERIFY_CA', 'VERIFY_IDENTITY' can be specified.
    
        :param dict ssl:
            dictionary or mapping contains SSL connection parameters;
            see the MySQL documentation for more details
            (mysql_ssl_set()).  If this is set, and the client does not
            support SSL, NotSupportedError will be raised.
    
        :param bool local_infile:
            enables LOAD LOCAL INFILE; zero disables
    
        :param bool autocommit:
            If False (default), autocommit is disabled.
            If True, autocommit is enabled.
            If None, autocommit isn't set and server default is used.
    
        :param bool binary_prefix:
            If set, the '_binary' prefix will be used for raw byte query
            arguments (e.g. Binary). This is disabled by default.
    
        There are a number of undocumented, non-standard methods. See the
        documentation for the MySQL C API for some hints on what they do.
        """
        from MySQLdb.constants import CLIENT, FIELD_TYPE
        from MySQLdb.converters import conversions, _bytes_or_str
        from weakref import proxy
    
        kwargs2 = kwargs.copy()
    
        if "database" in kwargs2:
            kwargs2["db"] = kwargs2.pop("database")
        if "password" in kwargs2:
            kwargs2["passwd"] = kwargs2.pop("password")
    
        if "conv" in kwargs:
            conv = kwargs["conv"]
        else:
            conv = conversions
    
        conv2 = {}
        for k, v in conv.items():
            if isinstance(k, int) and isinstance(v, list):
                conv2[k] = v[:]
            else:
                conv2[k] = v
        kwargs2["conv"] = conv2
    
        cursorclass = kwargs2.pop("cursorclass", self.default_cursor)
        charset = kwargs2.get("charset", "")
        use_unicode = kwargs2.pop("use_unicode", True)
        sql_mode = kwargs2.pop("sql_mode", "")
        self._binary_prefix = kwargs2.pop("binary_prefix", False)
    
        client_flag = kwargs.get("client_flag", 0)
        client_version = tuple(
            [numeric_part(n) for n in _mysql.get_client_info().split(".")[:2]]
        )
        if client_version >= (4, 1):
            client_flag |= CLIENT.MULTI_STATEMENTS
        if client_version >= (5, 0):
            client_flag |= CLIENT.MULTI_RESULTS
    
        kwargs2["client_flag"] = client_flag
    
        # PEP-249 requires autocommit to be initially off
        autocommit = kwargs2.pop("autocommit", False)
    
>       super().__init__(*args, **kwargs2)
E       MySQLdb._exceptions.OperationalError: (2002, "Can't connect to local MySQL server through socket '/tmp/pytest-of-ravi/pytest-49/pytest-mysql1/mysql.3308.sock' (2)")

../../.pyenv/versions/common-venv-3.8.2/lib/python3.8/site-packages/MySQLdb/connections.py:185: OperationalError
_______________________________________________________________________ ERROR at teardown of test_mysql_test_without_cursor ________________________________________________________________________

request = <SubRequest 'mysql_proc2' for <Function test_mysql_test_without_cursor>>
tmpdir_factory = TempdirFactory(_tmppath_factory=TempPathFactory(_given_basetemp=None, _trace=<pluggy._tracing.TagTracerSub object at 0x7f50195ad2e0>, _basetemp=PosixPath('/tmp/pytest-of-ravi/pytest-49')))

    @pytest.fixture(scope='session')
    def mysql_proc_fixture(request, tmpdir_factory):
        """
        Process fixture for MySQL server.
    
        #. Get config.
        #. Initialize MySQL data directory
        #. `Start a mysqld server
            <https://dev.mysql.com/doc/refman/5.0/en/mysqld-safe.html>`_
        #. Stop server and remove directory after tests.
            `See <https://dev.mysql.com/doc/refman/5.6/en/mysqladmin.html>`_
    
        :param FixtureRequest request: fixture request object
        :param tmpdir_factory: pytest fixture for temporary directories
        :rtype: pytest_dbfixtures.executors.TCPExecutor
        :returns: tcp executor
    
        """
        config = get_config(request)
        mysql_mysqld = mysqld_exec or config['mysqld']
        mysql_admin_exec = admin_executable or config['admin']
        mysql_mysqld_safe = mysqld_safe or config['mysqld_safe']
        mysql_port = get_port(port) or get_port(config['port'])
        mysql_host = host or config['host']
        mysql_params = params or config['params']
        mysql_install_db = install_db or config['install_db']
    
        tmpdir = tmpdir_factory.mktemp('pytest-mysql')
    
        logsdir = config['logsdir']
        logfile_path = os.path.join(
            logsdir,
            '{prefix}mysql-server.{port}.log'.format(
                prefix=logs_prefix,
                port=mysql_port
            )
        )
    
        mysql_executor = MySQLExecutor(
            mysqld_safe=mysql_mysqld_safe,
            mysqld=mysql_mysqld,
            admin_exec=mysql_admin_exec,
            logfile_path=logfile_path,
            base_directory=tmpdir,
            params=mysql_params,
            user=config['user'],
            host=mysql_host,
            port=mysql_port,
            install_db=mysql_install_db
        )
        with mysql_executor:
>           yield mysql_executor

../../.pyenv/versions/3.8.2/envs/common-venv-3.8.2/lib/python3.8/site-packages/pytest_mysql/factories/process.py:118: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../.pyenv/versions/common-venv-3.8.2/lib/python3.8/site-packages/mirakuru/base.py:179: in __exit__
    self.stop()
../../.pyenv/versions/3.8.2/envs/common-venv-3.8.2/lib/python3.8/site-packages/pytest_mysql/executor.py:191: in stop
    self.shutdown()
../../.pyenv/versions/3.8.2/envs/common-venv-3.8.2/lib/python3.8/site-packages/pytest_mysql/executor.py:187: in shutdown
    subprocess.check_output(shutdown_command, shell=True)
../../.pyenv/versions/3.8.2/lib/python3.8/subprocess.py:411: in check_output
    return run(*popenargs, stdout=PIPE, timeout=timeout, check=True,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

input = None, capture_output = False, timeout = None, check = True
popenargs = ('/usr/bin/mysqladmin --socket=/tmp/pytest-of-ravi/pytest-49/pytest-mysql1/mysql.3308.sock --user=root shutdown',), kwargs = {'shell': True, 'stdout': -1}
process = <subprocess.Popen object at 0x7f501958a730>, stdout = b'', stderr = None, retcode = 1

    def run(*popenargs,
            input=None, capture_output=False, timeout=None, check=False, **kwargs):
        """Run command with arguments and return a CompletedProcess instance.
    
        The returned instance will have attributes args, returncode, stdout and
        stderr. By default, stdout and stderr are not captured, and those attributes
        will be None. Pass stdout=PIPE and/or stderr=PIPE in order to capture them.
    
        If check is True and the exit code was non-zero, it raises a
        CalledProcessError. The CalledProcessError object will have the return code
        in the returncode attribute, and output & stderr attributes if those streams
        were captured.
    
        If timeout is given, and the process takes too long, a TimeoutExpired
        exception will be raised.
    
        There is an optional argument "input", allowing you to
        pass bytes or a string to the subprocess's stdin.  If you use this argument
        you may not also use the Popen constructor's "stdin" argument, as
        it will be used internally.
    
        By default, all communication is in bytes, and therefore any "input" should
        be bytes, and the stdout and stderr will be bytes. If in text mode, any
        "input" should be a string, and stdout and stderr will be strings decoded
        according to locale encoding, or by "encoding" if set. Text mode is
        triggered by setting any of text, encoding, errors or universal_newlines.
    
        The other arguments are the same as for the Popen constructor.
        """
        if input is not None:
            if kwargs.get('stdin') is not None:
                raise ValueError('stdin and input arguments may not both be used.')
            kwargs['stdin'] = PIPE
    
        if capture_output:
            if kwargs.get('stdout') is not None or kwargs.get('stderr') is not None:
                raise ValueError('stdout and stderr arguments may not be used '
                                 'with capture_output.')
            kwargs['stdout'] = PIPE
            kwargs['stderr'] = PIPE
    
        with Popen(*popenargs, **kwargs) as process:
            try:
                stdout, stderr = process.communicate(input, timeout=timeout)
            except TimeoutExpired as exc:
                process.kill()
                if _mswindows:
                    # Windows accumulates the output in a single blocking
                    # read() call run on child threads, with the timeout
                    # being done in a join() on those threads.  communicate()
                    # _after_ kill() is required to collect that and add it
                    # to the exception.
                    exc.stdout, exc.stderr = process.communicate()
                else:
                    # POSIX _communicate already populated the output so
                    # far into the TimeoutExpired exception.
                    process.wait()
                raise
            except:  # Including KeyboardInterrupt, communicate handled that.
                process.kill()
                # We don't call process.wait() as .__exit__ does that for us.
                raise
            retcode = process.poll()
            if check and retcode:
>               raise CalledProcessError(retcode, process.args,
                                         output=stdout, stderr=stderr)
E               subprocess.CalledProcessError: Command '/usr/bin/mysqladmin --socket=/tmp/pytest-of-ravi/pytest-49/pytest-mysql1/mysql.3308.sock --user=root shutdown' returned non-zero exit status 1.

../../.pyenv/versions/3.8.2/lib/python3.8/subprocess.py:512: CalledProcessError
------------------------------------------------------------------------------------- Captured stderr teardown -------------------------------------------------------------------------------------
mysqladmin: connect to server at 'localhost' failed
error: 'Lost connection to MySQL server at 'reading initial communication packet', system error: 104'
===================================================================================== short test summary info ======================================================================================
ERROR tests/test_mysql.py::test_mysql_test_without_cursor - MySQLdb._exceptions.OperationalError: (2002, "Can't connect to local MySQL server through socket '/tmp/pytest-of-ravi/pytest-49/py...
ERROR tests/test_mysql.py::test_mysql_test_without_cursor - subprocess.CalledProcessError: Command '/usr/bin/mysqladmin --socket=/tmp/pytest-of-ravi/pytest-49/pytest-mysql1/mysql.3308.sock -...
=================================================================================== 1 passed, 2 errors in 31.14s ===================================================================================
(common-venv-3.8.2) /home/ravi/cprog/pytest-mysql>


 ```
You might have to read a bit more: https://dev.mysql.com/doc/refman/5.6/en/error-lost-connection.html

Mind that the mysql isntance set up by pytest-mysql is completely new instance, not related to the one running by default in your system.

Which test passes (use -v paramter to pytest)? Could you run each test separately? using -k 'testname' parameter?
> You might have to read a bit more: https://dev.mysql.com/doc/refman/5.6/en/error-lost-connection.html
> 
> Mind that the mysql isntance set up by pytest-mysql is completely new instance, not related to the one running by default in your system.
> 
> Which test passes (use -v paramter to pytest)? Could you run each test separately? using -k 'testname' parameter?



thanks Grzegorz. I am able to resolve the issue now.however I have a suggestion here, we should have a db table assert feature also as we have variable assert in pytest (assert i == 4)
TableAssert tbl = { table_content.....}
Do we have any such thing ?
I don't think I understand, what you mean.
we use assert operator to compare the current variable value and expected
variable value. if not matches then it fails.

the same way is there a table assert operation which compares the actual
table content with the expected contents.
table assert operation should have features of comparing based on the give
cell/rows/columns values etc

On Mon, Jul 20, 2020 at 2:07 PM Grzegorz Śliwiński <notifications@github.com>
wrote:

> I don't think I understand, what you mean.
>
> —
> You are receiving this because you authored the thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/ClearcodeHQ/pytest-mysql/issues/242#issuecomment-660888189>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AF2MMSO2XJDLDPBLMI244ILR4P64BANCNFSM4OZYYXNA>
> .
>

no, you'll need to fetch the data to the list of dictionaries and prepare list of dictionaries with expected values (and order) yourself
thanks Grzegorz for your help and support.




On Mon, Jul 20, 2020 at 3:19 PM Grzegorz Śliwiński <notifications@github.com>
wrote:

> no, you'll need to fetch the data to the list of dictionaries and prepare
> list of dictionaries with expected values (and order) yourself
>
> —
> You are receiving this because you authored the thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/ClearcodeHQ/pytest-mysql/issues/242#issuecomment-660925282>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AF2MMSLCWEPN5TXAUKTOBTTR4QHJDANCNFSM4OZYYXNA>
> .
>

it'll help if you'd use sets instead of lists. then you'd have to keep track of their order if that's not important.
[![Coverage Status](https://coveralls.io/builds/8710940/badge)](https://coveralls.io/builds/8710940)

Changes Unknown when pulling **0e750b5fbc3ed3a54afa95ef4ab811cbbc9000e2 on fizyk:code_import** into *\* on ClearcodeHQ:master**.

[![Coverage Status](https://coveralls.io/builds/8710940/badge)](https://coveralls.io/builds/8710940)

Changes Unknown when pulling **0e750b5fbc3ed3a54afa95ef4ab811cbbc9000e2 on fizyk:code_import** into *\* on ClearcodeHQ:master**.

[![Coverage Status](https://coveralls.io/builds/8711277/badge)](https://coveralls.io/builds/8711277)

Changes Unknown when pulling **30eb4d764a9dd9e40c3f2d8312660aeec1f30ee9 on fizyk:code_import** into *\* on ClearcodeHQ:master**.

[![Coverage Status](https://coveralls.io/builds/8711414/badge)](https://coveralls.io/builds/8711414)

Changes Unknown when pulling **9f35673e3a2073ee6431247308eba5b844749878 on fizyk:code_import** into *\* on ClearcodeHQ:master**.


Fix patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 41a07fb8..243f3a77 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -7,6 +7,8 @@ CHANGELOG
 Feature
 +++++++
 
+- add `user` option to setup and tear down mysql process as non-privileged
+  user (without root permission).
 - `mysql_noproc` fixture to connect to already running mysql server
 - raise more meaningful error when the test database already exists
 
diff --git a/README.rst b/README.rst
index 7354eed2..c42c40a6 100644
--- a/README.rst
+++ b/README.rst
@@ -48,9 +48,10 @@ You can also create additional mysql client and process fixtures if you'd need t
 .. code-block:: python
 
     from pytest_mysql import factories
+    from getpass import getuser()
 
     mysql_my_proc = factories.mysql_proc(
-        port=None, logsdir='/tmp')
+        port=None, user=getuser())
     mysql_my = factories.mysql('mysql_my_proc')
 
 .. note::


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 83e21e3c..6523baa9 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -48,10 +48,10 @@ jobs:
         pip install -r requirements-test.txt
     - name: Run test
       run: |
-        pytest -n 0 -k "not mysqlnoproc" --cov-report=xml
+        pytest -n 0 -k "not mysqlnoproc" --cov-report=xml --mysql-user=root
     - name: Run xdist test
       run: |
-        pytest -n 1 -k "not mysqlnoproc" --cov-report=xml:coverage-xdist.xml
+        pytest -n 1 -k "not mysqlnoproc" --cov-report=xml:coverage-xdist.xml --mysql-user=root
     - name: Run noproc test
       run: |
         pytest -n 0 -k mysqlnoproc --cov-report=xml:coverage-noproc.xml --mysql-host="127.0.0.1" --mysql-port=3333
diff --git a/src/pytest_mysql/executor.py b/src/pytest_mysql/executor.py
index acd594c7..b8f9fad8 100644
--- a/src/pytest_mysql/executor.py
+++ b/src/pytest_mysql/executor.py
@@ -152,9 +152,18 @@ def start(self):
     def shutdown(self):
         """Send shutdown command to the server."""
         shutdown_command = (
-            f"{self.admin_exec} --socket={self.unixsocket} --user=root shutdown"
+            f"{self.admin_exec} --socket={self.unixsocket} "
+            f"--user={self.user} shutdown"
         )
-        subprocess.check_output(shutdown_command, shell=True)
+        try:
+            subprocess.check_output(shutdown_command, shell=True)
+        except subprocess.CalledProcessError:
+            # Fallback to using root user for shutdown
+            shutdown_command = (
+                f"{self.admin_exec} --socket={self.unixsocket} "
+                f"--user=root shutdown"
+            )
+            subprocess.check_output(shutdown_command, shell=True)
 
     def stop(self, sig=None, exp_sig=None):
         """Stop the server."""
diff --git a/src/pytest_mysql/factories/client.py b/src/pytest_mysql/factories/client.py
index 0067a171..2ea2a147 100644
--- a/src/pytest_mysql/factories/client.py
+++ b/src/pytest_mysql/factories/client.py
@@ -20,7 +20,7 @@
 
 import pytest
 import MySQLdb
-from MySQLdb import ProgrammingError
+from MySQLdb import ProgrammingError, OperationalError
 from _pytest.fixtures import FixtureRequest
 
 from pytest_mysql.config import get_config
@@ -58,6 +58,23 @@ def mysql(
     :rtype: func
     """
 
+    def _connect(connect_kwargs: dict, query_str: str) -> MySQLdb.Connection:
+        """Apply given query to a  given MySQLdb connection."""
+        mysql_conn: MySQLdb.Connection = MySQLdb.connect(**connect_kwargs)
+        try:
+            mysql_conn.query(query_str)
+        except ProgrammingError as e:
+            if "database exists" in str(e):
+                raise DatabaseExists(
+                    f"Database {mysql_db} already exists. There's some test "
+                    f"configuration error. Either you start your own server "
+                    f"with the database name used in tests, or you use two "
+                    f"fixtures with the same database name on the same "
+                    f"process fixture."
+                ) from e
+            raise
+        return mysql_conn
+
     @pytest.fixture
     def mysql_fixture(request: FixtureRequest) -> MySQLdb.Connection:
         """
@@ -81,7 +98,7 @@ def mysql_fixture(request: FixtureRequest) -> MySQLdb.Connection:
         if not process.running():
             process.start()
 
-        mysql_user = "root"
+        mysql_user = process.user
         mysql_passwd = passwd or config["passwd"]
         mysql_db = dbname or config["dbname"]
 
@@ -95,26 +112,22 @@ def mysql_fixture(request: FixtureRequest) -> MySQLdb.Connection:
         else:
             connection_kwargs["port"] = process.port
 
-        mysql_conn: MySQLdb.Connection = MySQLdb.connect(**connection_kwargs)
-
+        query_str = (
+            f"CREATE DATABASE {mysql_db} "
+            f"DEFAULT CHARACTER SET {charset} "
+            f"DEFAULT COLLATE {collation}"
+        )
         try:
-            mysql_conn.query(
-                f"CREATE DATABASE {mysql_db} "
-                f"DEFAULT CHARACTER SET {charset} "
-                f"DEFAULT COLLATE {collation}"
+            mysql_conn: MySQLdb.Connection = _connect(
+                connection_kwargs, query_str
             )
-        except ProgrammingError as e:
-            if "database exists" in str(e):
-                raise DatabaseExists(
-                    f"Database {mysql_db} already exists. There's some test "
-                    f"configuration error. Either you start your own server "
-                    f"with the database name used in tests, or you use two "
-                    f"fixtures with the same database name on the same process "
-                    f"fixture."
-                ) from e
-            raise
-        mysql_conn.query("USE %s" % mysql_db)
-
+        except OperationalError:
+            # Fallback to mysql connection with root user
+            connection_kwargs["user"] = "root"
+            mysql_conn: MySQLdb.Connection = _connect(
+                connection_kwargs, query_str
+            )
+        mysql_conn.query(f"USE {mysql_db}")
         yield mysql_conn
 
         # clean up after test that forgot to fetch selected data
diff --git a/src/pytest_mysql/factories/noprocess.py b/src/pytest_mysql/factories/noprocess.py
index 0ebb3522..6526f795 100644
--- a/src/pytest_mysql/factories/noprocess.py
+++ b/src/pytest_mysql/factories/noprocess.py
@@ -27,12 +27,13 @@
 from pytest_mysql.executor_noop import NoopMySQLExecutor
 
 
-def mysql_noproc(host=None, port=None):
+def mysql_noproc(host=None, port=None, user=None):
     """
     Process fixture factory for MySQL server.
 
     :param str host: hostname
-    :param int port:
+    :param int port: port name
+    :param str user: user name
     :rtype: func
     :returns: function which makes a redis process
 
@@ -53,9 +54,10 @@ def mysql_noproc_fixture(request: FixtureRequest):
         config = get_config(request)
         mysql_port = int(port or config["port"] or 3306)
         mysql_host = host or config["host"]
+        mysql_user = user or config["user"] or "root"
 
         mysql_executor = NoopMySQLExecutor(
-            user=config["user"],
+            user=mysql_user,
             host=mysql_host,
             port=mysql_port,
         )
diff --git a/src/pytest_mysql/factories/process.py b/src/pytest_mysql/factories/process.py
index ddb70679..56270ce9 100644
--- a/src/pytest_mysql/factories/process.py
+++ b/src/pytest_mysql/factories/process.py
@@ -34,6 +34,7 @@ def mysql_proc(
     admin_executable=None,
     mysqld_safe=None,
     host=None,
+    user=None,
     port=-1,
     params=None,
     logs_prefix="",
@@ -46,6 +47,7 @@ def mysql_proc(
     :param str admin_executable: path to mysql_admin executable
     :param str mysqld_safe: path to mysqld_safe executable
     :param str host: hostname
+    :param str user: user name
     :param str|int|tuple|set|list port:
         exact port (e.g. '8000', 8000)
         randomly selected port (None) - any random available port
@@ -123,7 +125,7 @@ def mysql_proc_fixture(
             logfile_path=logfile_path,
             base_directory=tmpdir,
             params=mysql_params,
-            user=config["user"],
+            user=user or config["user"] or "root",
             host=mysql_host,
             port=mysql_port,
             install_db=mysql_install_db,