Jacob Hinkle
d0f113645b
This is meant to hold schemas and migrations. It contains an `__init__.py` due to a limitation of `importlib.resources` outlined in detail here: https://github.com/python/importlib_resources/issues/58 Eventually the schema submodule will handle automatic migration of older databases when importing into new stores.
250 lines
10 KiB
Python
250 lines
10 KiB
Python
"""
|
|
Pure SQL tests that don't depend on nancy's Python code
|
|
"""
|
|
import pytest
|
|
|
|
import os
|
|
import sqlite3
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db():
|
|
"""Create an in-memory database that follow's the nancy schema"""
|
|
with sqlite3.connect(':memory:') as conn:
|
|
cur = conn.cursor()
|
|
|
|
from nancy import db
|
|
db.init_schema(cur)
|
|
|
|
yield cur
|
|
|
|
|
|
@pytest.fixture
|
|
def insert_machine(temp_db):
|
|
cur = temp_db
|
|
cur.executemany(
|
|
'INSERT INTO machine VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)',
|
|
[(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
'a5d97c08a15c4db69f5fded523a1bfe3', #machine_id TEXT, -- platform-dependent unique hardware id
|
|
'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, #id INTEGER PRIMARY KEY NOT NULL,
|
|
'afc9b06a23b74341b29d42b8312a4f8a',
|
|
'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
|
|
)],
|
|
)
|
|
return cur
|
|
|
|
def test_insert_machine(insert_machine):
|
|
cur = insert_machine
|
|
cur.execute('SELECT * FROM machine')
|
|
machines = cur.fetchall()
|
|
assert len(machines) == 2
|
|
|
|
@pytest.fixture
|
|
def insert_user(insert_machine):
|
|
cur = insert_machine
|
|
cur.executemany(
|
|
'INSERT INTO user VALUES '
|
|
'(?, ?, ?, ?, ?)',
|
|
[(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
'jacob', #username TEXT NOT NULL,
|
|
101, #userid INTEGER,
|
|
'Jacob Hinkle', #fullname TEXT,
|
|
1, #machine INTEGER NOT NULL,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
'jacob', #username TEXT NOT NULL,
|
|
10301, #userid INTEGER,
|
|
'Jacob Hinkle', #fullname TEXT,
|
|
2, #machine INTEGER NOT NULL,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
'bob', #username TEXT NOT NULL,
|
|
2035, #userid INTEGER,
|
|
'Just Bob', #fullname TEXT,
|
|
2, #machine INTEGER NOT NULL,
|
|
)],
|
|
)
|
|
return cur
|
|
def test_insert_user(insert_user):
|
|
cur = insert_user
|
|
cur.execute('SELECT * FROM user')
|
|
users = cur.fetchall()
|
|
assert len(users) == 3
|
|
def test_invalid_user_machine(insert_user):
|
|
cur = insert_user
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
# should fail foreign key constraint
|
|
cur.execute(
|
|
'INSERT INTO user VALUES '
|
|
'(?, ?, ?, ?, ?)',
|
|
(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
'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
|
|
cur.execute(
|
|
'INSERT INTO user VALUES '
|
|
'(?, ?, ?, ?, ?)',
|
|
(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
'jacob', #username TEXT NOT NULL,
|
|
101, #userid INTEGER,
|
|
'Bozo the Clown', #fullname TEXT,
|
|
1, #machine INTEGER NOT NULL,
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def insert_store(insert_machine):
|
|
cur = insert_machine
|
|
cur.executemany(
|
|
'INSERT INTO store VALUES '
|
|
'(?, ?, ?, ?)',
|
|
[(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, #machine INTEGER,
|
|
'/path/to/first/store', #dbpath TEXT NOT NULL,
|
|
False, #imported BOOL,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, #machine INTEGER,
|
|
'/path/to/dependency/store', #dbpath TEXT NOT NULL,
|
|
True, #imported BOOL,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
2, #machine INTEGER,
|
|
# same path but on a separate machine
|
|
'/path/to/first/store', #dbpath TEXT NOT NULL,
|
|
True, #imported BOOL,
|
|
)],
|
|
)
|
|
return cur
|
|
|
|
@pytest.fixture
|
|
def insert_directories(insert_store):
|
|
cur = insert_store
|
|
cur.executemany(
|
|
'INSERT INTO store_file VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)',
|
|
[(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, #store INTEGER NOT NULL,
|
|
'.', #filename TEXT, -- only a filename, not a path
|
|
None, #parent INTEGER REFERENCES store_file ON UPDATE CASCADE,
|
|
False, #frozen BOOL NOT NULL,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
'dr-xr-xr--', #unfrozen_perms TEXT, -- stat.filemode(os.stat(path).st_mode): '-rw-rw-r--'
|
|
None, #symlink_target TEXT, -- if this is a symlink, this is the (read but not fully resolved) target
|
|
'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855', #sha256 TEXT NOT NULL,
|
|
None, #source_program INTEGER, -- Note that this is redundant since datum points to a program...
|
|
None, #source_datum INTEGER,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, #store INTEGER NOT NULL,
|
|
'foo', #filename TEXT, -- only a filename, not a path
|
|
1, #parent INTEGER REFERENCES store_file ON UPDATE CASCADE,
|
|
False, #frozen BOOL NOT NULL,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
'dr-xr-xr--', #unfrozen_perms TEXT, -- stat.filemode(os.stat(path).st_mode): '-rw-rw-r--'
|
|
None, #symlink_target TEXT, -- if this is a symlink, this is the (read but not fully resolved) target
|
|
'5ad4e9e3090de8de781e4a35ce6ba16ad8eaba7a5456cd230a36f77143885396', #sha256 TEXT NOT NULL,
|
|
None, #source_program INTEGER, -- Note that this is redundant since datum points to a program...
|
|
None, #source_datum INTEGER,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
2, #store INTEGER NOT NULL,
|
|
'.', #filename TEXT, -- only a filename, not a path
|
|
None, #parent INTEGER REFERENCES store_file ON UPDATE CASCADE,
|
|
False, #frozen BOOL NOT NULL,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
'dr-xr-xr--', #unfrozen_perms TEXT, -- stat.filemode(os.stat(path).st_mode): '-rw-rw-r--'
|
|
None, #symlink_target TEXT, -- if this is a symlink, this is the (read but not fully resolved) target
|
|
'15c5e8d80a48803c18e72cd274532d608b8026dcbc192afc490fe1c289ec6ff1', #sha256 TEXT NOT NULL,
|
|
None, #source_program INTEGER, -- Note that this is redundant since datum points to a program...
|
|
None, #source_datum INTEGER,
|
|
)],
|
|
)
|
|
return cur
|
|
|
|
def test_crossstore_directory_insert(insert_directories):
|
|
cur = insert_directories
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
# declaring directory as belonging to store 2, but parent's store is 1
|
|
cur.execute(
|
|
'INSERT INTO store_file VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)',
|
|
(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
2, #store INTEGER NOT NULL,
|
|
'xstore_file', #filename TEXT, -- only a filename, not a path
|
|
1, #parent INTEGER REFERENCES store_file ON UPDATE CASCADE,
|
|
False, #frozen BOOL NOT NULL,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
'dr-xr-xr--', #unfrozen_perms TEXT, -- stat.filemode(os.stat(path).st_mode): '-rw-rw-r--'
|
|
None, #symlink_target TEXT, -- if this is a symlink, this is the (read but not fully resolved) target
|
|
'15c5e8d80a48803c18e72cd274532d608b8026dcbc192afc490fe1c289ec6ff1', #sha256 TEXT NOT NULL,
|
|
None, #source_program INTEGER, -- Note that this is redundant since datum points to a program...
|
|
None, #source_datum INTEGER,
|
|
))
|
|
for row in cur.connection.iterdump():
|
|
print(row)
|
|
cur.execute('SELECT * FROM store_file')
|
|
print(cur.fetchall())
|
|
|
|
|
|
@pytest.fixture
|
|
def insert_files(insert_directories):
|
|
cur = insert_directories
|
|
cur.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
|
|
'9aafde8f9dbec34c694b86333f746f58958c44247c474904e06d1f07f94292b4', #sha256 TEXT NOT NULL,
|
|
None, #source_program INTEGER,
|
|
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
|
|
'9add10cc3a6f0e4618dfed005ddfbeafdf268c58b773ba0021963c856d00235b', #sha256 TEXT NOT NULL,
|
|
None, #source_program INTEGER,
|
|
None, #source_datum INTEGER,
|
|
)]
|
|
)
|
|
return cur
|
|
|