Skip to content

Commit

Permalink
Skip certain tests on Windows
Browse files Browse the repository at this point in the history
Sending signals on Windows is surprisingly difficult. CTRL-C is handled by `SIGINT`, which is tested on other platforms and should behave the same. Improving test coverage for Windows will be a future task.
  • Loading branch information
RealOrangeOne committed Sep 13, 2024
1 parent a80f789 commit fe25e12
Showing 1 changed file with 6 additions and 4 deletions.
10 changes: 6 additions & 4 deletions tests/tests/test_database_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -1230,6 +1230,7 @@ def test_run_subprocess(self) -> None:

self.assertEqual(result.status, ResultStatus.COMPLETE)

@skipIf(sys.platform == "win32", "Terminate is always forceful on Windows")
def test_interrupt_no_tasks(self) -> None:
process = self.start_worker()

Expand All @@ -1240,6 +1241,7 @@ def test_interrupt_no_tasks(self) -> None:
process.wait(timeout=0.5)
self.assertEqual(process.returncode, 0)

@skipIf(sys.platform == "win32", "Cannot emulate CTRL-C on Windows")
def test_interrupt_signals(self) -> None:
for sig in [
signal.SIGINT, # ctrl-c
Expand All @@ -1266,6 +1268,7 @@ def test_interrupt_signals(self) -> None:

self.assertEqual(result.status, ResultStatus.COMPLETE)

@skipIf(sys.platform == "win32", "Cannot emulate CTRL-C on Windows")
def test_repeat_ctrl_c(self) -> None:
result = test_tasks.hang.enqueue()

Expand Down Expand Up @@ -1346,15 +1349,14 @@ def test_keyboard_interrupt_task(self) -> None:
self.assertIsInstance(result.exception, KeyboardInterrupt)

def test_multiple_workers(self) -> None:
results = [test_tasks.noop_task.enqueue() for _ in range(10)]
results = [test_tasks.sleep_for.enqueue(0.1) for _ in range(10)]

for _ in range(3):
self.start_worker()
self.start_worker(["--batch"])

time.sleep(self.WORKER_STARTUP_TIME)

for process in self.processes:
process.terminate()
process.wait(timeout=5)
self.assertIsNotNone(process.returncode)

Expand All @@ -1367,7 +1369,7 @@ def test_multiple_workers(self) -> None:
for process in self.processes:
stdout_text = process.stdout.read() # type:ignore[union-attr]
all_output += stdout_text
self.assertIn("shutting down gracefully", stdout_text)
self.assertIn("gracefully", stdout_text)

for result in results:
# Running and complete
Expand Down

0 comments on commit fe25e12

Please sign in to comment.