wrolpi/wrolpi/test/test_db_bootstrap.py

"""Tests for wrolpi.db_bootstrap: DB auto-creation, migration, and the drive-portability guard."""
import sqlite3

import mock
import pytest

from wrolpi import db_bootstrap
from wrolpi.db import get_db_file


@pytest.fixture(autouse=True)
def permissive_sqlite_floor():
    """CI images may ship a SQLite older than the production floor (Ubuntu 22.04 has 3.37; the
    floor targets Debian 12's 3.40).  The functional tests here exercise creation/migration, not
    the version gate; the gate itself is tested explicitly with mocked versions below."""
    with mock.patch.object(db_bootstrap, 'MINIMUM_SQLITE_VERSION', (3, 0, 0)):
        yield


def test_migration_keeps_logging(test_directory):
    """WROLPi migrates in-process at startup; alembic must not reconfigure the API's logging."""
    import logging

    existing = logging.getLogger('wrolpi.test.existing_logger')
    root_handlers = list(logging.getLogger().handlers)
    root_level = logging.getLogger().level

    assert db_bootstrap.ensure_db() is True

    assert existing.disabled is False
    assert logging.getLogger().handlers == root_handlers
    assert logging.getLogger().level == root_level


def test_ensure_db_creates_and_migrates(test_directory):
    """With no database file, ensure_db creates it and migrates to head."""
    db_file = get_db_file()
    assert not db_file.is_file()
    assert db_bootstrap.compare_db_version() == 'missing'

    assert db_bootstrap.ensure_db() is True

    assert db_file.is_file()
    assert db_bootstrap.compare_db_version() == 'current'
    with sqlite3.connect(db_file) as conn:
        version = conn.execute('SELECT version_num FROM alembic_version').fetchone()[0]
        assert version
        # Triggers and FTS5 tables were installed by the baseline.
        triggers = conn.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='trigger'").fetchone()[0]
        assert triggers > 0
        fts_tables = {i[0] for i in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('file_group_fts', 'doc_section_fts')")}
        assert fts_tables == {'file_group_fts', 'doc_section_fts'}
        file_group_cols = {r[1] for r in conn.execute('PRAGMA table_info(file_group)')}
        assert 'idempotency' not in file_group_cols
        assert 'stem' in file_group_cols

    # Running again is a no-op.
    assert db_bootstrap.ensure_db() is True


def test_ensure_db_refuses_newer_database(test_directory):
    """A database from a NEWER WROLPi (unknown alembic revision) must not be written."""
    assert db_bootstrap.ensure_db() is True
    db_file = get_db_file()
    with sqlite3.connect(db_file) as conn:
        conn.execute("UPDATE alembic_version SET version_num = 'future_revision_xyz'")
        conn.commit()

    assert db_bootstrap.compare_db_version() == 'ahead'
    assert db_bootstrap.ensure_db() is False
    # The database was not modified.
    with sqlite3.connect(db_file) as conn:
        assert conn.execute('SELECT version_num FROM alembic_version').fetchone()[0] == 'future_revision_xyz'


def test_compare_db_version_uninitialized_file(test_directory):
    """An empty/uninitialized file is treated like a fresh database."""
    db_file = get_db_file()
    db_file.parent.mkdir(parents=True, exist_ok=True)
    db_file.touch()
    assert db_bootstrap.compare_db_version() == 'missing'
    assert db_bootstrap.ensure_db() is True
    assert db_bootstrap.compare_db_version() == 'current'


def test_unique_stem_migration_collapses_duplicates(test_directory):
    """The unique-index migration merges duplicate (directory, stem) rows and keeps tags."""
    import sqlite3
    from alembic.config import Config
    from alembic import command
    from wrolpi.db import get_db_uri, get_db_file
    from wrolpi.vars import PROJECT_DIR

    (test_directory / 'config').mkdir(parents=True, exist_ok=True)
    config = Config(str(PROJECT_DIR / 'alembic.ini'))
    config.set_main_option('sqlalchemy.url', get_db_uri())
    command.upgrade(config, '2026_09_16_1200')

    db_file = get_db_file()
    directory = str(test_directory)
    with sqlite3.connect(db_file) as conn:
        conn.execute('PRAGMA foreign_keys=ON')
        conn.execute(
            'INSERT INTO file_group (directory, primary_path, files, stem, indexed) VALUES (?,?,?,?,0)',
            (directory, str(test_directory / 'dup.mp4'), '[]', 'dup'),
        )
        conn.execute(
            'INSERT INTO file_group (directory, primary_path, files, stem, indexed) VALUES (?,?,?,?,0)',
            (directory, str(test_directory / 'dup.info.json'), '[]', 'dup'),
        )
        first_id, second_id = [r[0] for r in conn.execute('SELECT id FROM file_group ORDER BY id').fetchall()]
        conn.execute("INSERT INTO tag (name, color) VALUES ('migrated', '#111111')")
        tag_id = conn.execute('SELECT id FROM tag').fetchone()[0]
        # Tag the lower-id row so it wins over the later untagged duplicate.
        conn.execute(
            'INSERT INTO tag_file (tag_id, file_group_id) VALUES (?, ?)',
            (tag_id, first_id),
        )
        conn.commit()
        assert conn.execute('SELECT COUNT(*) FROM file_group').fetchone()[0] == 2
        assert first_id < second_id

    command.upgrade(config, 'head')

    with sqlite3.connect(db_file) as conn:
        rows = list(conn.execute('SELECT id, stem FROM file_group'))
        assert len(rows) == 1
        assert rows[0][0] == first_id
        tagged = conn.execute('SELECT file_group_id FROM tag_file').fetchone()[0]
        assert tagged == first_id
        sql = conn.execute(
            "SELECT sql FROM sqlite_master WHERE type='index' AND name='file_group_directory_stem_idx'"
        ).fetchone()[0]
        assert 'UNIQUE' in sql.upper()


def test_drop_file_group_idempotency_preserves_fts(test_directory):
    """DROP COLUMN idempotency must not rebuild file_group or break file_group_fts."""
    from alembic.config import Config
    from alembic import command
    from wrolpi.db import get_db_uri
    from wrolpi.vars import PROJECT_DIR

    (test_directory / 'config').mkdir(parents=True, exist_ok=True)
    config = Config(str(PROJECT_DIR / 'alembic.ini'))
    config.set_main_option('sqlalchemy.url', get_db_uri())
    command.upgrade(config, '2026_09_18_1200')

    db_file = get_db_file()
    with sqlite3.connect(db_file) as conn:
        cols = {r[1] for r in conn.execute('PRAGMA table_info(file_group)')}
        assert 'idempotency' in cols
        conn.execute(
            'INSERT INTO file_group (directory, primary_path, files, stem, indexed, a_text) '
            'VALUES (?,?,?,?,0,?)',
            (str(test_directory), str(test_directory / 'searchable.txt'), '[]', 'searchable', 'alpha uniquephrase'),
        )
        conn.commit()
        hits = conn.execute(
            "SELECT rowid FROM file_group_fts WHERE file_group_fts MATCH 'uniquephrase'"
        ).fetchall()
        assert len(hits) == 1

    command.upgrade(config, 'head')

    with sqlite3.connect(db_file) as conn:
        cols = {r[1] for r in conn.execute('PRAGMA table_info(file_group)')}
        assert 'idempotency' not in cols
        hits = conn.execute(
            "SELECT rowid FROM file_group_fts WHERE file_group_fts MATCH 'uniquephrase'"
        ).fetchall()
        assert len(hits) == 1


def test_file_group_progress_downgrade_preserves_fts(test_directory):
    """Dropping file_group.progress/position must not rebuild file_group, which would drop its FTS triggers."""
    from alembic.config import Config
    from alembic import command
    from wrolpi.db import get_db_uri
    from wrolpi.vars import PROJECT_DIR

    (test_directory / 'config').mkdir(parents=True, exist_ok=True)
    config = Config(str(PROJECT_DIR / 'alembic.ini'))
    config.set_main_option('sqlalchemy.url', get_db_uri())
    command.upgrade(config, '2026_10_08_1200')
    command.downgrade(config, '2026_10_06_1500')

    with sqlite3.connect(get_db_file()) as conn:
        cols = {r[1] for r in conn.execute('PRAGMA table_info(file_group)')}
        assert not {'progress', 'position'} & cols
        triggers = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='trigger'")}
        assert {'file_group_fts_ai', 'file_group_fts_ad', 'file_group_fts_au'} <= triggers
        conn.execute(
            'INSERT INTO file_group (directory, primary_path, files, stem, indexed, a_text) VALUES (?,?,?,?,0,?)',
            (str(test_directory), str(test_directory / 'searchable.txt'), '[]', 'searchable', 'alpha uniquephrase'),
        )
        hits = conn.execute("SELECT rowid FROM file_group_fts WHERE file_group_fts MATCH 'uniquephrase'").fetchall()
        assert len(hits) == 1


def test_domain_collection_summary_migration_backfills(test_directory):
    """The domain-summary migration installs the archive triggers and backfills the summary
    columns of existing Domain Collections from their Archives."""
    import sqlite3
    from alembic.config import Config
    from alembic import command
    from wrolpi.db import get_db_uri, get_db_file
    from wrolpi.vars import PROJECT_DIR

    (test_directory / 'config').mkdir(parents=True, exist_ok=True)
    config = Config(str(PROJECT_DIR / 'alembic.ini'))
    config.set_main_option('sqlalchemy.url', get_db_uri())
    command.upgrade(config, '2026_09_20_1200')

    db_file = get_db_file()
    directory = str(test_directory)
    triggers = {'archive_insert_collection_summary', 'archive_delete_collection_summary',
                'archive_update_collection_summary', 'file_group_size_collection_summary'}
    with sqlite3.connect(db_file) as conn:
        conn.execute('PRAGMA foreign_keys=ON')
        # The baseline installs the current trigger DDL, so a fresh DB already has the archive
        # triggers.  A production DB created before them does not; emulate that.
        for name in triggers:
            conn.execute(f'DROP TRIGGER {name}')
        for id_, name, kind in ((1, 'example.com', 'domain'), (2, 'empty.org', 'domain'), (3, 'chan', 'channel')):
            conn.execute('INSERT INTO collection (id, name, kind, item_count, total_size) VALUES (?, ?, ?, 0, 0)',
                         (id_, name, kind))
        for fg_id, stem, size in ((10, 'one', 100), (11, 'two', 250)):
            conn.execute(
                'INSERT INTO file_group (id, directory, primary_path, files, stem, size, indexed)'
                ' VALUES (?,?,?,?,?,?,0)',
                (fg_id, directory, str(test_directory / f'{stem}.html'), '[]', stem, size),
            )
            conn.execute('INSERT INTO archive (collection_id, file_group_id) VALUES (1, ?)', (fg_id,))
        conn.commit()
        # Nothing maintains the columns before the migration.
        assert conn.execute('SELECT item_count, total_size FROM collection WHERE id = 1').fetchone() == (0, 0)

    command.upgrade(config, 'head')

    with sqlite3.connect(db_file) as conn:
        assert conn.execute('SELECT item_count, total_size FROM collection WHERE id = 1').fetchone() == (2, 350)
        assert conn.execute('SELECT item_count, total_size FROM collection WHERE id = 2').fetchone() == (0, 0)
        assert conn.execute('SELECT item_count, total_size FROM collection WHERE id = 3').fetchone() == (0, 0)
        installed = {r[0] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='trigger' AND name LIKE '%collection_summary'")}
        assert installed == triggers
        # The triggers are live after the migration.
        conn.execute('DELETE FROM archive WHERE file_group_id = 11')
        conn.commit()
        assert conn.execute('SELECT item_count, total_size FROM collection WHERE id = 1').fetchone() == (1, 100)


def test_markdown_mimetype_migration_retags_existing_rows(test_directory):
    """Markdown indexed before text/markdown existed is retagged and re-indexed.  A refresh skips a
    group whose files have not changed, so it would otherwise keep its old mimetype and indexer."""
    import sqlite3
    from alembic.config import Config
    from alembic import command
    from wrolpi.db import get_db_uri, get_db_file
    from wrolpi.vars import PROJECT_DIR

    (test_directory / 'config').mkdir(parents=True, exist_ok=True)
    config = Config(str(PROJECT_DIR / 'alembic.ini'))
    config.set_main_option('sqlalchemy.url', get_db_uri())
    command.upgrade(config, '2026_09_28_1200')

    rows = (
        (1, 'notes.md', 'text/plain'),
        # libmagic reports a programming language when a markdown file holds code.
        (2, 'code.markdown', 'text/x-c'),
        (3, 'UPPER.MD', 'text/html'),
        (4, 'already.md', 'text/markdown'),
        (5, 'readme.txt', 'text/plain'),
        # Not text, so not markdown, whatever its name.
        (6, 'binary.md', 'application/octet-stream'),
    )
    db_file = get_db_file()
    with sqlite3.connect(db_file) as conn:
        for id_, name, mimetype in rows:
            stem = name.rsplit('.', 1)[0]
            conn.execute(
                'INSERT INTO file_group (id, directory, primary_path, files, stem, mimetype, indexed)'
                ' VALUES (?,?,?,?,?,?,1)',
                (id_, str(test_directory), str(test_directory / name), '[]', stem, mimetype),
            )
        conn.commit()

    command.upgrade(config, 'head')

    with sqlite3.connect(db_file) as conn:
        result = {r[0]: (r[1], r[2]) for r in conn.execute('SELECT id, mimetype, indexed FROM file_group')}
    assert result == {
        1: ('text/markdown', 0),
        2: ('text/markdown', 0),
        3: ('text/markdown', 0),
        # Already correct, so not re-indexed.
        4: ('text/markdown', 1),
        5: ('text/plain', 1),
        6: ('application/octet-stream', 1),
    }


def test_strip_tag_names_migration_merges_tags(test_directory):
    """Whitespace is stripped from Tag names; Tags with the same stripped name are merged (keeping the first color,
    by id) in the database and in the configs which name them."""
    import json
    import sqlite3
    import yaml
    from alembic.config import Config
    from alembic import command
    from wrolpi.db import get_db_uri, get_db_file
    from wrolpi.vars import PROJECT_DIR

    config_directory = test_directory / 'config'
    config_directory.mkdir(parents=True, exist_ok=True)
    config = Config(str(PROJECT_DIR / 'alembic.ini'))
    config.set_main_option('sqlalchemy.url', get_db_uri())
    command.upgrade(config, '2026_10_24_1200')

    db_file = get_db_file()
    with sqlite3.connect(db_file) as conn:
        conn.executemany('INSERT INTO tag (id, name, color) VALUES (?,?,?)', (
            # Merged: the first has no color, so the second's is kept.
            (1, ' one', None), (2, 'one ', '#222222'), (3, 'one', '#333333'),
            # Merged: both have colors, so the first is kept.
            (4, 'two\t', '#444444'), (5, ' two ', '#555555'),
            (6, 'three ', None),
            (7, 'unchanged', '#777777'),
        ))
        for id_ in (10, 11):
            conn.execute('INSERT INTO file_group (id, directory, primary_path, files, stem, indexed)'
                         ' VALUES (?,?,?,?,?,0)',
                         (id_, str(test_directory), str(test_directory / f'{id_}.mp4'), '[]', str(id_)))
        conn.executemany('INSERT INTO tag_file (tag_id, file_group_id) VALUES (?,?)',
                         ((1, 10), (2, 10), (3, 11), (5, 11), (6, 10), (7, 10)))
        conn.execute("INSERT INTO collection (id, name, kind, tag_id, item_count, total_size) VALUES (20, 'channel', 'channel', 1, 0, 0)")
        conn.execute('INSERT INTO download (url, tag_names, settings) VALUES (?,?,?)', (
            'https://example.com/feed', json.dumps(['one ', 'one', 'three ']),
            json.dumps({'channel_tag_name': [' two'], 'video_resolutions': ['720p']})))
        conn.commit()

    (config_directory / 'tags.yaml').write_text(yaml.dump(dict(
        # Sorted, ' two ' comes before 'two\t'; the database's color (by id) is kept anyway.
        tags={' one': dict(color=None), 'one ': dict(color='#222222'), ' two ': dict(color='#555555'),
              'two\t': dict(color='#444444'), 'unchanged': dict(color='#777777')},
        tag_files=[[' one', '10.mp4', None], ['one ', '10.mp4', None], ['unchanged', '10.mp4', None]],
        tag_zims=[],
        version=3,
    )))
    (config_directory / 'channels.yaml').write_text(yaml.dump(dict(
        channels=[dict(name='channel', directory='videos/channel', tag_name=' one')], version=2)))
    # A config which names no padded Tag is not rewritten.
    domains = 'domains: []\nversion: 1\n'
    (config_directory / 'domains.yaml').write_text(domains)

    command.upgrade(config, 'head')

    with sqlite3.connect(db_file) as conn:
        assert sorted(conn.execute('SELECT id, name, color FROM tag')) == [
            (3, 'one', '#222222'), (4, 'two', '#444444'), (6, 'three', None), (7, 'unchanged', '#777777')]
        assert sorted(conn.execute('SELECT tag_id, file_group_id FROM tag_file')) == [
            (3, 10), (3, 11), (4, 11), (6, 10), (7, 10)]
        assert conn.execute('SELECT tag_id FROM collection').fetchone()[0] == 3
        tag_names, settings = conn.execute('SELECT tag_names, settings FROM download').fetchone()
        assert json.loads(tag_names) == ['one', 'three']
        assert json.loads(settings) == {'channel_tag_name': ['two'], 'video_resolutions': ['720p']}

    tags_config = yaml.safe_load((config_directory / 'tags.yaml').read_text())
    assert tags_config['tags'] == {'one': dict(color='#222222'), 'two': dict(color='#444444'),
                                   'unchanged': dict(color='#777777')}
    assert tags_config['tag_files'] == [['one', '10.mp4', None], ['unchanged', '10.mp4', None]]
    assert tags_config['version'] == 3
    channels_config = yaml.safe_load((config_directory / 'channels.yaml').read_text())
    assert channels_config['channels'][0]['tag_name'] == 'one'
    assert (config_directory / 'domains.yaml').read_text() == domains


def test_migrations_match_models(test_directory):
    """`alembic check`: a database migrated to head matches the SQLAlchemy models (tables, columns,
    indexes).  Fails when a model gains an index or column without a migration, or vice versa."""
    from alembic.autogenerate import compare_metadata
    from alembic.config import Config
    from alembic import command
    from alembic.migration import MigrationContext
    from wrolpi.common import Base
    from wrolpi.db import get_db_uri, create_wrolpi_engine
    from wrolpi.vars import PROJECT_DIR

    (test_directory / 'config').mkdir(parents=True, exist_ok=True)
    config = Config(str(PROJECT_DIR / 'alembic.ini'))
    config.set_main_option('sqlalchemy.url', get_db_uri())
    command.upgrade(config, 'head')

    # Compare with alembic/env.py's own exclusions (FTS tables), plus any table a test module
    # registered on the shared metadata (wrolpi/test/test_dates.py defines one).
    env_source = (PROJECT_DIR / 'alembic' / 'env.py').read_text()
    namespace = {}
    exec(env_source[env_source.index('def include_object'):env_source.index('def process_revision_directives')],
         namespace)
    model_tables = {name for name in Base.metadata.tables if not name.startswith('test_')}

    def include_object(object_, name, type_, reflected, compare_to):
        if type_ == 'table' and name not in model_tables and not reflected:
            return False
        return namespace['include_object'](object_, name, type_, reflected, compare_to)

    engine = create_wrolpi_engine(get_db_uri())
    with engine.connect() as conn:
        context = MigrationContext.configure(conn, opts={'include_object': include_object, 'render_as_batch': True})
        diffs = compare_metadata(context, Base.metadata)
    assert not diffs, diffs


def test_bootstrap_lock(test_directory):
    """Only one process can hold the bootstrap lock."""
    with db_bootstrap.bootstrap_lock() as acquired:
        assert acquired is True
        with db_bootstrap.bootstrap_lock() as second:
            # flock is per-file-descriptor, so a second open in the same process CAN acquire it;
            # this just proves the context manager doesn't error or deadlock.
            assert second in (True, False)


def test_check_sqlite_environment_version_gate(test_directory):
    """SQLite older than the floor is refused; at/above the floor is accepted.

    Mocked versions so the test is deterministic regardless of the machine's SQLite (CI images
    have shipped 3.37 while the production floor is 3.40)."""
    with mock.patch.object(db_bootstrap, 'MINIMUM_SQLITE_VERSION', (3, 40, 0)):
        with mock.patch.object(db_bootstrap.sqlite3, 'sqlite_version_info', (3, 39, 9)), \
                mock.patch.object(db_bootstrap.sqlite3, 'sqlite_version', '3.39.9'):
            error = db_bootstrap.check_sqlite_environment(get_db_file())
            assert error and 'too old' in error
        with mock.patch.object(db_bootstrap.sqlite3, 'sqlite_version_info', (3, 40, 0)):
            assert db_bootstrap.check_sqlite_environment(get_db_file()) is None
        with mock.patch.object(db_bootstrap.sqlite3, 'sqlite_version_info', (3, 46, 1)):
            assert db_bootstrap.check_sqlite_environment(get_db_file()) is None


@pytest.mark.parametrize('fs_type', ['vfat', 'exfat', 'msdos', 'ntfs', 'ntfs3'])
def test_check_sqlite_environment_allows_wal_incompatible_fs_with_warning(test_directory, fs_type):
    """A FAT/exFAT/NTFS drive on a native kernel driver is allowed (with a warning), not refused.

    These filesystems cannot support SQLite WAL — it needs mmap shared-memory they don't provide,
    so `PRAGMA journal_mode=WAL` raises `(sqlite3.OperationalError) disk I/O error` (the failure
    seen on 10.0.0.8; the tell-tale sign was `Media directory has the wrong permissions: 0o40777`,
    since these filesystems can't store Unix permission bits).  Rather than crash the whole API,
    the engine degrades to a rollback journal, so the environment check must NOT refuse the drive
    — it only warns that performance is reduced."""
    with mock.patch.object(db_bootstrap, '_media_fs_type', return_value=fs_type), \
            mock.patch.object(db_bootstrap.logger, 'warning') as mock_warning:
        assert db_bootstrap.check_sqlite_environment(get_db_file()) is None
    assert mock_warning.call_count == 1
    assert 'WAL is unavailable' in mock_warning.call_args.args[0], 'expected a WAL-unavailable warning'


@pytest.mark.parametrize('fs_type', ['ext4', 'btrfs', 'xfs', 'f2fs'])
def test_check_sqlite_environment_allows_wal_compatible_fs(test_directory, fs_type):
    """Ordinary Linux filesystems support WAL and must not be refused."""
    with mock.patch.object(db_bootstrap, '_media_fs_type', return_value=fs_type):
        assert db_bootstrap.check_sqlite_environment(get_db_file()) is None


@pytest.mark.parametrize('fs_type', ['nfs', 'nfs4', 'cifs', 'fuse.sshfs', 'fuseblk'])
def test_check_sqlite_environment_refuses_unsafe_locking_fs(test_directory, fs_type):
    """Filesystems with unreliable locking are refused: no journal mode makes SQLite safe there.

    Includes `fuseblk` (block-backed FUSE, e.g. ntfs-3g), whose byte-range locking is unreliable
    under the workers' concurrent access.  NTFS on a modern kernel mounts as native `ntfs3` instead
    and is allowed (with the WAL-unavailable warning)."""
    with mock.patch.object(db_bootstrap, '_media_fs_type', return_value=fs_type):
        error = db_bootstrap.check_sqlite_environment(get_db_file())
    assert error and fs_type in error


def test_unmounted_production_guard(test_directory):
    """An unmounted production media directory blocks DB creation/migration and db_up.

    Regression test for the shadow-database bug found on 10.0.0.8: after the media drive was
    unmounted, a worker's ORM connection created an empty wrolpi.db on the root filesystem and
    the bootstrap migrated it — because the old guard only refused when NO file existed."""
    import pathlib

    fake_media = mock.MagicMock(spec=pathlib.Path)
    fake_media.__str__ = lambda self: '/media/wrolpi'
    # Files beneath it (config/controller.yaml) come from the real, marker-less test directory.
    fake_media.__truediv__ = lambda self, other: test_directory / other
    fake_media.is_mount.return_value = False
    fake_media.is_dir.return_value = True

    with mock.patch.object(db_bootstrap, 'PYTEST', False), \
            mock.patch.object(db_bootstrap, 'DOCKERIZED', False), \
            mock.patch.object(db_bootstrap.sys, 'platform', 'linux'):
        # Unmounted /media/* on a production linux host: guarded.
        assert db_bootstrap.media_directory_is_unmounted_production(fake_media) is True
        # Mounted: fine.
        fake_media.is_mount.return_value = True
        assert db_bootstrap.media_directory_is_unmounted_production(fake_media) is False
        # Custom (non-/media) directories are exempt (dev boxes).
        fake_media.is_mount.return_value = False
        fake_media.__str__ = lambda self: '/home/user/media'
        assert db_bootstrap.media_directory_is_unmounted_production(fake_media) is False

    # The guard must refuse ensure_db EVEN IF a database file already exists (an ORM connection
    # creates an empty file implicitly, so file-existence cannot be the condition).
    assert db_bootstrap.ensure_db() is True  # create a real DB in the test directory first
    with mock.patch.object(db_bootstrap, 'media_directory_is_unmounted_production', return_value=True):
        assert db_bootstrap.ensure_db() is False


def test_unmounted_production_guard_library_on_internal_disk(test_directory):
    """A library the user chose to keep on the internal disk ("Run repair without external drive") is not an
    unmounted drive: the Controller's marker in config/controller.yaml lifts the guard."""
    import pathlib

    config_dir = test_directory / 'config'
    config_dir.mkdir(exist_ok=True)
    fake_media = mock.MagicMock(spec=pathlib.Path)
    fake_media.__str__ = lambda self: '/media/wrolpi'
    fake_media.__truediv__ = lambda self, other: test_directory / other
    fake_media.is_mount.return_value = False

    with mock.patch.object(db_bootstrap, 'PYTEST', False), \
            mock.patch.object(db_bootstrap, 'DOCKERIZED', False), \
            mock.patch.object(db_bootstrap.sys, 'platform', 'linux'):
        # A controller.yaml without the marker (a drive that failed to mount) stays guarded.
        (config_dir / 'controller.yaml').write_text('{}\n')
        assert db_bootstrap.media_directory_is_unmounted_production(fake_media) is True

        (config_dir / 'controller.yaml').write_text('drives:\n  library_on_internal_disk: true\n')
        assert db_bootstrap.media_directory_is_unmounted_production(fake_media) is False


@pytest.mark.parametrize('contents,expected', [
    (None, False),
    ('{}\n', False),
    ('drives:\n  library_on_internal_disk: false\n', False),
    ('drives:\n  library_on_internal_disk: true\n', True),
    ('drives: [not, a, dict]\n', False),
    ('drives: {library_on_internal_disk: true\n', False),  # Malformed YAML.
])
def test_library_on_internal_disk(test_directory, contents, expected):
    from wrolpi.common import library_on_internal_disk

    if contents is not None:
        (test_directory / 'config').mkdir(exist_ok=True)
        (test_directory / 'config' / 'controller.yaml').write_text(contents)
    assert library_on_internal_disk() is expected