Skip to content
Snippets Groups Projects

[AGIPD][TESTs]test_agipdlib AGIPDCtrl and get_bias_voltage for AGIPD1M and AGIPD500K

Merged Karim Ahmed requested to merge test/test_agipdlib into master
1 file
+ 21
26
Compare changes
  • Side-by-side
  • Inline
+ 21
26
@@ -126,32 +126,27 @@ async def test_wait_on_transfer(
else:
run_path.symlink_to(pnfs_dir)
with mock.patch("webservice.config.webservice") as config:
config["correct"] = {
"in-folder": f"{tmp_root}/{{instrument}}/{{cycle}}/p{{proposal}}/raw"
}
response = mock.Mock()
response.status_code = 200
response.json = lambda: {"runs": [{"repositories": {r: None for r in repositories}}]}
client = mock.Mock()
client.get_runs_by_proposal_number_api = mock.Mock(return_value=response)
_run_path = mock.Mock(spec=run_path)
_run_path.resolve = lambda: Path("/") / run_path.resolve().relative_to(tmp_path)
try:
await wait_on_transfer(
client,
run=run_no,
proposal=str(prop_no),
run_dir=_run_path,
max_tries=1,
sleep_completion=0,
)
except MigrationError as e:
if not expected_error:
raise e
response = mock.Mock()
response.status_code = 200
response.json = lambda: {"runs": [{"repositories": {r: None for r in repositories}}]}
client = mock.Mock()
client.get_runs_by_proposal_number_api = mock.Mock(return_value=response)
_run_path = mock.Mock(spec=run_path)
_run_path.resolve = lambda: Path("/") / run_path.resolve().relative_to(tmp_path)
try:
await wait_on_transfer(
client,
run=run_no,
proposal=str(prop_no),
run_dir=_run_path,
max_tries=1,
sleep_completion=0,
)
except MigrationError as e:
if not expected_error:
raise e
@pytest.mark.asyncio
@pytest.mark.parametrize(
Loading