349 lines
15 KiB
Python
349 lines
15 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())
|