gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

QTZPOC-20

Browse Source 
QTZPOC-20: Refactor test calls by extracting inline arguments into local variables for improved readability.
This commit is contained in:
Vitor Pimentel dos Santos
2026-09-10 09:34:06 -03:00
parent 7c45dbb407
commit 9e0da03717
2 changed files with 8 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/laborious/utils/bundle/test_reader.py
View File

@@ -486,10 +486,11 @@ def test_extract_member_rejects_a_stream_larger_than_its_declared_size(
monkeypatch.setattr(zipfile.ZipExtFile, 'read', _oversized_read) monkeypatch.setattr(zipfile.ZipExtFile, 'read', _oversized_read)
resolved_root = root.resolve()
with zipfile.ZipFile(io.BytesIO(buffer.getvalue())) as archive: with zipfile.ZipFile(io.BytesIO(buffer.getvalue())) as archive:
entry = archive.infolist()[0] entry = archive.infolist()[0]
with pytest.raises(BundleGateError, match='more than its declared') as error: with pytest.raises(BundleGateError, match='more than its declared') as error:
reader._extract_member(archive, entry, root, root.resolve()) reader._extract_member(archive, entry, root, resolved_root)
assert error.value.gate == 5 assert error.value.gate == 5

9
tests/laborious/workflows/test_import_model.py
View File

@@ -773,10 +773,11 @@ async def test_a_cancelled_terminal_write_propagates_instead_of_being_reported(
@mark.asyncio @mark.asyncio
async def test_a_cancelled_pipeline_failure_report_propagates(import_workflow: ImportModel): async def test_a_cancelled_pipeline_failure_report_propagates(import_workflow: ImportModel):
patcher = cancelling_activities() patcher = cancelling_activities()
error = RuntimeError('boom')
try: try:
with raises(asyncio.CancelledError): with raises(asyncio.CancelledError):
await import_workflow._report_failure( await import_workflow._report_failure(
BASE, RuntimeError('boom'), step=ImportStep.REGISTRATION.value, gate=None BASE, error, step=ImportStep.REGISTRATION.value, gate=None
) )
finally: finally:
patcher.stop() patcher.stop()
@@ -785,9 +786,10 @@ async def test_a_cancelled_pipeline_failure_report_propagates(import_workflow: I
@mark.asyncio @mark.asyncio
async def test_a_cancelled_status_write_failure_report_propagates(import_workflow: ImportModel): async def test_a_cancelled_status_write_failure_report_propagates(import_workflow: ImportModel):
patcher = cancelling_activities() patcher = cancelling_activities()
error = RuntimeError('db down')
try: try:
with raises(asyncio.CancelledError): with raises(asyncio.CancelledError):
await import_workflow._report_status_write_failure(BASE, RuntimeError('db down')) await import_workflow._report_status_write_failure(BASE, error)
finally: finally:
patcher.stop() patcher.stop()
@@ -795,9 +797,10 @@ async def test_a_cancelled_status_write_failure_report_propagates(import_workflo
@mark.asyncio @mark.asyncio
async def test_a_cancelled_retention_check_propagates(import_workflow: ImportModel): async def test_a_cancelled_retention_check_propagates(import_workflow: ImportModel):
patcher = cancelling_activities() patcher = cancelling_activities()
input_data = valid_input()
try: try:
with raises(asyncio.CancelledError): with raises(asyncio.CancelledError):
await import_workflow._ensure_retention(BASE, valid_input()) await import_workflow._ensure_retention(BASE, input_data)
finally: finally:
patcher.stop() patcher.stop()