Refactor parse_url to use the buffer protocol; add tests

Author: 1st1

httptools/parser/parser.pyx

@@ -310,9 +310,10 @@ cdef class URL:
                     self.query, self.fragment, self.userinfo))
 
 
-def parse_url(const char* url):
+def parse_url(url):
     cdef:
-        size_t url_len = len(url)
+        Py_buffer py_buf
+        char* buf_data
         cparser.http_parser_url* parsed
         int res
         bytes schema = None
@@ -330,44 +331,48 @@ def parse_url(const char* url):
                         PyMem_Malloc(sizeof(cparser.http_parser_url))
     cparser.http_parser_url_init(parsed)
 
+    PyObject_GetBuffer(url, &py_buf, PyBUF_SIMPLE)
     try:
-        res = cparser.http_parser_parse_url(url, url_len, 0, parsed)
+        buf_data = <char*>py_buf.buf
+        res = cparser.http_parser_parse_url(buf_data, py_buf.len, 0, parsed)
+
         if res == 0:
             if parsed.field_set & (1 << cparser.UF_SCHEMA):
                 off = parsed.field_data[<int>cparser.UF_SCHEMA].off
                 ln = parsed.field_data[<int>cparser.UF_SCHEMA].len
-                schema = url[off:off+ln]
+                schema = buf_data[off:off+ln]
 
             if parsed.field_set & (1 << cparser.UF_HOST):
                 off = parsed.field_data[<int>cparser.UF_HOST].off
                 ln = parsed.field_data[<int>cparser.UF_HOST].len
-                host = url[off:off+ln]
+                host = buf_data[off:off+ln]
 
             if parsed.field_set & (1 << cparser.UF_PORT):
                 port = parsed.port
 
             if parsed.field_set & (1 << cparser.UF_PATH):
                 off = parsed.field_data[<int>cparser.UF_PATH].off
                 ln = parsed.field_data[<int>cparser.UF_PATH].len
-                path = url[off:off+ln]
+                path = buf_data[off:off+ln]
 
             if parsed.field_set & (1 << cparser.UF_QUERY):
                 off = parsed.field_data[<int>cparser.UF_QUERY].off
                 ln = parsed.field_data[<int>cparser.UF_QUERY].len
-                query = url[off:off+ln]
+                query = buf_data[off:off+ln]
 
             if parsed.field_set & (1 << cparser.UF_FRAGMENT):
                 off = parsed.field_data[<int>cparser.UF_FRAGMENT].off
                 ln = parsed.field_data[<int>cparser.UF_FRAGMENT].len
-                fragment = url[off:off+ln]
+                fragment = buf_data[off:off+ln]
 
             if parsed.field_set & (1 << cparser.UF_USERINFO):
                 off = parsed.field_data[<int>cparser.UF_USERINFO].off
                 ln = parsed.field_data[<int>cparser.UF_USERINFO].len
-                userinfo = url[off:off+ln]
+                userinfo = buf_data[off:off+ln]
 
             return URL(schema, host, port, path, query, fragment, userinfo)
         else:
             raise HttpParserInvalidURLError("invalid url {!r}".format(url))
     finally:
+        PyBuffer_Release(&py_buf)
         PyMem_Free(parsed)

tests/test_parser.py

@@ -316,3 +316,12 @@ def test_parser_url_7(self):
     def test_parser_url_8(self):
         with self.assertRaises(TypeError):
             httptools.parse_url(None)
+
+    def test_parser_url_9(self):
+        with self.assertRaisesRegex(httptools.HttpParserInvalidURLError,
+                                    r'a\\x00aa'):
+            self.parse(b'dsf://a\x00aa')
+
+    def test_parser_url_10(self):
+        with self.assertRaisesRegex(TypeError, 'a bytes-like object'):
+            self.parse('dsf://aaa')