Added web interface, crawler and more work on local storage

spec/Crawler_spec.py

@@ -1,10 +1,22 @@
 from unittest import TestCase
 
+from parsing import GenericFileParser, Sha1CheckSumCalculator, ExtensionMimeGuesser
 from crawler import Crawler
 
 
 class CrawlerTest(TestCase):
 
     def test_dir_walk(self):
-        c = Crawler()
+        c = Crawler([GenericFileParser([Sha1CheckSumCalculator()], ExtensionMimeGuesser())])
 
+        c.crawl("test_folder")
+
+        self.assertEqual(len(c.documents), 28)
+
+    def test_get_parser_by_ext(self):
+
+        c = Crawler([GenericFileParser([Sha1CheckSumCalculator()], ExtensionMimeGuesser())])
+
+        self.assertIsInstance(c.get_parser_by_ext("any"), GenericFileParser)
+
+        # todo add more parsers here

spec/FileParser_spec.py

@@ -1,39 +1,52 @@
 import os
 from unittest import TestCase
 
-from crawler import GenericFileParser, Md5CheckSumCalculator, Sha1CheckSumCalculator, Sha256CheckSumCalculator
+
+from parsing import GenericFileParser, Md5CheckSumCalculator, Sha1CheckSumCalculator, Sha256CheckSumCalculator, ExtensionMimeGuesser
 
 
 class GenericFileParserTest(TestCase):
 
     def setUp(self):
-        if os.path.exists("test_parse"):
-            os.remove("test_parse")
+        if os.path.exists("test_parse.txt"):
+            os.remove("test_parse.txt")
 
-        test_file = open("test_parse", "w")
+        test_file = open("test_parse.txt", "w")
         test_file.write("12345678")
         test_file.close()
+        os.utime("test_parse.txt", (1330123456, 1330654321))
 
-        self.parser = GenericFileParser([Md5CheckSumCalculator()])
+        self.parser = GenericFileParser([Md5CheckSumCalculator()], ExtensionMimeGuesser())
 
     def tearDown(self):
-        os.remove("test_parse")
+        os.remove("test_parse.txt")
 
     def test_parse_size(self):
-        result = self.parser.parse("test_parse")
+        result = self.parser.parse("test_parse.txt")
 
         self.assertEqual(result["size"], 8)
 
     def test_parse_name(self):
-        result = self.parser.parse("test_parse")
+        result = self.parser.parse("test_parse.txt")
 
-        self.assertEqual(result["name"], "test_parse")
+        self.assertEqual(result["name"], "test_parse.txt")
 
     def test_parse_md5(self):
-        result = self.parser.parse("test_parse")
+        result = self.parser.parse("test_parse.txt")
 
         self.assertEqual(result["md5"], "25D55AD283AA400AF464C76D713C07AD")
 
+    def test_mtime(self):
+
+        result = self.parser.parse("test_parse.txt")
+
+        self.assertEqual(result["mtime"], 1330654321)
+
+    def test_mime(self):
+
+        result = self.parser.parse("test_parse.txt")
+        self.assertEqual(result["mime"], "text/plain")
+
 
 class Md5CheckSumCalculatorTest(TestCase):
 
@@ -128,4 +141,7 @@ class Sha256CheckSumCalculatorTest(TestCase):
         self.assertEqual(result, "DA7606DC763306B700685A71E2E72A2D95F1291209E5DA344B82DA2508FC27C5")
 
         result = self.calculator.checksum("test_sha256_2")
-        self.assertEqual(result, "C39C7E0E7D84C9692F3C9C22E1EA0327DEBF1BF531B5738EEA8E79FE27EBC570")
+        self.assertEqual(result, "C39C7E0E7D84C9692F3C9C22E1EA0327DEBF1BF531B5738EEA8E79FE27EBC570")
+
+
+

spec/Indexer_spec.py

@@ -0,0 +1,16 @@
+from unittest import TestCase
+from indexer import Indexer
+
+
+class IndexerTest(TestCase):
+
+    def test_create_bulk_query(self):
+
+        docs = [{"name": "doc1"}, {"name": "doc2"}]
+
+        result = Indexer.create_bulk_index_string(docs, "indexName")
+
+        self.assertEqual(result, '{"index":{"_index":"indexName","_type":"file"}}\n'
+                                 '{"name": "doc1"}\n'
+                                 '{"index":{"_index":"indexName","_type":"file"}}\n'
+                                 '{"name": "doc2"}\n')

spec/LocalStorage_spec.py

@@ -1,10 +1,13 @@
 from unittest import TestCase
-from storage import LocalStorage, Directory, DuplicateDirectoryException, User
+from storage import LocalStorage, Directory, DuplicateDirectoryException, User, DuplicateUserException, Option
+import os
 
 
 class LocalStorageTest(TestCase):
 
     def setUp(self):
+        if os.path.exists("test_database.db"):
+            os.remove("test_database.db")
 
         s = LocalStorage("test_database.db")
         s.init_db("../database.sql")
@@ -13,42 +16,59 @@ class LocalStorageTest(TestCase):
 
         storage = LocalStorage("test_database.db")
 
-        d = Directory("/some/directory", True, ["opt1", "opt2", "opt3"])
+        d = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val2")], "An excellent name")
 
-        storage.save_directory(d)
+        dir_id = storage.save_directory(d)
+
+        self.assertEqual(storage.dirs()[dir_id].enabled, True)
+        self.assertEqual(storage.dirs()[dir_id].options[0].key, "key1")
+        self.assertEqual(storage.dirs()[dir_id].options[0].value, "val1")
+        self.assertEqual(storage.dirs()[dir_id].options[0].dir_id, 1)
 
-        self.assertEqual(storage.dirs()["/some/directory"].enabled, True)
-        self.assertEqual(storage.dirs()["/some/directory"].options[0], "opt1")
 
     def test_save_and_retrieve_dir_persistent(self):
 
         s1 = LocalStorage("test_database.db")
 
-        d = Directory("/some/directory", True, ["opt1", "opt2", "opt3"])
+        d = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val2")], "An excellent name")
 
-        s1.save_directory(d)
+        dir_id = s1.save_directory(d)
 
         s2 = LocalStorage("test_database.db")
-        self.assertEqual(s2.dirs()["/some/directory"].enabled, True)
-        self.assertEqual(s2.dirs()["/some/directory"].options[0], "opt1")
+        self.assertEqual(s2.dirs()[dir_id].enabled, True)
+
+        self.assertEqual(s2.dirs()[dir_id].options[0].key, "key1")
+        self.assertEqual(s2.dirs()[dir_id].options[0].value, "val1")
+        self.assertEqual(s2.dirs()[dir_id].options[0].dir_id, 1)
 
     def test_reject_duplicate_path(self):
 
         s = LocalStorage("test_database.db")
 
-        d1 = Directory("/some/directory", True, ["opt1", "opt2"])
-        d2 = Directory("/some/directory", True, ["opt1", "opt2"])
+        d1 = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val2")], "An excellent name")
+        d2 = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val2")], "An excellent name")
 
         s.save_directory(d1)
 
         with self.assertRaises(DuplicateDirectoryException) as e:
             s.save_directory(d2)
 
+    def test_remove_dir(self):
+        s = LocalStorage("test_database.db")
+
+        d = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val3")], "An excellent name")
+        dir_id = s.save_directory(d)
+
+        s.remove_directory(dir_id)
+
+        with self.assertRaises(KeyError):
+            _ = s.dirs()[dir_id]
+
     def test_save_and_retrieve_user(self):
 
         s = LocalStorage("test_database.db")
 
-        u = User("bob", "anHashedPassword", True)
+        u = User("bob", b"anHashedPassword", True)
 
         s.save_user(u)
 
@@ -66,12 +86,97 @@ class LocalStorageTest(TestCase):
 
         s = LocalStorage("test_database.db")
 
-        u = User("bob", b'$2b$14$VZEMbwAdy/HvLL/zh0.Iv.8XYnoZMz/LU9V4VKXLiuS.pthcUly2O', True)
+        u = User("bob", b'$2b$10$RakMb.3n/tl76sK7iVahJuklNYkR7f2Y4dsf73tPANwYBkp4VuJ7.', True)
 
         s.save_user(u)
 
         self.assertTrue(s.auth_user("bob", "test"))
         self.assertFalse(s.auth_user("bob", "wrong"))
-        self.assertFalse(s.auth_user("wrong", "test"))
 
         pass
+
+    def test_reject_duplicate_user(self):
+
+        s = LocalStorage("test_database.db")
+
+        u1 = User("user1", b"anHashedPassword", True)
+        u2 = User("user1", b"anotherHashedPassword", True)
+
+        s.save_user(u1)
+
+        with self.assertRaises(DuplicateUserException) as e:
+            s.save_user(u2)
+
+    def test_update_user(self):
+        s = LocalStorage("test_database.db")
+
+        u = User("neil", b"anHashedPassword", True)
+
+        s.save_user(u)
+
+        u.admin = False
+        s.update_user(u)
+
+        self.assertFalse(s.users()["neil"].admin)
+
+    def test_remove_user(self):
+
+        s = LocalStorage("test_database.db")
+
+        u = User("martin", b"anHashedPassword", True)
+        s.save_user(u)
+
+        s.remove_user(u.username)
+
+        with self.assertRaises(KeyError):
+            _ = s.users()["martin"]
+
+    def test_update_directory(self):
+
+        s = LocalStorage("test_database.db")
+
+        d = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val2")], "An excellent name")
+
+        dir_id = s.save_directory(d)
+
+        d.name = "A modified name"
+        d.path = "/another/directory"
+        d.id = dir_id
+
+        s.update_directory(d)
+
+        s2 = LocalStorage("test_database.db")
+
+        self.assertEqual(s2.dirs()[dir_id].name, "A modified name")
+        self.assertEqual(len(s2.dirs()[dir_id].options), 2)
+        self.assertEqual(s2.dirs()[dir_id].path, "/another/directory")
+
+    def test_save_option(self):
+
+        s = LocalStorage("test_database.db")
+
+        d = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val2")], "An excellent name")
+        dir_id = s.save_directory(d)
+
+        opt_id = s.save_option(Option("key3", "val3", dir_id))
+
+        self.assertEqual(s.dirs()[dir_id].options[2].key, "key3")
+        self.assertEqual(s.dirs()[dir_id].options[2].value, "val3")
+        self.assertEqual(s.dirs()[dir_id].options[2].dir_id, dir_id)
+        self.assertEqual(opt_id, 3)
+
+    def test_del_option(self):
+
+        s = LocalStorage("test_database.db")
+
+        d = Directory("/some/directory", True, [Option("key1", "val1"), Option("key2", "val2")], "An excellent name")
+        dir_id = s.save_directory(d)
+        s.del_option(1)
+
+        self.assertEqual(len(s.dirs()[dir_id].options), 1)
+        self.assertEqual(s.dirs()[dir_id].options[0].key, "key2")
+        self.assertEqual(s.dirs()[dir_id].options[0].value, "val2")
+        self.assertEqual(s.dirs()[dir_id].options[0].dir_id, 1)
+
+
+

spec/MimeGuesser_spec.py

@@ -0,0 +1,23 @@
+from parsing import ContentMimeGuesser, ExtensionMimeGuesser
+from unittest import TestCase
+
+
+class MimeGuesserTest(TestCase):
+
+    def test_content_guesser(self):
+
+        guesser = ContentMimeGuesser()
+
+        self.assertEqual("text/x-shellscript", guesser.guess_mime("test_folder/test_utf8.sh"))
+        self.assertEqual("text/plain", guesser.guess_mime("test_folder/more_books.json"))
+        self.assertEqual("application/java-archive", guesser.guess_mime("test_folder/post.jar"))
+        self.assertEqual("image/jpeg", guesser.guess_mime("test_folder/sample_1.jpg"))
+
+    def test_extension_guesser(self):
+
+        guesser = ExtensionMimeGuesser()
+
+        self.assertEqual("text/x-sh", guesser.guess_mime("test_folder/test_utf8.sh"))
+        self.assertEqual("application/json", guesser.guess_mime("test_folder/more_books.json"))
+        self.assertEqual("application/java-archive", guesser.guess_mime("test_folder/post.jar"))
+        self.assertEqual("image/jpeg", guesser.guess_mime("test_folder/sample_1.jpg"))