1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
import os
import tempfile
import time
from pathlib import Path
from shutil import rmtree
import pytest
try:
from h5pyd import Folder
from hsds.hsds_app import HsdsApp
with_reqd_pkgs = True
except ImportError:
with_reqd_pkgs = False
@pytest.fixture(scope="session")
def hsds_up():
"""Provide HDF Highly Scalable Data Service (HSDS) for h5pyd testing."""
if not with_reqd_pkgs:
pytest.skip("Required packages h5pyd and hsds not available")
root_dir = Path(tempfile.mkdtemp(prefix="tmp-hsds-root-"))
bucket_name = "pytest"
os.environ["BUCKET_NAME"] = bucket_name
# need to create a directory for our bucket
(root_dir / bucket_name).mkdir()
kwargs = {
"username": "h5netcdf-pytest",
"password": "TestEarlyTestEverything",
"root_dir": str(root_dir),
"logfile": str(root_dir / "hsds.log"),
"log_level": "DEBUG",
"host": "localhost",
"sn_port": 5101,
}
os.environ.update(
{
"BUCKET_NAME": bucket_name,
"HS_USERNAME": kwargs["username"],
"HS_PASSWORD": kwargs["password"],
"HS_USE_HTTPS": "False",
}
)
hsds = HsdsApp(**kwargs)
try:
hsds.run()
timeout = time.time() + 60
while not hsds.ready:
if time.time() > timeout:
raise TimeoutError("HSDS server did not become ready in time")
time.sleep(1)
os.environ["HS_ENDPOINT"] = hsds.endpoint
# make folders expected by pytest
Folder("/home/", mode="w")
Folder("/home/h5netcdf-pytest/", mode="w")
yield True
except Exception as err:
log_path = kwargs["logfile"]
if os.path.exists(log_path):
with open(log_path) as f:
print("\n=== HSDS Log ===")
print(f.read())
else:
print(f"HSDS log not found at: {log_path}")
raise err
finally:
try:
hsds.check_processes()
hsds.stop()
except Exception:
pass
rmtree(root_dir, ignore_errors=True)
|
