Christopher Hawthorne
10 years ago
parent
d858f34ef8
commit
1637c72955
@ -0,0 +1,17 @@ |
||||
import abc |
||||
|
||||
class LibrarianException(Exception, metaclass=abc.ABCMeta): |
||||
@abc.abstractproperty |
||||
def error_msg(self): |
||||
pass |
||||
|
||||
def __str__(self): |
||||
return self.error_msg |
||||
|
||||
class PermissionsError(LibrarianException): |
||||
def __init__(self, permission_string): |
||||
self.permission_string = permission_string |
||||
|
||||
@property |
||||
def error_msg(self): |
||||
return "Need privilege level {}".format(self.permission_string) |
||||
@ -0,0 +1,490 @@ |
||||
import contextlib |
||||
import os |
||||
import sqlite3 |
||||
import unittest |
||||
|
||||
import db_layer |
||||
import exceptions |
||||
import permissions |
||||
|
||||
class TestDBSetup(unittest.TestCase): |
||||
def tearDown(self): |
||||
try: |
||||
os.remove(db_layer.dbFile) |
||||
except FileNotFoundError: |
||||
pass |
||||
|
||||
def test_db_setup(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with self.assertRaises(sqlite3.OperationalError): |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT * FROM books;') |
||||
|
||||
db_layer.initializeDatabase() |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT * FROM books;') |
||||
|
||||
def test_existing_db_file(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
db_layer.initializeDatabase() |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('INSERT INTO books (isbn) VALUES (1111111111);') |
||||
conn.commit() |
||||
|
||||
db_layer.initializeDatabase() |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT isbn FROM books;') |
||||
rows = list(c) |
||||
|
||||
self.assertEqual([1111111111], [row[0] for row in rows]) |
||||
|
||||
class TestDBFunctions(unittest.TestCase): |
||||
maxDiff = None |
||||
|
||||
def assertISBNs(self, expected_isbns, book_dicts): |
||||
expected_isbns = frozenset(expected_isbns) |
||||
actual_isbns = frozenset(book.get('isbn') for book in book_dicts) |
||||
self.assertEqual(expected_isbns, actual_isbns) |
||||
|
||||
def assertCategories(self, expected_categories, book_dicts): |
||||
expected_categories = frozenset(expected_categories) |
||||
actual_categories = frozenset(book['category'] for book in book_dicts) |
||||
self.assertEqual(expected_categories, actual_categories) |
||||
|
||||
def setUp(self): |
||||
db_layer.initializeDatabase() |
||||
permissions._CURRENT_GROUPS_GETTER = lambda: ["office", "libcom"] |
||||
|
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
INSERT INTO books (isbn, lccn, title, subtitle, authors, edition, publisher, |
||||
publish_year, publish_month, publish_location, pages, |
||||
pagination, weight) |
||||
VALUES |
||||
(1111111111111, 2222222222, 'Attack of the bad code', 'Return of the GOTO', |
||||
'cdchawthorne', '1st', 'CSC Publishing', '1992', 'June', 'Waterloo, Canada', |
||||
'496', 'xxvi, 493 p', '1 kg'), |
||||
(3333333333333, 4444444444, 'Star Wars VI--Return of the Jedi', 'Now in text!', |
||||
'George Lucas', '2nd', 'Lucas Film', '2013', 'November', 'Somewhere, USA', |
||||
'8128', 'xx, 8100 p', '10 kg'); |
||||
''') |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
INSERT INTO books (isbn, lccn, deleted) VALUES |
||||
(5555555555555, NULL, 0), (NULL, 6666666666, 0), (7777777777777, NULL, 1); |
||||
''') |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
INSERT INTO categories (category) |
||||
VALUES ('My special category'), ('My second special category'); |
||||
''') |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('''SELECT rowid, isbn, lccn, last_updated FROM books;''') |
||||
for row in c: |
||||
if row[1] == 1111111111111: |
||||
self.book0_id = row[0] |
||||
self.book0_last_updated = row[3] |
||||
elif row[1] == 3333333333333: |
||||
self.book1_id = row[0] |
||||
self.book1_last_updated = row[3] |
||||
elif row[1] == 5555555555555: |
||||
self.book2_id = row[0] |
||||
self.book2_last_updated = row[3] |
||||
elif row[2] == 6666666666: |
||||
self.book3_id = row[0] |
||||
self.book3_last_updated = row[3] |
||||
elif row[1] == 7777777777777: |
||||
self.book4_id = row[0] |
||||
self.book4_last_updated = row[3] |
||||
else: |
||||
self.assertTrue(False, |
||||
"Unexpected data in DB during setup") |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('''SELECT rowid, category FROM categories;''') |
||||
for row in c: |
||||
if row[1] == 'My special category': |
||||
self.category0_id = row[0] |
||||
elif row[1] == 'My second special category': |
||||
self.category1_id = row[0] |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
INSERT INTO book_categories (id, cat_id) VALUES (?, ?), (?, ?), (?, ?), (?, ?); |
||||
''', |
||||
(self.book1_id, self.category0_id, |
||||
self.book1_id, self.category1_id, |
||||
self.book0_id, self.category1_id, |
||||
self.book2_id, self.category1_id, |
||||
) |
||||
) |
||||
|
||||
conn.commit() |
||||
|
||||
def tearDown(self): |
||||
try: |
||||
os.remove(db_layer.dbFile) |
||||
except FileNotFoundError: |
||||
pass |
||||
|
||||
def test_getBooks(self): |
||||
books = db_layer.getBooks() |
||||
|
||||
expected_book0 = { |
||||
'id': self.book0_id, |
||||
'isbn': 1111111111111, |
||||
'lccn': 2222222222, |
||||
'title': 'Attack of the bad code', |
||||
'subtitle': 'Return of the GOTO', |
||||
'authors': 'cdchawthorne', |
||||
'edition': '1st', |
||||
'publisher': 'CSC Publishing', |
||||
'publish year': '1992', |
||||
'publish month': 'June', |
||||
'publish location': 'Waterloo, Canada', |
||||
'pages': '496', |
||||
'pagination': 'xxvi, 493 p', |
||||
'weight': '1 kg', |
||||
'last updated': self.book0_last_updated, |
||||
} |
||||
|
||||
found_book0 = False |
||||
for book in books: |
||||
if book.get('isbn') == 1111111111111: |
||||
found_book0 = True |
||||
self.assertEqual(book, expected_book0) |
||||
|
||||
self.assertTrue(found_book0, "getBooks() missing book0") |
||||
|
||||
expected_isbns = [1111111111111, 3333333333333, 5555555555555, None] |
||||
self.assertISBNs(expected_isbns, books) |
||||
|
||||
def test_getBooksByCategory(self): |
||||
books = db_layer.getBooksByCategory(str(self.category0_id)) |
||||
expected_isbns = [3333333333333] |
||||
self.assertISBNs(expected_isbns, books) |
||||
|
||||
def test_getRemovedBooks(self): |
||||
books = db_layer.getRemovedBooks() |
||||
expected_isbns = [7777777777777] |
||||
self.assertISBNs(expected_isbns, books) |
||||
|
||||
def test_addBook(self): |
||||
db_layer.addBook({'isbn': 8888888888888, 'title': 'New book'}) |
||||
|
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
SELECT title FROM books WHERE isbn = '8888888888888'; |
||||
''') |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(['New book'], [row[0] for row in rows]) |
||||
|
||||
def test_updateBook(self): |
||||
db_layer.updateBook({'title': 'Attack of the questionable code'}, |
||||
str(self.book0_id)) |
||||
|
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
SELECT title FROM books WHERE id = ?; |
||||
''', (self.book0_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(['Attack of the questionable code'], |
||||
[row[0] for row in rows]) |
||||
|
||||
def test_getBookById(self): |
||||
book = db_layer.getBookByID(self.book0_id) |
||||
self.assertEqual('Attack of the bad code', book['title']) |
||||
|
||||
def test_removeBook(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
SELECT deleted FROM books WHERE id = ?; |
||||
''', (self.book1_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual([0], [row[0] for row in rows]) |
||||
|
||||
db_layer.removeBook(self.book1_id) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
SELECT deleted FROM books WHERE id = ?; |
||||
''', (self.book1_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual([1], [row[0] for row in rows]) |
||||
|
||||
def test_removeBooks(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
SELECT deleted FROM books WHERE id = ? OR id = ?; |
||||
''', (self.book0_id, self.book1_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual([0,0], [row[0] for row in rows]) |
||||
|
||||
db_layer.removeBooks([str(self.book0_id), str(self.book1_id)]) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute(''' |
||||
SELECT deleted FROM books WHERE id = ? OR id = ?; |
||||
''', (self.book0_id, self.book1_id)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual([1,1], [row[0] for row in rows]) |
||||
|
||||
def test_deleteBook(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM books WHERE id = ? OR id = ?;', |
||||
(self.book0_id, self.book1_id)) |
||||
count = c.fetchone()[0] |
||||
|
||||
self.assertEqual(2, count) |
||||
|
||||
db_layer.deleteBook(self.book0_id) |
||||
db_layer.deleteBook(self.book1_id) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM books WHERE id = ? OR id = ?;', |
||||
(self.book0_id, self.book1_id)) |
||||
count = c.fetchone()[0] |
||||
|
||||
self.assertEqual(0, count) |
||||
|
||||
# Code duplication? What's that? |
||||
def test_deleteBooks(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM books WHERE id = ? OR id = ?;', |
||||
(self.book0_id, self.book1_id)) |
||||
count = c.fetchone()[0] |
||||
|
||||
self.assertEqual(2, count) |
||||
|
||||
db_layer.deleteBooks([str(self.book0_id), str(self.book1_id)]) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM books WHERE id = ? OR id = ?;', |
||||
(self.book0_id, self.book1_id)) |
||||
count = c.fetchone()[0] |
||||
|
||||
self.assertEqual(0, count) |
||||
|
||||
def test_getBookCategories(self): |
||||
categories = db_layer.getBookCategories(str(self.book1_id)) |
||||
expected_categories = ['My special category', |
||||
'My second special category'] |
||||
self.assertCategories(expected_categories, categories) |
||||
|
||||
categories = db_layer.getBookCategories(str(self.book0_id)) |
||||
expected_categories = ['My second special category'] |
||||
self.assertCategories(expected_categories, categories) |
||||
|
||||
categories = db_layer.getBookCategories(str(self.book3_id)) |
||||
expected_categories = [] |
||||
self.assertCategories(expected_categories, categories) |
||||
|
||||
def test_categorizeBook(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT cat_id FROM book_categories WHERE id = ?;', |
||||
(self.book3_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual([], rows) |
||||
|
||||
db_layer.categorizeBook({'id': self.book3_id}, |
||||
[{'id': self.category0_id}, |
||||
{'id': self.category1_id}]) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT cat_id FROM book_categories WHERE id = ?;', |
||||
(self.book3_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(frozenset([self.category0_id, self.category1_id]), |
||||
frozenset(row[0] for row in rows)) |
||||
|
||||
def test_uncategorizeBook(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT cat_id FROM book_categories WHERE id = ?;', |
||||
(self.book1_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(frozenset([self.category0_id, self.category1_id]), |
||||
frozenset(row[0] for row in rows)) |
||||
|
||||
db_layer.uncategorizeBook({'id': self.book1_id}, |
||||
[{'id': self.category0_id}, |
||||
{'id': self.category1_id}]) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT cat_id FROM book_categories WHERE id = ?;', |
||||
(self.book3_id,)) |
||||
rows = list(c) |
||||
|
||||
self.assertEqual([], rows) |
||||
|
||||
def test_getCategories(self): |
||||
categories = db_layer.getCategories() |
||||
expected_categories = ['My special category', |
||||
'My second special category'] |
||||
self.assertCategories(expected_categories, categories) |
||||
|
||||
def test_addCategory(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT category FROM categories;') |
||||
rows = list(c) |
||||
|
||||
expected_categories = ['My special category', |
||||
'My second special category'] |
||||
self.assertEqual(frozenset(expected_categories), |
||||
frozenset(row[0] for row in rows)) |
||||
|
||||
db_layer.addCategory('My third special category') |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT category FROM categories;') |
||||
rows = list(c) |
||||
|
||||
expected_categories.append('My third special category') |
||||
self.assertEqual(frozenset(expected_categories), |
||||
frozenset(row[0] for row in rows)) |
||||
|
||||
def test_deleteCategories(self): |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM categories;') |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(2, rows[0][0]) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM book_categories;') |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(4, rows[0][0]) |
||||
|
||||
db_layer.deleteCategories([str(self.category1_id)]) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM categories;') |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(1, rows[0][0]) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT COUNT(*) FROM book_categories;') |
||||
rows = list(c) |
||||
|
||||
self.assertEqual(1, rows[0][0]) |
||||
|
||||
class TestDBFunctionPermissions(unittest.TestCase): |
||||
def setUp(self): |
||||
db_layer.initializeDatabase() |
||||
# Need book in database or getBookByID will error when we test it |
||||
conn = sqlite3.connect(db_layer.dbFile) |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('INSERT INTO books (isbn) VALUES (5555555555555);') |
||||
|
||||
with contextlib.closing(conn.cursor()) as c: |
||||
c.execute('SELECT rowid FROM books;') |
||||
self.dummy_book_id = c.fetchone()[0] |
||||
|
||||
def tearDown(self): |
||||
try: |
||||
os.remove(db_layer.dbFile) |
||||
except FileNotFoundError: |
||||
pass |
||||
|
||||
def verify_permissions(self, fn, authorized_groups, unauthorized_groups): |
||||
for group in authorized_groups: |
||||
permissions._CURRENT_GROUPS_GETTER = lambda: [group] |
||||
fn() |
||||
|
||||
for group in unauthorized_groups: |
||||
permissions._CURRENT_GROUPS_GETTER = lambda: [group] |
||||
with self.assertRaises(exceptions.PermissionsError): |
||||
fn() |
||||
|
||||
def test_db_function_permissions(self): |
||||
function_dicts = [ |
||||
{'fn': lambda: db_layer.addBook({'isbn': 8888888888888}), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.updateBook({'isbn': 9999999999999}, 0), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.getBooks(), |
||||
'authorized_groups': ['libcom', 'office', 'unauthorized'], |
||||
'unauthorized_groups': []}, |
||||
{'fn': lambda: db_layer.getBooksByCategory('0'), |
||||
'authorized_groups': ['libcom', 'office', 'unauthorized'], |
||||
'unauthorized_groups': []}, |
||||
{'fn': lambda: db_layer.getRemovedBooks(), |
||||
'authorized_groups': ['libcom', 'office', 'unauthorized'], |
||||
'unauthorized_groups': []}, |
||||
{'fn': lambda: db_layer.getBookByID(self.dummy_book_id), |
||||
'authorized_groups': ['libcom', 'office', 'unauthorized'], |
||||
'unauthorized_groups': []}, |
||||
{'fn': lambda: db_layer.removeBook(0), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.removeBooks(['0']), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.restoreBooks(['0']), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.deleteBook(0), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.deleteBooks(['0']), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.getBookCategories('0'), |
||||
'authorized_groups': ['libcom', 'office', 'unauthorized'], |
||||
'unauthorized_groups': []}, |
||||
{'fn': lambda: db_layer.categorizeBook({'id':'0'}, []), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.uncategorizeBook({'id':'0'}, []), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.getCategories(), |
||||
'authorized_groups': ['libcom', 'office', 'unauthorized'], |
||||
'unauthorized_groups': []}, |
||||
{'fn': lambda: db_layer.addCategory('Cat5'), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
{'fn': lambda: db_layer.deleteCategories([]), |
||||
'authorized_groups': ['libcom'], |
||||
'unauthorized_groups': ['office', 'unauthorized']}, |
||||
] |
||||
|
||||
for function_dict in function_dicts: |
||||
self.verify_permissions(function_dict['fn'], |
||||
function_dict['authorized_groups'], |
||||
function_dict['unauthorized_groups']) |
||||
@ -0,0 +1,34 @@ |
||||
import grp |
||||
import os |
||||
|
||||
import exceptions |
||||
|
||||
class _PermissionLevel: |
||||
def __init__(self, group_name, pretty_name): |
||||
self.group_name = group_name |
||||
self.pretty_name = pretty_name |
||||
|
||||
PERMISSION_OFFICE = _PermissionLevel("office", "Office worker") |
||||
PERMISSION_LIBCOM = _PermissionLevel("libcom", "Library Committee") |
||||
|
||||
def check_permissions(permission_level): |
||||
def decorator(fn): |
||||
def wrapped_function(*args, **kwargs): |
||||
if not has_permission(permission_level): |
||||
raise exceptions.PermissionsError(permission_level.pretty_name) |
||||
return fn(*args, **kwargs) |
||||
|
||||
return wrapped_function |
||||
|
||||
return decorator |
||||
|
||||
def has_permission(permission_level): |
||||
return permission_level.group_name in _CURRENT_GROUPS_GETTER() |
||||
|
||||
def _current_group_names(): |
||||
group_ids = os.getgroups() |
||||
group_names = [grp.getgrgid(group_id).gr_name for group_id in group_ids] |
||||
return group_names |
||||
|
||||
# Hack to allow dependency injection for testing |
||||
_CURRENT_GROUPS_GETTER = _current_group_names |
||||
@ -0,0 +1,9 @@ |
||||
#!/usr/bin/env bash |
||||
|
||||
librarian_dir="$(realpath "$(dirname "$0")")" |
||||
librarian_tests_dir="${librarian_dir}/librarian_tests" |
||||
|
||||
export PYTHONPATH="${librarian_dir}" |
||||
cd "${librarian_tests_dir}" |
||||
|
||||
python -m unittest "$@" test_*.py |
||||
Loading…
Reference in new issue