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
84
85
86
87
88
89
90
91
92
93
94
95
96
|
"""Test fixtures to set up a test named graph for loading RDF data."""
import os
import tempfile
import subprocess
from string import Template
import psutil # type: ignore
import pytest
import requests
from requests.auth import HTTPDigestAuth
from tests.fixtures.virtuoso import VIRTUOSO_INI_FILE
SPARQL_CONF = {
"sparql_user": "dba",
"sparql_password": "dba",
"sparql_auth_uri": "http://localhost:8191/sparql-auth/",
"sparql_crud_auth_uri": "http://localhost:8191/sparql-graph-crud-auth/",
"sparql_endpoint": "http://localhost:8191/sparql/",
}
def get_process_id(name) -> list:
"""Return process ids found by (partial) name or regex.
>>> get_process_id('kthreadd')
[2]
>>> get_process_id('watchdog')
[10, 11, 16, 21, 26, 31, 36, 41, 46, 51, 56, 61] # ymmv
>>> get_process_id('non-existent process')
[]
"""
with subprocess.Popen(
["pgrep", "-f", name], stdout=subprocess.PIPE, shell=False
) as proc:
response = proc.communicate()[0]
return [int(pid) for pid in response.split()]
@pytest.fixture(scope="session")
def rdf_setup():
"""Upload RDF to a Virtuoso named graph"""
dir_path = os.path.dirname(__file__).split("fixtures")[0]
file_path = os.path.join(
dir_path,
"test_data/ttl-files/test-data.ttl",
)
# We intentionally use a temporary directory. This way, all the
# virtuoso database files are properly cleaned up after running
# tests.
with tempfile.TemporaryDirectory() as tmpdirname:
init_file = os.path.join(tmpdirname, "virtuoso.ini")
# Create the virtuoso init file which we use when
# bootstrapping virtuoso.
with open(init_file, "w", encoding="utf-8") as file_:
file_.write(Template(VIRTUOSO_INI_FILE).substitute(
dir_path=tmpdirname))
# Here we intentionally ignore the "+foreground" option to
# allow virtuoso to run in the background.
with subprocess.Popen(
[
"virtuoso-t",
"+wait",
"+no-checkpoint",
"+configfile",
init_file,
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
) as pid:
pid.wait()
# Define the query parameters and authentication
params = {"graph": "http://cd-test.genenetwork.org"}
auth = HTTPDigestAuth("dba", "dba")
# Make sure this graph does not exist before running anything
requests.delete(
SPARQL_CONF["sparql_crud_auth_uri"], params=params, auth=auth
)
# Open the file in binary mode and send the request
with open(file_path, "rb") as file:
response = requests.put(
SPARQL_CONF["sparql_crud_auth_uri"],
params=params,
auth=auth,
data=file,
)
yield response
requests.delete(
SPARQL_CONF["sparql_crud_auth_uri"], params=params, auth=auth
)
for pid_ in get_process_id(init_file):
psutil.Process(pid_).kill()
|
