Add Unit Tests

.github/workflows/coverage.yml

@@ -0,0 +1,39 @@
+name: Run automated tests
+
+on: push
+
+jobs:
+  coverage:
+    name: Check coverage with coverage.py
+    runs-on: ubuntu-latest
+    permissions:
+      pull-requests: write
+    steps:
+      - uses: actions/checkout@v4.2.2
+      - name: Install the latest version of uv
+        uses: astral-sh/setup-uv@v6.1.0
+      - name: Setup environment
+        run: uv sync --dev
+      - name: Run unit tests
+        run: uv run coverage run
+      - name: Report coverage
+        run: uv run coverage report --format markdown >> $GITHUB_STEP_SUMMARY
+      - name: Generate coverage artifacts
+        if: ${{ failure() }}
+        run: |
+          uv run coverage html --fail-under 0
+          uv run coverage xml --fail-under 0
+          uv run coverage json --fail-under 0
+          uv run coverage lcov --fail-under 0
+          uv run coverage annotate
+      - name: Upload artifacts
+        if: ${{ failure() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: Coverage reports
+          path: |
+            coverage.json
+            coverage.lcov
+            coverage.xml
+            htmlcov/
+            *.py,cover

.gitignore

@@ -0,0 +1,6 @@
+.venv/
+# Temporary unit testing artefacts
+_trial_temp/
+__pycache__/
+.coverage
+*.py,cover

pyproject.toml

@@ -3,18 +3,19 @@ name = "remote-server"
 version = "0.1.0"
 description = "NVDA Remote Access remote relay server."
 readme = "README.md"
-requires-python = ">=3.13.3"
+requires-python = "~=3.13"
 dependencies = [
-    "pyopenssl>=25.1.0",
-    "service-identity>=24.2.0",
-    "twisted>=24.11.0",
+    "pyopenssl~=25.1",
+    "service-identity~=24.2",
+    "twisted~=24.11",
 ]
 
 [dependency-groups]
 dev = [
-    "pre-commit>=4.2.0",
-    "pyright>=1.1.401",
-    "ruff>=0.11.12",
+    "coverage~=7.8",
+    "pre-commit~=4.2",
+    "pyright~=1.1",
+    "ruff~=0.11",
 ]
 
 [tool.pyright]
@@ -38,20 +39,15 @@ exclude = [
 ]
 
 # General config
-analyzeUnannotatedFunctions = false
+analyzeUnannotatedFunctions = true
 deprecateTypingAliases = true
 
-reportMissingTypeArgument=false
-reportUnknownVariableType=false
-reportAttributeAccessIssue=false
-reportUnknownMemberType=false
-reportUnknownParameterType=false
-reportUnknownArgumentType=false
-reportMissingParameterType=false
-reportUnknownLambdaType=false
-reportUnusedVariable=false
-reportOptionalMemberAccess=false
-reportDeprecated=false
+# The following options cause problems due to polymorphism in Twisted
+reportAttributeAccessIssue = false
+reportUnknownMemberType = false
+reportOptionalMemberAccess = false
+# The following option causes problems due to dynamic member access
+reportUnknownArgumentType = false
 
 [tool.ruff]
 line-length = 110
@@ -78,3 +74,13 @@ ignore = [
 	# indentation contains tabs
 	"W191",
 ]
+
+[tool.coverage.run]
+branch = true
+command_line = "-m twisted.trial ./test.py"
+
+[tool.coverage.report]
+fail_under = 100
+exclude_also = [
+    'if __name__ == .__main__.:',
+]

server.py

@@ -10,6 +10,7 @@
 
 from OpenSSL import crypto
 from twisted.internet import reactor, ssl
+from twisted.internet.interfaces import ITCPTransport
 from twisted.internet.protocol import Factory, defer
 from twisted.internet.task import LoopingCall
 from twisted.protocols.basic import LineReceiver
@@ -30,15 +31,15 @@ def __init__(self, key, server_state=None):
 		self.server_state = server_state
 
 	def add_client(self, client):
-		if client.protocol.protocol_version == 1:
+		if client.protocol.protocol_version == 1:  # pragma: no cover - protocol v1 is not tested
 			ids = [c.user_id for c in self.clients.values()]
 			msg = dict(type="channel_joined", channel=self.key, user_ids=ids, origin=client.user_id)
 		else:
 			clients = [i.as_dict() for i in self.clients.values()]
 			msg = dict(type="channel_joined", channel=self.key, origin=client.user_id, clients=clients)
 		client.send(**msg)
 		for existing_client in self.clients.values():
-			if existing_client.protocol.protocol_version == 1:
+			if existing_client.protocol.protocol_version == 1:  # pragma: no cover - protocol v1 is not tested
 				existing_client.send(type="client_joined", user_id=client.user_id)
 			else:
 				existing_client.send(type="client_joined", client=client.as_dict())
@@ -48,7 +49,7 @@ def remove_connection(self, con):
 		if con.user_id in self.clients:
 			del self.clients[con.user_id]
 		for client in self.clients.values():
-			if client.protocol.protocol_version == 1:
+			if client.protocol.protocol_version == 1:  # pragma: no cover - protocol v1 is not tested
 				client.send(type="client_left", user_id=con.user_id)
 			else:
 				client.send(type="client_left", client=con.as_dict())
@@ -77,7 +78,10 @@ def __init__(self):
 
 	def connectionMade(self):
 		logger.info("Connection %d from %s" % (self.connection_id, self.transport.getPeer()))
-		self.transport.setTcpNoDelay(True)
+		# We use a non-tcp transport for unit testing,
+		# which doesn't support setTcpNoDelay.
+		if isinstance(self.transport, ITCPTransport):  # pragma: no cover
+			self.transport.setTcpNoDelay(True)
 		self.bytes_sent = 0
 		self.bytes_received = 0
 		self.user = User(protocol=self)
@@ -90,7 +94,9 @@ def connectionLost(self, reason):
 			% (self.connection_id, self.bytes_sent, self.bytes_received),
 		)
 		self.user.connection_lost()
-		if self.cleanup_timer is not None and not self.cleanup_timer.cancelled:
+		if (
+			self.cleanup_timer is not None and not self.cleanup_timer.cancelled
+		):  # pragma: no cover - not sure how to trigger this
 			self.cleanup_timer.cancel()
 
 	def lineReceived(self, line):
@@ -169,12 +175,14 @@ def generate_key(self):
 		self.server_state.generated_keys.add(key)
 		self.server_state.generated_ips[ip] = time.time()
 		reactor.callLater(GENERATED_KEY_EXPIRATION_TIME, lambda: self.server_state.generated_keys.remove(key))
-		if key:
+		if key:  # pragma: no cover - I can't work out why this branch is here. When would this be False?
 			self.send(type="generate_key", key=key)
 		return key
 
 	def connection_lost(self):
-		if self.channel is not None:
+		if (
+			self.channel is not None
+		):  # pragma: no branch - we don't care about the alternative, as it's a no-op
 			self.channel.remove_connection(self)
 
 	def join(self, channel, connection_type):
@@ -185,7 +193,8 @@ def join(self, channel, connection_type):
 		self.channel = self.server_state.find_or_create_channel(channel)
 		self.channel.add_client(self)
 
-	def do_generate_key(self):
+	# TODO: Work out if this is ever called.
+	def do_generate_key(self):  # pragma: no cover
 		key = self.generate_key()
 		if key:
 			self.send(type="generate_key", key=key)
@@ -214,6 +223,7 @@ def __init__(self):
 		self.generated_keys = set()
 		# Dictionary of ips to generated time for people who have generated keys.
 		self.generated_ips = {}
+		self.motd: str | None = None
 
 	def remove_channel(self, channel):
 		del self.channels[channel]
@@ -238,7 +248,8 @@ class Options(usage.Options):
 	]
 
 
-def main():
+# Exclude from coverage as it's hard to unit test.
+def main():  # pragma: no cover
 	config = Options()
 	config.parseOptions()
 	privkey = open(config["privkey"]).read()