261 lines
9.6 KiB
Python
261 lines
9.6 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,
|
|
'Jacob Hinkle', #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 filedir 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 filedir ON UPDATE CASCADE,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #frozen BOOL NOT NULL,
|
|
False, #deleted BOOL NOT NULL,
|
|
), (
|
|
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 filedir ON UPDATE CASCADE,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #frozen BOOL NOT NULL,
|
|
False, #deleted BOOL NOT NULL,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
2, #store INTEGER NOT NULL,
|
|
'.', #filename TEXT, -- only a filename, not a path
|
|
None, #parent INTEGER REFERENCES filedir ON UPDATE CASCADE,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #frozen BOOL NOT NULL,
|
|
False, #deleted BOOL NOT NULL,
|
|
)],
|
|
)
|
|
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 filedir VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?)',
|
|
(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
2, #store INTEGER NOT NULL,
|
|
'some_dir', #filename TEXT, -- only a filename, not a path
|
|
1, #parent INTEGER REFERENCES filedir ON UPDATE CASCADE,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #frozen BOOL NOT NULL,
|
|
False, #deleted BOOL NOT NULL,
|
|
))
|
|
for row in cur.connection.iterdump():
|
|
print(row)
|
|
cur.execute('SELECT * FROM filedir')
|
|
print(cur.fetchall())
|
|
|
|
|
|
@pytest.fixture
|
|
def insert_files(insert_directories):
|
|
cur = insert_directories
|
|
cur.executemany(
|
|
'INSERT INTO filedir VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?)',
|
|
[(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, #store INTEGER NOT NULL,
|
|
'example.csv', #filename TEXT, -- only a filename, not a path
|
|
1, #parent INTEGER REFERENCES filedir ON UPDATE CASCADE,
|
|
'REG', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #frozen BOOL NOT NULL,
|
|
False, #deleted BOOL NOT NULL,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, #store INTEGER NOT NULL,
|
|
'plots.png', #filename TEXT, -- only a filename, not a path
|
|
2, #parent INTEGER REFERENCES filedir ON UPDATE CASCADE,
|
|
'REG', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #frozen BOOL NOT NULL,
|
|
False, #deleted BOOL NOT NULL,
|
|
)]
|
|
)
|
|
return cur
|
|
|
|
def test_nondir_parent_directory_insert(insert_files):
|
|
cur = insert_files
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
# declaring parent as 5, but 5 is a file (plots.png)
|
|
cur.execute(
|
|
'INSERT INTO filedir VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?)',
|
|
(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
2, #store INTEGER NOT NULL,
|
|
'some_filedir.txt', #filename TEXT, -- only a filename, not a path
|
|
5, #parent INTEGER REFERENCES filedir ON UPDATE CASCADE,
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #frozen BOOL NOT NULL,
|
|
False, #deleted BOOL NOT NULL,
|
|
))
|
|
for row in cur.connection.iterdump():
|
|
print(row)
|
|
cur.execute('SELECT * FROM filedir')
|
|
print(cur.fetchall())
|
|
|
|
|
|
def test_update_deleted_file(insert_files):
|
|
cur = insert_files
|
|
# first we set a file to deleted
|
|
cur.execute('UPDATE filedir SET deleted=True WHERE id=5')
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
# Now we try and update it, which should fail due to trigger
|
|
cur.execute('UPDATE filedir SET frozen=True WHERE id=5')
|