Commit 89e01c0e authored by Andrea Azzarone's avatar Andrea Azzarone 🚴

tests: Reformat in line with Python standards

Python has a recommended code style, known as PEP-8. The tests now follow this
style.

This commit was created using `autopep8` version 1.4.3, run with this command:

	find . -name "*.py" -exec autopep8 --ignore E501 --in-place {} \;
parent eb2b5c5c
......@@ -35,182 +35,183 @@ from common.utils.minertest import CommonTrackerMinerTest
NFO_DOCUMENT = 'http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#Document'
class MinerCrawlTest (CommonTrackerMinerTest):
"""
Test cases to check if miner is able to monitor files that are created, deleted or moved
"""
def __get_text_documents (self):
return self.tracker.query ("""
def __get_text_documents(self):
return self.tracker.query("""
SELECT ?url WHERE {
?u a nfo:TextDocument ;
nie:url ?url.
}
""")
def __get_parent_urn (self, filepath):
result = self.tracker.query ("""
def __get_parent_urn(self, filepath):
result = self.tracker.query("""
SELECT nfo:belongsToContainer(?u) WHERE {
?u a nfo:FileDataObject ;
nie:url \"%s\" .
}
""" % (self.uri (filepath)))
self.assertEqual (len (result), 1)
""" % (self.uri(filepath)))
self.assertEqual(len(result), 1)
return result[0][0]
def __get_file_urn (self, filepath):
result = self.tracker.query ("""
def __get_file_urn(self, filepath):
result = self.tracker.query("""
SELECT ?u WHERE {
?u a nfo:FileDataObject ;
nie:url \"%s\" .
}
""" % (self.uri (filepath)))
self.assertEqual (len (result), 1)
""" % (self.uri(filepath)))
self.assertEqual(len(result), 1)
return result[0][0]
"""
Boot the miner with the correct configuration and check everything is fine
"""
def test_01_initial_crawling (self):
def test_01_initial_crawling(self):
"""
The precreated files and folders should be there
"""
result = self.__get_text_documents ()
self.assertEqual (len (result), 3)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 3)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
# We don't check (yet) folders, because Applications module is injecting results
## class copy(TestUpdate):
## FIXME all tests in one class because the miner-fs restarting takes some time (~5 sec)
## Maybe we can move the miner-fs initialization to setUpModule and then move these
## tests to different classes
# class copy(TestUpdate):
# FIXME all tests in one class because the miner-fs restarting takes some time (~5 sec)
# Maybe we can move the miner-fs initialization to setUpModule and then move these
# tests to different classes
def test_02_copy_from_unmonitored_to_monitored (self):
def test_02_copy_from_unmonitored_to_monitored(self):
"""
Copy an file from unmonitored directory to monitored directory
and verify if data base is updated accordingly
"""
source = os.path.join (self.workdir, "test-no-monitored", "file0.txt")
dest = os.path.join (self.workdir, "test-monitored", "file0.txt")
shutil.copyfile (source, dest)
source = os.path.join(self.workdir, "test-no-monitored", "file0.txt")
dest = os.path.join(self.workdir, "test-monitored", "file0.txt")
shutil.copyfile(source, dest)
dest_id, dest_urn = self.system.store.await_resource_inserted (NFO_DOCUMENT, self.uri(dest))
dest_id, dest_urn = self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(dest))
# verify if miner indexed this file.
result = self.__get_text_documents ()
self.assertEqual (len (result), 4)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/file0.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 4)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/file0.txt"), unpacked_result)
# Clean the new file so the test directory is as before
log ("Remove and wait")
os.remove (dest)
self.system.store.await_resource_deleted (NFO_DOCUMENT, dest_id)
log("Remove and wait")
os.remove(dest)
self.system.store.await_resource_deleted(NFO_DOCUMENT, dest_id)
def test_03_copy_from_monitored_to_unmonitored (self):
def test_03_copy_from_monitored_to_unmonitored(self):
"""
Copy an file from a monitored location to an unmonitored location
Nothing should change
"""
# Copy from monitored to unmonitored
source = os.path.join (self.workdir, "test-monitored", "file1.txt")
dest = os.path.join (self.workdir, "test-no-monitored", "file1.txt")
shutil.copyfile (source, dest)
source = os.path.join(self.workdir, "test-monitored", "file1.txt")
dest = os.path.join(self.workdir, "test-no-monitored", "file1.txt")
shutil.copyfile(source, dest)
time.sleep (1)
time.sleep(1)
# Nothing changed
result = self.__get_text_documents ()
self.assertEqual (len (result), 3, "Results:" + str(result))
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 3, "Results:" + str(result))
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
# Clean the file
os.remove (dest)
os.remove(dest)
def test_04_copy_from_monitored_to_monitored (self):
def test_04_copy_from_monitored_to_monitored(self):
"""
Copy a file between monitored directories
"""
source = os.path.join (self.workdir, "test-monitored", "file1.txt")
dest = os.path.join (self.workdir, "test-monitored", "dir1", "dir2", "file-test04.txt")
shutil.copyfile (source, dest)
source = os.path.join(self.workdir, "test-monitored", "file1.txt")
dest = os.path.join(self.workdir, "test-monitored", "dir1", "dir2", "file-test04.txt")
shutil.copyfile(source, dest)
dest_id, dest_urn = self.system.store.await_resource_inserted (NFO_DOCUMENT, self.uri(dest))
dest_id, dest_urn = self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(dest))
result = self.__get_text_documents ()
self.assertEqual (len (result), 4)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file-test04.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 4)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file-test04.txt"), unpacked_result)
# Clean the file
os.remove (dest)
self.system.store.await_resource_deleted (NFO_DOCUMENT, dest_id)
self.assertEqual (3, self.tracker.count_instances ("nfo:TextDocument"))
os.remove(dest)
self.system.store.await_resource_deleted(NFO_DOCUMENT, dest_id)
self.assertEqual(3, self.tracker.count_instances("nfo:TextDocument"))
@ut.skip("https://gitlab.gnome.org/GNOME/tracker-miners/issues/56")
def test_05_move_from_unmonitored_to_monitored (self):
def test_05_move_from_unmonitored_to_monitored(self):
"""
Move a file from unmonitored to monitored directory
"""
source = os.path.join (self.workdir, "test-no-monitored", "file0.txt")
dest = os.path.join (self.workdir, "test-monitored", "dir1", "file-test05.txt")
shutil.move (source, dest)
dest_id, dest_urn = self.system.store.await_resource_inserted (NFO_DOCUMENT, self.uri(dest))
result = self.__get_text_documents ()
self.assertEqual (len (result), 4)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/file-test05.txt"), unpacked_result)
source = os.path.join(self.workdir, "test-no-monitored", "file0.txt")
dest = os.path.join(self.workdir, "test-monitored", "dir1", "file-test05.txt")
shutil.move(source, dest)
dest_id, dest_urn = self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(dest))
result = self.__get_text_documents()
self.assertEqual(len(result), 4)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/file-test05.txt"), unpacked_result)
# Clean the file
os.remove (dest)
self.system.store.await_resource_deleted (NFO_DOCUMENT, dest_id)
self.assertEqual (3, self.tracker.count_instances ("nfo:TextDocument"))
os.remove(dest)
self.system.store.await_resource_deleted(NFO_DOCUMENT, dest_id)
self.assertEqual(3, self.tracker.count_instances("nfo:TextDocument"))
## """ move operation and tracker-miner response test cases """
## class move(TestUpdate):
# class move(TestUpdate):
def test_06_move_from_monitored_to_unmonitored (self):
def test_06_move_from_monitored_to_unmonitored(self):
"""
Move a file from monitored to unmonitored directory
"""
source = self.path("test-monitored/dir1/file2.txt")
dest = self.path("test-no-monitored/file2.txt")
source_id = self.system.store.get_resource_id (self.uri(source))
shutil.move (source, dest)
self.system.store.await_resource_deleted (NFO_DOCUMENT, source_id)
source_id = self.system.store.get_resource_id(self.uri(source))
shutil.move(source, dest)
self.system.store.await_resource_deleted(NFO_DOCUMENT, source_id)
result = self.__get_text_documents ()
self.assertEqual (len (result), 2)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 2)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
# Restore the file
shutil.move (dest, source)
self.system.store.await_resource_inserted (NFO_DOCUMENT, self.uri(source))
self.assertEqual (3, self.tracker.count_instances ("nfo:TextDocument"))
shutil.move(dest, source)
self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(source))
self.assertEqual(3, self.tracker.count_instances("nfo:TextDocument"))
def test_07_move_from_monitored_to_monitored (self):
def test_07_move_from_monitored_to_monitored(self):
"""
Move a file between monitored directories
"""
......@@ -220,89 +221,89 @@ class MinerCrawlTest (CommonTrackerMinerTest):
resource_id = self.tracker.get_resource_id(url=self.uri(source))
source_dir_urn = self.__get_file_urn (os.path.dirname(source))
parent_before = self.__get_parent_urn (source)
self.assertEqual (source_dir_urn, parent_before)
source_dir_urn = self.__get_file_urn(os.path.dirname(source))
parent_before = self.__get_parent_urn(source)
self.assertEqual(source_dir_urn, parent_before)
shutil.move (source, dest)
shutil.move(source, dest)
self.tracker.await_property_changed(NFO_DOCUMENT, resource_id, 'nie:url')
# Checking fix for NB#214413: After a move operation, nfo:belongsToContainer
# should be changed to the new one
dest_dir_urn = self.__get_file_urn (os.path.dirname(dest))
parent_after = self.__get_parent_urn (dest)
self.assertNotEqual (parent_before, parent_after)
self.assertEqual (dest_dir_urn, parent_after)
result = self.__get_text_documents ()
self.assertEqual (len (result), 3)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/file2.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
dest_dir_urn = self.__get_file_urn(os.path.dirname(dest))
parent_after = self.__get_parent_urn(dest)
self.assertNotEqual(parent_before, parent_after)
self.assertEqual(dest_dir_urn, parent_after)
result = self.__get_text_documents()
self.assertEqual(len(result), 3)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/file2.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
# Restore the file
shutil.move (dest, source)
shutil.move(dest, source)
self.tracker.await_property_changed(NFO_DOCUMENT, resource_id, 'nie:url')
result = self.__get_text_documents ()
self.assertEqual (len (result), 3)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/dir1/file2.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 3)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
def test_08_deletion_single_file (self):
def test_08_deletion_single_file(self):
"""
Delete one of the files
"""
victim = self.path("test-monitored/dir1/file2.txt")
victim_id = self.system.store.get_resource_id (self.uri(victim))
os.remove (victim)
self.system.store.await_resource_deleted (NFO_DOCUMENT, victim_id)
victim_id = self.system.store.get_resource_id(self.uri(victim))
os.remove(victim)
self.system.store.await_resource_deleted(NFO_DOCUMENT, victim_id)
result = self.__get_text_documents ()
self.assertEqual (len (result), 2)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
self.assertIn ( self.uri ("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 2)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
# Restore the file
f = open (victim, "w")
f.write ("Don't panic, everything is fine")
f.close ()
self.system.store.await_resource_inserted (NFO_DOCUMENT, self.uri(victim))
f = open(victim, "w")
f.write("Don't panic, everything is fine")
f.close()
self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(victim))
def test_09_deletion_directory (self):
def test_09_deletion_directory(self):
"""
Delete a directory
"""
victim = self.path("test-monitored/dir1")
victim_id = self.system.store.get_resource_id (self.uri(victim))
shutil.rmtree (victim)
victim_id = self.system.store.get_resource_id(self.uri(victim))
shutil.rmtree(victim)
file_inside_victim_url = self.uri (os.path.join (victim, "file2.txt"))
file_inside_victim_id = self.system.store.get_resource_id (file_inside_victim_url)
self.system.store.await_resource_deleted (NFO_DOCUMENT, file_inside_victim_id)
file_inside_victim_url = self.uri(os.path.join(victim, "file2.txt"))
file_inside_victim_id = self.system.store.get_resource_id(file_inside_victim_url)
self.system.store.await_resource_deleted(NFO_DOCUMENT, file_inside_victim_id)
result = self.__get_text_documents ()
self.assertEqual (len (result), 1)
unpacked_result = [ r[0] for r in result]
self.assertIn ( self.uri ("test-monitored/file1.txt"), unpacked_result)
result = self.__get_text_documents()
self.assertEqual(len(result), 1)
unpacked_result = [r[0] for r in result]
self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
# Restore the dirs
os.makedirs (self.path("test-monitored/dir1"))
os.makedirs (self.path("test-monitored/dir1/dir2"))
os.makedirs(self.path("test-monitored/dir1"))
os.makedirs(self.path("test-monitored/dir1/dir2"))
for f in ["test-monitored/dir1/file2.txt",
"test-monitored/dir1/dir2/file3.txt"]:
filename = self.path(f)
writer = open (filename, "w")
writer.write ("Don't panic, everything is fine")
writer.close ()
self.system.store.await_resource_inserted (NFO_DOCUMENT, self.uri(f))
writer = open(filename, "w")
writer.write("Don't panic, everything is fine")
writer.close()
self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(f))
# Check everything is fine
result = self.__get_text_documents ()
self.assertEqual (len (result), 3)
result = self.__get_text_documents()
self.assertEqual(len(result), 3)
if __name__ == "__main__":
ut.main(failfast=True)
......@@ -37,57 +37,57 @@ NMM_MUSIC_PIECE = 'http://www.tracker-project.org/temp/nmm#MusicPiece'
class MinerResourceRemovalTest (CommonTrackerMinerTest):
def prepare_directories (self):
def prepare_directories(self):
# Override content from the base class
pass
def create_test_content (self, file_urn, title):
def create_test_content(self, file_urn, title):
sparql = "INSERT { \
_:ie a nmm:MusicPiece ; \
nie:title \"%s\" ; \
nie:isStoredAs <%s> \
} " % (title, file_urn)
self.tracker.update (sparql)
self.tracker.update(sparql)
return self.tracker.await_resource_inserted (rdf_class = NMM_MUSIC_PIECE,
title = title)
return self.tracker.await_resource_inserted(rdf_class=NMM_MUSIC_PIECE,
title=title)
def create_test_file (self, file_name):
def create_test_file(self, file_name):
file_path = self.path(file_name)
file = open (file_path, 'w')
file.write ("Test")
file.close ()
file = open(file_path, 'w')
file.write("Test")
file.close()
return self.tracker.await_resource_inserted (rdf_class = NFO_DOCUMENT,
url = self.uri(file_name))
return self.tracker.await_resource_inserted(rdf_class=NFO_DOCUMENT,
url=self.uri(file_name))
@ut.skip("https://gitlab.gnome.org/GNOME/tracker-miners/issues/57")
def test_01_file_deletion (self):
def test_01_file_deletion(self):
"""
Ensure every logical resource (nie:InformationElement) contained with
in a file is deleted when the file is deleted.
"""
(file_1_id, file_1_urn) = self.create_test_file ("test-monitored/test_1.txt")
(file_2_id, file_2_urn) = self.create_test_file ("test-monitored/test_2.txt")
(ie_1_id, ie_1_urn) = self.create_test_content (file_1_urn, "Test resource 1")
(ie_2_id, ie_2_urn) = self.create_test_content (file_2_urn, "Test resource 2")
(file_1_id, file_1_urn) = self.create_test_file("test-monitored/test_1.txt")
(file_2_id, file_2_urn) = self.create_test_file("test-monitored/test_2.txt")
(ie_1_id, ie_1_urn) = self.create_test_content(file_1_urn, "Test resource 1")
(ie_2_id, ie_2_urn) = self.create_test_content(file_2_urn, "Test resource 2")
os.unlink (self.path ("test-monitored/test_1.txt"))
os.unlink(self.path("test-monitored/test_1.txt"))
self.tracker.await_resource_deleted (NFO_DOCUMENT, file_1_id)
self.tracker.await_resource_deleted(NFO_DOCUMENT, file_1_id)
self.assertResourceMissing (file_1_urn)
self.assertResourceMissing(file_1_urn)
# Ensure the logical resource is deleted when the relevant file is
# removed.
self.assertResourceMissing (ie_1_urn)
self.assertResourceMissing(ie_1_urn)
self.assertResourceExists (file_2_urn)
self.assertResourceExists (ie_2_urn)
self.assertResourceExists(file_2_urn)
self.assertResourceExists(ie_2_urn)
#def test_02_removable_device_data (self):
# def test_02_removable_device_data (self):
# """
# Tracker does periodic cleanups of data on removable volumes that haven't
# been seen since 'removable-days-threshold', and will also remove all data
......
#!/usr/bin/env python3
#-*- coding: utf-8 -*-
# -*- coding: utf-8 -*-
# Copyright (C) 2010, Nokia (ivan.frade@nokia.com)
#
......@@ -42,67 +42,66 @@ class MinerFTSBasicTest (CommonTrackerMinerFTSTest):
Tests different contents in a single file
"""
def test_01_single_word (self):
def test_01_single_word(self):
TEXT = "automobile"
self.basic_test (TEXT, TEXT)
self.basic_test(TEXT, TEXT)
def test_02_multiple_words (self):
def test_02_multiple_words(self):
TEXT = "automobile with unlimited power"
self.set_text (TEXT)
results = self.search_word ("automobile")
self.assertEqual (len (results), 1)
self.assertIn (self.uri (self.testfile), results)
self.set_text(TEXT)
results = self.search_word ("unlimited")
self.assertEqual (len (results), 1)
self.assertIn (self.uri (self.testfile), results)
results = self.search_word("automobile")
self.assertEqual(len(results), 1)
self.assertIn(self.uri(self.testfile), results)
results = self.search_word("unlimited")
self.assertEqual(len(results), 1)
self.assertIn(self.uri(self.testfile), results)
def test_03_long_word (self):
def test_03_long_word(self):
# TEXT is longer than the 20 characters specified in the fts configuration
TEXT = "fsfsfsdfskfweeqrewqkmnbbvkdasdjefjewriqjfnc"
self.set_text (TEXT)
self.set_text(TEXT)
results = self.search_word (TEXT)
self.assertEqual (len (results), 0)
results = self.search_word(TEXT)
self.assertEqual(len(results), 0)
def test_04_non_existent_word (self):
def test_04_non_existent_word(self):
TEXT = "This a trick"
self.set_text (TEXT)
results = self.search_word ("trikc")
self.assertEqual (len (results), 0)
self.set_text(TEXT)
results = self.search_word("trikc")
self.assertEqual(len(results), 0)
def test_05_word_multiple_times_in_file (self):
def test_05_word_multiple_times_in_file(self):