switch tests to pytest

.github/workflows/ci.yml

@@ -15,9 +15,16 @@ jobs:
  auto-config: "false"
  use-mathlib-cache: "false"
  build: "true"
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: '3.x'
  - name: Run integration tests
  run: |
- ./test_daemon.py
+ python -m pip install --upgrade pip
+ pip install pytest
+ pytest test_daemon.py
 
  # - name: Create package
  # run: |

flake.nix

@@ -11,6 +11,9 @@
  devShells.x86_64-linux.default = pkgs.mkShell {
  buildInputs = [
  pkgs.lean4
+ (pkgs.python3.withPackages (python-pkgs: [
+ python-pkgs.pytest
+ ]))
  ];
  };
  };

src/Sand/SandDaemon.lean

@@ -208,6 +208,7 @@ def envFd : BaseIO (Except EnvFdError UInt32) := do
 def systemdSockFd : UInt32 := 3
 
 def SandDaemon.main (_args : List String) : IO α := do
+ IO.eprintln "Starting Sand daemon."
 
  let fd? ← envFd
  let fd : UInt32 ← match fd? with

test_daemon.py

@@ -3,7 +3,8 @@
 '''
 Sand integration tests.
 
-If this gets too convoluted, we'll switch to a proper test framework.
+TODO:
+- take better advantage of pytest's features, eg fixtures
 '''
 
 import time
@@ -13,14 +14,12 @@
 import fcntl
 import subprocess
 import json
-import codecs
+import pytest
 
 from contextlib import contextmanager
 
 SOCKET_PATH = "./test.sock"
 
-failure = False
-
 '''
 Remove the socket file if it already exists
 '''
@@ -57,13 +56,15 @@ def daemon():
  daemon_proc = subprocess.Popen(
  [daemon_command] + daemon_args,
  pass_fds=(sock.fileno(),),
- stdout=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL,
+ env={"SAND_SOCKFD": str(sock.fileno())},
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
  )
 
  print(f"-- Daemon started with PID {daemon_proc.pid}")
  # Close the socket in the parent process
  sock.close()
+
  yield daemon_proc
  finally:
  print(f"-- Terminating daemon with PID {daemon_proc.pid}")
@@ -75,96 +76,36 @@ def daemon():
  ensure_socket_deleted()
 
 
-def main():
- print("--------------------------")
- print("Starting integration tests")
- print("--------------------------")
-
- with daemon():
- # wait a moment for the daemon to start
- time.sleep(0.1)
- run_client_tests()
-
 @contextmanager
 def client_socket():
  with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client_sock:
  client_sock.connect(SOCKET_PATH)
  yield client_sock
 
-def test_msg_and_response(test_name, msg, expected):
- global failure
- try:
- msg_bytes = bytes(json.dumps(msg), encoding='utf-8')
-
- with client_socket() as client_sock:
- client_sock.send(msg_bytes)
- resp_bytes = client_sock.recv(1024)
+def msg_and_response(msg):
+ msg_bytes = bytes(json.dumps(msg), encoding='utf-8')
 
- response = json.loads(resp_bytes.decode('utf-8'))
+ with client_socket() as client_sock:
+ client_sock.send(msg_bytes)
+ resp_bytes = client_sock.recv(1024)
 
- if response != expected:
- print()
- print(f'-- test {test_name} failed.')
- print(f'sent: {msg}')
- print(f'expected: {expected}')
- print(f'received: {response}')
+ response = json.loads(resp_bytes.decode('utf-8'))
+ return response
 
- failure = True
- print('❌', end='', flush=True)
- return
-
- print('✔️', end='', flush=True)
- except Exception as e:
- print()
- print(f'-- test {test_name} failed.')
- print(f'sent: {msg}')
- print(f'expected: {expected}')
- print(f'but got exception: {e}')
-
- failure = True
- print('❌', end='', flush=True)
- return
+def test_list():
+ with daemon():
+ msg = 'list'
+ expected = {'ok': {'timers': []}}
+ response = msg_and_response(msg)
+ assert response == expected, f"Test 'list' failed. Expected {expected}, got {response}"
+ assert False
 
-'''
-format for tests:
-{
- 'test_name': Name of the test,
- 'msg': Message to send to the daemon. Will be serialised with 
- json.dumps().
- 'expected': Expected response from the daemon. Will be deserialised with 
- json.loads().
-}
-'''
-test_cases = [
- {
- 'test_name': 'list',
- 'msg': 'list',
- 'expected': {'ok': {'timers': []}}
- },
- {
- 'test_name': 'add',
- 'msg': {'addTimer': {'duration': {'millis': 60000}}},
- 'expected': 'ok'
- },
-]
-
-def run_client_tests():
- print(f'-- Running client tests against {SOCKET_PATH}...')
-
- for test_case in test_cases:
- test_msg_and_response(**test_case)
- print()
-
- if failure:
- print("-------------------")
- print("Some tests failed")
- print("-------------------")
- sys.exit(1)
- else:
- print("-------------------")
- print("All tests passed")
- print("-------------------")
+def test_add():
+ with daemon():
+ msg = {'addTimer': {'duration': {'millis': 60000}}}
+ expected = 'ok'
+ response = msg_and_response(msg)
+ assert response == expected, f"Test 'add' failed. Expected {expected}, got {response}"
 
 if __name__ == "__main__":
- main()
-
+ pytest.main([__file__])