import datetime
from datetime import date
from pathlib import Path
import pytest
from requests.exceptions import HTTPError
from .utils import generate
[docs]
def test_login(icatplus_restricted_client):
client, messages = icatplus_restricted_client
with pytest.raises(RuntimeError, match="Login is required."):
_ = client.session_id
with pytest.raises(HTTPError, match="403 Client Error: Authentication failed"):
_ = client.login("wrong")
result = client.login("correct")
assert result["sessionId"] == client.session_id
assert messages.empty()
[docs]
def test_login_with_username_plugin(icatplus_restricted_client):
client, messages = icatplus_restricted_client
result = client.login(password="correct", username="john", plugin="db")
assert result["sessionId"] == client.session_id
assert messages.empty()
[docs]
def test_get_investigations_by(icatplus_restricted_client, icat_metadata_client):
client, messages = icatplus_restricted_client
mclient, mmessages = icat_metadata_client
with pytest.raises(RuntimeError, match="Login is required."):
_ = client.get_investigations_by()
assert client.login("correct")
investigations = client.get_investigations_by()
assert isinstance(investigations, list)
mclient.start_investigation(proposal="hg123", beamline="id00")
message = mmessages.get(timeout=10)
expected_release_date = _release_date_from_start_date(message["startDate"])
investigations = client.get_investigations_by()
expected = [
{
"experiment": "hg123",
"id": 0,
"instrument": {"name": "id00"},
"proposal": "hg123",
"startDate": message["startDate"],
"releaseDate": expected_release_date.strftime("%Y-%m-%d"),
}
]
assert investigations == expected
assert messages.empty()
assert mmessages.empty()
[docs]
def test_get_investigations_by_release_dates(
icatplus_restricted_client, icat_metadata_client, icat_data_dir
):
client, messages = icatplus_restricted_client
mclient, mmessages = icat_metadata_client
assert client.login("correct")
mclient.start_investigation(proposal="hg123", beamline="id00")
message = mmessages.get(timeout=10)
release_date = _release_date_from_start_date(message["startDate"])
# match
investigations = client.get_investigations_by(release_dates=[release_date])
assert len(investigations) == 1
assert investigations[0]["releaseDate"] == release_date.strftime("%Y-%m-%d")
# no match
investigations = client.get_investigations_by(release_dates=[date(2000, 1, 1)])
assert investigations == []
assert messages.empty()
assert mmessages.empty()
[docs]
def test_get_datasets_by(icatplus_restricted_client, icat_metadata_client):
client, messages = icatplus_restricted_client
mclient, mmessages = icat_metadata_client
investigation_id = generate.icat_id()
with pytest.raises(RuntimeError, match="Login is required."):
client.get_datasets_by(investigation_id=investigation_id)
assert client.login("correct")
# Create an investigation with datasets
mclient.start_investigation(proposal="hg123", beamline="id00")
mmessages.get(timeout=10)
investigations = client.get_investigations_by()
investigation_id = investigations[0]["id"]
nb_datasets = 4
for i in range(nb_datasets):
icat_metadata_client[0].send_metadata(
proposal="hg123",
beamline="id00",
dataset=f"dataset_{i}",
path=Path.home() / f"dataset_{i}",
metadata={
"Sample_name": "samplename",
},
)
icat_metadata_client[1].get(timeout=10)
datasets = client.get_datasets_by(investigation_id=investigation_id)
assert isinstance(datasets, list)
assert len(datasets) == nb_datasets
paginated = client.get_datasets_by(
investigation_id=investigation_id,
offset=0,
limit=2,
)
assert isinstance(paginated, list)
assert paginated == datasets[:2]
assert messages.empty()
assert mmessages.empty()
[docs]
def test_get_parcels_by(icatplus_restricted_client):
client, messages = icatplus_restricted_client
investigation_id = generate.icat_id()
with pytest.raises(RuntimeError, match="Login is required."):
_ = client.get_parcels_by(investigation_id)
assert client.login("correct")
parcels = client.get_parcels_by(investigation_id)
assert parcels == []
assert messages.empty()
[docs]
def test_get_samples_by(icatplus_restricted_client):
client, messages = icatplus_restricted_client
investigation_id = generate.icat_id()
with pytest.raises(RuntimeError, match="Login is required."):
_ = client.get_samples_by(investigation_id)
assert client.login("correct")
samples = client.get_samples_by(investigation_id)
assert samples == []
assert messages.empty()
[docs]
@pytest.mark.parametrize(
"params, expected_ids",
[
({"investigation_id": "mx415"}, ["sa", "sb"]),
({"investigation_id": "mx2012"}, ["sc"]),
({"sample_ids": "sc"}, ["sc"]),
({"investigation_id": "mx415", "sample_ids": "sa"}, ["sa"]),
({"investigation_id": "mx100"}, []),
({"sample_ids": "sd"}, []),
],
)
def test_get_samples_by_returns_expected_data(
icatplus_restricted_client, params, expected_ids
):
client, messages = icatplus_restricted_client
client.login("correct")
samples = client.get_samples_by(**params)
returned_ids = [s["id"] for s in samples]
assert returned_ids == expected_ids
assert messages.empty()
[docs]
@pytest.mark.parametrize(
"use_chunks, chunk_size",
[
(False, None),
(True, 1024),
],
)
def test_download_file_by_sample_parametrized(
icatplus_restricted_client, use_chunks, chunk_size
):
client, messages = icatplus_restricted_client
sample_id = generate.icat_id()
resource_id = generate.icat_id()
if use_chunks:
with pytest.raises(RuntimeError, match="Login is required."):
_ = client.download_file_by(
sample_id, resource_id, use_chunks=use_chunks, chunk_size=chunk_size
)
else:
with pytest.raises(RuntimeError, match="Login is required."):
_ = client.download_file_by(sample_id, resource_id)
assert client.login("correct")
result = client.download_file_by(
sample_id, resource_id, use_chunks=use_chunks, chunk_size=chunk_size
)
assert isinstance(result, bytes)
assert result == b"fake file content"
assert messages.empty()
[docs]
def test_get_investigation_users(icatplus_restricted_client):
client, messages = icatplus_restricted_client
investigation_id = generate.icat_id()
with pytest.raises(RuntimeError, match="Login is required."):
_ = client.get_investigation_users(investigation_id)
assert client.login("correct")
users = client.get_investigation_users(investigation_id)
assert users == []
assert messages.empty()
def _release_date_from_start_date(start_date_str: str) -> datetime.date:
start = datetime.datetime.fromisoformat(start_date_str[:19])
return start.date().replace(year=start.year + 3)