import gzip import http.server import threading import unittest import wfdb.io._url class TestNetFiles(unittest.TestCase): """ Test accessing remote files. """ def test_requests(self): """ Test reading a remote file using various APIs. This tests that we can create a file object using wfdb.io._url.openurl(), and tests that the object implements the standard Python API functions for a file of the appropriate type. Parameters ---------- N/A Returns ------- N/A """ text_data = """ BERNARDO: Who's there? FRANCISCO: Nay, answer me: stand, and unfold yourself. BERNARDO: Long live the king! FRANCISCO: Bernardo? BERNARDO: He. FRANCISCO: You come most carefully upon your hour. BERNARDO: 'Tis now struck twelve; get thee to bed, Francisco. """ binary_data = text_data.encode() file_content = {"/foo.txt": binary_data} # Test all possible combinations of: # - whether or not the server supports compression # - whether or not the server supports random access # - chosen buffering policy for allow_gzip in (False, True): for allow_range in (False, True): with DummyHTTPServer( file_content=file_content, allow_gzip=allow_gzip, allow_range=allow_range, ) as server: url = server.url("/foo.txt") for buffering in (-2, -1, 0, 20): self._test_text(url, text_data, buffering) self._test_binary(url, binary_data, buffering) def _test_text(self, url, content, buffering): """ Test reading a URL using text-mode file APIs. Parameters ---------- url : str URL of the remote resource. content : str Expected content of the resource. buffering : int Buffering policy for openurl(). Returns ------- N/A """ # read(-1), readable(), seekable() with wfdb.io._url.openurl(url, "r", buffering=buffering) as tf: self.assertTrue(tf.readable()) self.assertTrue(tf.seekable()) self.assertEqual(tf.read(), content) self.assertEqual(tf.read(), "") # read(10) with wfdb.io._url.openurl(url, "r", buffering=buffering) as tf: result = "" while True: chunk = tf.read(10) result += chunk if len(chunk) < 10: break self.assertEqual(result, content) # readline(), seek(), tell() with wfdb.io._url.openurl(url, "r", buffering=buffering) as tf: result = "" while True: rpos = tf.tell() tf.seek(0) tf.seek(rpos) chunk = tf.readline() result += chunk if len(chunk) == 0: break self.assertEqual(result, content) def _test_binary(self, url, content, buffering): """ Test reading a URL using binary-mode file APIs. Parameters ---------- url : str URL of the remote resource. content : bytes Expected content of the resource. buffering : int Buffering policy for openurl(). Returns ------- N/A """ # read(-1), readable(), seekable() with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: self.assertTrue(bf.readable()) self.assertTrue(bf.seekable()) self.assertEqual(bf.read(), content) self.assertEqual(bf.read(), b"") self.assertEqual(bf.tell(), len(content)) # read(10) with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: result = b"" while True: chunk = bf.read(10) result += chunk if len(chunk) < 10: break self.assertEqual(result, content) self.assertEqual(bf.tell(), len(content)) # readline() with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: result = b"" while True: chunk = bf.readline() result += chunk if len(chunk) == 0: break self.assertEqual(result, content) self.assertEqual(bf.tell(), len(content)) # read1(10), seek(), tell() with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: bf.seek(0, 2) self.assertEqual(bf.tell(), len(content)) bf.seek(0) result = b"" while True: rpos = bf.tell() bf.seek(0) bf.seek(rpos) chunk = bf.read1(10) result += chunk if len(chunk) == 0: break self.assertEqual(result, content) self.assertEqual(bf.tell(), len(content)) # readinto(bytearray(10)) with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: result = b"" chunk = bytearray(10) while True: count = bf.readinto(chunk) result += chunk[:count] if count < 10: break self.assertEqual(result, content) self.assertEqual(bf.tell(), len(content)) # readinto1(bytearray(10)) with wfdb.io._url.openurl(url, "rb", buffering=buffering) as bf: result = b"" chunk = bytearray(10) while True: count = bf.readinto1(chunk) result += chunk[:count] if count == 0: break self.assertEqual(result, content) self.assertEqual(bf.tell(), len(content)) class TestRemoteFLACFiles(unittest.TestCase): """ Test reading FLAC files over HTTP. """ def test_whole_file(self): """ Test reading a complete FLAC file using local and HTTP APIs. This tests that we can read the file 'sample-data/flacformats.d2' (a 24-bit FLAC stream) using the soundfile library, first by reading the file from the local filesystem, and then using wfdb.io._url.openurl() to access it through a simulated web server. This is meant to verify that the soundfile library works using only the standard Python file object API (as implemented by wfdb.io._url.NetFile), and doesn't require the input file to be an actual io.FileIO object. Parameters ---------- N/A Returns ------- N/A """ import soundfile import numpy as np data_file_path = "sample-data/flacformats.d2" expected_format = "FLAC" expected_subtype = "PCM_24" # Read the file using standard file I/O sf1 = soundfile.SoundFile(data_file_path) self.assertEqual(sf1.format, expected_format) self.assertEqual(sf1.subtype, expected_subtype) data1 = sf1.read() # Read the file using HTTP with open(data_file_path, "rb") as f: file_content = {"/foo.dat": f.read()} with DummyHTTPServer(file_content) as server: url = server.url("/foo.dat") file2 = wfdb.io._url.openurl(url, "rb") sf2 = soundfile.SoundFile(file2) self.assertEqual(sf2.format, expected_format) self.assertEqual(sf2.subtype, expected_subtype) data2 = sf2.read() # Check that results are equal np.testing.assert_array_equal(data1, data2) class DummyHTTPServer(http.server.HTTPServer): """ HTTPServer used to simulate a web server for testing. The server may be used as a context manager (using "with"); during execution of the "with" block, a background thread runs that listens for and handles client requests. Attributes ---------- file_content : dict Dictionary containing the content of each file on the server. The keys are absolute paths (such as "/foo.txt"); the values are the corresponding content (bytes). allow_gzip : bool, optional True if the server should return compressed responses (using "Content-Encoding: gzip") when the client requests them (using "Accept-Encoding: gzip"). allow_range : bool, optional True if the server should return partial responses (using 206 Partial Content and "Content-Range") when the client requests them (using "Range"). server_address : tuple (str, int), optional A tuple specifying the address and port number where the server should listen for connections. If the port is 0, an arbitrary unused port is selected. The default address is "127.0.0.1" and the default port is 0. """ def __init__( self, file_content, allow_gzip=True, allow_range=True, server_address=("127.0.0.1", 0), ): super().__init__(server_address, DummyHTTPRequestHandler) self.file_content = file_content self.allow_gzip = allow_gzip self.allow_range = allow_range def url(self, path="/"): """ Generate a URL that points to a file on this server. Parameters ---------- path : str, optional Path of the file on the server. Returns ------- url : str Absolute URL for the specified file. """ return "http://127.0.0.1:%d/%s" % ( self.server_address[1], path.lstrip("/"), ) def __enter__(self): super().__enter__() self.thread = threading.Thread(target=self.serve_forever) self.thread.start() return self def __exit__(self, exc_type, exc_val, exc_tb): self.shutdown() self.thread.join() self.thread = None return super().__exit__(exc_type, exc_val, exc_tb) class DummyHTTPRequestHandler(http.server.BaseHTTPRequestHandler): """ HTTPRequestHandler used to simulate a web server for testing. """ def do_HEAD(self): self.send_head() def do_GET(self): body = self.send_head() self.wfile.write(body) def log_message(self, message, *args): pass def send_head(self): content = self.server.file_content.get(self.path) if content is None: self.send_error(404) return b"" headers = {"Content-Type": "text/plain"} status = 200 if self.server.allow_gzip: headers["Vary"] = "Accept-Encoding" if "gzip" in self.headers.get("Accept-Encoding", ""): content = gzip.compress(content) headers["Content-Encoding"] = "gzip" if self.server.allow_range: headers["Accept-Ranges"] = "bytes" req_range = self.headers.get("Range", "") if req_range.startswith("bytes="): start, end = req_range.split("=")[1].split("-") start = int(start) if end == "": end = len(content) else: end = min(len(content), int(end) + 1) if start < end: status = 206 resp_range = "bytes %d-%d/%d" % ( start, end - 1, len(content), ) content = content[start:end] else: status = 416 resp_range = "bytes */%d" % len(content) content = b"" headers["Content-Range"] = resp_range headers["Content-Length"] = len(content) self.send_response(status) for h, v in sorted(headers.items()): self.send_header(h, v) self.end_headers() return content if __name__ == "__main__": unittest.main()