Commit b5ded9fa authored by Sam Thursfield's avatar Sam Thursfield

Merge branch 'python3-fixes' into 'master'

Move tests to python3 and reformat with autopep8

See merge request !55
parents 67958708 89e01c0e
Pipeline #69327 passed with stage
in 3 minutes and 19 seconds
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Copyright (C) 2009, Nokia <ivan.frade@nokia.com>
#
......
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Copyright (C) 2009, Nokia <ivan.frade@nokia.com>
#
......@@ -64,14 +64,14 @@ class AbstractTextEngine:
def send_data_cb (self):
sparql = self.get_insert_sparql ()
self.iface.SparqlUpdate (sparql)
print int(time.time()), self.msgsize, self.publicname
print(int(time.time()), self.msgsize, self.publicname)
return True
def get_insert_sparql (self):
print "Implement this method in a subclass!!!"
print("Implement this method in a subclass!!!")
assert False
def get_running_label (self):
print "Implement this method in a subclass!!!"
print("Implement this method in a subclass!!!")
assert False
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Copyright (C) 2009, Nokia <ivan.frade@nokia.com>
#
......@@ -47,23 +47,23 @@ def run_query ():
start = time.time ()
results = iface.SparqlQuery (SPARQL_QUERY % ("10"))
end = time.time ()
print int (time.time()), "%f" % (end - start)
print(int (time.time()), "%f" % (end - start))
return True
def exit_cb ():
sys.exit (0)
def usage ():
print "Usage:"
print " client.py [OPTION...] - Run periodically a query on tracker"
print ""
print "Help Options:"
print " -h, --help Show help options"
print ""
print "Application Options:"
print " -p, --period=NUM Time (in sec) between queries"
print " -t, --timeout=NUM Switch off the program after NUM seconds"
print ""
print("Usage:")
print(" client.py [OPTION...] - Run periodically a query on tracker")
print("")
print("Help Options:")
print(" -h, --help Show help options")
print("")
print("Application Options:")
print(" -p, --period=NUM Time (in sec) between queries")
print(" -t, --timeout=NUM Switch off the program after NUM seconds")
print("")
if __name__ == "__main__":
......
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Copyright (C) 2009, Nokia <ivan.frade@nokia.com>
#
......
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Copyright (C) 2009, Nokia <ivan.frade@nokia.com>
#
......@@ -22,18 +22,18 @@ import getopt
import sys
def usage ():
print "Usage:"
print " xxx.py [OPTION...] - Input data into tracker"
print ""
print "Help Options:"
print " -h, --help Show help options"
print ""
print "Application Options:"
print " -g, --graphics Enable GTK interface"
print " -p, --period=NUM Time (in sec) between insertion message"
print " -s, --size=NUM Amount of instances in the message"
print " -t, --timeout=NUM Switch off the program after NUM seconds"
print ""
print("Usage:")
print(" xxx.py [OPTION...] - Input data into tracker")
print("")
print("Help Options:")
print(" -h, --help Show help options")
print("")
print("Application Options:")
print(" -g, --graphics Enable GTK interface")
print(" -p, --period=NUM Time (in sec) between insertion message")
print(" -s, --size=NUM Amount of instances in the message")
print(" -t, --timeout=NUM Switch off the program after NUM seconds")
print("")
def parse_options (graphic_mode=False, period=1, msgsize=1, timeout=0):
......@@ -41,9 +41,9 @@ def parse_options (graphic_mode=False, period=1, msgsize=1, timeout=0):
opts, args = getopt.getopt(sys.argv[1:],
"gp:s:t:h",
["graphics", "period", "size", "timeout", "help"])
except getopt.GetoptError, err:
except getopt.GetoptError as err:
# print help information and exit:
print str (err) # will print something like "option -a not recognized"
print(str (err)) # will print something like "option -a not recognized"
usage ()
sys.exit (2)
......
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Copyright (C) 2009, Nokia <ivan.frade@nokia.com>
#
......
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Copyright (C) 2009, Nokia <ivan.frade@nokia.com>
#
......
#!/usr/bin/python
#!/usr/bin/env python3
# Copyright (C) 2010, Nokia (ivan.frade@nokia.com)
#
......@@ -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/python
#-*- coding: utf-8 -*-
#!/usr/bin/env python3
# -*- 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.assertEquals (len (results), 1)
self.assertIn (self.uri (self.testfile), results)
self.set_text(TEXT)
results = self.search_word ("unlimited")
self.assertEquals (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.assertEquals (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.assertEquals (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):
TEXT = "automobile is red. automobile is big. automobile is great!"
self.basic_test (TEXT, "automobile")
self.basic_test(TEXT, "automobile")
def test_06_sentence (self):
def test_06_sentence(self):
TEXT = "plastic is fantastic"
self.basic_test (TEXT, TEXT)
self.basic_test(TEXT, TEXT)
def test_07_partial_sentence (self):
def test_07_partial_sentence(self):
TEXT = "plastic is fantastic"
self.basic_test (TEXT, "is fantastic")
self.basic_test(TEXT, "is fantastic")
@ut.skip("Currently fails with: fts5: syntax error near '.'")
def test_08_strange_word (self):
def test_08_strange_word(self):
# FIXME Not sure what are we testing here
TEXT = "'summer.time'"
self.basic_test (TEXT, "summer.time")
self.basic_test(TEXT, "summer.time")
# Skip the test 'search for .'
def test_09_mixed_letters_and_numbers (self):
def test_09_mixed_letters_and_numbers(self):
TEXT = "abc123"
self.basic_test (TEXT, "abc123")
self.basic_test(TEXT, "abc123")
def test_10_ignore_numbers (self):
def test_10_ignore_numbers(self):
TEXT = "palabra 123123"
self.set_text (TEXT)
results = self.search_word ("123123")
self.assertEquals (len (results), 0)
self.set_text(TEXT)
results = self.search_word("123123")
self.assertEqual(len(results), 0)
if __name__ == "__main__":
ut.main (failfast=True)
ut.main(failfast=True)
#!/usr/bin/python
#-*- coding: utf-8 -*-
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2010, Nokia (ivan.frade@nokia.com)
#
......@@ -45,73 +45,72 @@ class MinerFTSFileOperationsTest (CommonTrackerMinerFTSTest):
Move, update, delete the files and check the text indexes are updated accordingly.
"""
def test_01_removal_of_file (self):
def test_01_removal_of_file(self):
"""
When removing the file, its text contents disappear from the index
"""
TEXT = "automobile is red and big and whatnot"
self.basic_test (TEXT, "automobile")
self.basic_test(TEXT, "automobile")
id = self._query_id (self.uri (self.testfile))
os.remove (self.path (self.testfile))
self.tracker.await_resource_deleted (NFO_DOCUMENT, id)
id = self._query_id(self.uri(self.testfile))
os.remove(self.path(self.testfile))
self.tracker.await_resource_deleted(NFO_DOCUMENT, id)
results = self.search_word ("automobile")
self.assertEquals (len (results), 0)
results = self.search_word("automobile")
self.assertEqual(len(results), 0)
def test_02_empty_the_file (self):
def test_02_empty_the_file(self):
"""
Emptying the file, the indexed words are also removed
"""
TEXT = "automobile is red and big and whatnot"
self.basic_test (TEXT, "automobile")
self.basic_test(TEXT, "automobile")
self.set_text ("")
results = self.search_word ("automobile")
self.assertEquals (len (results), 0)
self.set_text("")
results = self.search_word("automobile")
self.assertEqual(len(results), 0)
@ut.skip("FIXME: this test fails!")
def test_03_update_the_file (self):
def test_03_update_the_file(self):
"""
Changing the contents of the file, updates the index
"""
TEXT = "automobile is red and big and whatnot"
self.basic_test (TEXT, "automobile")
self.basic_test(TEXT, "automobile")
self.set_text ("airplane is blue and small and wonderful")
self.set_text("airplane is blue and small and wonderful")
results = self.search_word ("automobile")
self.assertEquals (len (results), 0)
results = self.search_word("automobile")
self.assertEqual(len(results), 0)
results = self.search_word ("airplane")
self.assertEquals (len (results), 1)
results = self.search_word("airplane")
self.assertEqual(len(results), 1)
# Skip the test_text_13... feel, feet, fee in three diff files and search feet
def __recreate_file (self, filename, content):
if os.path.exists (filename):
os.remove (filename)
def __recreate_file(self, filename, content):
if os.path.exists(filename):
os.remove(filename)
f = open (filename, "w")
f.write (content)
f.close ()
f = open(filename, "w")
f.write(content)
f.close()
def test_04_on_unmonitored_file (self):
def test_04_on_unmonitored_file(self):
"""
Set text in an unmonitored file. There should be no results.
"""
TEXT = "automobile is red"
TEST_15_FILE = "test-no-monitored/fts-indexing-test-15.txt"
self.__recreate_file (self.path (TEST_15_FILE), TEXT)
self.__recreate_file(self.path(TEST_15_FILE), TEXT)
results = self.search_word ("automobile")
self.assertEquals (len (results), 0)
results = self.search_word("automobile")
self.assertEqual(len(results), 0)
os.remove (self.path (TEST_15_FILE))
os.remove(self.path(TEST_15_FILE))
def test_05_move_file_unmonitored_monitored (self):
def test_05_move_file_unmonitored_monitored(self):
"""
Move file from unmonitored location to monitored location and index should be updated
"""
......@@ -119,27 +118,27 @@ class MinerFTSFileOperationsTest (CommonTrackerMinerFTSTest):
TEXT = "airplane is beautiful"
TEST_16_SOURCE = "test-no-monitored/fts-indexing-text-16.txt"
TEST_16_DEST = "test-monitored/fts-indexing-text-16.txt"
self.__recreate_file (self.path (TEST_16_SOURCE), TEXT)
self.__recreate_file(self.path(TEST_16_SOURCE), TEXT)
# the file is supposed to be ignored by tracker, so there is no notification..
time.sleep (2)
time.sleep(2)
results = self.search_word ("airplane")
self.assertEquals (len (results), 0)
results = self.search_word("airplane")
self.assertEqual(len(results), 0)
shutil.copyfile (self.path (TEST_16_SOURCE), self.path (TEST_16_DEST))
self.tracker.await_resource_inserted (rdf_class = NFO_DOCUMENT,
url = self.uri(TEST_16_DEST),
required_property = 'nie:plainTextContent')
shutil.copyfile(self.path(TEST_16_SOURCE), self.path(TEST_16_DEST))
self.tracker.await_resource_inserted(rdf_class=NFO_DOCUMENT,
url=self.uri(TEST_16_DEST),
required_property='nie:plainTextContent')
results = self.search_word ("airplane")
self.assertEquals (len (results), 1)
results = self.search_word("airplane")
self.assertEqual(len(results), 1)
os.remove (self.path (TEST_16_SOURCE))
os.remove (self.path (TEST_16_DEST))
os.remove(self.path(TEST_16_SOURCE))
os.remove(self.path(TEST_16_DEST))
# skip test for a file in a hidden directory
if __name__ == "__main__":
ut.main (failfast=True)
ut.main(failfast=True)
#!/usr/bin/python
#-*- coding: utf-8 -*-
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2010, Nokia (ivan.frade@nokia.com)
#
......@@ -36,50 +36,51 @@ from common.utils.helpers import log
from common.utils.minertest import CommonTrackerMinerFTSTest, DEFAULT_TEXT
from common.utils import configuration as cfg
class MinerFTSStopwordsTest (CommonTrackerMinerFTSTest):
"""
Search for stopwords in a file
"""
def __get_some_stopwords (self):
def __get_some_stopwords(self):
langcode, encoding = locale.getdefaultlocale ()
langcode, encoding = locale.getdefaultlocale()
if "_" in langcode:
langcode = langcode.split ("_")[0]
langcode = langcode.split("_")[0]
stopwordsfile = os.path.join (cfg.DATADIR, "tracker", "stop-words", "stopwords." + langcode)
stopwordsfile = os.path.join(cfg.DATADIR, "tracker", "stop-words", "stopwords." + langcode)
if not os.path.exists (stopwordsfile):
self.skipTest ("No stopwords for the current locale ('%s' doesn't exist)" % (stopwordsfile))
if not os.path.exists(stopwordsfile):
self.skipTest("No stopwords for the current locale ('%s' doesn't exist)" % (stopwordsfile))
return []
stopwords = []
counter = 0
for line in open (stopwordsfile, "r"):
if len (line) > 4:
stopwords.append (line[:-1])
for line in open(stopwordsfile, "r"):
if len(line) > 4:
stopwords.append(line[:-1])
counter += 1
if counter > 5:
break
return stopwords
def test_01_stopwords (self):
stopwords = self.__get_some_stopwords ()
TEXT = " ".join (["this a completely normal text automobile"] + stopwords)
self.set_text (TEXT)
results = self.search_word ("automobile")
self.assertEquals (len (results), 1)
log ("Stopwords: %s" % stopwords)
for i in range (0, len (stopwords)):
results = self.search_word (stopwords[i])
self.assertEquals (len (results), 0)
## FIXME add all the special character tests!
## http://git.gnome.org/browse/tracker/commit/?id=81c0d3bd754a6b20ac72323481767dc5b4a6217b
def test_01_stopwords(self):
stopwords = self.__get_some_stopwords()
TEXT = " ".join(["this a completely normal text automobile"] + stopwords)
self.set_text(TEXT)
results = self.search_word("automobile")
self.assertEqual(len(results), 1)
log("Stopwords: %s" % stopwords)
for i in range(0, len(stopwords)):
results = self.search_word(stopwords[i])
self.assertEqual(len(results), 0)
# FIXME add all the special character tests!
# http://git.gnome.org/browse/tracker/commit/?id=81c0d3bd754a6b20ac72323481767dc5b4a6217b
if __name__ == "__main__":
ut.main (failfast=True)
ut.main(failfast=True)
#!/usr/bin/python
#!/usr/bin/env python3
#
# Copyright (C) 2010, Nokia <ivan.frade@nokia.com>
# Copyright (C) 2018, Sam Thursfield <sam@afuera.me.uk>
......@@ -32,16 +32,16 @@ import os
import sys
class ExtractionTestCase (ut.TestCase):
"""
Test checks if the tracker extractor is able to retrieve metadata
"""
def __init__ (self, methodName='runTest', descfile=None):
def __init__(self, methodName='runTest', descfile=None):
"""
Descfile is the description file in a relative path
"""
ut.TestCase.__init__ (self, methodName)
ut.TestCase.__init__(self, methodName)
self.descfile = descfile
try:
......@@ -51,155 +51,152 @@ class ExtractionTestCase (ut.TestCase):
self.fail("Error loading %s: %s" % (descfile, e))
# Add a method to the class called after the description file
methodName = descfile.lower()[:-len(".expected")].replace (" ", "_")[-60:]
methodName = descfile.lower()[:-len(".expected")].replace(" ", "_")[-60:]
if (self.spec['test'].get('ExpectedFailure', False)):
setattr (self,
setattr(self,
methodName,
self.expected_failure_test_extraction)
else:
setattr (self, methodName, self.generic_test_extraction)
setattr(self, methodName, self.generic_test_extraction)
# unittest framework will run the test called "self._testMethodName"
# So we set that variable to our new name
self._testMethodName = methodName
def runTest (self):
def runTest(self):
"""
Empty function pointer, that should NEVER be called. It is required to exist by unittest.
"""
assert False
def __get_bugnumber (self):
def __get_bugnumber(self):
return self.spec['test'].get('Bugzilla')
def generic_test_extraction (self):
abs_description = os.path.abspath (self.descfile)
def generic_test_extraction(self):
abs_description = os.path.abspath(self.descfile)
# Filename contains the file to extract, in a relative path to the description file
desc_root, desc_file = os.path.split (abs_description)
desc_root, desc_file = os.path.split(abs_description)
filename_to_extract = self.spec['test']['Filename']
self.file_to_extract = os.path.join (desc_root, filename_to_extract)
self.file_to_extract = os.path.join(desc_root, filename_to_extract)
result = get_tracker_extract_jsonld_output(self.file_to_extract)
self.__assert_extraction_ok (result)
self.__assert_extraction_ok(result)
def expected_failure_test_extraction (self):
try:
self.generic_test_extraction ()
except Exception:
raise ut.case._ExpectedFailure(sys.exc_info())
@ut.expectedFailure
def expected_failure_test_extraction(self):
self.generic_test_extraction()
if self.__get_bugnumber ():
raise Exception ("Unexpected success. Maybe bug: " + self.__get_bugnumber () + " has been fixed?")
if self.__get_bugnumber():
raise Exception("Unexpected success. Maybe bug: " + self.__get_bugnumber() + " has been fixed?")
else:
raise Exception ("Unexpected success. Check " + self.rel_description)
raise Exception("Unexpected success. Check " + self.rel_description)
def assertDictHasKey (self, d, key, msg=None):
def assertDictHasKey(self, d, key, msg=None):
if not isinstance(d, dict):
self.fail ("Expected dict, got %s" % d)
if not d.has_key (key):
self.fail("Expected dict, got %s" % d)
if key not in d:
standardMsg = "Missing: %s\n" % (key)
self.fail (self._formatMessage (msg, standardMsg))
self.fail(self._formatMessage(msg, standardMsg))
else:
return
def assertIsURN (self, supposed_uuid, msg=None):
def assertIsURN(self, supposed_uuid, msg=None):
import uuid
try:
if (supposed_uuid.startswith ("<") and supposed_uuid.endswith (">")):