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

Fix TempCo integration tests with thread-based async server

Browse Source 
Redesign integration test architecture to eliminate async/sync deadlock:
- Run SimulationServer in dedicated background thread with own event loop
- Rewrite TempCo tests as fully synchronous (no @pytest.mark.asyncio)
- Add ServerThread fixture in tests/integration/conftest.py
- Fix Unicode encoding errors (replace deg, mu, +/- with ASCII)
- Optimize temperature points for faster settling (23C, 25C, 27C)

All 3 TempCo integration tests now passing in ~5 minutes total.
This commit is contained in:
Kai Chappell
2025-10-12 17:59:48 +00:00
parent 3fdaba500d
commit 42356efce2
3 changed files with 299 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

353
tests/integration/test_tempco.py
View File

@@ -7,108 +7,85 @@ from __future__ import annotations
from pathlib import Path
import pytest
from py_dvt_ate.data.models import TestStatus
from py_dvt_ate.data.repository import SQLiteRepository
from py_dvt_ate.framework.context import TestContext
from py_dvt_ate.framework.logger import TestLogger
from py_dvt_ate.instruments.factory import InstrumentConfig, InstrumentFactory
from py_dvt_ate.simulation.server import ServerConfig, SimulationServer
from py_dvt_ate.simulation.server import ServerConfig
from py_dvt_ate.tests.thermal.tempco import TempCoTest
@pytest.mark.asyncio(loop_scope="function")
class TestTempCoIntegration:
"""Integration tests for TempCo test with simulator."""
async def test_tempco_runs_successfully(self, tmp_path: Path) -> None:
def test_tempco_runs_successfully(
self, tmp_path: Path, simulation_server: ServerConfig
) -> None:
"""Test TempCo test runs end-to-end with simulator."""
import asyncio
# Start simulation server
server_config = ServerConfig(
host="127.0.0.1",
chamber_port=17000,
psu_port=17001,
dmm_port=17002,
physics_rate_hz=100.0,
# Create instrument set connected to simulator
instrument_config = InstrumentConfig(
backend="simulator",
simulator_host=simulation_server.host,
chamber_port=simulation_server.chamber_port,
psu_port=simulation_server.psu_port,
dmm_port=simulation_server.dmm_port,
)
server = SimulationServer(server_config)
await server.start()
instruments = InstrumentFactory.create(instrument_config)
# Give server time to fully initialize
await asyncio.sleep(0.1)
# Create test repository
db_path = tmp_path / "test.db"
repository = SQLiteRepository(db_path)
# Create test run
run_id = repository.create_run(
test_name="tempco",
config={
"temperatures": [23.0, 25.0, 27.0], # Close to start temp for fast settling
"input_voltage": 5.0,
"load_current": 0.1,
"settle_time": 0.2, # Short since temps close to start
"num_samples": 3, # Reduced for faster test
"tempco_limit": 100.0, # Relaxed for testing
},
description="Integration test of TempCo",
)
# Create test logger
logger = TestLogger(run_id, repository)
# Create test context
context = TestContext(
run_id=run_id,
instruments=instruments,
logger=logger,
config={
"temperatures": [23.0, 25.0, 27.0],
"input_voltage": 5.0,
"load_current": 0.1,
"settle_time": 0.2,
"num_samples": 3,
"tempco_limit": 100.0,
},
)
# Create test
test = TempCoTest()
assert test.name == "tempco"
assert test.description == "Output voltage temperature coefficient"
# Connect to instruments
instruments.chamber.connect() # type: ignore[attr-defined]
instruments.psu.connect() # type: ignore[attr-defined]
instruments.dmm.connect() # type: ignore[attr-defined]
try:
# Create instrument set connected to simulator
instrument_config = InstrumentConfig(
backend="simulator",
simulator_host="127.0.0.1",
chamber_port=17000,
psu_port=17001,
dmm_port=17002,
)
instruments = InstrumentFactory.create(instrument_config)
# 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)
# Create test run
run_id = repository.create_run(
test_name="tempco",
config={
"temperatures": [0.0, 25.0, 50.0], # Reduced for faster test
"input_voltage": 5.0,
"load_current": 0.1,
"settle_time": 0.5, # Reduced for faster test
"num_samples": 3, # Reduced for faster test
"tempco_limit": 100.0, # Relaxed for testing
},
description="Integration test of TempCo",
)
# Create test logger
logger = TestLogger(run_id, repository)
# Create test context
context = TestContext(
run_id=run_id,
instruments=instruments,
logger=logger,
config={
"temperatures": [0.0, 25.0, 50.0],
"input_voltage": 5.0,
"load_current": 0.1,
"settle_time": 0.5,
"num_samples": 3,
"tempco_limit": 100.0,
},
)
# Create test
test = TempCoTest()
assert test.name == "tempco"
assert test.description == "Output voltage temperature coefficient"
# 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)
# Run test
status = test.execute(context)
# Verify test completed
assert status in (TestStatus.PASSED, TestStatus.FAILED)
@@ -126,7 +103,7 @@ class TestTempCoIntegration:
# Find TempCo result
tempco_result = next(r for r in results if r.parameter == "temp_co")
assert tempco_result is not None
assert tempco_result.unit == "ppm/°C"
assert tempco_result.unit == "ppm/C"
assert tempco_result.lower_limit == -100.0
assert tempco_result.upper_limit == 100.0
@@ -145,71 +122,58 @@ class TestTempCoIntegration:
assert len(temps_recorded) >= 3
finally:
await server.stop()
# Disconnect from instruments
instruments.chamber.disconnect() # type: ignore[attr-defined]
instruments.psu.disconnect() # type: ignore[attr-defined]
instruments.dmm.disconnect() # type: ignore[attr-defined]
async def test_tempco_with_minimal_config(self, tmp_path: Path) -> None:
def test_tempco_with_minimal_config(
self, tmp_path: Path, simulation_server: ServerConfig
) -> None:
"""Test TempCo uses default configuration when not specified."""
import asyncio
# Start simulation server
server_config = ServerConfig(
host="127.0.0.1",
chamber_port=17100,
psu_port=17101,
dmm_port=17102,
# Create instrument set
instrument_config = InstrumentConfig(
backend="simulator",
simulator_host=simulation_server.host,
chamber_port=simulation_server.chamber_port,
psu_port=simulation_server.psu_port,
dmm_port=simulation_server.dmm_port,
)
server = SimulationServer(server_config)
await server.start()
instruments = InstrumentFactory.create(instrument_config)
# Give server time to fully initialize
await asyncio.sleep(0.1)
# Create repository
db_path = tmp_path / "test_minimal.db"
repository = SQLiteRepository(db_path)
run_id = repository.create_run(
test_name="tempco",
config={}, # Empty config - should use defaults
)
# Create logger and context with minimal config
logger = TestLogger(run_id, repository)
context = TestContext(
run_id=run_id,
instruments=instruments,
logger=logger,
config={
# Override temperatures for faster test
"temperatures": [24.0, 26.0],
"settle_time": 0.2,
"num_samples": 2,
},
)
# Execute test
test = TempCoTest()
# Connect to instruments
instruments.chamber.connect() # type: ignore[attr-defined]
instruments.psu.connect() # type: ignore[attr-defined]
instruments.dmm.connect() # type: ignore[attr-defined]
try:
# Create instrument set
instrument_config = InstrumentConfig(
backend="simulator",
simulator_host="127.0.0.1",
chamber_port=17100,
psu_port=17101,
dmm_port=17102,
)
instruments = InstrumentFactory.create(instrument_config)
# Create repository
db_path = tmp_path / "test_minimal.db"
repository = SQLiteRepository(db_path)
run_id = repository.create_run(
test_name="tempco",
config={}, # Empty config - should use defaults
)
# Create logger and context with minimal config
logger = TestLogger(run_id, repository)
context = TestContext(
run_id=run_id,
instruments=instruments,
logger=logger,
config={
# Override temperatures for faster test
"temperatures": [25.0, 50.0],
"settle_time": 0.2,
"num_samples": 2,
},
)
# 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)
status = await loop.run_in_executor(None, run_test)
# Run test
status = test.execute(context)
# Should complete without error
assert status in (TestStatus.PASSED, TestStatus.FAILED, TestStatus.ERROR)
@@ -222,69 +186,55 @@ class TestTempCoIntegration:
assert len(results) >= 1
finally:
await server.stop()
# Disconnect from instruments
instruments.chamber.disconnect() # type: ignore[attr-defined]
instruments.psu.disconnect() # type: ignore[attr-defined]
instruments.dmm.disconnect() # type: ignore[attr-defined]
async def test_tempco_handles_errors_gracefully(self, tmp_path: Path) -> None:
def test_tempco_handles_errors_gracefully(
self, tmp_path: Path, simulation_server: ServerConfig
) -> None:
"""Test TempCo returns ERROR status when instruments fail."""
import asyncio
# Start simulation server
server_config = ServerConfig(
host="127.0.0.1",
chamber_port=17200,
psu_port=17201,
dmm_port=17202,
# Create instrument set
instrument_config = InstrumentConfig(
backend="simulator",
simulator_host=simulation_server.host,
chamber_port=simulation_server.chamber_port,
psu_port=simulation_server.psu_port,
dmm_port=simulation_server.dmm_port,
)
server = SimulationServer(server_config)
await server.start()
instruments = InstrumentFactory.create(instrument_config)
# Give server time to fully initialize
await asyncio.sleep(0.1)
# Create repository
db_path = tmp_path / "test_error.db"
repository = SQLiteRepository(db_path)
run_id = repository.create_run(test_name="tempco", config={})
# Create logger and context
logger = TestLogger(run_id, repository)
context = TestContext(
run_id=run_id,
instruments=instruments,
logger=logger,
config={
"temperatures": [], # Invalid: empty temperature list
"settle_time": 0.1,
},
)
# Execute test
test = TempCoTest()
# Connect to instruments
instruments.chamber.connect() # type: ignore[attr-defined]
instruments.psu.connect() # type: ignore[attr-defined]
instruments.dmm.connect() # type: ignore[attr-defined]
try:
# Create instrument set
instrument_config = InstrumentConfig(
backend="simulator",
simulator_host="127.0.0.1",
chamber_port=17200,
psu_port=17201,
dmm_port=17202,
)
instruments = InstrumentFactory.create(instrument_config)
# Create repository
db_path = tmp_path / "test_error.db"
repository = SQLiteRepository(db_path)
run_id = repository.create_run(test_name="tempco", config={})
# Create logger and context
logger = TestLogger(run_id, repository)
context = TestContext(
run_id=run_id,
instruments=instruments,
logger=logger,
config={
"temperatures": [], # Invalid: empty temperature list
"settle_time": 0.1,
},
)
# 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 = await loop.run_in_executor(None, run_test)
status = test.execute(context)
# If it completes, it should indicate an error or failure
assert status in (TestStatus.ERROR, TestStatus.FAILED)
except Exception:
@@ -294,4 +244,7 @@ class TestTempCoIntegration:
logger.flush()
finally:
await server.stop()
# Disconnect from instruments
instruments.chamber.disconnect() # type: ignore[attr-defined]
instruments.psu.disconnect() # type: ignore[attr-defined]
instruments.dmm.disconnect() # type: ignore[attr-defined]