nancy/tests/test_db.py

"""
Pure SQL tests that don't depend on nancy's Python code
"""
import pytest

import os
import sqlite3


@pytest.fixture
def db():
    """Create an in-memory database that follow's the nancy schema"""
    conn = sqlite3.connect(':memory:')
    cur = conn.cursor()
    #import importlib
    #schema = importlib.resources.open_text("nancy", "schema.sql").read()
    schema = open(os.path.join(
            os.path.dirname(__file__),
            '../src/nancy/schema.sql',
        ), 'r').read()
    cur.executescript(schema)
    yield cur
    conn.close()


@pytest.fixture(autouse=True)
def insert_machine(db):
    db.executemany(
        'INSERT INTO machine VALUES '
        '(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)',
        [(
            None,
            'lucky', #hostname TEXT, -- platform.node(): 'lucky'
            '', #processor TEXT, -- platform.processor():
            'Linux', #system TEXT, -- platform.system(): 'Linux'
            '5.15.64', #release TEXT, -- platform.release(): '5.15.64'
            'aarch64', #machine TEXT, -- platform.machine(): 'x86_64'
            'EDT', #timezone TEXT, -- timezone, for interpreting event times
            '', #freedesktop_os_release TEXT, -- requires python 3.10
            '', #win32_ver TEXT, -- platform.win32_ver() as JSON
            '', #mac_ver TEXT -- platform.mac_ver() as JSON
        ), (
            None,
            'a100', #hostname TEXT, -- platform.node(): 'lucky'
            '', #processor TEXT, -- platform.processor():
            'Linux', #system TEXT, -- platform.system(): 'Linux'
            '5.15.63', #release TEXT, -- platform.release(): '5.15.64'
            'x86_64', #machine TEXT, -- platform.machine(): 'x86_64'
            'EST', #timezone TEXT, -- timezone, for interpreting event times
            '', #freedesktop_os_release TEXT, -- requires python 3.10
            '', #win32_ver TEXT, -- platform.win32_ver() as JSON
            '', #mac_ver TEXT -- platform.mac_ver() as JSON
        )],
    )
def test_insert_machine(db):
    db.execute('SELECT * FROM machine')
    machines = db.fetchall()
    assert len(machines) == 2

@pytest.fixture(autouse=True)
def insert_user(db):
    db.executemany(
        'INSERT INTO user VALUES '
        '(?, ?, ?, ?, ?)',
        [(
            None,
            'jacob', #username TEXT NOT NULL,
            101, #userid INTEGER,
            'Jacob Hinkle', #fullname TEXT,
            1, #machine INTEGER NOT NULL,
        ), (
            None,
            'jacob', #username TEXT NOT NULL,
            10301, #userid INTEGER,
            'Jacob Hinkle', #fullname TEXT,
            2, #machine INTEGER NOT NULL,
        ), (
            None,
            'bob', #username TEXT NOT NULL,
            2035, #userid INTEGER,
            'Just Bob', #fullname TEXT,
            2, #machine INTEGER NOT NULL,
        )],
    )
def test_insert_user(db):
    db.execute('SELECT * FROM user')
    users = db.fetchall()
    assert len(users) == 3
def test_invalid_user_machine(db):
    with pytest.raises(sqlite3.IntegrityError):
        # should fail foreign key constraint
        db.execute(
            'INSERT INTO user VALUES '
            '(?, ?, ?, ?, ?)',
            (
                None,
                'bozo', #username TEXT NOT NULL,
                100, #userid INTEGER,
                'Bozo the Clown', #fullname TEXT,
                3, #machine INTEGER NOT NULL,
            ),
        )
    with pytest.raises(sqlite3.IntegrityError):
        # should fail uniqueness constraint
        db.execute(
            'INSERT INTO user VALUES '
            '(?, ?, ?, ?, ?)',
            (
                None,
                'jacob', #username TEXT NOT NULL,
                101, #userid INTEGER,
                'Bozo the Clown', #fullname TEXT,
                1, #machine INTEGER NOT NULL,
            ),
        )


@pytest.fixture(autouse=True)
def insert_store(db):
    db.executemany(
        'INSERT INTO store VALUES '
        '(?, ?, ?, ?)',
        [(
            None,
            1, #machine INTEGER,
            '/path/to/first/store', #dbpath TEXT NOT NULL,
            False, #imported BOOL,
        ), (
            None,
            1, #machine INTEGER,
            '/path/to/dependencys/store', #dbpath TEXT NOT NULL,
            True, #imported BOOL,
        ), (
            None,
            2, #machine INTEGER,
            # same path but on a separate machine
            '/path/to/first/store', #dbpath TEXT NOT NULL,
            True, #imported BOOL,
        )],
    )
@pytest.fixture(autouse=True)
def insert_store_directory(db):
    db.executemany(
        'INSERT INTO store_directory VALUES '
        '(?, ?, ?, ?, ?)',
        [(
            None,
            '.', #name TEXT,
            1, #store INTEGER NOT NULL,
            None, #parent INTEGER, -- parent directory (should be) in same store
            False, #frozen BOOL NOT NULL,
        ), (
            None,
            'foo', #name TEXT,
            1, #store INTEGER NOT NULL,
            1, #parent INTEGER, -- parent directory (should be) in same store
            False, #frozen BOOL NOT NULL,
        ), (
            None,
            '.', #name TEXT,
            2, #store INTEGER NOT NULL,
            None, #parent INTEGER, -- parent directory (should be) in same store
            False, #frozen BOOL NOT NULL,
        )],
    )

def test_crossstore_directory_insert(db):
    with pytest.raises(sqlite3.IntegrityError):
        # declaring directory as belonging to store 2, but parent's store is 1
        db.execute(
            'INSERT INTO store_directory VALUES '
            '(?, ?, ?, ?, ?)',
            (
                None,
                '.', #name TEXT,
                2, #store INTEGER NOT NULL,
                1, #parent INTEGER, -- parent directory (should be) in same store
                False, #frozen BOOL NOT NULL,
            ))


@pytest.fixture(autouse=True)
def insert_store_file(db):
    db.executemany(
        'INSERT INTO store_file VALUES '
        '(?, ?, ?, ?, ?, ?)',
        [(
            None,
            1, #directory INTEGER NOT NULL
            'example.csv', #filename TEXT
            1, #frozen BOOL NOT NULL
            '-rw-rw-r--', #unfrozen_perms TEXT
            None, #source_datum INTEGER,
        ), (
            None,
            2, #directory INTEGER NOT NULL
            'plots.png', #filename TEXT
            1, #frozen BOOL NOT NULL
            '-rw-r--r--', #unfrozen_perms TEXT
            None, #source_datum INTEGER,
        )]
    )