317 lines
14 KiB
Python
317 lines
14 KiB
Python
"""
|
|
Pure SQL tests that don't depend on nancy's Python code
|
|
"""
|
|
import pytest
|
|
|
|
import datetime
|
|
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):
|
|
import uuid
|
|
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,
|
|
str(uuid.uuid4()), #-- UUID generated by str(uuid.uuid4())
|
|
False, #imported BOOL,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, #machine INTEGER,
|
|
'/path/to/dependency/store', #dbpath TEXT NOT NULL,
|
|
str(uuid.uuid4()), #-- UUID generated by str(uuid.uuid4())
|
|
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,
|
|
str(uuid.uuid4()), #-- UUID generated by str(uuid.uuid4())
|
|
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,
|
|
False, #frozen 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,
|
|
False, #frozen 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,
|
|
False, #frozen BOOL NOT NULL,
|
|
)],
|
|
)
|
|
cur.executemany(
|
|
'INSERT INTO filedir_version VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?, ?, ?)',
|
|
[(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, # INTEGER REFERENCES filedir ON UPDATE CASCADE, -- parent filedir entry
|
|
datetime.datetime.now().timestamp(),
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #deleted BOOL NOT NULL, -- set True when recording a deleted file
|
|
'drwxrwxr-x', #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. I.e. this is the "content" of the symlink.
|
|
'a84ed33864d06615a87bc8da5258d841163f1e7969367ecd07b041ae1a18febd', #sha256 TEXT,
|
|
None, #source_task INTEGER,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, # INTEGER REFERENCES filedir ON UPDATE CASCADE, -- parent filedir entry
|
|
datetime.datetime.now().timestamp(),
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #deleted BOOL NOT NULL, -- set True when recording a deleted file
|
|
'drwxrwxr-x', #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. I.e. this is the "content" of the symlink.
|
|
'a84ed33864d06615a87bc8da5258d841163f1e7969367ecd07b041ae1a18febd', #sha256 TEXT,
|
|
None, #source_task INTEGER,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
1, # INTEGER REFERENCES filedir ON UPDATE CASCADE, -- parent filedir entry
|
|
datetime.datetime.now().timestamp(),
|
|
'DIR', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #deleted BOOL NOT NULL, -- set True when recording a deleted file
|
|
'drwxrwxr-x', #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. I.e. this is the "content" of the symlink.
|
|
'a84ed33864d06615a87bc8da5258d841163f1e7969367ecd07b041ae1a18febd', #sha256 TEXT,
|
|
None, #source_task 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 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,
|
|
False, #frozen 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.execute('SELECT COUNT(*) FROM filedir')
|
|
nprev, = cur.fetchone()
|
|
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,
|
|
False, #frozen 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,
|
|
False, #frozen BOOL NOT NULL,
|
|
)]
|
|
)
|
|
cur.executemany(
|
|
'INSERT INTO filedir_version VALUES '
|
|
'(?, ?, ?, ?, ?, ?, ?, ?, ?)',
|
|
[(
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
nprev + 1, # INTEGER REFERENCES filedir ON UPDATE CASCADE, -- parent filedir entry
|
|
datetime.datetime.now().timestamp(),
|
|
'REG', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #deleted BOOL NOT NULL, -- set True when recording a deleted file
|
|
'drwxrwxr-x', #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. I.e. this is the "content" of the symlink.
|
|
'a84ed33864d06615a87bc8da5258d841163f1e7969367ecd07b041ae1a18febd', #sha256 TEXT,
|
|
None, #source_task INTEGER,
|
|
), ( # second version of first file
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
nprev + 1, # INTEGER REFERENCES filedir ON UPDATE CASCADE, -- parent filedir entry
|
|
datetime.datetime.now().timestamp(),
|
|
'REG', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #deleted BOOL NOT NULL, -- set True when recording a deleted file
|
|
'drwxr-xr-x', #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. I.e. this is the "content" of the symlink.
|
|
'a94ed33864d06615a87bc8da5258d841163f1e7969367ecd07b041ae1a18febd', #sha256 TEXT,
|
|
None, #source_task INTEGER,
|
|
), (
|
|
None, #id INTEGER PRIMARY KEY NOT NULL,
|
|
nprev + 2, # INTEGER REFERENCES filedir ON UPDATE CASCADE, -- parent filedir entry
|
|
datetime.datetime.now().timestamp(),
|
|
'REG', #filetype TEXT, -- One of 'LNK', 'DIR', 'REG', etc. See store.FSEntry.from_path for details
|
|
False, #deleted BOOL NOT NULL, -- set True when recording a deleted file
|
|
'drwxr-xr-x', #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. I.e. this is the "content" of the symlink.
|
|
'a94ed33864d06615a87bc8da5258d841163f1e7969367ecd07b041ae1a18febd', #sha256 TEXT,
|
|
None, #source_task INTEGER,
|
|
)])
|
|
return cur
|
|
|
|
# TODO: This test is disabled until triggers are added to check for these types
|
|
# of constraints. These became much more complicated to check when I added
|
|
# filedir_version.
|
|
def disabled_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,
|
|
1, #store INTEGER NOT NULL,
|
|
'some_filedir.txt', #filename TEXT, -- only a filename, not a path
|
|
5, #parent INTEGER REFERENCES filedir ON UPDATE CASCADE,
|
|
False, #frozen BOOL NOT NULL,
|
|
))
|
|
for row in cur.connection.iterdump():
|
|
print(row)
|
|
cur.execute('SELECT * FROM filedir')
|
|
print(cur.fetchall())
|