test_sqlitedb.py

# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt

"""Tests for coverage.sqlitedb"""

from __future__ import annotations

from typing import NoReturn
from unittest import mock

import pytest

import coverage.sqlitedb
from coverage.exceptions import DataError
from coverage.sqlitedb import SqliteDb

from tests.coveragetest import CoverageTest
from tests.helpers import DebugControlString, FailingProxy

DB_INIT = """\
create table name (first text, last text);
insert into name (first, last) values ("pablo", "picasso");
"""

class SqliteDbTest(CoverageTest):
    """Tests of tricky parts of SqliteDb."""

    def test_error_reporting(self) -> None:
        msg = "Couldn't use data file 'test.db': no such table: bar"
        with SqliteDb("test.db", DebugControlString(options=["sql"])) as db:
            with pytest.raises(DataError, match=msg):
                with db.execute("select foo from bar"):
                    # Entering the context manager raises the error, this line doesn't run:
                    pass    # pragma: not covered

    def test_retry_execute(self) -> None:
        with SqliteDb("test.db", DebugControlString(options=["sql"])) as db:
            db.executescript(DB_INIT)
            proxy = FailingProxy(db.con, "execute", [Exception("WUT")])
            with mock.patch.object(db, "con", proxy):
                with db.execute("select first from name order by 1") as cur:
                    assert list(cur) == [("pablo",)]

    def test_retry_execute_failure(self) -> None:
        with SqliteDb("test.db", DebugControlString(options=["sql"])) as db:
            db.executescript(DB_INIT)
            proxy = FailingProxy(db.con, "execute", [Exception("WUT"), RuntimeError("Fake")])
            with mock.patch.object(db, "con", proxy):
                with pytest.raises(RuntimeError, match="Fake"):
                    with db.execute("select first from name order by 1"):
                        # Entering the context manager raises the error, this line doesn't run:
                        pass    # pragma: not covered

    def test_retry_executemany_void(self) -> None:
        with SqliteDb("test.db", DebugControlString(options=["sql"])) as db:
            db.executescript(DB_INIT)
            proxy = FailingProxy(db.con, "executemany", [Exception("WUT")])
            with mock.patch.object(db, "con", proxy):
                db.executemany_void(
                    "insert into name (first, last) values (?, ?)",
                    [("vincent", "van gogh")],
                )
            with db.execute("select first from name order by 1") as cur:
                assert list(cur) == [("pablo",), ("vincent",)]

    def test_retry_executemany_void_failure(self) -> None:
        with SqliteDb("test.db", DebugControlString(options=["sql"])) as db:
            db.executescript(DB_INIT)
            proxy = FailingProxy(db.con, "executemany", [Exception("WUT"), RuntimeError("Fake")])
            with mock.patch.object(db, "con", proxy):
                with pytest.raises(RuntimeError, match="Fake"):
                    db.executemany_void(
                        "insert into name (first, last) values (?, ?)",
                        [("vincent", "van gogh")],
                    )

    def test_open_fails_on_bad_db(self) -> None:
        self.make_file("bad.db", "boogers")
        def fake_failing_open(filename: str, mode: str) -> NoReturn:
            assert (filename, mode) == ("bad.db", "rb")
            raise RuntimeError("No you can't!")
        with mock.patch.object(coverage.sqlitedb, "open", fake_failing_open):
            msg = "Couldn't use data file 'bad.db': file is not a database"
            with pytest.raises(DataError, match=msg):
                with SqliteDb("bad.db", DebugControlString(options=["sql"])):
                    pass    # pragma: not covered

    def test_execute_void_can_allow_failure(self) -> None:
        with SqliteDb("fail.db", DebugControlString(options=["sql"])) as db:
            db.executescript(DB_INIT)
            proxy = FailingProxy(db.con, "execute", [Exception("WUT")])
            with mock.patch.object(db, "con", proxy):
                db.execute_void("select x from nosuchtable", fail_ok=True)

    def test_execute_void_can_refuse_failure(self) -> None:
        with SqliteDb("fail.db", DebugControlString(options=["sql"])) as db:
            db.executescript(DB_INIT)
            proxy = FailingProxy(db.con, "execute", [Exception("WUT")])
            with mock.patch.object(db, "con", proxy):
                msg = "Couldn't use data file 'fail.db': no such table: nosuchtable"
                with pytest.raises(DataError, match=msg):
                    db.execute_void("select x from nosuchtable", fail_ok=False)