kschappell/py-dvt-ate
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

WIP: Use thread pool executor for integration tests

Browse Source 
Move synchronous test execution to thread pool executor to avoid
blocking the async event loop. This prevents deadlocks when sync
client code tries to communicate with async server in same loop.

Note: Integration tests still experiencing timeouts - needs further
investigation. Unit tests and TCP server communication are working.
This commit is contained in:
Kai Chappell
2025-10-08 16:16:13 +00:00
parent e33bddd7a0
commit ae3de8484e
1 changed files with 56 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
tests/integration/test_tempco.py
View File

@@ -24,6 +24,8 @@ class TestTempCoIntegration:
async def test_tempco_runs_successfully(self, tmp_path: Path) -> None:
"""Test TempCo test runs end-to-end with simulator."""
import asyncio
# Start simulation server
server_config = ServerConfig(
host="127.0.0.1",
@@ -35,6 +37,9 @@ class TestTempCoIntegration:
server = SimulationServer(server_config)
await server.start()
# Give server time to fully initialize
await asyncio.sleep(0.1)
try:
# Create instrument set connected to simulator
instrument_config = InstrumentConfig(
@@ -46,15 +51,6 @@ class TestTempCoIntegration:
)
instruments = InstrumentFactory.create(instrument_config)
# Connect to instruments
instruments.chamber.connect()
instruments.psu.connect()
instruments.dmm.connect()
# Configure instruments
instruments.chamber.set_ramp_rate(10.0) # Fast ramp for testing
instruments.psu.enable_output(1, False) # Ensure off initially
# Create test repository
db_path = tmp_path / "test.db"
repository = SQLiteRepository(db_path)
@@ -91,13 +87,28 @@ class TestTempCoIntegration:
},
)
# Create and execute test
# Create test
test = TempCoTest()
assert test.name == "tempco"
assert test.description == "Output voltage temperature coefficient"
# Run test (this is synchronous, but simulation runs async in background)
status = test.execute(context)
# Run synchronous test code in thread pool to avoid blocking event loop
loop = asyncio.get_running_loop()
def run_test():
# Connect to instruments
instruments.chamber.connect()
instruments.psu.connect()
instruments.dmm.connect()
# Configure instruments
instruments.chamber.set_ramp_rate(10.0) # Fast ramp for testing
instruments.psu.enable_output(1, False) # Ensure off initially
# Run test
return test.execute(context)
status = await loop.run_in_executor(None, run_test)
# Verify test completed
assert status in (TestStatus.PASSED, TestStatus.FAILED)
@@ -138,6 +149,8 @@ class TestTempCoIntegration:
async def test_tempco_with_minimal_config(self, tmp_path: Path) -> None:
"""Test TempCo uses default configuration when not specified."""
import asyncio
# Start simulation server
server_config = ServerConfig(
host="127.0.0.1",
@@ -148,6 +161,9 @@ class TestTempCoIntegration:
server = SimulationServer(server_config)
await server.start()
# Give server time to fully initialize
await asyncio.sleep(0.1)
try:
# Create instrument set
instrument_config = InstrumentConfig(
@@ -159,11 +175,6 @@ class TestTempCoIntegration:
)
instruments = InstrumentFactory.create(instrument_config)
# Connect to instruments
instruments.chamber.connect()
instruments.psu.connect()
instruments.dmm.connect()
# Create repository
db_path = tmp_path / "test_minimal.db"
repository = SQLiteRepository(db_path)
@@ -186,9 +197,19 @@ class TestTempCoIntegration:
},
)
# Execute test
# Execute test in thread pool
test = TempCoTest()
status = test.execute(context)
loop = asyncio.get_running_loop()
def run_test():
# Connect to instruments
instruments.chamber.connect()
instruments.psu.connect()
instruments.dmm.connect()
# Run test
return test.execute(context)
status = await loop.run_in_executor(None, run_test)
# Should complete without error
assert status in (TestStatus.PASSED, TestStatus.FAILED, TestStatus.ERROR)
@@ -205,6 +226,8 @@ class TestTempCoIntegration:
async def test_tempco_handles_errors_gracefully(self, tmp_path: Path) -> None:
"""Test TempCo returns ERROR status when instruments fail."""
import asyncio
# Start simulation server
server_config = ServerConfig(
host="127.0.0.1",
@@ -215,6 +238,9 @@ class TestTempCoIntegration:
server = SimulationServer(server_config)
await server.start()
# Give server time to fully initialize
await asyncio.sleep(0.1)
try:
# Create instrument set
instrument_config = InstrumentConfig(
@@ -226,11 +252,6 @@ class TestTempCoIntegration:
)
instruments = InstrumentFactory.create(instrument_config)
# Connect to instruments
instruments.chamber.connect()
instruments.psu.connect()
instruments.dmm.connect()
# Create repository
db_path = tmp_path / "test_error.db"
repository = SQLiteRepository(db_path)
@@ -248,13 +269,22 @@ class TestTempCoIntegration:
},
)
# Execute test
# Execute test in thread pool
test = TempCoTest()
loop = asyncio.get_running_loop()
def run_test():
# Connect to instruments
instruments.chamber.connect()
instruments.psu.connect()
instruments.dmm.connect()
# Run test
return test.execute(context)
# Should handle gracefully (may return FAILED or ERROR)
# The test should not raise an unhandled exception
try:
status = test.execute(context)
status = await loop.run_in_executor(None, run_test)
# If it completes, it should indicate an error or failure
assert status in (TestStatus.ERROR, TestStatus.FAILED)
except Exception: