- tor-commits - lists.torproject.org

[stem/master] Unit tests for the load_processed_files() function
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 5e6623ce92e0bee08df9af18aacdd30548d47ad7 Author: Damian Johnson <atagar(a)torproject.org> Date: Wed Mar 7 20:28:44 2012 -0800 Unit tests for the load_processed_files() function Unit tests for stem.descriptor.reader's load_processed_files() function. This mocks the open call to simulate file reads and exercise a variety of good and malformed contents. This does not yet test for use cases where the file doesn't exists or fails to be read due to permissions, but those will be integ tests. --- run_tests.py | 2 + test/unit/descriptor/__init__.py | 6 ++ test/unit/descriptor/reader.py | 96 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 104 insertions(+), 0 deletions(-) diff --git a/run_tests.py b/run_tests.py index b59d3c3..7cf5117 100755 --- a/run_tests.py +++ b/run_tests.py @@ -17,6 +17,7 @@ import test.unit.connection.authentication import test.unit.connection.protocolinfo import test.unit.socket.control_line import test.unit.socket.control_message +import test.unit.descriptor.reader import test.unit.util.conf import test.unit.util.enum import test.unit.util.system @@ -80,6 +81,7 @@ UNIT_TESTS = ( test.unit.util.enum.TestEnum, test.unit.util.conf.TestConf, test.unit.util.system.TestSystem, + test.unit.descriptor.reader.TestDescriptorReader, test.unit.version.TestVersion, test.unit.socket.control_message.TestControlMessage, test.unit.socket.control_line.TestControlLine, diff --git a/test/unit/descriptor/__init__.py b/test/unit/descriptor/__init__.py new file mode 100644 index 0000000..ddcd821 --- /dev/null +++ b/test/unit/descriptor/__init__.py @@ -0,0 +1,6 @@ +""" +Unit tests for stem.descriptor. +""" + +__all__ = ["reader"] + diff --git a/test/unit/descriptor/reader.py b/test/unit/descriptor/reader.py new file mode 100644 index 0000000..cf11c23 --- /dev/null +++ b/test/unit/descriptor/reader.py @@ -0,0 +1,96 @@ +""" +Unit tests for stem.descriptor.reader. +""" + +import unittest +import StringIO + +import stem.descriptor.reader +import test.mocking as mocking + +class TestDescriptorReader(unittest.TestCase): + def tearDown(self): + mocking.revert_mocking() + + def test_load_processed_files(self): + """ + Successful load of content. + """ + + test_lines = ( + "/dir/ 0", + "/dir/file 12345", + "/dir/file with spaces 7138743", + " /dir/with extra space 12345 ", + " \t ", + "", + "/dir/after empty line 12345", + ) + + expected_value = { + "/dir/": 0, + "/dir/file": 12345, + "/dir/file with spaces": 7138743, + "/dir/with extra space": 12345, + "/dir/after empty line": 12345, + } + + test_content = StringIO.StringIO("\n".join(test_lines)) + mocking.support_with(test_content) + mocking.mock(open, mocking.return_value(test_content)) + self.assertEquals(expected_value, stem.descriptor.reader.load_processed_files("")) + + def test_load_processed_files_empty(self): + """ + Tests the load_processed_files() function with an empty file. + """ + + test_content = StringIO.StringIO("") + mocking.support_with(test_content) + mocking.mock(open, mocking.return_value(test_content)) + self.assertEquals({}, stem.descriptor.reader.load_processed_files("")) + + def test_load_processed_files_no_file(self): + """ + Tests the load_processed_files() function content that is malformed because + it is missing the file path. + """ + + test_content = StringIO.StringIO(" 12345") + mocking.support_with(test_content) + mocking.mock(open, mocking.return_value(test_content)) + self.assertRaises(TypeError, stem.descriptor.reader.load_processed_files, "") + + def test_load_processed_files_no_timestamp(self): + """ + Tests the load_processed_files() function content that is malformed because + it is missing the timestamp. + """ + + test_content = StringIO.StringIO("/dir/file ") + mocking.support_with(test_content) + mocking.mock(open, mocking.return_value(test_content)) + self.assertRaises(TypeError, stem.descriptor.reader.load_processed_files, "") + + def test_load_processed_files_malformed_file(self): + """ + Tests the load_processed_files() function content that is malformed because + it has an invalid file path. + """ + + test_content = StringIO.StringIO("not_an_absolute_file 12345") + mocking.support_with(test_content) + mocking.mock(open, mocking.return_value(test_content)) + self.assertRaises(TypeError, stem.descriptor.reader.load_processed_files, "") + + def test_load_processed_files_malformed_timestamp(self): + """ + Tests the load_processed_files() function content that is malformed because + it has a non-numeric timestamp. + """ + + test_content = StringIO.StringIO("/dir/file 123a") + mocking.support_with(test_content) + mocking.mock(open, mocking.return_value(test_content)) + self.assertRaises(TypeError, stem.descriptor.reader.load_processed_files, "") +

1 0

[stem/master] Integration tests for load_processed_files()
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit c547ed5688da5015dbddbb091118e351fef31bdd Author: Damian Johnson <atagar(a)torproject.org> Date: Thu Mar 8 08:27:51 2012 -0800 Integration tests for load_processed_files() Basic set of integ tests exercising file reading when successful, non-existant, or lacking permission. --- run_tests.py | 2 + test/integ/descriptor/__init__.py | 6 +++ test/integ/descriptor/reader.py | 74 +++++++++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 0 deletions(-) diff --git a/run_tests.py b/run_tests.py index 7cf5117..60118af 100755 --- a/run_tests.py +++ b/run_tests.py @@ -28,6 +28,7 @@ import test.integ.connection.protocolinfo import test.integ.control.base_controller import test.integ.socket.control_message import test.integ.socket.control_socket +import test.integ.descriptor.reader import test.integ.util.conf import test.integ.util.system import test.integ.version @@ -92,6 +93,7 @@ UNIT_TESTS = ( INTEG_TESTS = ( test.integ.util.conf.TestConf, test.integ.util.system.TestSystem, + test.integ.descriptor.reader.TestDescriptorReader, test.integ.version.TestVersion, test.integ.socket.control_socket.TestControlSocket, test.integ.socket.control_message.TestControlMessage, diff --git a/test/integ/descriptor/__init__.py b/test/integ/descriptor/__init__.py new file mode 100644 index 0000000..e1095ca --- /dev/null +++ b/test/integ/descriptor/__init__.py @@ -0,0 +1,6 @@ +""" +Integration tests for stem.descriptor.* contents. +""" + +__all__ = ["reader"] + diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py new file mode 100644 index 0000000..c8b93b0 --- /dev/null +++ b/test/integ/descriptor/reader.py @@ -0,0 +1,74 @@ +""" +Integration tests for stem.descriptor.reader. +""" + +import os +import unittest + +import stem.descriptor.reader +import test.runner + +BASIC_LISTING = """ +/tmp 123 +/bin/grep 4567 +/file with spaces/and \\ stuff 890 +""" + +def _get_processed_files_path(): + return os.path.join(test.runner.get_runner().get_test_dir(), "descriptor_processed_files") + +def _make_processed_files_listing(contents): + """ + Writes the given 'processed file' listing to disk, returning the path where + it is located. + """ + + test_listing_path = _get_processed_files_path() + + test_listing_file = open(test_listing_path, "w") + test_listing_file.write(contents) + test_listing_file.close() + + return test_listing_path + +class TestDescriptorReader(unittest.TestCase): + def tearDown(self): + # cleans up 'processed file' listings that we made + test_listing_path = _get_processed_files_path() + + if os.path.exists(test_listing_path): + os.remove(test_listing_path) + + def test_load_processed_files(self): + """ + Basic sanity test for loading a processed files listing from disk. + """ + + test_listing_path = _make_processed_files_listing(BASIC_LISTING) + loaded_listing = stem.descriptor.reader.load_processed_files(test_listing_path) + + expected_listing = { + "/tmp": 123, + "/bin/grep": 4567, + "/file with spaces/and \\ stuff": 890, + } + + self.assertEquals(expected_listing, loaded_listing) + + def test_load_processed_files_missing(self): + """ + Tests the load_processed_files() function with a file that doesn't exist. + """ + + self.assertRaises(IOError, stem.descriptor.reader.load_processed_files, "/non-existant/path") + + def test_load_processed_files_permissions(self): + """ + Tests the load_processed_files() function with a file that can't be read + due to permissions. + """ + + test_listing_path = _make_processed_files_listing(BASIC_LISTING) + os.chmod(test_listing_path, 0077) # remove read permissions + self.assertRaises(IOError, stem.descriptor.reader.load_processed_files, test_listing_path) +

1 0

[stem/master] Rewriting reader documentation
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 02d51e37e54aeefc77d906792d12e1bb711b89c5 Author: Damian Johnson <atagar(a)torproject.org> Date: Fri Mar 9 19:51:38 2012 -0800 Rewriting reader documentation Moving all of the examples and substantial documentation up into the header, and minorly revising how I intend for the module to be used. Not all of the methods in the header have been added yet. --- stem/descriptor/reader.py | 112 +++++++++++++++++++++++++++------------------ 1 files changed, 67 insertions(+), 45 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index a3ad418..b0c8f9a 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -1,7 +1,8 @@ """ -Reads descriptors from local directories and archives. +Utilities for reading descriptors from local directories and archives. This is +mostly done through the DescriptorReader class, which is an iterator for the +descriptor data in a series of destinations. For example... -Example: my_descriptors = [ "/tmp/server-descriptors-2012-03.tar.bz2", "/tmp/archived_descriptors/", @@ -9,10 +10,53 @@ Example: reader = DescriptorReader(my_descriptors) + # prints the contents of all the descriptor files with reader: - # prints all of the descriptor contents for descriptor in reader: print descriptor + +This ignores files that cannot be processed due to read errors or unparsable +content. To be notified of skipped files you can register a listener with +register_skip_listener(). + +The DescriptorReader keeps track of the last modified timestamps for descriptor +files that it has read so it can skip unchanged files if ran again. This +listing of processed files can also be persisted and applied to other +DescriptorReaders. For instance, the following prints descriptors as they're +changed over the course of a minute, and picks up where it left off if ran +again... + + reader = DescriptorReader(["/tmp/descriptor_data"]) + + try: + processed_files = load_processed_files("/tmp/used_descriptors") + reader.set_processed_files(processed_files) + except: pass # could not load, mabye this is the first run + + with reader: + start_time = time.time() + + while time.time() - start_time < 60: + # prints any descriptors that have changed since last checked + for descriptor in reader: + print descriptor + + time.sleep(1) + + save_processed_files(reader.get_processed_files(), "/tmp/used_descriptors") + + +load_processed_files - Loads a listing of processed files. +save_processed_files - Saves a listing of processed files. + +DescriptorReader - Iterator for descriptor data on the local file system. + |- get_processed_files - provides the listing of files that we've processed + |- set_processed_files - sets our tracking of the files we have processed + |- start - begins reading descriptor data + |- stop - stops reading descriptor data + |- join - joins on the thread used to process descriptor data + |- __enter__ / __exit__ - manages the descriptor reader thread in the context + +- __iter__ - iterates over descriptor data in unread files """ import os @@ -29,7 +73,7 @@ def load_processed_files(path): path (str) - location to load the processed files dictionary from Returns: - dict of 'path (str) => last modified timestamp (int)' mappings + dict of 'path (str) => last modified unix timestamp (int)' mappings Raises: IOError if unable to read the file @@ -90,32 +134,6 @@ class DescriptorReader(threading.Thread): """ Iterator for the descriptor data on the local file system. This can process text files, tarball archives (gzip or bzip2), or recurse directories. - - This keeps track the last modified timestamps for descriptor files we have - used, and if you call restart() then this will only provide descriptors from - new files or files that have changed since them. - - You can also save this listing of processed files and later apply it another - DescriptorReader. For instance, to only print the descriptors that have - changed since the last ran... - - reader = DescriptorReader(["/tmp/descriptor_data"]) - - try: - processed_files = load_processed_files("/tmp/used_descriptors") - reader.set_processed_files(processed_files) - except: pass # could not load, mabye this is the first run - - # only prints descriptors that have changed since we last ran - with reader: - for descriptor in reader: - print descriptor - - save_processed_files(reader.get_processed_files(), "/tmp/used_descriptors") - - This ignores files that cannot be processed (either due to read errors or - because they don't contain descriptor data). The caller can be notified of - files that are skipped by restering a listener with register_skip_listener(). """ def __init__(self, targets): @@ -124,19 +142,12 @@ class DescriptorReader(threading.Thread): self.processed_files = {} self._stop_event = threading.Event() - def stop(self): - """ - Stops further reading of descriptors. - """ - - self._stop_event.set() - def get_processed_files(self): """ - For each file we have provided descriptor data for this provides a mapping - of the form... + For each file that we have read descriptor data from this provides a + mapping of the form... - absolute_path (str) => modified_time (int) + absolute path (str) => last modified unix timestamp (int) This includes entries set through the set_processed_files() method. @@ -149,17 +160,16 @@ class DescriptorReader(threading.Thread): def set_processed_files(self, processed_files): """ - Appends a dictionary of 'path => modified timestamp' mappings to our - listing of processed files. With the get_processed_files() method this can - be used to skip descriptors that we have already read. For instance... - + Sets the listing of the files we have processed. Most often this is useful + as a method for pre-populating the listing of descriptor files that we have + seen. Arguments: processed_files (dict) - mapping of absolute paths (str) to unix timestamps for the last modified time (int) """ - self.processed_files.update(processed_files) + self.processed_files = dict(processed_files) def register_skip_listener(self, listener): """ @@ -176,6 +186,13 @@ class DescriptorReader(threading.Thread): self.skip_listeners.append(listener) + def stop(self): + """ + Stops further reading of descriptor files. + """ + + self._stop_event.set() + def run(self): # os.walk(path, followlinks = True) # @@ -184,6 +201,11 @@ class DescriptorReader(threading.Thread): # # >>> mimetypes.guess_type("/home/atagar/Desktop/server-descriptors-2012-03.tar.bz2") # ('application/x-tar', 'bzip2') + # + # This only checks the file extension. To actually check the content (like + # the 'file' command) an option would be pymagic... + # https://github.com/cloudburst/pymagic + while not self._stop_event.isSet(): pass # TODO: implement

1 0

[stem/master] Integration tests for save_processed_files()
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 86b90e84e26ac91e1f876ec96df4d983c5ab384f Author: Damian Johnson <atagar(a)torproject.org> Date: Thu Mar 8 09:04:47 2012 -0800 Integration tests for save_processed_files() Basic integ tests for the save function, checking that we can load persisted files and that it has some basic input validation. --- stem/descriptor/reader.py | 5 ++++- test/integ/descriptor/reader.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 1 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index ec6a2b2..a3ad418 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -81,7 +81,10 @@ def save_processed_files(processed_files, path): with open(path, "w") as output_file: for path, timestamp in processed_files.items(): - output_file.write("%s %i" % (path, timestamp)) + if not os.path.isabs(path): + raise TypeError("Only absolute paths are acceptable: %s" % path) + + output_file.write("%s %i\n" % (path, timestamp)) class DescriptorReader(threading.Thread): """ diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index c8b93b0..6341153 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -71,4 +71,34 @@ class TestDescriptorReader(unittest.TestCase): test_listing_path = _make_processed_files_listing(BASIC_LISTING) os.chmod(test_listing_path, 0077) # remove read permissions self.assertRaises(IOError, stem.descriptor.reader.load_processed_files, test_listing_path) + + def test_save_processed_files(self): + """ + Basic sanity test for persisting files listings to disk. + """ + + initial_listing = { + "/tmp": 123, + "/bin/grep": 4567, + "/file with spaces/and \\ stuff": 890, + } + + # saves the initial_listing to a file then reloads it + test_listing_path = _get_processed_files_path() + stem.descriptor.reader.save_processed_files(initial_listing, test_listing_path) + loaded_listing = stem.descriptor.reader.load_processed_files(test_listing_path) + + self.assertEquals(initial_listing, loaded_listing) + + def test_save_processed_files_malformed(self): + """ + Tests the save_processed_files() function with malformed data. + """ + + missing_filename = {"": 123} + relative_filename = {"foobar": 123} + string_timestamp = {"/tmp": "123a"} + + for listing in (missing_filename, relative_filename, string_timestamp): + self.assertRaises(TypeError, stem.descriptor.reader.save_processed_files, listing, "/tmp/foo")

1 0

[stem/master] Integ test for basic usage example
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 50ac2381a069a9e72094ff210954f9db1bdad0cb Author: Damian Johnson <atagar(a)torproject.org> Date: Fri Mar 9 20:12:42 2012 -0800 Integ test for basic usage example The header documentation contains two examples - the equivilant to 'hello world' with this class and an example of making use of the high water mark the DescriptorReader tracks. This adds an integ test for the first. At present this integ test dies in a fiery... well, fire. This is because the DescriptorReader isn't yet implemented. That's next... --- test/integ/descriptor/data/example_descriptor | 30 +++++++++++++++++++++++++ test/integ/descriptor/reader.py | 15 ++++++++++++ 2 files changed, 45 insertions(+), 0 deletions(-) diff --git a/test/integ/descriptor/data/example_descriptor b/test/integ/descriptor/data/example_descriptor new file mode 100644 index 0000000..5ab4f57 --- /dev/null +++ b/test/integ/descriptor/data/example_descriptor @@ -0,0 +1,30 @@ +router caerSidi 71.35.133.197 9001 0 0 +platform Tor 0.2.1.30 on Linux x86_64 +opt protocols Link 1 2 Circuit 1 +published 2012-03-01 17:15:27 +opt fingerprint A756 9A83 B570 6AB1 B1A9 CB52 EFF7 D2D3 2E45 53EB +uptime 588217 +bandwidth 153600 256000 104590 +opt extra-info-digest D225B728768D7EA4B5587C13A7A9D22EBBEE6E66 +onion-key +-----BEGIN RSA PUBLIC KEY----- +MIGJAoGBAJv5IIWQ+WDWYUdyA/0L8qbIkEVH/cwryZWoIaPAzINfrw1WfNZGtBmg +skFtXhOHHqTRN4GPPrZsAIUOQGzQtGb66IQgT4tO/pj+P6QmSCCdTfhvGfgTCsC+ +WPi4Fl2qryzTb3QO5r5x7T8OsG2IBUET1bLQzmtbC560SYR49IvVAgMBAAE= +-----END RSA PUBLIC KEY----- +signing-key +-----BEGIN RSA PUBLIC KEY----- +MIGJAoGBAKwvOXyztVKnuYvpTKt+nS3XIKeO8dVungi8qGoeS+6gkR6lDtGfBTjd +uE9UIkdAl9zi8/1Ic2wsUNHE9jiS0VgeupITGZY8YOyMJJ/xtV1cqgiWhq1dUYaq +51TOtUogtAPgXPh4J+V8HbFFIcCzIh3qCO/xXo+DSHhv7SSif1VpAgMBAAE= +-----END RSA PUBLIC KEY----- +family $0CE3CFB1E9CC47B63EA8869813BF6FAB7D4540C1 $1FD187E8F69A9B74C9202DC16A25B9E7744AB9F6 $74FB5EFA6A46DE4060431D515DC9A790E6AD9A7C $77001D8DA9BF445B0F81AA427A675F570D222E6A $B6D83EC2D9E18B0A7A33428F8CFA9C536769E209 $D2F37F46182C23AB747787FD657E680B34EAF892 $E0BD57A11F00041A9789577C53A1B784473669E4 $E5E3E9A472EAF7BE9682B86E92305DB4C71048EF +opt hidden-service-dir +contact www.atagar.com/contact +reject *:* +router-signature +-----BEGIN SIGNATURE----- +dskLSPz8beUW7bzwDjR6EVNGpyoZde83Ejvau+5F2c6cGnlu91fiZN3suE88iE6e +758b9ldq5eh5mapb8vuuV3uO+0Xsud7IEOqfxdkmk0GKnUX8ouru7DSIUzUL0zqq +Qlx9HNCqCY877ztFRC624ja2ql6A2hBcuoYMbkHjcQ4= +-----END SIGNATURE----- diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index 6341153..4ab2792 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -14,6 +14,9 @@ BASIC_LISTING = """ /file with spaces/and \\ stuff 890 """ +my_dir = os.path.dirname(__file__) +DESCRIPTOR_TEST_DATA = os.path.join(my_dir, "data") + def _get_processed_files_path(): return os.path.join(test.runner.get_runner().get_test_dir(), "descriptor_processed_files") @@ -101,4 +104,16 @@ class TestDescriptorReader(unittest.TestCase): for listing in (missing_filename, relative_filename, string_timestamp): self.assertRaises(TypeError, stem.descriptor.reader.save_processed_files, listing, "/tmp/foo") + + def test_basic_example(self): + """ + Exercises something similar to the first example in the header + documentation, checking that the contetns match the actual file. + """ + + reader = stem.descriptor.reader.DescriptorReader([DESCRIPTOR_TEST_DATA]) + + with reader: + for descriptor in reader: + print descriptor # TODO: change to be an automated check

1 0

[stem/master] Finishing basic DescriptorReader implementation
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 0aba5ba4704acb620915ba5f2115b7d036c9477a Author: Damian Johnson <atagar(a)torproject.org> Date: Sat Mar 10 22:55:50 2012 -0800 Finishing basic DescriptorReader implementation Fixing the minor issue that was preventing the DescriptorReader from working and making its first integration test do automated verification, rather than needing a manual check. --- stem/descriptor/reader.py | 8 +++++--- test/integ/descriptor/reader.py | 22 ++++++++++++++++++++-- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 8bdb5b4..0784ce4 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -171,6 +171,7 @@ class DescriptorReader(threading.Thread): # TODO: implement # flag that's set when we're done self._stop_event = threading.Event() + self._finished_reading = threading.Event() def get_processed_files(self): """ @@ -266,9 +267,7 @@ class DescriptorReader(threading.Thread): elif target_type[1] == 'bzip2': pass # TODO: implement - # TODO: bug: __iter__ should finish with the _unreturned_descriptors - # contents. Could be fixed by adding a 'is done reading' event. - self._stop_event.set() + self._finished_reading.set() self._iter_notice.set() def __iter__(self): @@ -277,6 +276,9 @@ class DescriptorReader(threading.Thread): try: yield self._unreturned_descriptors.get_nowait() except Queue.Empty: + # if we've finished and there aren't any descriptors then we're done + if self._finished_reading.isSet(): break + self._iter_notice.wait() self._iter_notice.clear() diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index 4ab2792..ca77e26 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -108,12 +108,30 @@ class TestDescriptorReader(unittest.TestCase): def test_basic_example(self): """ Exercises something similar to the first example in the header - documentation, checking that the contetns match the actual file. + documentation, checking that some of the contents match what we'd expect. """ + # snag some of the plaintext descriptors so we can later make sure that we + # iterate over them + + descriptor_entries = [] + + descriptor_path = os.path.join(DESCRIPTOR_TEST_DATA, "example_descriptor") + with open(descriptor_path) as descriptor_file: + descriptor_entries.append(descriptor_file.read()) + reader = stem.descriptor.reader.DescriptorReader([DESCRIPTOR_TEST_DATA]) with reader: for descriptor in reader: - print descriptor # TODO: change to be an automated check + descriptor_str = str(descriptor) + + if descriptor_str in descriptor_entries: + descriptor_entries.remove(descriptor_str) + else: + # iterator is providing output that we didn't expect + self.fail() + + # check that we've seen all of the descriptor_entries + self.assertTrue(len(descriptor_entries) == 0)

1 0

[stem/master] Implementation of DescriptorReader concurrency
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 2d620558fa0885c532e8759c497881146a89ba0a Author: Damian Johnson <atagar(a)torproject.org> Date: Sat Mar 10 18:09:52 2012 -0800 Implementation of DescriptorReader concurrency After much hair pulling figured out a relatively producer/consumer simple model for this class. It should be trivial to add stop() later, but making this re-runable would greatly complicate the class and probably isn't worth it. This isn't yet working, but this is a decent breaking point. --- stem/descriptor/reader.py | 114 ++++++++++++++++++++++++++++++++++++--------- 1 files changed, 92 insertions(+), 22 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index b0c8f9a..8bdb5b4 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -52,6 +52,7 @@ save_processed_files - Saves a listing of processed files. DescriptorReader - Iterator for descriptor data on the local file system. |- get_processed_files - provides the listing of files that we've processed |- set_processed_files - sets our tracking of the files we have processed + |- register_skip_listener - adds a listener that's notified of skipped files |- start - begins reading descriptor data |- stop - stops reading descriptor data |- join - joins on the thread used to process descriptor data @@ -64,6 +65,17 @@ import threading import mimetypes import Queue +# TODO: Unimplemented concurrency features... +# - stop() +# - restarting when __iter__ is called additional times +# - maximum read-ahead + +# Maximum number of descriptors that we'll read ahead before waiting for our +# caller to fetch some of them. This is included to avoid unbounded memory +# usage. This condition will be removed if set to zero. + +MAX_STORED_DESCRIPTORS = 20 + def load_processed_files(path): """ Loads a dictionary of 'path => last modified timestamp' mappings, as @@ -136,10 +148,28 @@ class DescriptorReader(threading.Thread): text files, tarball archives (gzip or bzip2), or recurse directories. """ - def __init__(self, targets): - self.targets = targets - self.skip_listeners = [] - self.processed_files = {} + def __init__(self, targets, follow_links = False): + threading.Thread.__init__(self, name="Descriptor Reader") + self.setDaemon(True) + + # flag to indicate if we'll follow symlinks when transversing directories + self._follow_links = follow_links + + self._targets = targets + self._skip_listeners = [] + self._processed_files = {} + + self._iter_lock = threading.RLock() + self._iter_notice = threading.Event() + + # targets that remain to be read in this iteration + self._remaining_files = list(self._targets) + + # descriptors that we have read, but not yet provided to the caller + self._unreturned_descriptors = Queue.Queue() + + # TODO: implement + # flag that's set when we're done self._stop_event = threading.Event() def get_processed_files(self): @@ -156,7 +186,7 @@ class DescriptorReader(threading.Thread): files we have processed """ - return dict(self.processed_files) + return dict(self._processed_files) def set_processed_files(self, processed_files): """ @@ -169,7 +199,7 @@ class DescriptorReader(threading.Thread): timestamps for the last modified time (int) """ - self.processed_files = dict(processed_files) + self._processed_files = dict(processed_files) def register_skip_listener(self, listener): """ @@ -184,7 +214,7 @@ class DescriptorReader(threading.Thread): valid descriptor data """ - self.skip_listeners.append(listener) + self._skip_listeners.append(listener) def stop(self): """ @@ -192,23 +222,63 @@ class DescriptorReader(threading.Thread): """ self._stop_event.set() + self._iter_notice.set() def run(self): - # os.walk(path, followlinks = True) - # - # >>> mimetypes.guess_type("/home/atagar/Desktop/control-spec.txt") - # ('text/plain', None) - # - # >>> mimetypes.guess_type("/home/atagar/Desktop/server-descriptors-2012-03.tar.bz2") - # ('application/x-tar', 'bzip2') - # - # This only checks the file extension. To actually check the content (like - # the 'file' command) an option would be pymagic... - # https://github.com/cloudburst/pymagic - - - while not self._stop_event.isSet(): - pass # TODO: implement + while self._remaining_files: + target = self._remaining_files.pop(0) + if not os.path.exists(target): continue + + if os.path.isdir(target): + # adds all of the files that it contains + for root, _, files in os.walk(target, followlinks = self._follow_links): + for filename in files: + self._remaining_files.append(os.path.join(root, filename)) + else: + # This is a file. Register it's last modified timestamp and check if + # it's a file that we should skip. + + last_modified = os.stat(target).st_mtime + last_used = self._processed_files.get(target) + + if last_used and last_used >= last_modified: + continue + else: + self._processed_files[target] = last_modified + + # The mimetypes module only checks the file extension. To actually + # check the content (like the 'file' command) we'd need something like + # pymagic (https://github.com/cloudburst/pymagic) + + target_type = mimetypes.guess_type(target) + + if target_type[0] in (None, 'text/plain'): + # if either a '.txt' or unknown type then try to process it as a + # descriptor file + + with open(target) as target_file: + # TODO: replace with actual descriptor parsing when we have it + self._unreturned_descriptors.put(target_file.read()) + self._iter_notice.set() + elif target_type[0] == 'application/x-tar': + if target_type[1] == 'gzip': + pass # TODO: implement + elif target_type[1] == 'bzip2': + pass # TODO: implement + + # TODO: bug: __iter__ should finish with the _unreturned_descriptors + # contents. Could be fixed by adding a 'is done reading' event. + self._stop_event.set() + self._iter_notice.set() + + def __iter__(self): + with self._iter_lock: + while not self._stop_event.isSet(): + try: + yield self._unreturned_descriptors.get_nowait() + except Queue.Empty: + self._iter_notice.wait() + self._iter_notice.clear() def _notify_skip_listener(self, path, exception): for listener in self.skip_listeners:

1 0

[stem/master] Only provide absolute paths with get_processed_files()
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 0c021dc30ac9b4066e7c647f4a60bfda4511c9c3 Author: Damian Johnson <atagar(a)torproject.org> Date: Sun Mar 11 12:54:50 2012 -0700 Only provide absolute paths with get_processed_files() The DescriptorReader's targets can be relative paths, and this is fine. However, the load and save functions for processed_files expect absolute paths so making get_processed_files() expand relative paths. I'm on the fencepost about if we should do this or make the load/save functions accept relative paths. Going with this until someone produces a use case where this makes them sad. --- stem/descriptor/reader.py | 7 ++++--- 1 files changed, 4 insertions(+), 3 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index b003dbd..018adc7 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -185,11 +185,12 @@ class DescriptorReader(threading.Thread): This includes entries set through the set_processed_files() method. Returns: - dict with the paths and unix timestamp for the last modified times of the - files we have processed + dict with the absolute paths and unix timestamp for the last modified + times of the files we have processed """ - return dict(self._processed_files) + # make sure that we only provide back absolute paths + return dict((os.path.abspath(k), v) for (k, v) in self._processed_files.items()) def set_processed_files(self, processed_files): """

1 0

[stem/master] Inverting the 'finished_reading' flag
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit c16df469c9d49f50ed07bdb8b6c823e8aa31f6ce Author: Damian Johnson <atagar(a)torproject.org> Date: Sun Mar 11 12:39:47 2012 -0700 Inverting the 'finished_reading' flag Code is more readable when flags say what we're doing rather than what we're *not* doing. --- stem/descriptor/reader.py | 14 ++++++++++++-- 1 files changed, 12 insertions(+), 2 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index c3bf1aa..cc19ed6 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -69,6 +69,12 @@ import Queue # - restarting when __iter__ is called additional times # - maximum read-ahead +# TODO: Remianing impementation items... +# - integ test that we skip the 'processed files' items +# - impelment skip listening and add a test for it +# - remove start and join methods from header? +# - implement gzip and bz2 reading + # Maximum number of descriptors that we'll read ahead before waiting for our # caller to fetch some of them. This is included to avoid unbounded memory # usage. This condition will be removed if set to zero. @@ -160,6 +166,7 @@ class DescriptorReader(threading.Thread): self._iter_lock = threading.RLock() self._iter_notice = threading.Event() + self._is_reading = threading.Event() # files or directories that remain to be read self._remaining_files = list(self._targets) @@ -226,6 +233,8 @@ class DescriptorReader(threading.Thread): self._iter_notice.set() def run(self): + self._is_reading.set() + while self._remaining_files and not self._stop_called.isSet(): target = self._remaining_files.pop(0) if not os.path.exists(target): continue @@ -262,6 +271,7 @@ class DescriptorReader(threading.Thread): with open(target) as target_file: # TODO: replace with actual descriptor parsing when we have it + # TODO: impement skip listening self._unreturned_descriptors.put(target_file.read()) self._iter_notice.set() elif target_type[0] == 'application/x-tar': @@ -270,7 +280,7 @@ class DescriptorReader(threading.Thread): elif target_type[1] == 'bzip2': pass # TODO: implement - self._finished_reading.set() + self._is_reading.clear() self._iter_notice.set() def __iter__(self): @@ -280,7 +290,7 @@ class DescriptorReader(threading.Thread): yield self._unreturned_descriptors.get_nowait() except Queue.Empty: # if we've finished and there aren't any descriptors then we're done - if self._finished_reading.isSet(): break + if not self._is_reading.isSet(): break self._iter_notice.wait() self._iter_notice.clear()

1 0

[stem/master] Fix and test for DescriptorReader stop() method
by atagar＠torproject.org 26 Mar '12

26 Mar '12

commit 932ded67f961c67c515750bd6e1c9ddb735f8559 Author: Damian Johnson <atagar(a)torproject.org> Date: Sat Mar 10 23:30:28 2012 -0800 Fix and test for DescriptorReader stop() method Adding an integraion test and some fixes for the stop() method of the DescriptorReader class. --- stem/descriptor/reader.py | 19 +++++++++++-------- test/integ/descriptor/reader.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 8 deletions(-) diff --git a/stem/descriptor/reader.py b/stem/descriptor/reader.py index 0784ce4..c3bf1aa 100644 --- a/stem/descriptor/reader.py +++ b/stem/descriptor/reader.py @@ -66,7 +66,6 @@ import mimetypes import Queue # TODO: Unimplemented concurrency features... -# - stop() # - restarting when __iter__ is called additional times # - maximum read-ahead @@ -162,15 +161,16 @@ class DescriptorReader(threading.Thread): self._iter_lock = threading.RLock() self._iter_notice = threading.Event() - # targets that remain to be read in this iteration + # files or directories that remain to be read self._remaining_files = list(self._targets) # descriptors that we have read, but not yet provided to the caller self._unreturned_descriptors = Queue.Queue() - # TODO: implement - # flag that's set when we're done - self._stop_event = threading.Event() + # flag that's set when stop() is called + self._stop_called = threading.Event() + + # flag that's set when we're out of descriptor files to read self._finished_reading = threading.Event() def get_processed_files(self): @@ -222,11 +222,11 @@ class DescriptorReader(threading.Thread): Stops further reading of descriptor files. """ - self._stop_event.set() + self._stop_called.set() self._iter_notice.set() def run(self): - while self._remaining_files: + while self._remaining_files and not self._stop_called.isSet(): target = self._remaining_files.pop(0) if not os.path.exists(target): continue @@ -235,6 +235,9 @@ class DescriptorReader(threading.Thread): for root, _, files in os.walk(target, followlinks = self._follow_links): for filename in files: self._remaining_files.append(os.path.join(root, filename)) + + # this can take a while if, say, we're including the root directory + if self._stop_called.isSet(): break else: # This is a file. Register it's last modified timestamp and check if # it's a file that we should skip. @@ -272,7 +275,7 @@ class DescriptorReader(threading.Thread): def __iter__(self): with self._iter_lock: - while not self._stop_event.isSet(): + while not self._stop_called.isSet(): try: yield self._unreturned_descriptors.get_nowait() except Queue.Empty: diff --git a/test/integ/descriptor/reader.py b/test/integ/descriptor/reader.py index ca77e26..5fa61b8 100644 --- a/test/integ/descriptor/reader.py +++ b/test/integ/descriptor/reader.py @@ -3,6 +3,8 @@ Integration tests for stem.descriptor.reader. """ import os +import time +import signal import unittest import stem.descriptor.reader @@ -134,4 +136,30 @@ class TestDescriptorReader(unittest.TestCase): # check that we've seen all of the descriptor_entries self.assertTrue(len(descriptor_entries) == 0) + + def test_stop(self): + """ + Runs a DescriptorReader over the root directory, then checks that calling + stop() makes it terminate in a timely fashion. + """ + + is_test_running = True + reader = stem.descriptor.reader.DescriptorReader(["/"]) + + # Fails the test after a couple seconds if we don't finish successfully. + # Depending on what we're blocked on this might not work when the test + # fails, requiring that we give a manual kill to the test. + + def timeout_handler(signum, frame): + if is_test_running: + self.fail() + + signal.signal(signal.SIGALRM, timeout_handler) + signal.alarm(2) + + reader.start() + time.sleep(0.1) + reader.stop() + reader.join() + is_test_running = False

1 0