Pytest

.envrc

@@ -0,0 +1 @@
+use flake

.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: |

.gitignore

@@ -1,3 +1,5 @@
 /.lake
 /release/
 /release.tar.zst
+.direnv/
+__pycache__

flake.nix

@@ -0,0 +1,20 @@
+{
+ inputs = {
+ nixpkgs.url = "github:nixos/nixpkgs?ref=nixos-unstable";
+ };
+
+ outputs = { self, nixpkgs }: 
+ let 
+ pkgs = nixpkgs.legacyPackages.x86_64-linux;
+ in
+ {
+ devShells.x86_64-linux.default = pkgs.mkShell {
+ buildInputs = [
+ pkgs.lean4
+ (pkgs.python3.withPackages (python-pkgs: [
+ python-pkgs.pytest
+ ]))
+ ];
+ };
+ };
+}

test_daemon.py

@@ -3,7 +3,9 @@
 '''
 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
+- try make it a bit faster
 '''
 
 import time
@@ -13,14 +15,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,14 +57,24 @@ def daemon():
  daemon_proc = subprocess.Popen(
  [daemon_command] + daemon_args,
  pass_fds=(sock.fileno(),),
- stdout=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
  )
 
  print(f"-- Daemon started with PID {daemon_proc.pid}")
  # Close the socket in the parent process
  sock.close()
+
+ # Read stdout and stderr
+ stdout, stderr = daemon_proc.communicate(timeout=1)
+ if stdout:
+ print(f"-- Daemon stdout:\n{stdout.decode()}")
+ if stderr:
+ print(f"-- Daemon stderr:\n{stderr.decode()}")
+
  yield daemon_proc
+ except Exception as e:
+ print(f"-- Error starting daemon: {e}")
  finally:
  print(f"-- Terminating daemon with PID {daemon_proc.pid}")
  daemon_proc.terminate()
@@ -75,96 +85,37 @@ 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():
+ time.sleep(1) # Increased sleep duration
+ msg = 'list'
+ expected = {'ok': {'timers': []}}
+ response = msg_and_response(msg)
+ assert response == expected, f"Test 'list' failed. Expected {expected}, got {response}"
 
-'''
-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():
+ time.sleep(1) # Increased sleep duration
+ 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__])