write unit tests

Author: saviorand

README.md

@@ -126,12 +126,12 @@ Once you have a Mojo project set up locally,
     fn main() raises:
         var server = Server()
         var handler = Welcome()
-        server.listen_and_serve("0.0.0.0:8080", handler)
+        server.listen_and_serve("localhost:8080", handler)
     ```
 
 Feel free to change the settings in `listen_and_serve()` to serve on a particular host and port.
 
-Now send a request `0.0.0.0:8080`. You should see some details about the request printed out to the console.
+Now send a request `localhost:8080`. You should see some details about the request printed out to the console.
 
 Congrats 🥳 You're using Lightbug!
 

benchmark/bench_server.mojo

@@ -6,7 +6,7 @@ def main():
     try:
         var server = Server(tcp_keep_alive=True)
         var handler = TechEmpowerRouter()
-        server.listen_and_serve("0.0.0.0:8080", handler)
+        server.listen_and_serve("localhost:8080", handler)
     except e:
         print("Error starting server: " + e.__str__())
         return

lightbug.🔥

@@ -3,4 +3,4 @@ from lightbug_http import Welcome, Server
 fn main() raises:
     var server = Server()
     var handler = Welcome()
-    server.listen_and_serve("0.0.0.0:8080", handler)
+    server.listen_and_serve("localhost:8080", handler)

lightbug_http/_libc.mojo

@@ -574,7 +574,7 @@ fn inet_pton[address_family: Int32](src: UnsafePointer[c_char]) raises -> c_uint
     * This function is valid for `AF_INET` and `AF_INET6`.
     """
     constrained[
-        int(address_family) in [AF_INET, AF_INET6], "Address family must be either INET_ADDRSTRLEN or INET6_ADDRSTRLEN."
+        int(address_family) in [AF_INET, AF_INET6], "Address family must be either AF_INET or AF_INET6."
     ]()
     var ip_buffer: UnsafePointer[c_void]
 

lightbug_http/address.mojo

@@ -313,11 +313,11 @@ fn is_ip_protocol(network: NetworkType) -> Bool:
 
 fn is_ipv4(network: NetworkType) -> Bool:
     """Check if the network type is IPv4."""
-    return network in (NetworkType.tcp4, NetworkType.udp4)
+    return network in (NetworkType.tcp4, NetworkType.udp4, NetworkType.ip4)
 
 fn is_ipv6(network: NetworkType) -> Bool:
     """Check if the network type is IPv6."""
-    return network in (NetworkType.tcp6, NetworkType.udp6)
+    return network in (NetworkType.tcp6, NetworkType.udp6, NetworkType.ip6)
 
 fn resolve_localhost(host: String, network: NetworkType) -> String:
     """Resolve localhost to the appropriate IP address based on network type."""
@@ -383,45 +383,51 @@ fn parse_address(network: NetworkType, address: String) raises -> (String, UInt1
     Returns:
         Tuple containing the host and port
     """
-    # Handle IP protocols separately
     if is_ip_protocol(network):
-        if address.find(":") != -1:
-            raise Error("IP protocol addresses should not include ports")
-        
         var host = resolve_localhost(address, network)
         if host == AddressConstants.EMPTY:
             raise Error("missing host")
+            
+        # For IPv6 addresses in IP protocol mode, we need to handle the address as-is
+        if network == NetworkType.ip6 and host.find(":") != -1:
+            return host, DEFAULT_IP_PORT
+            
+        # For other IP protocols, no colons allowed
+        if host.find(":") != -1:
+            raise Error("IP protocol addresses should not include ports")
+            
         return host, DEFAULT_IP_PORT
 
-    # Parse regular addresses
     var colon_index = address.rfind(":")
     if colon_index == -1:
         raise MissingPortError
 
     var host: String
-    var bracket_offset: Int
+    var bracket_offset: Int = 0
 
     # Handle IPv6 addresses
-    try:
-        (host, bracket_offset) = parse_ipv6_bracketed_address(address)
-    except e:
-        raise e
-
-    # Validate no unexpected brackets
-    validate_no_brackets(address, bracket_offset)
+    if address[0] == "[":
+        try:
+            (host, bracket_offset) = parse_ipv6_bracketed_address(address)
+        except e:
+            raise e
+        
+        validate_no_brackets(address, bracket_offset)
+    else:
+        # For IPv4, simply split at the last colon
+        host = address[:colon_index]
+        if host.find(":") != -1:
+            raise TooManyColonsError
 
-    # Parse and validate port
     var port = parse_port(address[colon_index + 1:])
 
-    # Resolve localhost if needed
     host = resolve_localhost(host, network)
     if host == AddressConstants.EMPTY:
         raise Error("missing host")
 
     return host, port
 
 
-
 # TODO: Support IPv6 long form.
 fn join_host_port(host: String, port: String) -> String:
     if host.find(":") != -1:  # must be IPv6 literal

lightbug_http/server.mojo

@@ -91,8 +91,8 @@ struct Server(Movable):
             address: The address (host:port) to listen on.
             handler: An object that handles incoming HTTP requests.
         """
-        var net = ListenConfig()
-        var listener = net.listen(address)
+        var config = ListenConfig()
+        var listener = config.listen(address)
         self.set_address(address)
         self.serve(listener^, handler)
 

scripts/bench_server.sh

@@ -9,7 +9,7 @@ echo "running server..."
 sleep 2
 
 echo "Running benchmark"
-wrk -t1 -c1 -d10s http://0.0.0.0:8080/ --header "User-Agent: wrk"
+wrk -t1 -c1 -d10s http://localhost:8080/ --header "User-Agent: wrk"
 
 kill $!
 wait $! 2>/dev/null

tests/lightbug_http/test_host_port.mojo

@@ -4,21 +4,107 @@ from lightbug_http.strings import NetworkType
 
 
 def test_split_host_port():
-    # IPv4
-    var hp = parse_address("127.0.0.1:8080")
+    # TCP4
+    var hp = parse_address(NetworkType.tcp4, "127.0.0.1:8080")
     testing.assert_equal(hp[0], "127.0.0.1")
     testing.assert_equal(hp[1], 8080)
 
-    # IPv6
-    hp = parse_address("[::1]:8080")
+    # TCP4 with localhost
+    hp = parse_address(NetworkType.tcp4, "localhost:8080")
+    testing.assert_equal(hp[0], "127.0.0.1")
+    testing.assert_equal(hp[1], 8080)
+
+    # TCP6
+    hp = parse_address(NetworkType.tcp6, "[::1]:8080")
     testing.assert_equal(hp[0], "::1")
     testing.assert_equal(hp[1], 8080)
 
-    # # TODO: IPv6 long form - Not supported yet.
+    # TCP6 with localhost
+    hp = parse_address(NetworkType.tcp6, "localhost:8080")
+    testing.assert_equal(hp[0], "::1")
+    testing.assert_equal(hp[1], 8080)
+
+    # UDP4
+    hp = parse_address(NetworkType.udp4, "192.168.1.1:53")
+    testing.assert_equal(hp[0], "192.168.1.1")
+    testing.assert_equal(hp[1], 53)
+
+    # UDP4 with localhost
+    hp = parse_address(NetworkType.udp4, "localhost:53")
+    testing.assert_equal(hp[0], "127.0.0.1")
+    testing.assert_equal(hp[1], 53)
+
+    # UDP6
+    hp = parse_address(NetworkType.udp6, "[2001:db8::1]:53")
+    testing.assert_equal(hp[0], "2001:db8::1")
+    testing.assert_equal(hp[1], 53)
+
+    # UDP6 with localhost
+    hp = parse_address(NetworkType.udp6, "localhost:53")
+    testing.assert_equal(hp[0], "::1")
+    testing.assert_equal(hp[1], 53)
+
+    # IP4 (no port)
+    hp = parse_address(NetworkType.ip4, "192.168.1.1")
+    testing.assert_equal(hp[0], "192.168.1.1")
+    testing.assert_equal(hp[1], 0)
+
+    # IP4 with localhost
+    hp = parse_address(NetworkType.ip4, "localhost")
+    testing.assert_equal(hp[0], "127.0.0.1")
+    testing.assert_equal(hp[1], 0)
+
+    # IP6 (no port)
+    hp = parse_address(NetworkType.ip6, "2001:db8::1")
+    testing.assert_equal(hp[0], "2001:db8::1")
+    testing.assert_equal(hp[1], 0)
+
+    # IP6 with localhost
+    hp = parse_address(NetworkType.ip6, "localhost")
+    testing.assert_equal(hp[0], "::1")
+    testing.assert_equal(hp[1], 0)
+
+    # TODO: IPv6 long form - Not supported yet.
     # hp = parse_address("0:0:0:0:0:0:0:1:8080")
     # testing.assert_equal(hp[0], "0:0:0:0:0:0:0:1")
     # testing.assert_equal(hp[1], 8080)
 
+    # Error cases
+    # IP protocol with port
+    try:
+        _ = parse_address(NetworkType.ip4, "192.168.1.1:80")
+        testing.assert_false("Should have raised an error for IP protocol with port")
+    except Error:
+        testing.assert_true(True)
+
+    # Missing port
+    try:
+        _ = parse_address(NetworkType.tcp4, "192.168.1.1")
+        testing.assert_false("Should have raised MissingPortError")
+    except MissingPortError:
+        testing.assert_true(True)
+
+    # Missing port
+    try:
+        _ = parse_address(NetworkType.tcp6, "[::1]")
+        testing.assert_false("Should have raised MissingPortError")
+    except MissingPortError:
+        testing.assert_true(True)
+
+    # Port out of range
+    try:
+        _ = parse_address(NetworkType.tcp4, "192.168.1.1:70000")
+        testing.assert_false("Should have raised error for invalid port")
+    except Error:
+        testing.assert_true(True)
+    
+    # Missing closing bracket
+    try:
+        _ = parse_address(NetworkType.tcp6, "[::1:8080")
+        testing.assert_false("Should have raised error for missing bracket")
+    except Error:
+        testing.assert_true(True)
+
 
 def test_join_host_port():
     # IPv4