Adding tests (#52)

* Adding tests

* fix python matrix strings

.github/workflows/test.yml

@@ -0,0 +1,32 @@
+name: Jetforce
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: [ "3.7", "3.8", "3.9", "3.10-dev" ]
+    steps:
+    - name: Check out repository
+      uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install packages
+      run: |
+        pip install .
+        pip install mypy pytest
+    - name: Check types
+      run: |
+        mypy --ignore-missing-imports jetforce/
+    - name: Run tests
+      run: |
+        pytest -v tests/

jetforce/app/static.py

@@ -172,6 +172,7 @@ class StaticDirectoryApplication(JetforceApplication):
             env=cgi_env,
             bufsize=0,
         )
+        proc.stdout = typing.cast(typing.IO[bytes], proc.stdout)
 
         status_line = proc.stdout.readline(self.CGI_MAX_RESPONSE_HEADER_SIZE)
         if len(status_line) == self.CGI_MAX_RESPONSE_HEADER_SIZE:
@@ -194,6 +195,8 @@ class StaticDirectoryApplication(JetforceApplication):
         Non-blocking read from the stdout of the CGI process and pipe it
         to the socket transport.
         """
+        proc.stdout = typing.cast(typing.IO[bytes], proc.stdout)
+
         while True:
             proc.poll()
 

jetforce/server.py

@@ -105,15 +105,10 @@ class GeminiServer(Factory):
         """
         return GeminiProtocol(self, self.app)
 
-    def run(self) -> None:
+    def initialize(self) -> None:
         """
-        This is the main server loop.
+        Install the server into the twisted reactor.
         """
-        self.log_message(ABOUT)
-        self.log_message(f"Server hostname is {self.hostname}")
-        self.log_message(f"TLS Certificate File: {self.certfile}")
-        self.log_message(f"TLS Private Key File: {self.keyfile}")
-
         certificate_options = GeminiCertificateOptions(
             certfile=self.certfile,
             keyfile=self.keyfile,
@@ -131,4 +126,13 @@ class GeminiServer(Factory):
             )
             endpoint.listen(self).addCallback(self.on_bind_interface)
 
+    def run(self) -> None:
+        """
+        This is the main server loop.
+        """
+        self.log_message(ABOUT)
+        self.log_message(f"Server hostname is {self.hostname}")
+        self.log_message(f"TLS Certificate File: {self.certfile}")
+        self.log_message(f"TLS Private Key File: {self.keyfile}")
+        self.initialize()
         self.reactor.run()

jetforce_client.py

@@ -39,10 +39,6 @@ def fetch(
                 sys.stdout.buffer.flush()
                 data = fp.read(1024)
 
-            # Send a close_notify alert
-            # ssock.setblocking(False)
-            # ssock.unwrap()
-
 
 def run_client() -> None:
     # fmt: off
@@ -66,7 +62,7 @@ def run_client() -> None:
 
     if args.tls_keylog:
         # This is a "private" variable that the stdlib exposes for debugging
-        context.keylog_filename = args.tls_keylog
+        context.keylog_filename = args.tls_keylog  # type: ignore
 
     fetch(args.url, args.host, args.port, args.tls_enable_sni)
 

tests/data/cgi-bin/echo.cgi

@@ -0,0 +1,3 @@
+#!/bin/bash
+echo "20 text/plain"
+echo $QUERY_STRING

tests/data/index.gmi

@@ -0,0 +1 @@
+Jetforce rules!

tests/test_jetforce.py

@@ -0,0 +1,101 @@
+import os
+import ssl
+import socket
+import unittest
+from threading import Thread
+
+from twisted.internet import reactor
+from jetforce import StaticDirectoryApplication, GeminiServer
+
+ROOT_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "data")
+
+
+class GeminiTestServer(GeminiServer):
+
+    real_port: int
+
+    def on_bind_interface(self, port):
+        """
+        Capture the port number that the test server actually binds to.
+        """
+        sock_ip, sock_port, *_ = port.socket.getsockname()
+        self.real_port = sock_port
+
+    def log_access(self, message: str) -> None:
+        """Suppress logging"""
+
+    def log_message(self, message: str) -> None:
+        """Suppress logging"""
+
+
+class FunctionalTestCase(unittest.TestCase):
+    """
+    This class will spin up a complete test jetforce server and serve it
+    on a local TCP port in a new thread. The tests will send real gemini
+    connection strings to the server and check the validity of the response
+    body from end-to-end.
+    """
+
+    server: GeminiTestServer
+    thread: Thread
+
+    @classmethod
+    def setUpClass(cls):
+        app = StaticDirectoryApplication(root_directory=ROOT_DIR)
+        cls.server = GeminiTestServer(app=app, port=0)
+        cls.server.initialize()
+
+        cls.thread = Thread(target=reactor.run, args=(False,))
+        cls.thread.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        reactor.callFromThread(reactor.stop)
+        cls.thread.join(timeout=5)
+
+    @classmethod
+    def request(cls, data):
+        """
+        Send bytes to the server using a TCP/IP socket.
+        """
+        context = ssl.create_default_context()
+        context.check_hostname = False
+        context.verify_mode = ssl.CERT_NONE
+
+        with socket.create_connection((cls.server.host, cls.server.real_port)) as sock:
+            with context.wrap_socket(sock) as ssock:
+                ssock.sendall(data)
+                fp = ssock.makefile("rb")
+                return fp.read()
+
+    def test_index(self):
+        resp = self.request(b"gemini://localhost\r\n")
+        self.assertEqual(resp, b"20 text/gemini\r\nJetforce rules!\n")
+
+    def test_invalid_path(self):
+        resp = self.request(b"gemini://localhost/invalid\r\n")
+        self.assertEqual(resp, b"51 Not Found\r\n")
+
+    def test_invalid_hostname(self):
+        resp = self.request(b"gemini://example.com\r\n")
+        self.assertEqual(resp, b"53 This server does not allow proxy requests\r\n")
+
+    def test_invalid_port(self):
+        resp = self.request(b"gemini://localhost:1111\r\n")
+        self.assertEqual(resp, b"53 This server does not allow proxy requests\r\n")
+
+    def test_directory_redirect(self):
+        resp = self.request(b"gemini://localhost/cgi-bin\r\n")
+        self.assertEqual(resp, b"31 gemini://localhost/cgi-bin/\r\n")
+
+    def test_directory(self):
+        resp = self.request(b"gemini://localhost/cgi-bin/\r\n")
+        self.assertEqual(resp.splitlines(keepends=True)[0], b"20 text/gemini\r\n")
+
+    def test_cgi_script(self):
+        resp = self.request(b"gemini://localhost/cgi-bin/echo.cgi?hello%20world\r\n")
+        self.assertEqual(resp, b"20 text/plain\r\nhello%20world\n")
+
+
+if __name__ == "__main__":
+    unittest.main()