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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
# SPDX-FileCopyrightText: All Contributors to the ITango project
# SPDX-License-Identifier: LGPL-3.0-or-later
import contextlib
import os
import socket
import subprocess
import sys
import time
from unittest import mock
import pytest
import tango
from IPython.terminal.interactiveshell import TerminalInteractiveShell
def get_free_port():
"""Return a free port
There is a race condition as another process could take
the port after the function returns,
but it's probably good enough.
"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
# Bind to a free port provided by the OS
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@pytest.fixture(scope="session")
def mock_tango_host():
"""Mock TANGO_HOST env variable"""
PYTANGO_HOST = f"127.0.0.1:{get_free_port()}"
with mock.patch.dict(os.environ, {"TANGO_HOST": PYTANGO_HOST}):
yield PYTANGO_HOST
@pytest.fixture(scope="session")
def pytango_db(mock_tango_host):
"""Start PyTango database and TangoTest"""
# Copy env to inherit PATH
# When passing env to subprocess, it fully replaces os.environ
env = os.environ.copy()
# Don't write to disk
env["PYTANGO_DATABASE_NAME"] = ":memory:"
try:
databaseds = subprocess.Popen(
[sys.executable, "-m", "tango.databaseds.database", "2"],
stderr=subprocess.PIPE,
env=env,
)
waited = 0
dt = 0.3
while True:
time.sleep(dt)
waited += dt
if databaseds.poll() is not None:
stderr = databaseds.stderr.read().decode()
print("------------------STDERR------------------")
print(stderr)
print("------------------------------------------")
raise RuntimeError(f"Database stopped: {databaseds.returncode}")
try:
host, port = mock_tango_host.split(":")
db = tango.Database(host, int(port))
db.get_info()
break
except tango.DevFailed as exc:
if waited > 10:
raise RuntimeError(
f"Tired of waiting for database...{exc}"
) from exc
# Start TangoTest
tango_test = subprocess.Popen(
["TangoTest", "test"],
stderr=subprocess.PIPE,
)
waited = 0
dt = 0.3
while True:
time.sleep(dt)
waited += dt
if tango_test.poll() is not None:
stderr = tango_test.stderr.read().decode()
print("------------------STDERR------------------")
print(stderr)
print("------------------------------------------")
raise RuntimeError(f"TangoTest stopped: {tango_test.returncode}")
try:
proxy = tango.DeviceProxy(f"tango://{mock_tango_host}/sys/tg_test/1")
proxy.ping()
if proxy.read_attribute("State").value == tango.DevState.RUNNING:
break
except tango.DevFailed as exc:
if waited > 10:
raise RuntimeError(
f"Tired of waiting for device proxy...{exc}"
) from exc
yield mock_tango_host
finally:
# Clean up
try:
tango_test.kill()
except Exception:
pass
with contextlib.suppress(Exception):
databaseds.kill()
@pytest.fixture(scope="session")
def ipython_shell(pytango_db):
# Create and return a fresh IPython shell instance
shell = TerminalInteractiveShell.instance()
shell.extension_manager.load_extension("itango")
yield shell
|
