QTZPOC-20
QTZPOC-20: Improve error handling and classification tests for model import activities and add validation for OPC repository security.
This commit is contained in:
@@ -646,6 +646,77 @@ async def test_a_gate_failure_inside_the_fused_activity_still_names_its_step_and
|
||||
assert failure.value.details[0]['gate'] == 3
|
||||
|
||||
|
||||
async def _fake_download(_input_data: dict) -> dict:
|
||||
return {'path': 'p', 'size': 1, 'digest': 'd'}
|
||||
|
||||
|
||||
async def _fake_open(_input_data: dict) -> dict:
|
||||
return {
|
||||
'extracted_dir': 'e',
|
||||
'metadata': {
|
||||
'model_name': 'factory_model',
|
||||
'model_version': '1',
|
||||
'experiment_name': 'factory_experiment',
|
||||
},
|
||||
'parameters': {},
|
||||
'digest': 'd',
|
||||
}
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_a_misconfigured_repository_at_experiment_creation_is_not_wrapped(
|
||||
import_activity: ModelImport,
|
||||
):
|
||||
"""`ImportMisconfiguredError` is a config problem, not a broken dependency — it must reach the
|
||||
workflow untouched, not get rewrapped into a generic `ApplicationError`."""
|
||||
import_activity.import_mlflow_repository = None
|
||||
|
||||
with (
|
||||
patch.object(import_activity, '_download_bundle', side_effect=_fake_download),
|
||||
patch.object(import_activity, '_open_bundle', side_effect=_fake_open),
|
||||
raises(ImportMisconfiguredError),
|
||||
):
|
||||
await import_activity.prepare_and_upload_import_bundle({'metadata': METADATA})
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_a_broken_dependency_at_experiment_creation_is_tagged_with_its_step(
|
||||
import_activity: ModelImport,
|
||||
):
|
||||
"""A raw MLflow/network failure carries no location of its own, so the fused activity must tag
|
||||
it with the step it actually broke at — otherwise the workflow's fallback misclassifies it."""
|
||||
|
||||
async def fake_create_experiment(_input_data: dict) -> dict:
|
||||
raise RuntimeError('mlflow unreachable')
|
||||
|
||||
with (
|
||||
patch.object(import_activity, '_download_bundle', side_effect=_fake_download),
|
||||
patch.object(import_activity, '_open_bundle', side_effect=_fake_open),
|
||||
patch.object(import_activity, '_create_experiment', side_effect=fake_create_experiment),
|
||||
raises(ApplicationError, match='mlflow unreachable') as failure,
|
||||
):
|
||||
await import_activity.prepare_and_upload_import_bundle({'metadata': METADATA})
|
||||
|
||||
assert failure.value.details[0]['step'] == ImportStep.EXPERIMENT_CREATION.value
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_a_misconfigured_repository_at_artifact_upload_is_not_wrapped(
|
||||
import_activity: ModelImport,
|
||||
):
|
||||
async def fake_create_experiment(_input_data: dict) -> dict:
|
||||
import_activity.import_mlflow_repository = None
|
||||
return {'experiment_id': '1', 'experiment_name': 'factory_experiment'}
|
||||
|
||||
with (
|
||||
patch.object(import_activity, '_download_bundle', side_effect=_fake_download),
|
||||
patch.object(import_activity, '_open_bundle', side_effect=_fake_open),
|
||||
patch.object(import_activity, '_create_experiment', side_effect=fake_create_experiment),
|
||||
raises(ImportMisconfiguredError),
|
||||
):
|
||||
await import_activity.prepare_and_upload_import_bundle({'metadata': METADATA})
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_a_retry_of_the_fused_activity_redoes_all_four_phases(import_activity: ModelImport):
|
||||
"""A retry re-runs download, open, experiment creation and upload from the start.
|
||||
|
||||
@@ -442,6 +442,24 @@ async def test_the_claim_writes_before_it_reads(
|
||||
assert engine.statements[1].verb == 'SELECT'
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_a_refused_claim_statement_is_reported_and_reraised():
|
||||
"""The database refusing the claim write is a config problem (bad column, CHECK violation), not
|
||||
a retryable one — it must be reported, and it must still propagate to fail the activity."""
|
||||
engine = FakeStatusEngine(
|
||||
write_errors=[IntegrityError('UPDATE ...', {}, Exception('refused'))],
|
||||
)
|
||||
activity = build_activity(engine)
|
||||
|
||||
with raises(ImportStatusRejectedError) as failure:
|
||||
await activity.claim_import_status(claim_input())
|
||||
|
||||
assert failure.value.non_retryable is True
|
||||
logged = activity.logger.custom_error.call_args_list[0].args[0]
|
||||
assert ImportErrorCode.STATUS_NOT_RECORDED.value in logged
|
||||
assert 'status_write' in logged
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
@mark.parametrize(
|
||||
('row', 'expected'),
|
||||
|
||||
@@ -144,6 +144,16 @@ async def test_set_security_missing_certificates(opc_repository):
|
||||
assert str(e) == 'Certificate and private key paths must be provided for secure connection.'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_set_security_missing_server_uri(opc_repository):
|
||||
opc_repository.server_uri = None
|
||||
|
||||
try:
|
||||
await opc_repository.set_security()
|
||||
except ValueError as e:
|
||||
assert str(e) == 'server_uri must be provided for secure connection.'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_set_security_missing_client(opc_repository):
|
||||
opc_repository.client = None
|
||||
|
||||
Reference in New Issue
Block a user