Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Ensure that FilesystemSynchronousBackend.get_{reader,writer} always r…

…eturn a Deferred.
  • Loading branch information...
commit 2ea495b6a48b1905183a86588c5777c80a8648d2 1 parent 36c080c
@allenap authored
Showing with 27 additions and 16 deletions.
  1. +5 −5 tftp/backend.py
  2. +22 −11 tftp/test/test_backend.py
View
10 tftp/backend.py
@@ -3,7 +3,7 @@
'''
from os import fstat
from tftp.errors import Unsupported, FileExists, AccessViolation, FileNotFound
-from twisted.internet.defer import succeed
+from tftp.util import deferred
from twisted.python.filepath import FilePath, InsecurePath
import shutil
import tempfile
@@ -256,6 +256,7 @@ def __init__(self, base_path, can_read=True, can_write=True):
self.base = FilePath(base_path)
self.can_read, self.can_write = can_read, can_write
+ @deferred
def get_reader(self, file_name):
"""
@see: L{IBackend.get_reader}
@@ -269,9 +270,9 @@ def get_reader(self, file_name):
target_path = self.base.child(file_name)
except InsecurePath, e:
raise AccessViolation("Insecure path: %s" % e)
- reader = FilesystemReader(target_path)
- return succeed(reader)
+ return FilesystemReader(target_path)
+ @deferred
def get_writer(self, file_name):
"""
@see: L{IBackend.get_writer}
@@ -285,5 +286,4 @@ def get_writer(self, file_name):
target_path = self.base.child(file_name)
except InsecurePath, e:
raise AccessViolation("Insecure path: %s" % e)
- writer = FilesystemWriter(target_path)
- return succeed(writer)
+ return FilesystemWriter(target_path)
View
33 tftp/test/test_backend.py
@@ -2,7 +2,7 @@
@author: shylent
'''
from tftp.backend import (FilesystemSynchronousBackend, FilesystemReader,
- FilesystemWriter)
+ FilesystemWriter, IReader, IWriter)
from tftp.errors import Unsupported, AccessViolation, FileNotFound, FileExists
from twisted.python.filepath import FilePath
from twisted.trial import unittest
@@ -24,21 +24,31 @@ def setUp(self):
with open(self.existing_file_name, 'w') as f:
f.write(self.test_data)
- def test_unsupported(self):
+ def test_read_supported_by_default(self):
+ b = FilesystemSynchronousBackend(self.temp_dir)
+ return b.get_reader('foo').addCallback(IReader.providedBy)
+
+ def test_write_supported_by_default(self):
+ b = FilesystemSynchronousBackend(self.temp_dir)
+ return b.get_writer('bar').addCallback(IWriter.providedBy)
+
+ def test_read_unsupported(self):
b = FilesystemSynchronousBackend(self.temp_dir, can_read=False)
- self.assertRaises(Unsupported, b.get_reader, 'foo')
- self.assert_(b.get_writer('bar'),
- "A writer should be dispatched")
+ return self.assertFailure(b.get_reader('foo'), Unsupported)
+
+ def test_write_unsupported(self):
b = FilesystemSynchronousBackend(self.temp_dir, can_write=False)
- self.assertRaises(Unsupported, b.get_writer, 'bar')
- self.assert_(b.get_reader('foo'),
- "A reader should be dispatched")
+ return self.assertFailure(b.get_writer('bar'), Unsupported)
- def test_insecure(self):
+ def test_insecure_reader(self):
b = FilesystemSynchronousBackend(self.temp_dir)
- self.assertRaises(AccessViolation, b.get_reader, '../foo')
+ return self.assertFailure(
+ b.get_reader('../foo'), AccessViolation)
+
+ def test_insecure_writer(self):
b = FilesystemSynchronousBackend(self.temp_dir)
- self.assertRaises(AccessViolation, b.get_writer, '../foo')
+ return self.assertFailure(
+ b.get_writer('../foo'), AccessViolation)
def tearDown(self):
shutil.rmtree(self.temp_dir)
@@ -94,6 +104,7 @@ def test_cancel(self):
r.read(3)
r.finish()
self.failUnless(r.file_obj.closed,
+
"The session has been finished, so the file object should be in the closed state")
r.finish()
Please sign in to comment.
Something went wrong with that request. Please try again.